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.linuxtools.tmf.component.TmfEventProvider;
  21 import org.eclipse.linuxtools.tmf.event.TmfEvent;
  22 import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  23 import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
  24 import org.eclipse.linuxtools.tmf.request.ITmfDataRequest;
  25 import org.eclipse.linuxtools.tmf.request.ITmfDataRequest.ExecutionType;
  26 import org.eclipse.linuxtools.tmf.request.ITmfEventRequest;
  27 import org.eclipse.linuxtools.tmf.request.TmfDataRequest;
  28 import org.eclipse.linuxtools.tmf.request.TmfEventRequest;
  29 import org.eclipse.linuxtools.tmf.signal.TmfSignalHandler;
  30 import org.eclipse.linuxtools.tmf.signal.TmfTraceOpenedSignal;
  31 import org.eclipse.linuxtools.tmf.signal.TmfTraceUpdatedSignal;
  32
  33 /**
  34  * <b><u>TmfTrace</u></b>
  35  * <p>
  36  * Abstract implementation of ITmfTrace. It should be sufficient to extend this
  37  * class and provide implementation for <code>getCurrentLocation()</code> and
  38  * <code>seekLocation()</code>, as well as a proper parser, to have a working
  39  * concrete implementation.
  40  * <p>
  41  * Note: The notion of event rank is still under heavy discussion. Although
  42  * used by the Events View and probably useful in the general case, there
  43  * is no easy way to implement it for LTTng (actually  a strong case is being
  44  * made that this is useless).
  45  * <p>
  46  * That it is not supported by LTTng does by no mean indicate that it is not
  47  * useful for (just about) every other tracing tool. Therefore, this class
  48  * provides a minimal (and partial) implementation of rank. However, the current
  49  * implementation should not be relied on in the general case.
  50  *
  51  * TODO: Add support for live streaming (notifications, incremental indexing, ...)
  52  */
  53 public abstract class TmfTrace<T extends TmfEvent> extends TmfEventProvider<T> implements ITmfTrace, Cloneable {
  54
  55     // ------------------------------------------------------------------------
  56     // Constants
  57     // ------------------------------------------------------------------------
  58
  59     // The default number of events to cache
  60         // TODO: Make the DEFAULT_CACHE_SIZE a preference
  61     public static final int DEFAULT_INDEX_PAGE_SIZE = 1000;
  62
  63     // ------------------------------------------------------------------------
  64     // Attributes
  65     // ------------------------------------------------------------------------
  66
  67     // The trace path
  68     private final String fPath;
  69
  70     // The cache page size AND checkpoints interval
  71     protected int fIndexPageSize;
  72
  73     // The set of event stream checkpoints (for random access)
  74     protected Vector<TmfCheckpoint> fCheckpoints = new Vector<TmfCheckpoint>();
  75
  76     // The number of events collected
  77     protected long fNbEvents = 0;
  78
  79     // The time span of the event stream
  80     private TmfTimestamp fStartTime = TmfTimestamp.BigCrunch;
  81     private TmfTimestamp fEndTime   = TmfTimestamp.BigBang;
  82
  83     // ------------------------------------------------------------------------
  84     // Constructors
  85     // ------------------------------------------------------------------------
  86
  87     /**
  88      * @param path
  89      * @throws FileNotFoundException
  90      */
  91     protected TmfTrace(String name, Class<T> type, String path) throws FileNotFoundException {
  92         this(name, type, path, DEFAULT_INDEX_PAGE_SIZE);
  93     }
  94
  95     /**
  96      * @param path
  97      * @param cacheSize
  98      * @throws FileNotFoundException
  99      */
 100     protected TmfTrace(String name, Class<T> type, String path, int cacheSize) throws FileNotFoundException {
 101         super(name, type);
 102         int sep = path.lastIndexOf(File.separator);
 103         String simpleName = (sep >= 0) ? path.substring(sep + 1) : path;
 104         setName(simpleName);
 105         fPath = path;
 106         fIndexPageSize = (cacheSize > 0) ? cacheSize : DEFAULT_INDEX_PAGE_SIZE;
 107     }
 108
 109     /* (non-Javadoc)
 110      * @see java.lang.Object#clone()
 111      */
 112     @SuppressWarnings("unchecked")
 113         @Override
 114         public TmfTrace<T> clone() throws CloneNotSupportedException {
 115         TmfTrace<T> clone = (TmfTrace<T>) super.clone();
 116         clone.fCheckpoints = (Vector<TmfCheckpoint>) fCheckpoints;
 117         clone.fStartTime = new TmfTimestamp(fStartTime);
 118         clone.fEndTime   = new TmfTimestamp(fEndTime);
 119         return clone;
 120     }
 121
 122     // ------------------------------------------------------------------------
 123     // Accessors
 124     // ------------------------------------------------------------------------
 125
 126     /**
 127      * @return the trace path
 128      */
 129     public String getPath() {
 130         return fPath;
 131     }
 132
 133     /* (non-Javadoc)
 134      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getNbEvents()
 135      */
 136     public long getNbEvents() {
 137         return fNbEvents;
 138     }
 139
 140     /**
 141      * @return the size of the cache
 142      */
 143     public int getCacheSize() {
 144         return fIndexPageSize;
 145     }
 146
 147     /* (non-Javadoc)
 148      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getTimeRange()
 149      */
 150     public TmfTimeRange getTimeRange() {
 151         return new TmfTimeRange(fStartTime, fEndTime);
 152     }
 153
 154     /* (non-Javadoc)
 155      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getStartTime()
 156      */
 157     public TmfTimestamp getStartTime() {
 158         return fStartTime;
 159     }
 160
 161     /* (non-Javadoc)
 162      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getEndTime()
 163      */
 164     public TmfTimestamp getEndTime() {
 165         return fEndTime;
 166     }
 167
 168     @SuppressWarnings("unchecked")
 169         public Vector<TmfCheckpoint> getCheckpoints() {
 170         return (Vector<TmfCheckpoint>) fCheckpoints.clone();
 171     }
 172
 173     /**
 174      * Returns the rank of the first event with the requested timestamp.
 175      * If none, returns the index of the next event (if any).
 176      *
 177      * @param timestamp
 178      * @return
 179      */
 180     public long getRank(TmfTimestamp timestamp) {
 181         TmfContext context = seekEvent(timestamp);
 182         return context.getRank();
 183     }
 184
 185     // ------------------------------------------------------------------------
 186     // Operators
 187     // ------------------------------------------------------------------------
 188
 189     protected void setTimeRange(TmfTimeRange range) {
 190         fStartTime = range.getStartTime();
 191         fEndTime   = range.getEndTime();
 192     }
 193
 194     protected void setStartTime(TmfTimestamp startTime) {
 195         fStartTime = startTime;
 196     }
 197
 198     protected void setEndTime(TmfTimestamp endTime) {
 199         fEndTime = endTime;
 200     }
 201
 202         // ------------------------------------------------------------------------
 203         // TmfProvider
 204         // ------------------------------------------------------------------------
 205
 206         @Override
 207         public ITmfContext armRequest(ITmfDataRequest<T> request) {
 208                 if (request instanceof ITmfEventRequest<?>) {
 209                         return seekEvent(((ITmfEventRequest<T>) request).getRange().getStartTime());
 210                 }
 211                 return seekEvent(request.getIndex());
 212         }
 213
 214         /**
 215          * Return the next piece of data based on the context supplied. The context
 216          * would typically be updated for the subsequent read.
 217          *
 218          * @param context
 219          * @return
 220          */
 221         @SuppressWarnings("unchecked")
 222         @Override
 223         public T getNext(ITmfContext context) {
 224                 if (context instanceof TmfContext) {
 225                         return (T) getNextEvent((TmfContext) context);
 226                 }
 227                 return null;
 228         }
 229
 230         // ------------------------------------------------------------------------
 231         // ITmfTrace
 232         // ------------------------------------------------------------------------
 233
 234     /* (non-Javadoc)
 235      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(org.eclipse.linuxtools.tmf.event.TmfTimestamp)
 236      */
 237     public TmfContext seekEvent(TmfTimestamp timestamp) {
 238
 239         if (timestamp == null) {
 240                 timestamp = TmfTimestamp.BigBang;
 241         }
 242
 243         // First, find the right checkpoint
 244         int index = Collections.binarySearch(fCheckpoints, new TmfCheckpoint(timestamp, null));
 245
 246         // In the very likely case that the checkpoint was not found, bsearch
 247         // returns its negated would-be location (not an offset...). From that
 248         // index, we can then position the stream and get the event.
 249         if (index < 0) {
 250             index = Math.max(0, -(index + 2));
 251         }
 252
 253         // Position the stream at the checkpoint
 254         ITmfLocation<?> location;
 255         synchronized (fCheckpoints) {
 256                 if (fCheckpoints.size() > 0) {
 257                         if (index >= fCheckpoints.size()) {
 258                                 index = fCheckpoints.size() - 1;
 259                         }
 260                         location = fCheckpoints.elementAt(index).getLocation();
 261                 }
 262                 else {
 263                         location = null;
 264                 }
 265         }
 266         TmfContext context = seekLocation(location);
 267         context.setRank(index * fIndexPageSize);
 268
 269         // And locate the event
 270         TmfContext nextEventContext = context.clone(); // Must use clone() to get the right subtype...
 271         TmfEvent event = getNextEvent(nextEventContext);
 272         while (event != null && event.getTimestamp().compareTo(timestamp, false) < 0) {
 273                 context.setLocation(nextEventContext.getLocation().clone());
 274                 context.updateRank(1);
 275                 event = getNextEvent(nextEventContext);
 276         }
 277
 278         return context;
 279     }
 280
 281     /* (non-Javadoc)
 282      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(int)
 283      */
 284     public TmfContext seekEvent(long rank) {
 285
 286         // Position the stream at the previous checkpoint
 287         int index = (int) rank / fIndexPageSize;
 288         ITmfLocation<?> location;
 289         synchronized (fCheckpoints) {
 290                 if (fCheckpoints.size() == 0) {
 291                         location = null;
 292                 }
 293                 else {
 294                         if (index >= fCheckpoints.size()) {
 295                                 index  = fCheckpoints.size() - 1;
 296                         }
 297                         location = fCheckpoints.elementAt(index).getLocation();
 298                 }
 299         }
 300
 301         TmfContext context = seekLocation(location);
 302         long pos = index * fIndexPageSize;
 303         context.setRank(pos);
 304
 305         if (pos < rank) {
 306             TmfEvent event = getNextEvent(context);
 307             while (event != null && ++pos < rank) {
 308                 event = getNextEvent(context);
 309             }
 310         }
 311
 312         return context;
 313     }
 314
 315         /* (non-Javadoc)
 316          * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getNextEvent(org.eclipse.linuxtools.tmf.trace.ITmfTrace.TraceContext)
 317          */
 318         public synchronized TmfEvent getNextEvent(TmfContext context) {
 319                 // parseEvent() does not update the context
 320                 TmfEvent event = parseEvent(context);
 321                 if (event != null) {
 322                         updateIndex(context, context.getRank(), event.getTimestamp());
 323                         context.setLocation(getCurrentLocation());
 324                         context.updateRank(1);
 325                         processEvent(event);
 326                 }
 327         return event;
 328         }
 329
 330         protected synchronized void updateIndex(ITmfContext context, long rank, TmfTimestamp timestamp) {
 331                 if (fStartTime.compareTo(timestamp, false) > 0) fStartTime = timestamp;
 332                 if (fEndTime.compareTo(timestamp, false) < 0) fEndTime = timestamp;
 333                 if (context.isValidRank()) {
 334                         if (fNbEvents <= rank)
 335                                 fNbEvents = rank + 1;
 336                         // Build the index as we go along
 337                         if ((rank % fIndexPageSize) == 0) {
 338                                 // Determine the table position
 339                                 long position = rank / fIndexPageSize;
 340                                 // Add new entry at proper location (if empty)
 341                                 if (fCheckpoints.size() == position) {
 342                                         ITmfLocation<?> location = context.getLocation().clone();
 343                                         fCheckpoints.add(new TmfCheckpoint(timestamp, location));
 344 //                                      System.out.println(getName() + "[" + (fCheckpoints.size() - 1) + "] " + timestamp + ", " + location.toString());
 345                                 }
 346                         }
 347                 }
 348         }
 349
 350     /**
 351          * Hook for "special" processing by the concrete class
 352          * (called by getNextEvent())
 353          *
 354          * @param event
 355          */
 356         protected void processEvent(TmfEvent event) {
 357                 // Do nothing by default
 358         }
 359
 360     /**
 361      * To be implemented by the concrete class
 362      */
 363     public abstract TmfContext seekLocation(ITmfLocation<?> location);
 364         public abstract ITmfLocation<?> getCurrentLocation();
 365     public abstract TmfEvent parseEvent(TmfContext context);
 366
 367         // ------------------------------------------------------------------------
 368         // toString
 369         // ------------------------------------------------------------------------
 370
 371         /* (non-Javadoc)
 372          * @see java.lang.Object#toString()
 373          */
 374         @Override
 375         public String toString() {
 376                 return "[TmfTrace (" + getName() + ")]";
 377         }
 378
 379     // ------------------------------------------------------------------------
 380     // Indexing
 381     // ------------------------------------------------------------------------
 382
 383         /*
 384          * The purpose of the index is to keep the information needed to rapidly
 385          * restore the traces contexts at regular intervals (every INDEX_PAGE_SIZE
 386          * event).
 387          */
 388
 389         @SuppressWarnings("unchecked")
 390         private void indexTrace(boolean waitForCompletion) {
 391
 392                 fCheckpoints.clear();
 393
 394                 ITmfEventRequest<TmfEvent> request = new TmfEventRequest<TmfEvent>(TmfEvent.class, TmfTimeRange.Eternity, TmfDataRequest.ALL_DATA, 1, ITmfDataRequest.ExecutionType.LONG) {
 395
 396                         TmfTimestamp startTime =  null;
 397                         TmfTimestamp lastTime  =  null;
 398
 399                         @Override
 400                         public void handleData() {
 401                                 TmfEvent[] events = getData();
 402                                 if (events.length > 0) {
 403                                         TmfTimestamp ts = events[0].getTimestamp();
 404                                         if (startTime == null) {
 405                                                 startTime = new TmfTimestamp(ts);
 406                                                 fStartTime = startTime;
 407                                         }
 408                                         lastTime = new TmfTimestamp(ts);
 409
 410                                         if ((fNbRead % DEFAULT_INDEX_PAGE_SIZE) == 0) {
 411                                                 updateTraceData();
 412                                         }
 413                                 }
 414                         }
 415
 416                         @Override
 417                         public void handleSuccess() {
 418                                 updateTraceData();
 419                         }
 420
 421                         private void updateTraceData() {
 422                                 fEndTime  = new TmfTimestamp(lastTime);
 423                                 fNbEvents = fNbRead;
 424                                 notifyListeners();
 425                         }
 426                 };
 427
 428                 sendRequest((ITmfDataRequest<T>) request);
 429                 if (waitForCompletion)
 430                         try {
 431                                 request.waitForCompletion();
 432                         } catch (InterruptedException e) {
 433                                 e.printStackTrace();
 434                         }
 435         }
 436
 437         protected void notifyListeners() {
 438         broadcast(new TmfTraceUpdatedSignal(this, this, new TmfTimeRange(fStartTime, fEndTime)));
 439         }
 440
 441         // ------------------------------------------------------------------------
 442         // TmfDataProvider
 443         // ------------------------------------------------------------------------
 444
 445         @Override
 446     protected void queueLongRequest(final ITmfDataRequest<T> request) {
 447
 448         // TODO: Handle the data requests also...
 449         if (!(request instanceof ITmfEventRequest<?>)) {
 450             super.queueRequest(request);
 451             return;
 452         }
 453
 454         final TmfTrace<T> trace = this;
 455
 456         Thread thread = new Thread() {
 457             @Override
 458             public void run() {
 459
 460 //              final long requestStart = System.nanoTime();
 461
 462                 final Integer[] CHUNK_SIZE = new Integer[1];
 463                 CHUNK_SIZE[0] = fIndexPageSize + 1;
 464
 465                 final ITmfEventRequest<T> req = (ITmfEventRequest<T>) request;
 466
 467                 final Integer[] nbRead = new Integer[1];
 468                 nbRead[0] = 0;
 469
 470 //              final TmfTimestamp[] timestamp = new TmfTimestamp[1];
 471 //              timestamp[0] = new TmfTimestamp(req.getRange().getStartTime());
 472 //              final TmfTimestamp endTS = req.getRange().getEndTime();
 473
 474                 final Boolean[] isFinished = new Boolean[1];
 475                 isFinished[0] = Boolean.FALSE;
 476
 477                 while (!isFinished[0]) {
 478
 479 //                  TmfEventRequest<T> subRequest = new TmfEventRequest<T>(req.getDataType(), new TmfTimeRange(timestamp[0], endTS),
 480 //                          requestedSize, req.getBlockize(), ExecutionType.LONG)
 481                     TmfDataRequest<T> subRequest = new TmfDataRequest<T>(req.getDataType(), nbRead[0], CHUNK_SIZE[0],
 482                             req.getBlockize(), ExecutionType.LONG)
 483                     {
 484 //                      int count = 0;
 485                         @Override
 486                         public void handleData() {
 487                             T[] data = getData();
 488 //                          if (count == 0) {
 489 //                              System.out.println("First event of the block: " + data[0].getTimestamp());
 490 //                          }
 491 //                          count++;
 492 //                          Tracer.trace("Ndx: " + ((TmfEvent) data[0]).getTimestamp());
 493                             req.setData(data);
 494                             req.handleData();
 495                             if (fNbRead == CHUNK_SIZE[0]) {
 496                                 nbRead[0] += fNbRead;
 497 //                              System.out.println("fNbRead=" + fNbRead + ", count=" + count +", total=" + nbRead[0] + ", TS=" + data[0].getTimestamp());
 498                             }
 499                             if (fNbRead > CHUNK_SIZE[0]) {
 500                                 System.out.println("ERROR - Read too many events");
 501                             }
 502                         }
 503                         @Override
 504                         public void handleCompleted() {
 505 //                          System.out.println("Request completed at: " + timestamp[0]);
 506                             if (fNbRead < CHUNK_SIZE[0]) {
 507                                 req.done();
 508                                 isFinished[0] = Boolean.TRUE;
 509                                 nbRead[0] += fNbRead;
 510 //                              System.out.println("fNbRead=" + fNbRead + ", count=" + count +", total=" + nbRead[0]);
 511                             }
 512                             super.handleCompleted();
 513                         }
 514                     };
 515
 516                     if (!isFinished[0]) {
 517                         trace.queueRequest(subRequest);
 518
 519                         try {
 520                             subRequest.waitForCompletion();
 521 //                          System.out.println("Finished at " + timestamp[0]);
 522                         } catch (InterruptedException e) {
 523                             e.printStackTrace();
 524                         }
 525
 526 //                      TmfTimestamp newTS = new TmfTimestamp(timestamp[0].getValue() + 1, timestamp[0].getScale(), timestamp[0].getPrecision());
 527 //                      timestamp[0] = newTS;
 528                         CHUNK_SIZE[0] = fIndexPageSize;
 529 //                      System.out.println("New timestamp: " + timestamp[0]);
 530                     }
 531                 }
 532 //              final long requestEnded = System.nanoTime();
 533 //              System.out.println("Background request completed. Elapsed= " + (requestEnded * 1.0 - requestStart) / 1000000000);
 534             }
 535         };
 536         thread.start();
 537     }
 538
 539         // ------------------------------------------------------------------------
 540         // Signal handlers
 541         // ------------------------------------------------------------------------
 542
 543         @TmfSignalHandler
 544         public void traceOpened(TmfTraceOpenedSignal signal) {
 545             ITmfTrace trace = signal.getTrace();
 546             if (trace == this) {
 547                 indexTrace(false);
 548             }
 549         }
 550 }