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