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