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