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