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