org.eclipse.linuxtools.ctf.core/src/org/eclipse/linuxtools/ctf/core/trace/CTFTraceReader.java

   1 /*******************************************************************************
   2  * Copyright (c) 2011-2012 Ericsson, Ecole Polytechnique de Montreal and others
   3  *
   4  * All rights reserved. This program and the accompanying materials are made
   5  * available under the terms of the Eclipse Public License v1.0 which
   6  * accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors: Matthew Khouzam - Initial API and implementation
  10  * Contributors: Alexandre Montplaisir - Initial API and implementation
  11  *******************************************************************************/
  12
  13 package org.eclipse.linuxtools.ctf.core.trace;
  14
  15 import java.util.Collection;
  16 import java.util.PriorityQueue;
  17 import java.util.Set;
  18 import java.util.Vector;
  19
  20 import org.eclipse.linuxtools.ctf.core.event.EventDefinition;
  21 import org.eclipse.linuxtools.internal.ctf.core.Activator;
  22 import org.eclipse.linuxtools.internal.ctf.core.trace.StreamInputPacketIndexEntry;
  23
  24 /**
  25  * Reads the events of a trace.
  26  */
  27
  28 public class CTFTraceReader {
  29
  30     // ------------------------------------------------------------------------
  31     // Attributes
  32     // ------------------------------------------------------------------------
  33
  34     /**
  35      * The trace to read from.
  36      */
  37     private final CTFTrace trace;
  38
  39     /**
  40      * Vector of all the trace file readers.
  41      */
  42     private final Vector<StreamInputReader> streamInputReaders = new Vector<StreamInputReader>();
  43
  44     /**
  45      * Priority queue to order the trace file readers by timestamp.
  46      */
  47     protected PriorityQueue<StreamInputReader> prio;
  48
  49     /**
  50      * Array to count the number of event per trace file.
  51      */
  52     private int[] eventCountPerTraceFile;
  53
  54     /**
  55      * Timestamp of the first event in the trace
  56      */
  57     private long startTime;
  58
  59     /**
  60      * Timestamp of the last event read so far
  61      */
  62     private long endTime;
  63
  64     /**
  65      * Current event index
  66      */
  67     private long index;
  68
  69     private final long startIndex[];
  70
  71     // ------------------------------------------------------------------------
  72     // Constructors
  73     // ------------------------------------------------------------------------
  74
  75     /**
  76      * Constructs a TraceReader to read a trace.
  77      *
  78      * @param trace
  79      *            The trace to read from.
  80      * @throws CTFReaderException
  81      */
  82     public CTFTraceReader(CTFTrace trace) {
  83         this.trace = trace;
  84
  85         /**
  86          * Create the trace file readers.
  87          */
  88         createStreamInputReaders();
  89
  90         /**
  91          * Populate the timestamp-based priority queue.
  92          */
  93         populateStreamInputReaderHeap();
  94
  95         /**
  96          * Get the start Time of this trace
  97          */
  98         this.startTime = prio.peek().getCurrentEvent().timestamp;
  99         this.endTime = this.startTime;
 100         this.index = 0;
 101         startIndex = new long[prio.size()];
 102         for (int i = 0; i < prio.size(); i++) {
 103             startIndex[i] = 0;
 104         }
 105     }
 106
 107     /**
 108      * Copy constructor
 109      */
 110     public CTFTraceReader copyFrom() {
 111         CTFTraceReader newReader = null;
 112
 113         newReader = new CTFTraceReader(this.trace);
 114         newReader.startTime = this.startTime;
 115         newReader.endTime = this.endTime;
 116         return newReader;
 117     }
 118
 119     // ------------------------------------------------------------------------
 120     // Getters/Setters/Predicates
 121     // ------------------------------------------------------------------------
 122
 123     /**
 124      * Return the start time of this trace (== timestamp of the first event)
 125      *
 126      * @return the trace start time
 127      */
 128     public long getStartTime() {
 129         return this.startTime;
 130     }
 131
 132     /**
 133      * @return the index
 134      */
 135     public long getIndex() {
 136         return index;
 137     }
 138
 139     // ------------------------------------------------------------------------
 140     // Operations
 141     // ------------------------------------------------------------------------
 142
 143     /**
 144      * Creates one trace file reader per trace file contained in the trace.
 145      */
 146     private void createStreamInputReaders() {
 147         Collection<Stream> streams = this.trace.getStreams().values();
 148
 149         /*
 150          * For each stream.
 151          */
 152         for (Stream stream : streams) {
 153             Set<StreamInput> streamInputs = stream.getStreamInputs();
 154
 155             /*
 156              * For each trace file of the stream.
 157              */
 158             for (StreamInput streamInput : streamInputs) {
 159                 /*
 160                  * Create a reader.
 161                  */
 162                 StreamInputReader streamInputReader = new StreamInputReader(
 163                         streamInput);
 164
 165                 /*
 166                  * Add it to the group.
 167                  */
 168                 this.streamInputReaders.add(streamInputReader);
 169             }
 170         }
 171
 172         /*
 173          * Create the array to count the number of event per trace file.
 174          */
 175         this.eventCountPerTraceFile = new int[this.streamInputReaders.size()];
 176     }
 177
 178     /**
 179      * Initializes the priority queue used to choose the trace file with the
 180      * lower next event timestamp.
 181      */
 182     private void populateStreamInputReaderHeap() {
 183         /*
 184          * Create the priority queue with a size twice as bigger as the number
 185          * of reader in order to avoid constant resizing.
 186          */
 187         this.prio = new PriorityQueue<StreamInputReader>(
 188                 this.streamInputReaders.size() * 2,
 189                 new StreamInputReaderTimestampComparator());
 190
 191         int pos = 0;
 192
 193         for (StreamInputReader reader : this.streamInputReaders) {
 194             /*
 195              * Add each trace file reader in the priority queue, if we are able
 196              * to read an event from it.
 197              */
 198             if (reader.readNextEvent()) {
 199                 this.prio.add(reader);
 200
 201                 this.eventCountPerTraceFile[pos] = 0;
 202                 reader.setName(pos);
 203
 204                 pos++;
 205             }
 206         }
 207     }
 208
 209     /**
 210      * Get the current event, which is the current event of the trace file
 211      * reader with the lowest timestamp.
 212      *
 213      * @return An event definition, or null of the trace reader reached the end
 214      *         of the trace.
 215      */
 216     public EventDefinition getCurrentEventDef() {
 217         StreamInputReader top = getTopStream();
 218
 219         return (top != null) ? top.getCurrentEvent() : null;
 220     }
 221
 222     /**
 223      * Go to the next event.
 224      *
 225      * @return True if an event was read.
 226      */
 227     public boolean advance() {
 228         /*
 229          * Index the
 230          */
 231         /*
 232          * Remove the reader from the top of the priority queue.
 233          */
 234         StreamInputReader top = this.prio.poll();
 235
 236         /*
 237          * If the queue was empty.
 238          */
 239         if (top == null) {
 240             return false;
 241         }
 242         /*
 243          * index if needed
 244          */
 245         if (hasMoreEvents()) {
 246             StreamInputPacketReader packetReader = top.getPacketReader();
 247             boolean packetHasMoreEvents = packetReader.hasMoreEvents();
 248             StreamInputPacketIndexEntry currentPacket = packetReader
 249                     .getCurrentPacket();
 250             if (!packetHasMoreEvents) {
 251                 int n = this.streamInputReaders.indexOf(top);
 252                 currentPacket.setIndexBegin(startIndex[n]);
 253                 currentPacket.setIndexEnd(index);
 254                 startIndex[n] = index + 1;
 255             }
 256         }
 257         /*
 258          * Read the next event of this reader.
 259          */
 260         if (top.readNextEvent()) {
 261             /*
 262              * Add it back in the queue.
 263              */
 264             this.prio.add(top);
 265             final long topEnd = top.getCurrentEvent().timestamp;
 266             this.endTime = Math.max(topEnd, this.endTime);
 267             this.eventCountPerTraceFile[top.getName()]++;
 268             /*
 269              * increment the index
 270              */
 271             index++;
 272         }
 273         boolean hasMoreEvents = hasMoreEvents();
 274
 275         /*
 276          * If there is no reader in the queue, it means the trace reader reached
 277          * the end of the trace.
 278          */
 279         return hasMoreEvents;
 280     }
 281
 282     /**
 283      * Go to the last event in the trace.
 284      *
 285      * @throws CTFReaderException
 286      */
 287     public void goToLastEvent() throws CTFReaderException {
 288
 289         this.seek(0);
 290         for (StreamInputReader streamInputReader : this.streamInputReaders) {
 291             /*
 292              * Seek the trace reader.
 293              */
 294             streamInputReader.goToLastEvent();
 295         }
 296         int count = prio.size();
 297         for (int i = 0; i < (count); i++) {
 298             advance();
 299         }
 300     }
 301
 302     /**
 303      * Seeks to a given timestamp It will go to the event just after the
 304      * timestamp or the timestamp itself. if a if a trace is 10 20 30 40 and
 305      * you're looking for 19, it'll give you 20, it you want 20, you'll get 20,
 306      * if you want 21, you'll get 30. You want -inf, you'll get the first
 307      * element, you want +inf, you'll get the end of the file with no events.
 308      *
 309      * @param timestamp
 310      *            the timestamp to seek to
 311      * @return true if the trace has more events following the timestamp
 312      */
 313     public boolean seek(long timestamp) {
 314         /*
 315          * Remove all the trace readers from the priority queue
 316          */
 317         this.prio.clear();
 318         index = 0;
 319         long offset = 0;
 320         for (StreamInputReader streamInputReader : this.streamInputReaders) {
 321             /*
 322              * Seek the trace reader.
 323              */
 324             offset += streamInputReader.seek(timestamp);
 325
 326             /*
 327              * Add it to the priority queue if there is a current event.
 328              */
 329
 330         }
 331         for (StreamInputReader streamInputReader : this.streamInputReaders) {
 332             if (streamInputReader.getCurrentEvent() != null) {
 333                 this.prio.add(streamInputReader);
 334                 index = Math.max(index, streamInputReader.getPacketReader()
 335                         .getCurrentPacket().getIndexBegin()
 336                         + offset);
 337             }
 338         }
 339         return hasMoreEvents();
 340     }
 341
 342     public boolean seekIndex(long index) {
 343         this.prio.clear();
 344
 345         long tempIndex = Long.MIN_VALUE;
 346         long tempTimestamp = Long.MIN_VALUE;
 347         try {
 348             for (StreamInputReader streamInputReader : this.streamInputReaders) {
 349                 /*
 350                  * Seek the trace reader.
 351                  */
 352                 final long streamIndex = streamInputReader.seekIndex(index);
 353                 tempIndex = Math.max(tempIndex, streamIndex);
 354                 tempTimestamp = Math.max(tempTimestamp,
 355                         streamInputReader.getCurrentEvent().timestamp);
 356
 357             }
 358         } catch (CTFReaderException e) {
 359             /*
 360              * Important, if it failed, it's because it's not yet indexed, so we
 361              * have to manually advance to the right value.
 362              */
 363             for (StreamInputReader streamInputReader : this.streamInputReaders) {
 364                 /*
 365                  * Seek the trace reader.
 366                  */
 367                 streamInputReader.seek(0);
 368             }
 369             tempIndex = 0;
 370         }
 371         for (StreamInputReader streamInputReader : this.streamInputReaders) {
 372             /*
 373              * Add it to the priority queue if there is a current event.
 374              */
 375
 376             if (streamInputReader.getCurrentEvent() != null) {
 377                 this.prio.add(streamInputReader);
 378             }
 379         }
 380         if (tempIndex == Long.MAX_VALUE) {
 381             tempIndex = 0;
 382         }
 383         long pos = tempIndex;
 384         if (index > tempIndex) {
 385             /*
 386              * advance for offset
 387              */
 388             while ((prio.peek().getCurrentEvent().timestamp < tempTimestamp)
 389                     && hasMoreEvents()) {
 390                 this.advance();
 391             }
 392
 393             for (pos = tempIndex; (pos < index) && hasMoreEvents(); pos++) {
 394                 this.advance();
 395             }
 396         }
 397         this.index = pos;
 398         return hasMoreEvents();
 399     }
 400
 401     public StreamInputReader getTopStream() {
 402         return this.prio.peek();
 403     }
 404
 405     /**
 406      * Does the trace have more events?
 407      *
 408      * @return true if yes.
 409      */
 410     public boolean hasMoreEvents() {
 411         return this.prio.size() > 0;
 412     }
 413
 414     /**
 415      * Prints the event count stats.
 416      */
 417     public void printStats() {
 418         printStats(60);
 419     }
 420
 421     /**
 422      * Prints the event count stats.
 423      *
 424      * @param width
 425      *            Width of the display.
 426      */
 427     public void printStats(int width) {
 428         int numEvents = 0;
 429         if (width == 0) {
 430             return;
 431         }
 432
 433         for (int i : this.eventCountPerTraceFile) {
 434             numEvents += i;
 435         }
 436
 437         for (int j = 0; j < this.eventCountPerTraceFile.length; j++) {
 438             StreamInputReader se = this.streamInputReaders.get(j);
 439
 440             int len = (width * this.eventCountPerTraceFile[se.getName()])
 441                     / numEvents;
 442
 443             StringBuilder sb = new StringBuilder(se.getStreamInput()
 444                     .getFilename() + "\t["); //$NON-NLS-1$
 445
 446             for (int i = 0; i < len; i++) {
 447                 sb.append('+');
 448             }
 449
 450             for (int i = len; i < width; i++) {
 451                 sb.append(' ');
 452             }
 453
 454             sb.append("]\t" + this.eventCountPerTraceFile[se.getName()] + " Events"); //$NON-NLS-1$//$NON-NLS-2$
 455             Activator.getDefault().log(sb.toString());
 456         }
 457     }
 458
 459     public long getEndTime() {
 460         return this.endTime;
 461     }
 462
 463     @Override
 464     public int hashCode() {
 465         final int prime = 31;
 466         int result = 1;
 467         result = (prime * result) + (int) (endTime ^ (endTime >>> 32));
 468         result = (prime * result) + (int) (startTime ^ (startTime >>> 32));
 469         result = (prime * result)
 470                 + ((streamInputReaders == null) ? 0 : streamInputReaders
 471                         .hashCode());
 472         result = (prime * result) + ((trace == null) ? 0 : trace.hashCode());
 473         return result;
 474     }
 475
 476     @Override
 477     public boolean equals(Object obj) {
 478         if (this == obj) {
 479             return true;
 480         }
 481         if (obj == null) {
 482             return false;
 483         }
 484         if (getClass() != obj.getClass()) {
 485             return false;
 486         }
 487         CTFTraceReader other = (CTFTraceReader) obj;
 488         if (endTime != other.endTime) {
 489             return false;
 490         }
 491         if (startTime != other.startTime) {
 492             return false;
 493         }
 494         if (streamInputReaders == null) {
 495             if (other.streamInputReaders != null) {
 496                 return false;
 497             }
 498         } else if (!streamInputReaders.equals(other.streamInputReaders)) {
 499             return false;
 500         }
 501         if (trace == null) {
 502             if (other.trace != null) {
 503                 return false;
 504             }
 505         } else if (!trace.equals(other.trace)) {
 506             return false;
 507         }
 508         return true;
 509     }
 510
 511     /*
 512      * (non-Javadoc)
 513      *
 514      * @see java.lang.Object#toString()
 515      */
 516     @Override
 517     public String toString() {
 518         /* Only for debugging, shouldn't be externalized */
 519         return "CTFTraceReader [trace=" + trace + ']'; //$NON-NLS-1$
 520     }
 521
 522     public CTFTrace getTrace() {
 523         return trace;
 524     }
 525 }