org.eclipse.linuxtools.tmf/src/org/eclipse/linuxtools/tmf/trace/TmfTrace.java

   1 /*******************************************************************************
   2  * Copyright (c) 2009 Ericsson
   3  *
   4  * All rights reserved. This program and the accompanying materials are
   5  * made 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  *   Francois Chouinard - Initial API and implementation
  11  *******************************************************************************/
  12
  13 package org.eclipse.linuxtools.tmf.trace;
  14
  15 import java.io.File;
  16 import java.io.FileNotFoundException;
  17 import java.util.Collections;
  18 import java.util.Vector;
  19
  20 import org.eclipse.core.runtime.IProgressMonitor;
  21 import org.eclipse.core.runtime.IStatus;
  22 import org.eclipse.core.runtime.Status;
  23 import org.eclipse.core.runtime.jobs.Job;
  24 import org.eclipse.linuxtools.tmf.component.TmfComponent;
  25 import org.eclipse.linuxtools.tmf.event.TmfEvent;
  26 import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  27 import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
  28 import org.eclipse.linuxtools.tmf.request.ITmfRequestHandler;
  29 import org.eclipse.linuxtools.tmf.request.TmfDataRequest;
  30
  31 /**
  32  * <b><u>TmfTrace</u></b>
  33  * <p>
  34  * Abstract implementation of ITmfTrace. It should be sufficient to extend this
  35  * class and provide implementation for <code>getCurrentLocation()</code> and
  36  * <code>seekLocation()</code>, as well as a proper parser, to have a working
  37  * concrete implementation.
  38  *
  39  * TODO: Add support for live streaming (notifications, incremental indexing, ...)
  40  */
  41 public abstract class TmfTrace extends TmfComponent implements ITmfTrace, ITmfRequestHandler<TmfEvent> {
  42
  43     // ========================================================================
  44     // Constants
  45     // ========================================================================
  46
  47     // The default number of events to cache
  48     public static final int DEFAULT_CACHE_SIZE = 1000;
  49
  50     // ========================================================================
  51     // Attributes
  52     // ========================================================================
  53
  54     // The trace path
  55     private final String fPath;
  56
  57     // The trace name
  58     private final String fName;
  59
  60     // The cache page size AND checkpoints interval
  61     private final int fCacheSize;
  62
  63     // Indicate if the stream should be pre-indexed
  64     private final boolean fWaitForIndexCompletion;
  65
  66     // The set of event stream checkpoints (for random access)
  67     protected Vector<TmfTraceCheckpoint> fCheckpoints = new Vector<TmfTraceCheckpoint>();
  68
  69     // The number of events collected
  70     private long fNbEvents = 0;
  71
  72     // The time span of the event stream
  73     private TmfTimeRange fTimeRange = new TmfTimeRange(TmfTimestamp.BigBang, TmfTimestamp.BigBang);
  74
  75     // ========================================================================
  76     // Constructors
  77     // ========================================================================
  78
  79     /**
  80      * @param name
  81      * @param pageSize
  82      * @param data.index
  83      * @throws FileNotFoundException
  84      */
  85     protected TmfTrace(String path, int pageSize, boolean waitForIndexCompletion) throws FileNotFoundException {
  86         super();
  87         int sep = path.lastIndexOf(File.separator);
  88         fName = (sep >= 0) ? path.substring(sep + 1) : path;
  89         fPath = path;
  90         fCacheSize = (pageSize > 0) ? pageSize : DEFAULT_CACHE_SIZE;
  91         fWaitForIndexCompletion = waitForIndexCompletion;
  92     }
  93
  94     /**
  95      * @param name
  96      * @param cacheSize
  97      * @throws FileNotFoundException
  98      */
  99     protected TmfTrace(String name, boolean waitForIndexCompletion) throws FileNotFoundException {
 100         this(name, DEFAULT_CACHE_SIZE, waitForIndexCompletion);
 101     }
 102
 103     /**
 104      * @param name
 105      * @param cacheSize
 106      * @throws FileNotFoundException
 107      */
 108     protected TmfTrace(String name, int pageSize) throws FileNotFoundException {
 109         this(name, pageSize, false);
 110     }
 111
 112     /**
 113      * @param name
 114      * @throws FileNotFoundException
 115      */
 116     protected TmfTrace(String name) throws FileNotFoundException {
 117         this(name, DEFAULT_CACHE_SIZE, false);
 118     }
 119
 120     // ========================================================================
 121     // Accessors
 122     // ========================================================================
 123
 124     /**
 125      * @return the trace path
 126      */
 127     public String getPath() {
 128         return fPath;
 129     }
 130
 131     /**
 132      * @return the trace name
 133      */
 134     public String getName() {
 135         return fName;
 136     }
 137
 138     /* (non-Javadoc)
 139      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getNbEvents()
 140      */
 141     public long getNbEvents() {
 142         return fNbEvents;
 143     }
 144
 145     /**
 146      * @return the size of the cache
 147      */
 148     public int getCacheSize() {
 149         return fCacheSize;
 150     }
 151
 152     /* (non-Javadoc)
 153      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getTimeRange()
 154      */
 155     public TmfTimeRange getTimeRange() {
 156         return fTimeRange;
 157     }
 158
 159     public TmfTimestamp getStartTime() {
 160         return fTimeRange.getStartTime();
 161     }
 162
 163     public TmfTimestamp getEndTime() {
 164         return fTimeRange.getEndTime();
 165     }
 166
 167     protected long getIndex(TmfTimestamp timestamp) {
 168         TmfTraceContext context = seekEvent(timestamp);
 169         return context.getIndex();
 170     }
 171
 172     protected TmfTimestamp getTimestamp(int index) {
 173         TmfTraceContext context = seekEvent(index);
 174         return context.getTimestamp();
 175     }
 176
 177     // ========================================================================
 178     // Operators
 179     // ========================================================================
 180
 181     protected void setTimeRange(TmfTimeRange range) {
 182         fTimeRange = range;
 183     }
 184
 185     protected void setStartTime(TmfTimestamp startTime) {
 186         fTimeRange = new TmfTimeRange(startTime, fTimeRange.getEndTime());
 187     }
 188
 189     protected void setEndTime(TmfTimestamp endTime) {
 190         fTimeRange = new TmfTimeRange(fTimeRange.getStartTime(), endTime);
 191     }
 192
 193     /* (non-Javadoc)
 194      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(org.eclipse.linuxtools.tmf.event.TmfTimestamp)
 195      */
 196     public TmfTraceContext seekEvent(TmfTimestamp timestamp) {
 197
 198         if (timestamp == null) {
 199                 timestamp = TmfTimestamp.BigBang;
 200         }
 201
 202         // First, find the right checkpoint
 203         int index = Collections.binarySearch(fCheckpoints, new TmfTraceCheckpoint(timestamp, 0));
 204
 205         // In the very likely case that the checkpoint was not found, bsearch
 206         // returns its negated would-be location (not an offset...). From that
 207         // index, we can then position the stream and get the event.
 208         if (index < 0) {
 209             index = Math.max(0, -(index + 2));
 210         }
 211
 212         // Position the stream at the checkpoint
 213         Object location;
 214         synchronized (fCheckpoints) { //Just in case we are re-indexing
 215             location = (index < fCheckpoints.size()) ? fCheckpoints.elementAt(index).getLocation() : null;
 216         }
 217         TmfTraceContext nextEventContext = seekLocation(location);
 218         nextEventContext.setIndex(index * fCacheSize);
 219         TmfTraceContext currentEventContext = new TmfTraceContext(nextEventContext);
 220
 221         // And get the event
 222         TmfEvent event = getNextEvent(nextEventContext);
 223         while (event != null && event.getTimestamp().compareTo(timestamp, false) < 0) {
 224                 currentEventContext.setLocation(nextEventContext.getLocation());
 225                 currentEventContext.incrIndex();
 226                 event = getNextEvent(nextEventContext);
 227         }
 228
 229         currentEventContext.setTimestamp((event != null) ? event.getTimestamp() : null);
 230         return currentEventContext;
 231     }
 232
 233     /* (non-Javadoc)
 234      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(int)
 235      */
 236     public TmfTraceContext seekEvent(long position) {
 237
 238         // Position the stream at the previous checkpoint
 239         int index = (int) position / fCacheSize;
 240         Object location;
 241         synchronized (fCheckpoints) { //Just in case we are re-indexing
 242             location = (index < fCheckpoints.size()) ? fCheckpoints.elementAt(index).getLocation() : null;
 243         }
 244         TmfTraceContext nextEventContext = seekLocation(location);
 245         nextEventContext.setIndex(index * fCacheSize);
 246         TmfTraceContext currentEventContext = new TmfTraceContext(nextEventContext);
 247
 248         // And locate the event (if it exists)
 249         TmfEvent event = getNextEvent(nextEventContext);
 250         while (event != null && currentEventContext.getIndex() < position) {
 251                 currentEventContext.setLocation(nextEventContext.getLocation());
 252                 currentEventContext.setTimestamp(event.getTimestamp());
 253                 currentEventContext.incrIndex();
 254                 event = getNextEvent(nextEventContext);
 255         }
 256
 257         return currentEventContext;
 258     }
 259
 260         /* (non-Javadoc)
 261          * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getNextEvent(org.eclipse.linuxtools.tmf.trace.ITmfTrace.TraceContext)
 262          */
 263         public TmfEvent getNextEvent(TmfTraceContext context) {
 264                 // parseEvent updates the context
 265                 TmfEvent event = parseEvent(context);
 266                 if (event != null) {
 267                         processEvent(event);
 268                 }
 269         return event;
 270         }
 271
 272     /**
 273      * To be implemented by the subclass.
 274      */
 275         public abstract Object getCurrentLocation();
 276     public abstract TmfEvent parseEvent(TmfTraceContext context);
 277
 278         /**
 279          * Hook for "special" processing by the extending class
 280          * @param event
 281          */
 282         public void processEvent(TmfEvent event) {
 283                 // Do nothing by default
 284         }
 285
 286     // ========================================================================
 287     // ITmfRequestHandler
 288     // ========================================================================
 289
 290     /* (non-Javadoc)
 291      * @see org.eclipse.linuxtools.tmf.eventlog.ITmfRequestHandler#processRequest(org.eclipse.linuxtools.tmf.eventlog.TmfDataRequest, boolean)
 292      */
 293         public void processRequest(TmfDataRequest<TmfEvent> request, boolean waitForCompletion) {
 294
 295                 // Process the request
 296                 processDataRequest(request);
 297
 298                 // Wait for completion if needed
 299         if (waitForCompletion) {
 300                         request.waitForCompletion();
 301                 }
 302         }
 303
 304         /**
 305          * Process a data request
 306          *
 307          * @param request
 308          */
 309         private void processDataRequest(final TmfDataRequest<TmfEvent> request) {
 310
 311                 // Initialize the trace context
 312                 final TmfTraceContext context = (request.getRange() != null) ?
 313                                 seekEvent(request.getRange().getStartTime()) :
 314                             seekEvent(request.getIndex());
 315
 316                 final TmfTimestamp endTime = (request.getRange() != null) ?
 317                                 request.getRange().getEndTime() :
 318                                 TmfTimestamp.BigCrunch;
 319
 320                 // Process the request
 321                 Thread thread = new Thread() {
 322
 323                         @Override
 324                         public void run() {
 325                                 // Extract the general request information
 326                                 int blockSize = request.getBlockize();
 327                                 int nbRequestedEvents = request.getNbRequestedEvents();
 328                                 if (nbRequestedEvents == -1) {
 329                                         nbRequestedEvents = Integer.MAX_VALUE;
 330                                 }
 331
 332                                 // Create the result buffer
 333                                 Vector<TmfEvent> events = new Vector<TmfEvent>();
 334                                 int nbEvents = 0;
 335
 336                                 // Get the ordered events
 337                                 TmfEvent event = getNextEvent(context);
 338                                 while (!request.isCancelled() && nbEvents < nbRequestedEvents && event != null
 339                                                 && event.getTimestamp().compareTo(endTime, false) <= 0)
 340                                 {
 341                                         events.add(event);
 342                                         if (++nbEvents % blockSize == 0) {
 343                                                 pushData(request, events);
 344                                         }
 345                                         // To avoid an unnecessary read passed the last event requested
 346                                         if (nbEvents < nbRequestedEvents)
 347                                                 event = getNextEvent(context);
 348                                 }
 349                                 pushData(request, events);
 350                                 request.done();
 351                         }
 352                 };
 353                 thread.start();
 354         }
 355
 356         /**
 357          * Format the result data and notify the requester.
 358          * Note: after handling, the data is *removed*.
 359          *
 360          * @param request
 361          * @param events
 362          */
 363         private void pushData(TmfDataRequest<TmfEvent> request, Vector<TmfEvent> events) {
 364                 TmfEvent[] result = new TmfEvent[events.size()];
 365                 events.toArray(result);
 366                 request.setData(result);
 367                 request.handleData();
 368                 events.removeAllElements();
 369         }
 370
 371         /* (non-Javadoc)
 372          * @see java.lang.Object#toString()
 373          */
 374         @Override
 375         public String toString() {
 376                 return "[TmfTrace (" + fName + "]";
 377         }
 378
 379         // ========================================================================
 380     // Trace indexing. Essentially, parse the stream asynchronously and build
 381         // the checkpoints index. This index is used to quickly find an event based
 382         // on a timestamp or an index.
 383     // ========================================================================
 384
 385         private Boolean fIndexing = false;
 386     public void indexStream() {
 387         synchronized (fIndexing) {
 388                 if (fIndexing) {
 389                         return;
 390                 }
 391                 fIndexing = true;
 392         }
 393
 394         final IndexingJob job = new IndexingJob("Indexing " + fName);
 395         job.schedule();
 396
 397         if (fWaitForIndexCompletion) {
 398                 try {
 399                         job.join();
 400                 } catch (InterruptedException e) {
 401                         // TODO Auto-generated catch block
 402                         e.printStackTrace();
 403                 }
 404         }
 405     }
 406
 407     private class IndexingJob extends Job {
 408
 409                 public IndexingJob(String name) {
 410                         super(name);
 411                 }
 412
 413                 /* (non-Javadoc)
 414                  * @see org.eclipse.core.runtime.jobs.Job#run(org.eclipse.core.runtime.IProgressMonitor)
 415                  */
 416                 @Override
 417                 protected IStatus run(IProgressMonitor monitor) {
 418
 419             int nbEvents = 0;
 420             TmfTimestamp startTime = new TmfTimestamp();
 421             TmfTimestamp lastTime  = new TmfTimestamp();
 422
 423             monitor.beginTask("Indexing " + fName, IProgressMonitor.UNKNOWN);
 424
 425             try {
 426                 // Position the trace at the beginning
 427                 TmfTraceContext context = seekLocation(null);
 428                 if (context.getTimestamp() == null) {
 429                         return Status.OK_STATUS;
 430                 }
 431                 // FIXME: LTTng hack - start
 432 //                      fCheckpoints.add(new TmfTraceCheckpoint(context.getTimestamp(), context.getLocation()));        // TMF
 433                 // FIXME: LTTng hack - end
 434
 435                 TmfEvent event;
 436                 startTime = context.getTimestamp();
 437                 lastTime  = context.getTimestamp();
 438                 while ((event = getNextEvent(context)) != null) {
 439                         TmfTimestamp timestamp = context.getTimestamp();
 440                         if (timestamp != null) {
 441                                 lastTime = timestamp;
 442                         }
 443                     // FIXME: LTTng hack - start
 444 //                    if (((++nbEvents % fCacheSize) == 0) && (timestamp != null)) {    // TMF
 445                     if (((nbEvents++ % fCacheSize) == 0) && (timestamp != null)) {      // LTTng
 446                         // FIXME: LTTng hack - end
 447                                 fCheckpoints.add(new TmfTraceCheckpoint(timestamp, context.getLocation()));
 448                                 fNbEvents = nbEvents - 1;
 449                                 lastTime = context.getTimestamp();
 450                                 fTimeRange = new TmfTimeRange(startTime, lastTime);
 451                                 notifyListeners(new TmfTimeRange(startTime, lastTime));
 452
 453                         monitor.worked(1);
 454
 455                         // Check monitor *after* fCheckpoints has been updated
 456                         if (monitor.isCanceled()) {
 457                             monitor.done();
 458                                 return Status.CANCEL_STATUS;
 459                         }
 460                     }
 461
 462                     // Do whatever
 463                     processEvent(event);
 464                 }
 465             }
 466             finally {
 467                 synchronized(this) {
 468                         fNbEvents = nbEvents;
 469                         fTimeRange = new TmfTimeRange(startTime, lastTime);
 470                         fIndexing = false;
 471                 }
 472                 notifyListeners(new TmfTimeRange(startTime, lastTime));
 473                 monitor.done();
 474
 475             }
 476
 477 //            createOffsetsFile();
 478 //            dumpCheckpoints();
 479
 480             return Status.OK_STATUS;
 481                 }
 482     }
 483
 484     private void notifyListeners(TmfTimeRange range) {
 485         broadcastSignal(new TmfTraceUpdatedSignal(this, this, range));
 486         }
 487
 488 //      /**
 489 //       * Dump the trace checkpoints
 490 //       */
 491 //      private void dumpCheckpoints() {
 492 //              System.out.println("-----");
 493 //              System.out.println("Checkpoints of " + fName);
 494 //              for (int i = 0; i < fCheckpoints.size(); i++) {
 495 //                      TmfTraceCheckpoint checkpoint = fCheckpoints.get(i);
 496 //                      TmfTraceContext context = new TmfTraceContext(checkpoint.getLocation());
 497 //                      TmfEvent event = getNextEvent(context);
 498 //                      System.out.println("  Entry: " + i + " timestamp: " + checkpoint.getTimestamp() + ", event: " + event.getTimestamp());
 499 //                      assert((checkpoint.getTimestamp().compareTo(event.getTimestamp(), false) == 0));
 500 //              }
 501 //              System.out.println();
 502 //      }
 503
 504 //      private void createOffsetsFile() {
 505 //
 506 //          try {
 507 //              ObjectOutputStream  out = new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream("LTTngOffsets.dat")));
 508 //
 509 //              TmfTraceContext context = null;
 510 //              context = seekLocation(null);
 511 //                      out.writeObject(context.getLocation());
 512 //
 513 //                      int nbEvents = 0;
 514 //            while (getNextEvent(context) != null) {
 515 //                      out.writeObject(context.getLocation());
 516 //                      nbEvents++;
 517 //            }
 518 //            out.close();
 519 //            System.out.println("TmfTrace wrote " + nbEvents + " events");
 520 //              } catch (FileNotFoundException e) {
 521 //                      // TODO Auto-generated catch block
 522 //                      e.printStackTrace();
 523 //              } catch (IOException e) {
 524 //                      // TODO Auto-generated catch block
 525 //                      e.printStackTrace();
 526 //              }
 527 //      }
 528
 529 //      private void createOffsetsFile() {
 530 //
 531 //              try {
 532 //                      DataOutputStream out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream("LTTngOffsets.dat")));
 533 //
 534 //                      TmfTraceContext context = null;
 535 //                      context = seekLocation(null);
 536 //
 537 //                      TmfEvent event;
 538 //                      int nbEvents = 0;
 539 //                      while ((event = getNextEvent(context)) != null) {
 540 //                              out.writeUTF(event.getTimestamp().toString());
 541 //                              nbEvents++;
 542 //                      }
 543 //                      out.close();
 544 //                      System.out.println("TmfTrace wrote " + nbEvents + " events");
 545 //              } catch (FileNotFoundException e) {
 546 //                      // TODO Auto-generated catch block
 547 //                      e.printStackTrace();
 548 //              } catch (IOException e) {
 549 //                      // TODO Auto-generated catch block
 550 //                      e.printStackTrace();
 551 //              }
 552 //      }
 553
 554 }