ctf/org.eclipse.tracecompass.ctf.core/src/org/eclipse/tracecompass/ctf/core/trace/CTFTraceReader.java

   1 /*******************************************************************************
   2  * Copyright (c) 2011, 2014 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:
  10  *     Matthew Khouzam - Initial API and implementation
  11  *     Alexandre Montplaisir - Initial API and implementation
  12  *******************************************************************************/
  13
  14 package org.eclipse.tracecompass.ctf.core.trace;
  15
  16 import static org.eclipse.tracecompass.common.core.NonNullUtils.checkNotNull;
  17
  18 import java.io.IOException;
  19 import java.util.ArrayList;
  20 import java.util.Collections;
  21 import java.util.HashSet;
  22 import java.util.List;
  23 import java.util.PriorityQueue;
  24 import java.util.Set;
  25
  26 import org.eclipse.tracecompass.ctf.core.CTFException;
  27 import org.eclipse.tracecompass.ctf.core.event.EventDefinition;
  28 import org.eclipse.tracecompass.ctf.core.event.IEventDeclaration;
  29 import org.eclipse.tracecompass.internal.ctf.core.Activator;
  30 import org.eclipse.tracecompass.internal.ctf.core.trace.StreamInputReaderTimestampComparator;
  31
  32 import com.google.common.collect.ImmutableSet;
  33 import com.google.common.collect.ImmutableSet.Builder;
  34
  35 /**
  36  * A CTF trace reader. Reads the events of a trace.
  37  *
  38  * @version 1.0
  39  * @author Matthew Khouzam
  40  * @author Alexandre Montplaisir
  41  */
  42 public class CTFTraceReader implements AutoCloseable {
  43
  44     private static final int LINE_LENGTH = 60;
  45
  46     private static final int MIN_PRIO_SIZE = 16;
  47
  48     // ------------------------------------------------------------------------
  49     // Attributes
  50     // ------------------------------------------------------------------------
  51
  52     /**
  53      * The trace to read from.
  54      */
  55     private final CTFTrace fTrace;
  56
  57     /**
  58      * Vector of all the trace file readers.
  59      */
  60     private final List<CTFStreamInputReader> fStreamInputReaders = Collections.synchronizedList(new ArrayList<CTFStreamInputReader>());
  61
  62     /**
  63      * Priority queue to order the trace file readers by timestamp.
  64      */
  65     private PriorityQueue<CTFStreamInputReader> fPrio;
  66
  67     /**
  68      * Array to count the number of event per trace file.
  69      */
  70     private long[] fEventCountPerTraceFile;
  71
  72     /**
  73      * Timestamp of the first event in the trace
  74      */
  75     private long fStartTime;
  76
  77     /**
  78      * Timestamp of the last event read so far
  79      */
  80     private long fEndTime;
  81
  82     /**
  83      * Boolean to indicate if the CTFTraceReader has been closed
  84      */
  85     private boolean fClosed = false;
  86
  87     // ------------------------------------------------------------------------
  88     // Constructors
  89     // ------------------------------------------------------------------------
  90
  91     /**
  92      * Constructs a TraceReader to read a trace.
  93      *
  94      * @param trace
  95      *            The trace to read from.
  96      * @throws CTFException
  97      *             if an error occurs
  98      */
  99     public CTFTraceReader(CTFTrace trace) throws CTFException {
 100         fTrace = trace;
 101         fStreamInputReaders.clear();
 102
 103         /**
 104          * Create the trace file readers.
 105          */
 106         createStreamInputReaders();
 107
 108         /**
 109          * Populate the timestamp-based priority queue.
 110          */
 111         populateStreamInputReaderHeap();
 112
 113         /**
 114          * Get the start Time of this trace bear in mind that the trace could be
 115          * empty.
 116          */
 117         fStartTime = 0;
 118         if (hasMoreEvents()) {
 119             EventDefinition currentEvent = getTopStream().getCurrentEvent();
 120             if (currentEvent != null) {
 121                 fStartTime = currentEvent.getTimestamp();
 122                 setEndTime(fStartTime);
 123             }
 124         }
 125     }
 126
 127     /**
 128      * Copy constructor
 129      *
 130      * @return The new CTFTraceReader
 131      * @throws CTFException
 132      *             if an error occurs
 133      */
 134     public CTFTraceReader copyFrom() throws CTFException {
 135         CTFTraceReader newReader = null;
 136
 137         newReader = new CTFTraceReader(fTrace);
 138         newReader.fStartTime = fStartTime;
 139         newReader.setEndTime(fEndTime);
 140         return newReader;
 141     }
 142
 143     /**
 144      * Dispose the CTFTraceReader
 145      */
 146     @Override
 147     public void close() {
 148         synchronized (fStreamInputReaders) {
 149             for (CTFStreamInputReader reader : fStreamInputReaders) {
 150                 if (reader != null) {
 151                     try {
 152                         reader.close();
 153                     } catch (IOException e) {
 154                         Activator.logError(e.getMessage(), e);
 155                     }
 156                 }
 157             }
 158             fStreamInputReaders.clear();
 159         }
 160         fPrio.clear();
 161         fClosed = true;
 162     }
 163
 164     // ------------------------------------------------------------------------
 165     // Getters/Setters/Predicates
 166     // ------------------------------------------------------------------------
 167
 168     /**
 169      * Return the start time of this trace (== timestamp of the first event)
 170      *
 171      * @return the trace start time
 172      */
 173     public long getStartTime() {
 174         return fStartTime;
 175     }
 176
 177     /**
 178      * Set the trace's end time
 179      *
 180      * @param endTime
 181      *            The end time to use
 182      */
 183     protected final void setEndTime(long endTime) {
 184         fEndTime = endTime;
 185     }
 186
 187     /**
 188      * Get the priority queue of this trace reader.
 189      *
 190      * @return The priority queue of input readers
 191      */
 192     protected PriorityQueue<CTFStreamInputReader> getPrio() {
 193         return fPrio;
 194     }
 195
 196     // ------------------------------------------------------------------------
 197     // Operations
 198     // ------------------------------------------------------------------------
 199
 200     /**
 201      * Creates one trace file reader per trace file contained in the trace.
 202      *
 203      * @throws CTFException
 204      *             if an error occurs
 205      */
 206     private void createStreamInputReaders() throws CTFException {
 207         /*
 208          * For each stream.
 209          */
 210         for (CTFStream stream : fTrace.getStreams()) {
 211             Set<CTFStreamInput> streamInputs = stream.getStreamInputs();
 212
 213             /*
 214              * For each trace file of the stream.
 215              */
 216             for (CTFStreamInput streamInput : streamInputs) {
 217
 218                 /*
 219                  * Create a reader and add it to the group.
 220                  */
 221                 fStreamInputReaders.add(new CTFStreamInputReader(checkNotNull(streamInput)));
 222             }
 223         }
 224
 225         /*
 226          * Create the array to count the number of event per trace file.
 227          */
 228         fEventCountPerTraceFile = new long[fStreamInputReaders.size()];
 229     }
 230
 231     /**
 232      * Returns whether or not this CTFTraceReader has been closed
 233      *
 234      * @return true if it has been closed, false else
 235      * @since 1.1
 236      */
 237     public boolean isClosed() {
 238         return fClosed;
 239     }
 240
 241     /**
 242      * Update the priority queue to make it match the parent trace
 243      *
 244      * @throws CTFException
 245      *             An error occured
 246      */
 247     public void update() throws CTFException {
 248         Set<CTFStreamInputReader> readers = new HashSet<>();
 249         for (CTFStream stream : fTrace.getStreams()) {
 250             Set<CTFStreamInput> streamInputs = stream.getStreamInputs();
 251             for (CTFStreamInput streamInput : streamInputs) {
 252                 /*
 253                  * Create a reader.
 254                  */
 255                 CTFStreamInputReader streamInputReader = new CTFStreamInputReader(checkNotNull(streamInput));
 256
 257                 /*
 258                  * Add it to the group.
 259                  */
 260                 if (!fStreamInputReaders.contains(streamInputReader)) {
 261                     streamInputReader.readNextEvent();
 262                     fStreamInputReaders.add(streamInputReader);
 263                     readers.add(streamInputReader);
 264                 }
 265             }
 266         }
 267         long[] temp = fEventCountPerTraceFile;
 268         fEventCountPerTraceFile = new long[readers.size() + temp.length];
 269         for (CTFStreamInputReader reader : readers) {
 270             fPrio.add(reader);
 271         }
 272         for (int i = 0; i < temp.length; i++) {
 273             fEventCountPerTraceFile[i] = temp[i];
 274         }
 275     }
 276
 277     /**
 278      * Gets an iterable of the stream input readers, useful for foreaches
 279      *
 280      * @return the iterable of the stream input readers
 281      */
 282     public Iterable<IEventDeclaration> getEventDeclarations() {
 283         ImmutableSet.Builder<IEventDeclaration> builder = new Builder<>();
 284         for (CTFStreamInputReader sir : fStreamInputReaders) {
 285             builder.addAll(sir.getEventDeclarations());
 286         }
 287         return builder.build();
 288     }
 289
 290     /**
 291      * Initializes the priority queue used to choose the trace file with the
 292      * lower next event timestamp.
 293      *
 294      * @throws CTFException
 295      *             if an error occurs
 296      */
 297     private void populateStreamInputReaderHeap() throws CTFException {
 298         if (fStreamInputReaders.isEmpty()) {
 299             fPrio = new PriorityQueue<>(MIN_PRIO_SIZE,
 300                     new StreamInputReaderTimestampComparator());
 301             return;
 302         }
 303
 304         /*
 305          * Create the priority queue with a size twice as bigger as the number
 306          * of reader in order to avoid constant resizing.
 307          */
 308         fPrio = new PriorityQueue<>(
 309                 Math.max(fStreamInputReaders.size() * 2, MIN_PRIO_SIZE),
 310                 new StreamInputReaderTimestampComparator());
 311
 312         int pos = 0;
 313
 314         for (CTFStreamInputReader reader : fStreamInputReaders) {
 315             /*
 316              * Add each trace file reader in the priority queue, if we are able
 317              * to read an event from it.
 318              */
 319             CTFResponse readNextEvent = reader.readNextEvent();
 320             if (readNextEvent == CTFResponse.OK || readNextEvent == CTFResponse.WAIT) {
 321                 fPrio.add(reader);
 322
 323                 fEventCountPerTraceFile[pos] = 0;
 324                 reader.setName(pos);
 325
 326                 pos++;
 327             }
 328         }
 329     }
 330
 331     /**
 332      * Get the current event, which is the current event of the trace file
 333      * reader with the lowest timestamp.
 334      *
 335      * @return An event definition, or null of the trace reader reached the end
 336      *         of the trace.
 337      */
 338     public EventDefinition getCurrentEventDef() {
 339         CTFStreamInputReader top = getTopStream();
 340         return (top != null) ? top.getCurrentEvent() : null;
 341     }
 342
 343     /**
 344      * Go to the next event.
 345      *
 346      * @return True if an event was read.
 347      * @throws CTFException
 348      *             if an error occurs
 349      */
 350     public boolean advance() throws CTFException {
 351         /*
 352          * Remove the reader from the top of the priority queue.
 353          */
 354         CTFStreamInputReader top = fPrio.poll();
 355
 356         /*
 357          * If the queue was empty.
 358          */
 359         if (top == null) {
 360             return false;
 361         }
 362         /*
 363          * Read the next event of this reader.
 364          */
 365         switch (top.readNextEvent()) {
 366         case OK: {
 367             /*
 368              * Add it back in the queue.
 369              */
 370             fPrio.add(top);
 371             /*
 372              * We're in OK, there's a guaranteed top#getCurrentEvent() unless another
 373              * thread does something bad.
 374              */
 375             EventDefinition currentEvent = checkNotNull(top.getCurrentEvent());
 376             final long topEnd = fTrace.timestampCyclesToNanos(currentEvent.getTimestamp());
 377             setEndTime(Math.max(topEnd, getEndTime()));
 378             fEventCountPerTraceFile[top.getName()]++;
 379             fEndTime = Math.max(currentEvent.getTimestamp(), fEndTime);
 380             break;
 381         }
 382         case WAIT: {
 383             fPrio.add(top);
 384             break;
 385         }
 386         case FINISH:
 387             break;
 388         case ERROR:
 389         default:
 390             // something bad happend
 391         }
 392         /*
 393          * If there is no reader in the queue, it means the trace reader reached
 394          * the end of the trace.
 395          */
 396         return hasMoreEvents();
 397     }
 398
 399     /**
 400      * Go to the last event in the trace.
 401      *
 402      * @throws CTFException
 403      *             if an error occurs
 404      */
 405     public void goToLastEvent() throws CTFException {
 406         seek(getEndTime());
 407         while (fPrio.size() > 1) {
 408             advance();
 409         }
 410     }
 411
 412     /**
 413      * Seeks to a given timestamp. It will seek to the nearest event greater or
 414      * equal to timestamp. If a trace is [10 20 30 40] and you are looking for
 415      * 19, it will give you 20. If you want 20, you will get 20, if you want 21,
 416      * you will get 30. The value -inf will seek to the first element and the
 417      * value +inf will seek to the end of the file (past the last event).
 418      *
 419      * @param timestamp
 420      *            the timestamp to seek to
 421      * @return true if there are events above or equal the seek timestamp, false
 422      *         if seek at the end of the trace (no valid event).
 423      * @throws CTFException
 424      *             if an error occurs
 425      */
 426     public boolean seek(long timestamp) throws CTFException {
 427         /*
 428          * Remove all the trace readers from the priority queue
 429          */
 430         fPrio.clear();
 431         for (CTFStreamInputReader streamInputReader : fStreamInputReaders) {
 432             /*
 433              * Seek the trace reader.
 434              */
 435             streamInputReader.seek(timestamp);
 436
 437             /*
 438              * Add it to the priority queue if there is a current event.
 439              */
 440             if (streamInputReader.getCurrentEvent() != null) {
 441                 fPrio.add(streamInputReader);
 442             }
 443         }
 444         return hasMoreEvents();
 445     }
 446
 447     /**
 448      * Gets the stream with the oldest event
 449      *
 450      * @return the stream with the oldest event
 451      */
 452     public CTFStreamInputReader getTopStream() {
 453         return fPrio.peek();
 454     }
 455
 456     /**
 457      * Does the trace have more events?
 458      *
 459      * @return true if yes.
 460      */
 461     public final boolean hasMoreEvents() {
 462         return fPrio.size() > 0;
 463     }
 464
 465     /**
 466      * Prints the event count stats.
 467      */
 468     public void printStats() {
 469         printStats(LINE_LENGTH);
 470     }
 471
 472     /**
 473      * Prints the event count stats.
 474      *
 475      * @param width
 476      *            Width of the display.
 477      */
 478     public void printStats(int width) {
 479         int numEvents = 0;
 480         if (width == 0) {
 481             return;
 482         }
 483
 484         for (long i : fEventCountPerTraceFile) {
 485             numEvents += i;
 486         }
 487
 488         for (int j = 0; j < fEventCountPerTraceFile.length; j++) {
 489             CTFStreamInputReader se = fStreamInputReaders.get(j);
 490
 491             long len = (width * fEventCountPerTraceFile[se.getName()])
 492                     / numEvents;
 493
 494             StringBuilder sb = new StringBuilder(se.getFilename());
 495             sb.append("\t["); //$NON-NLS-1$
 496
 497             for (int i = 0; i < len; i++) {
 498                 sb.append('+');
 499             }
 500
 501             for (long i = len; i < width; i++) {
 502                 sb.append(' ');
 503             }
 504
 505             sb.append("]\t" + fEventCountPerTraceFile[se.getName()] + " Events"); //$NON-NLS-1$//$NON-NLS-2$
 506             Activator.log(sb.toString());
 507         }
 508     }
 509
 510     /**
 511      * Gets the last event timestamp that was read. This is NOT necessarily the
 512      * last event in a trace, just the last one read so far.
 513      *
 514      * @return the last event
 515      */
 516     public long getEndTime() {
 517         return fEndTime;
 518     }
 519
 520     /**
 521      * Sets a trace to be live or not
 522      *
 523      * @param live
 524      *            whether the trace is live
 525      */
 526     public void setLive(boolean live) {
 527         for (CTFStreamInputReader s : fPrio) {
 528             s.setLive(live);
 529         }
 530     }
 531
 532     /**
 533      * Get if the trace is to read live or not
 534      *
 535      * @return whether the trace is live or not
 536      */
 537     public boolean isLive() {
 538         return getTopStream().isLive();
 539     }
 540
 541     @Override
 542     public int hashCode() {
 543         final int prime = 31;
 544         int result = 1;
 545         result = (prime * result) + (int) (fStartTime ^ (fStartTime >>> 32));
 546         result = (prime * result) + fStreamInputReaders.hashCode();
 547         result = (prime * result) + ((fTrace == null) ? 0 : fTrace.hashCode());
 548         return result;
 549     }
 550
 551     @Override
 552     public boolean equals(Object obj) {
 553         if (this == obj) {
 554             return true;
 555         }
 556         if (obj == null) {
 557             return false;
 558         }
 559         if (!(obj instanceof CTFTraceReader)) {
 560             return false;
 561         }
 562         CTFTraceReader other = (CTFTraceReader) obj;
 563         if (!fStreamInputReaders.equals(other.fStreamInputReaders)) {
 564             return false;
 565         }
 566         if (fTrace == null) {
 567             if (other.fTrace != null) {
 568                 return false;
 569             }
 570         } else if (!fTrace.equals(other.fTrace)) {
 571             return false;
 572         }
 573         return true;
 574     }
 575
 576     @Override
 577     public String toString() {
 578         /* Only for debugging, shouldn't be externalized */
 579         return "CTFTraceReader [trace=" + fTrace + ']'; //$NON-NLS-1$
 580     }
 581
 582     /**
 583      * Gets the parent trace
 584      *
 585      * @return the parent trace
 586      */
 587     public CTFTrace getTrace() {
 588         return fTrace;
 589     }
 590
 591     /**
 592      * This will read the entire trace and populate all the indexes. The reader
 593      * will then be reset to the first event in the trace.
 594      *
 595      * Do not call in the fast path.
 596      *
 597      * @throws CTFException
 598      *             A trace reading error occurred
 599      * @since 1.0
 600      */
 601     public void populateIndex() throws CTFException {
 602         for (CTFStreamInputReader sir : fPrio) {
 603             sir.goToLastEvent();
 604         }
 605         seek(0);
 606
 607     }
 608 }