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