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

   1 /*******************************************************************************
   2  * Copyright (c) 2009, 2010 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.TmfEventProvider;
  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.TmfCoalescedEventRequest;
  29 import org.eclipse.linuxtools.tmf.request.TmfDataRequest;
  30 import org.eclipse.linuxtools.tmf.request.TmfEventRequest;
  31
  32 /**
  33  * <b><u>TmfTrace</u></b>
  34  * <p>
  35  * Abstract implementation of ITmfTrace. It should be sufficient to extend this
  36  * class and provide implementation for <code>getCurrentLocation()</code> and
  37  * <code>seekLocation()</code>, as well as a proper parser, to have a working
  38  * concrete implementation.
  39  *
  40  * Note: The notion of event rank is still under heavy discussion. Although
  41  * used for by the Events View and probably useful in then general case, there
  42  * is no easy way to implement it for LTTng (actually  a strong case is being
  43  * made that this is useless). Therefore, this is a minimal and partial
  44  * implementation and rank should not be relied upon in the general case (i.e.
  45  *  it was hacked to work for the Events View).
  46  *
  47  * TODO: Add support for live streaming (notifications, incremental indexing, ...)
  48  */
  49 public abstract class TmfTrace<T extends TmfEvent> extends TmfEventProvider<T> implements ITmfTrace {
  50
  51     // ------------------------------------------------------------------------
  52     // Constants
  53     // ------------------------------------------------------------------------
  54
  55     // The default number of events to cache
  56         // TODO: Make the DEFAULT_CACHE_SIZE a preference
  57     public static final int DEFAULT_CACHE_SIZE = 1000;
  58
  59     // ------------------------------------------------------------------------
  60     // Attributes
  61     // ------------------------------------------------------------------------
  62
  63     // The trace path
  64     private final String fPath;
  65
  66     // The trace name
  67     private final String fName;
  68
  69     // The cache page size AND checkpoints interval
  70     protected int fIndexPageSize;
  71
  72     // The set of event stream checkpoints (for random access)
  73     protected Vector<TmfCheckpoint> fCheckpoints = new Vector<TmfCheckpoint>();
  74
  75     // The number of events collected
  76     protected long fNbEvents = 0;
  77
  78     // The time span of the event stream
  79     private TmfTimeRange fTimeRange = new TmfTimeRange(TmfTimestamp.BigBang, TmfTimestamp.BigBang);
  80
  81     // ------------------------------------------------------------------------
  82     // Constructors
  83     // ------------------------------------------------------------------------
  84
  85     /**
  86      * @param path
  87      * @param cacheSize
  88      * @throws FileNotFoundException
  89      */
  90     protected TmfTrace(Class<T> type, String path, int cacheSize) throws FileNotFoundException {
  91         super(type);
  92         int sep = path.lastIndexOf(File.separator);
  93         fName = (sep >= 0) ? path.substring(sep + 1) : path;
  94         fPath = path;
  95         fIndexPageSize = (cacheSize > 0) ? cacheSize : DEFAULT_CACHE_SIZE;
  96     }
  97
  98     /**
  99      * @param path
 100      * @throws FileNotFoundException
 101      */
 102     protected TmfTrace(Class<T> type, String path) throws FileNotFoundException {
 103         this(type, path, DEFAULT_CACHE_SIZE);
 104     }
 105
 106     // ------------------------------------------------------------------------
 107     // Accessors
 108     // ------------------------------------------------------------------------
 109
 110     /**
 111      * @return the trace path
 112      */
 113     public String getPath() {
 114         return fPath;
 115     }
 116
 117     /**
 118      * @return the trace name
 119      */
 120     @Override
 121         public String getName() {
 122         return fName;
 123     }
 124
 125     /* (non-Javadoc)
 126      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getNbEvents()
 127      */
 128     public long getNbEvents() {
 129         return fNbEvents;
 130     }
 131
 132     /**
 133      * @return the size of the cache
 134      */
 135     public int getCacheSize() {
 136         return fIndexPageSize;
 137     }
 138
 139     /* (non-Javadoc)
 140      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getTimeRange()
 141      */
 142     public TmfTimeRange getTimeRange() {
 143         return fTimeRange;
 144     }
 145
 146     /* (non-Javadoc)
 147      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getStartTime()
 148      */
 149     public TmfTimestamp getStartTime() {
 150         return fTimeRange.getStartTime();
 151     }
 152
 153     /* (non-Javadoc)
 154      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getEndTime()
 155      */
 156     public TmfTimestamp getEndTime() {
 157         return fTimeRange.getEndTime();
 158     }
 159
 160     public Vector<TmfCheckpoint> getCheckpoints() {
 161         return fCheckpoints;
 162     }
 163
 164     // ------------------------------------------------------------------------
 165     // Operators
 166     // ------------------------------------------------------------------------
 167
 168     protected void setTimeRange(TmfTimeRange range) {
 169         fTimeRange = range;
 170     }
 171
 172     protected void setStartTime(TmfTimestamp startTime) {
 173         fTimeRange = new TmfTimeRange(startTime, fTimeRange.getEndTime());
 174     }
 175
 176     protected void setEndTime(TmfTimestamp endTime) {
 177         fTimeRange = new TmfTimeRange(fTimeRange.getStartTime(), endTime);
 178     }
 179
 180         // ------------------------------------------------------------------------
 181         // TmfProvider
 182         // ------------------------------------------------------------------------
 183
 184         @Override
 185         public ITmfContext armRequest(TmfDataRequest<T> request) {
 186                 if (request instanceof TmfEventRequest<?>) {
 187                         return seekEvent(((TmfEventRequest<T>) request).getRange().getStartTime());
 188                 }
 189                 if (request instanceof TmfCoalescedEventRequest<?>) {
 190                         return seekEvent(((TmfCoalescedEventRequest<T>) request).getRange().getStartTime());
 191                 }
 192                 return null;
 193         }
 194
 195         /**
 196          * Return the next piece of data based on the context supplied. The context
 197          * would typically be updated for the subsequent read.
 198          *
 199          * @param context
 200          * @return
 201          */
 202         @SuppressWarnings("unchecked")
 203         @Override
 204         public T getNext(ITmfContext context) {
 205                 if (context instanceof TmfContext) {
 206                         return (T) getNextEvent((TmfContext) context);
 207                 }
 208                 return null;
 209         }
 210
 211         // ------------------------------------------------------------------------
 212         // ITmfTrace
 213         // ------------------------------------------------------------------------
 214
 215     /* (non-Javadoc)
 216      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(org.eclipse.linuxtools.tmf.event.TmfTimestamp)
 217      */
 218     public TmfContext seekEvent(TmfTimestamp timestamp) {
 219
 220         if (timestamp == null) {
 221                 timestamp = TmfTimestamp.BigBang;
 222         }
 223
 224         // First, find the right checkpoint
 225         int index = Collections.binarySearch(fCheckpoints, new TmfCheckpoint(timestamp, null));
 226
 227         // In the very likely case that the checkpoint was not found, bsearch
 228         // returns its negated would-be location (not an offset...). From that
 229         // index, we can then position the stream and get the event.
 230         if (index < 0) {
 231             index = Math.max(0, -(index + 2));
 232         }
 233
 234         // Position the stream at the checkpoint
 235         ITmfLocation<?> location;
 236         synchronized (fCheckpoints) {
 237                 if (fCheckpoints.size() > 0) {
 238                         if (index >= fCheckpoints.size()) {
 239                                 index = fCheckpoints.size() - 1;
 240                         }
 241                         location = fCheckpoints.elementAt(index).getLocation();
 242                 }
 243                 else {
 244                         location = null;
 245                 }
 246         }
 247         TmfContext context = seekLocation(location);
 248         context.setRank(index * fIndexPageSize);
 249
 250         // And locate the event
 251         TmfContext nextEventContext = new TmfContext(context);
 252         TmfEvent event = getNextEvent(nextEventContext);
 253         while (event != null && event.getTimestamp().compareTo(timestamp, false) < 0) {
 254                 context.setLocation(nextEventContext.getLocation().clone());
 255                 context.updateRank(1);
 256                 event = getNextEvent(nextEventContext);
 257         }
 258
 259         return context;
 260     }
 261
 262     /* (non-Javadoc)
 263      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(int)
 264      */
 265     public TmfContext seekEvent(long rank) {
 266
 267         // Position the stream at the previous checkpoint
 268         int index = (int) rank / fIndexPageSize;
 269         ITmfLocation<?> location;
 270         synchronized (fCheckpoints) {
 271                 if (fCheckpoints.size() == 0) {
 272                         location = null;
 273                 }
 274                 else {
 275                         if (index >= fCheckpoints.size()) {
 276                                 index  = fCheckpoints.size() - 1;
 277                         }
 278                         location = fCheckpoints.elementAt(index).getLocation();
 279                 }
 280         }
 281
 282         TmfContext context = seekLocation(location);
 283         long pos = index * fIndexPageSize;
 284         context.setRank(pos);
 285
 286         if (pos < rank) {
 287             TmfEvent event = getNextEvent(context);
 288             while (event != null && ++pos < rank) {
 289                 event = getNextEvent(context);
 290             }
 291         }
 292
 293         return new TmfContext(context.getLocation(), context.getRank());
 294     }
 295
 296         /* (non-Javadoc)
 297          * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getNextEvent(org.eclipse.linuxtools.tmf.trace.ITmfTrace.TraceContext)
 298          */
 299         public synchronized TmfEvent getNextEvent(TmfContext context) {
 300                 // parseEvent() does not update the context
 301                 TmfEvent event = parseEvent(context);
 302                 if (event != null) {
 303                         context.setLocation(getCurrentLocation());
 304                         context.updateRank(1);
 305                         processEvent(event);
 306                 }
 307         return event;
 308         }
 309
 310     /**
 311          * Hook for "special" processing by the concrete class
 312          * (called by getNextEvent())
 313          *
 314          * @param event
 315          */
 316         public void processEvent(TmfEvent event) {
 317                 // Do nothing by default
 318         }
 319
 320     /**
 321      * To be implemented by the concrete class
 322      */
 323     public abstract TmfContext seekLocation(ITmfLocation<?> location);
 324         public abstract ITmfLocation<?> getCurrentLocation();
 325     public abstract TmfEvent parseEvent(TmfContext context);
 326
 327         // ------------------------------------------------------------------------
 328         // toString
 329         // ------------------------------------------------------------------------
 330
 331         /* (non-Javadoc)
 332          * @see java.lang.Object#toString()
 333          */
 334         @Override
 335         public String toString() {
 336                 return "[TmfTrace (" + fName + "]";
 337         }
 338
 339     // ------------------------------------------------------------------------
 340     // Indexing
 341     // ------------------------------------------------------------------------
 342
 343         /*
 344          * The purpose of the index is to keep the information needed to rapidly
 345          * access a trace event based on its timestamp or rank.
 346          *
 347          * NOTE: As it is, doesn't work for streaming traces.
 348          */
 349
 350         private IndexingJob job;
 351
 352         // Indicates that an indexing job is already running
 353         private Object  fIndexingLock = new Object();
 354         private boolean fIndexing = false;
 355         private Boolean fIndexed  = false;
 356
 357         public void indexTrace(boolean waitForCompletion) {
 358         synchronized (fIndexingLock) {
 359                         if (fIndexed || fIndexing) {
 360                         return;
 361                 }
 362                 fIndexing = true;
 363         }
 364
 365         job = new IndexingJob("Indexing " + fName);
 366         job.schedule();
 367
 368         if (waitForCompletion) {
 369                 try {
 370                         job.join();
 371                 } catch (InterruptedException e) {
 372                         e.printStackTrace();
 373                 }
 374         }
 375     }
 376
 377     private class IndexingJob extends Job {
 378
 379                 public IndexingJob(String name) {
 380                         super(name);
 381                 }
 382
 383                 /* (non-Javadoc)
 384                  * @see org.eclipse.core.runtime.jobs.Job#run(org.eclipse.core.runtime.IProgressMonitor)
 385                  */
 386                 @Override
 387                 protected IStatus run(IProgressMonitor monitor) {
 388
 389             monitor.beginTask("Indexing " + fName, IProgressMonitor.UNKNOWN);
 390
 391             int nbEvents = 0;
 392             TmfTimestamp startTime = null;
 393             TmfTimestamp lastTime  = null;
 394
 395             // Reset the index
 396             fCheckpoints = new Vector<TmfCheckpoint>();
 397
 398             try {
 399                 // Position the trace at the beginning
 400                 TmfContext context = seekLocation(null);
 401                 ITmfLocation<?> location = context.getLocation().clone();
 402
 403                 // Get the first event
 404                 TmfEvent event = getNextEvent(context);
 405                 if (event != null) {
 406                     startTime = new TmfTimestamp(event.getTimestamp());
 407                 }
 408
 409                 // Index the trace
 410                 while (event != null) {
 411                         lastTime = event.getTimestamp();
 412                                 if ((nbEvents++ % fIndexPageSize) == 0) {
 413                                         lastTime = new TmfTimestamp(event.getTimestamp());
 414                                 fCheckpoints.add(new TmfCheckpoint(lastTime, location.clone()));
 415                                 fNbEvents = nbEvents;
 416                                 fTimeRange = new TmfTimeRange(startTime, lastTime);
 417                                 notifyListeners(new TmfTimeRange(startTime, lastTime));
 418
 419                         monitor.worked(1);
 420
 421                         // Check monitor *after* fCheckpoints has been updated
 422                         if (monitor.isCanceled()) {
 423                             monitor.done();
 424                                 return Status.CANCEL_STATUS;
 425                         }
 426                     }
 427
 428                     // We will need this location at the next iteration
 429                     if ((nbEvents % fIndexPageSize) == 0) {
 430                         location = context.getLocation().clone();
 431                                 }
 432
 433                     event = getNextEvent(context);
 434                 }
 435             }
 436             finally {
 437                 synchronized(this) {
 438                         fNbEvents = nbEvents;
 439                         fTimeRange = new TmfTimeRange(startTime, lastTime);
 440                         fIndexing = false;
 441                         fIndexed = true;
 442                 }
 443                 notifyListeners(new TmfTimeRange(startTime, lastTime));
 444                 monitor.done();
 445             }
 446
 447             return Status.OK_STATUS;
 448                 }
 449     }
 450
 451     protected void notifyListeners(TmfTimeRange range) {
 452         broadcast(new TmfTraceUpdatedSignal(this, this, range));
 453         }
 454
 455 }