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