org.eclipse.linuxtools.tmf.core/src/org/eclipse/linuxtools/tmf/core/experiment/TmfExperiment.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.core.experiment;
  14
  15 import java.util.Collections;
  16 import java.util.Vector;
  17
  18 import org.eclipse.core.resources.IFile;
  19 import org.eclipse.core.resources.IProject;
  20 import org.eclipse.core.resources.IResource;
  21 import org.eclipse.core.runtime.IProgressMonitor;
  22 import org.eclipse.core.runtime.IStatus;
  23 import org.eclipse.core.runtime.Status;
  24 import org.eclipse.core.runtime.jobs.Job;
  25 import org.eclipse.linuxtools.tmf.core.component.TmfEventProvider;
  26 import org.eclipse.linuxtools.tmf.core.event.ITmfEvent;
  27 import org.eclipse.linuxtools.tmf.core.event.ITmfTimestamp;
  28 import org.eclipse.linuxtools.tmf.core.event.TmfTimeRange;
  29 import org.eclipse.linuxtools.tmf.core.event.TmfTimestamp;
  30 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest;
  31 import org.eclipse.linuxtools.tmf.core.request.ITmfEventRequest;
  32 import org.eclipse.linuxtools.tmf.core.request.TmfDataRequest;
  33 import org.eclipse.linuxtools.tmf.core.request.TmfEventRequest;
  34 import org.eclipse.linuxtools.tmf.core.signal.TmfEndSynchSignal;
  35 import org.eclipse.linuxtools.tmf.core.signal.TmfExperimentDisposedSignal;
  36 import org.eclipse.linuxtools.tmf.core.signal.TmfExperimentRangeUpdatedSignal;
  37 import org.eclipse.linuxtools.tmf.core.signal.TmfExperimentSelectedSignal;
  38 import org.eclipse.linuxtools.tmf.core.signal.TmfExperimentUpdatedSignal;
  39 import org.eclipse.linuxtools.tmf.core.signal.TmfSignalHandler;
  40 import org.eclipse.linuxtools.tmf.core.signal.TmfTraceUpdatedSignal;
  41 import org.eclipse.linuxtools.tmf.core.trace.ITmfContext;
  42 import org.eclipse.linuxtools.tmf.core.trace.ITmfLocation;
  43 import org.eclipse.linuxtools.tmf.core.trace.ITmfTrace;
  44 import org.eclipse.linuxtools.tmf.core.trace.TmfCheckpoint;
  45 import org.eclipse.linuxtools.tmf.core.trace.TmfContext;
  46
  47 /**
  48  * <b><u>TmfExperiment</u></b>
  49  * <p>
  50  * TmfExperiment presents a time-ordered, unified view of a set of TmfTraces
  51  * that are part of a tracing experiment.
  52  * <p>
  53  */
  54 public class TmfExperiment<T extends ITmfEvent> extends TmfEventProvider<T> implements ITmfTrace<T> {
  55
  56     // ------------------------------------------------------------------------
  57     // Attributes
  58     // ------------------------------------------------------------------------
  59
  60     // The currently selected experiment
  61     protected static TmfExperiment<?> fCurrentExperiment = null;
  62
  63     // The set of traces that constitute the experiment
  64     protected ITmfTrace<T>[] fTraces;
  65
  66     // The total number of events
  67     protected long fNbEvents;
  68
  69     // The experiment time range
  70     protected TmfTimeRange fTimeRange;
  71
  72     // The experiment reference timestamp (default: ZERO)
  73     protected ITmfTimestamp fEpoch;
  74
  75     // The experiment index
  76     protected Vector<TmfCheckpoint> fCheckpoints = new Vector<TmfCheckpoint>();
  77
  78     // The current experiment context
  79     protected TmfExperimentContext fExperimentContext;
  80
  81     // Flag to initialize only once
  82     private boolean fInitialized = false;
  83
  84     // The experiment bookmarks file
  85     private IFile fBookmarksFile;
  86
  87     // The properties resource
  88     private IResource fResource;
  89
  90     // ------------------------------------------------------------------------
  91     // Constructors
  92     // ------------------------------------------------------------------------
  93
  94     @Override
  95     public TmfExperiment<T> clone() throws CloneNotSupportedException {
  96         throw new CloneNotSupportedException();
  97     }
  98
  99     @Override
 100     public boolean validate(final IProject project, final String path) {
 101         return true;
 102     }
 103
 104     @Override
 105     public void initTrace(final String name, final String path, final Class<T> eventType) {
 106     }
 107
 108     /**
 109      * @param type
 110      * @param id
 111      * @param traces
 112      * @param epoch
 113      * @param indexPageSize
 114      */
 115     public TmfExperiment(final Class<T> type, final String id, final ITmfTrace<T>[] traces, final ITmfTimestamp epoch,
 116             final int indexPageSize) {
 117         this(type, id, traces, TmfTimestamp.ZERO, indexPageSize, false);
 118     }
 119
 120     public TmfExperiment(final Class<T> type, final String id, final ITmfTrace<T>[] traces, final ITmfTimestamp epoch,
 121             final int indexPageSize, final boolean preIndexExperiment) {
 122         super(id, type);
 123
 124         fTraces = traces;
 125         fEpoch = epoch;
 126         fIndexPageSize = indexPageSize;
 127         fTimeRange = TmfTimeRange.NULL_RANGE;
 128
 129         if (preIndexExperiment) {
 130             indexExperiment(true, 0, TmfTimeRange.ETERNITY);
 131             updateTimeRange();
 132         }
 133     }
 134
 135     protected TmfExperiment(final String id, final Class<T> type) {
 136         super(id, type);
 137     }
 138
 139     /**
 140      * @param type
 141      * @param id
 142      * @param traces
 143      */
 144     public TmfExperiment(final Class<T> type, final String id, final ITmfTrace<T>[] traces) {
 145         this(type, id, traces, TmfTimestamp.ZERO, DEFAULT_INDEX_PAGE_SIZE);
 146     }
 147
 148     /**
 149      * @param type
 150      * @param id
 151      * @param traces
 152      * @param indexPageSize
 153      */
 154     public TmfExperiment(final Class<T> type, final String id, final ITmfTrace<T>[] traces, final int indexPageSize) {
 155         this(type, id, traces, TmfTimestamp.ZERO, indexPageSize);
 156     }
 157
 158     /**
 159      * Clears the experiment
 160      */
 161     @Override
 162     @SuppressWarnings("rawtypes")
 163     public synchronized void dispose() {
 164
 165         final TmfExperimentDisposedSignal<T> signal = new TmfExperimentDisposedSignal<T>(this, this);
 166         broadcast(signal);
 167         if (fCurrentExperiment == this)
 168             fCurrentExperiment = null;
 169
 170         if (fTraces != null) {
 171             for (final ITmfTrace trace : fTraces)
 172                 trace.dispose();
 173             fTraces = null;
 174         }
 175         if (fCheckpoints != null)
 176             fCheckpoints.clear();
 177         super.dispose();
 178     }
 179
 180     // ------------------------------------------------------------------------
 181     // ITmfTrace
 182     // ------------------------------------------------------------------------
 183
 184     @Override
 185     public long getNbEvents() {
 186         return fNbEvents;
 187     }
 188
 189     @Override
 190     public int getIndexPageSize() {
 191         return fIndexPageSize;
 192     }
 193
 194     @Override
 195     public TmfTimeRange getTimeRange() {
 196         return fTimeRange;
 197     }
 198
 199     @Override
 200     public ITmfTimestamp getStartTime() {
 201         return fTimeRange.getStartTime();
 202     }
 203
 204     @Override
 205     public ITmfTimestamp getEndTime() {
 206         return fTimeRange.getEndTime();
 207     }
 208
 209     public Vector<TmfCheckpoint> getCheckpoints() {
 210         return fCheckpoints;
 211     }
 212
 213     // ------------------------------------------------------------------------
 214     // Accessors
 215     // ------------------------------------------------------------------------
 216
 217     public static void setCurrentExperiment(final TmfExperiment<?> experiment) {
 218         if (fCurrentExperiment != null && fCurrentExperiment != experiment)
 219             fCurrentExperiment.dispose();
 220         fCurrentExperiment = experiment;
 221     }
 222
 223     public static TmfExperiment<?> getCurrentExperiment() {
 224         return fCurrentExperiment;
 225     }
 226
 227     public ITmfTimestamp getEpoch() {
 228         return fEpoch;
 229     }
 230
 231     public ITmfTrace<T>[] getTraces() {
 232         return fTraces;
 233     }
 234
 235     /**
 236      * Returns the rank of the first event with the requested timestamp. If
 237      * none, returns the index of the next event (if any).
 238      *
 239      * @param timestamp the event timestamp
 240      * @return the corresponding event rank
 241      */
 242     @Override
 243     public long getRank(final ITmfTimestamp timestamp) {
 244         final TmfExperimentContext context = seekEvent(timestamp);
 245         return context.getRank();
 246     }
 247
 248     /**
 249      * Returns the timestamp of the event at the requested index. If none,
 250      * returns null.
 251      *
 252      * @param index the event index (rank)
 253      * @return the corresponding event timestamp
 254      */
 255     public ITmfTimestamp getTimestamp(final int index) {
 256         final TmfExperimentContext context = seekEvent(index);
 257         final ITmfEvent event = getNextEvent(context);
 258         return (event != null) ? event.getTimestamp() : null;
 259     }
 260
 261     // ------------------------------------------------------------------------
 262     // Operators
 263     // ------------------------------------------------------------------------
 264
 265     /**
 266      * Update the global time range
 267      */
 268     protected void updateTimeRange() {
 269         ITmfTimestamp startTime = fTimeRange != TmfTimeRange.NULL_RANGE ? fTimeRange.getStartTime()
 270                 : TmfTimestamp.BIG_CRUNCH;
 271         ITmfTimestamp endTime = fTimeRange != TmfTimeRange.NULL_RANGE ? fTimeRange.getEndTime() : TmfTimestamp.BIG_BANG;
 272
 273         for (final ITmfTrace<T> trace : fTraces) {
 274             final ITmfTimestamp traceStartTime = trace.getStartTime();
 275             if (traceStartTime.compareTo(startTime, true) < 0)
 276                 startTime = traceStartTime;
 277             final ITmfTimestamp traceEndTime = trace.getEndTime();
 278             if (traceEndTime.compareTo(endTime, true) > 0)
 279                 endTime = traceEndTime;
 280         }
 281         fTimeRange = new TmfTimeRange(startTime, endTime);
 282     }
 283
 284     // ------------------------------------------------------------------------
 285     // TmfProvider
 286     // ------------------------------------------------------------------------
 287     @Override
 288     public ITmfContext armRequest(final ITmfDataRequest<T> request) {
 289         //              Tracer.trace("Ctx: Arming request - start");
 290         ITmfTimestamp timestamp = (request instanceof ITmfEventRequest<?>) ? ((ITmfEventRequest<T>) request).getRange().getStartTime() : null;
 291         if (TmfTimestamp.BIG_BANG.equals(timestamp) || request.getIndex() > 0)
 292             timestamp = null; // use request index
 293         TmfExperimentContext context = null;
 294         if (timestamp != null) {
 295             // seek by timestamp
 296             context = seekEvent(timestamp);
 297             ((ITmfEventRequest<T>) request).setStartIndex((int) context.getRank());
 298         } else // Seek by rank
 299             if ((fExperimentContext != null) && fExperimentContext.getRank() == request.getIndex())
 300                 // We are already at the right context -> no need to seek
 301                 context = fExperimentContext;
 302             else
 303                 context = seekEvent(request.getIndex());
 304         // Tracer.trace("Ctx: Arming request - done");
 305         return context;
 306     }
 307
 308     @SuppressWarnings("unchecked")
 309     @Override
 310     public T getNext(final ITmfContext context) {
 311         if (context instanceof TmfExperimentContext)
 312             return (T) getNextEvent(context);
 313         return null;
 314     }
 315
 316     // ------------------------------------------------------------------------
 317     // ITmfTrace trace positioning
 318     // ------------------------------------------------------------------------
 319
 320     // Returns a brand new context based on the location provided
 321     // and initializes the event queues
 322     @Override
 323     public synchronized TmfExperimentContext seekLocation(final ITmfLocation<?> location) {
 324         // Validate the location
 325         if (location != null && !(location instanceof TmfExperimentLocation))
 326             return null; // Throw an exception?
 327
 328         if (fTraces == null)
 329             return null;
 330
 331         // Instantiate the location
 332         final TmfExperimentLocation expLocation = (location == null) ? new TmfExperimentLocation(new TmfLocationArray(
 333                 new ITmfLocation<?>[fTraces.length]), new long[fTraces.length]) : (TmfExperimentLocation) location.clone();
 334
 335         // Create and populate the context's traces contexts
 336         final TmfExperimentContext context = new TmfExperimentContext(fTraces, new ITmfContext[fTraces.length]);
 337         // Tracer.trace("Ctx: SeekLocation - start");
 338
 339         long rank = 0;
 340         for (int i = 0; i < fTraces.length; i++) {
 341             // Get the relevant trace attributes
 342             final ITmfLocation<?> traceLocation = expLocation.getLocation().locations[i];
 343             final long traceRank = expLocation.getRanks()[i];
 344
 345             // Set the corresponding sub-context
 346             context.getContexts()[i] = fTraces[i].seekLocation(traceLocation);
 347             context.getContexts()[i].setRank(traceRank);
 348             rank += traceRank;
 349
 350             // Set the trace location and read the corresponding event
 351             /*
 352              * The (TmfContext) cast should be safe since we created 'context'
 353              * ourselves higher up.
 354              */
 355             expLocation.getLocation().locations[i] = context.getContexts()[i].getLocation().clone();
 356             context.getEvents()[i] = fTraces[i].getNextEvent(context.getContexts()[i]);
 357         }
 358
 359         // Tracer.trace("Ctx: SeekLocation - done");
 360
 361         // Finalize context
 362         context.setLocation(expLocation);
 363         context.setLastTrace(TmfExperimentContext.NO_TRACE);
 364         context.setRank(rank);
 365
 366         fExperimentContext = context;
 367
 368         return context;
 369     }
 370
 371     /*
 372      * (non-Javadoc)
 373      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(org.eclipse.linuxtools .tmf.event.TmfTimestamp)
 374      */
 375     @Override
 376     public synchronized TmfExperimentContext seekEvent(ITmfTimestamp timestamp) {
 377
 378         //              Tracer.trace("Ctx: seekEvent(TS) - start");
 379
 380         if (timestamp == null)
 381             timestamp = TmfTimestamp.BIG_BANG;
 382
 383         // First, find the right checkpoint
 384         int index = Collections.binarySearch(fCheckpoints, new TmfCheckpoint(timestamp, null));
 385
 386         // In the very likely case that the checkpoint was not found, bsearch
 387         // returns its negated would-be location (not an offset...). From that
 388         // index, we can then position the stream and get the event.
 389         if (index < 0)
 390             index = Math.max(0, -(index + 2));
 391
 392         // Position the experiment at the checkpoint
 393         ITmfLocation<?> location;
 394         synchronized (fCheckpoints) {
 395             if (fCheckpoints.size() > 0) {
 396                 if (index >= fCheckpoints.size())
 397                     index = fCheckpoints.size() - 1;
 398                 location = fCheckpoints.elementAt(index).getLocation();
 399             } else
 400                 location = null;
 401         }
 402
 403         final TmfExperimentContext context = seekLocation(location);
 404         context.setRank((long) index * fIndexPageSize);
 405
 406         // And locate the event
 407         ITmfEvent event = parseEvent(context);
 408         while ((event != null) && (event.getTimestamp().compareTo(timestamp, false) < 0)) {
 409             getNextEvent(context);
 410             event = parseEvent(context);
 411         }
 412
 413         if (event == null) {
 414             context.setLocation(null);
 415             context.setRank(ITmfContext.UNKNOWN_RANK);
 416         }
 417
 418         return context;
 419     }
 420
 421     /*
 422      * (non-Javadoc)
 423      *
 424      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(long)
 425      */
 426     @Override
 427     public synchronized TmfExperimentContext seekEvent(final long rank) {
 428
 429         //              Tracer.trace("Ctx: seekEvent(rank) - start");
 430
 431         // Position the stream at the previous checkpoint
 432         int index = (int) rank / fIndexPageSize;
 433         ITmfLocation<?> location;
 434         synchronized (fCheckpoints) {
 435             if (fCheckpoints.size() == 0)
 436                 location = null;
 437             else {
 438                 if (index >= fCheckpoints.size())
 439                     index = fCheckpoints.size() - 1;
 440                 location = fCheckpoints.elementAt(index).getLocation();
 441             }
 442         }
 443
 444         final TmfExperimentContext context = seekLocation(location);
 445         context.setRank((long) index * fIndexPageSize);
 446
 447         // And locate the event
 448         ITmfEvent event = parseEvent(context);
 449         long pos = context.getRank();
 450         while ((event != null) && (pos++ < rank)) {
 451             getNextEvent(context);
 452             event = parseEvent(context);
 453         }
 454
 455         if (event == null) {
 456             context.setLocation(null);
 457             context.setRank(ITmfContext.UNKNOWN_RANK);
 458         }
 459
 460         return context;
 461     }
 462
 463     @Override
 464     public TmfContext seekLocation(final double ratio) {
 465         final TmfContext context = seekEvent((long) (ratio * getNbEvents()));
 466         return context;
 467     }
 468
 469     @Override
 470     public double getLocationRatio(final ITmfLocation<?> location) {
 471         if (location instanceof TmfExperimentLocation)
 472             return (double) seekLocation(location).getRank() / getNbEvents();
 473         return 0;
 474     }
 475
 476     @Override
 477     public ITmfLocation<?> getCurrentLocation() {
 478         if (fExperimentContext != null)
 479             return fExperimentContext.getLocation();
 480         return null;
 481     }
 482
 483     //  private void dumpContext(TmfExperimentContext context, boolean isBefore) {
 484
 485     //          TmfContext context0 = context.getContexts()[0];
 486     //          TmfEvent   event0   = context.getEvents()[0];
 487     //          TmfExperimentLocation location0 = (TmfExperimentLocation) context.getLocation();
 488     //          long       rank0    = context.getRank();
 489     //          int        trace    = context.getLastTrace();
 490     //
 491     //          StringBuffer result = new StringBuffer("Ctx: " + (isBefore ? "B " : "A "));
 492     //
 493     //          result.append("[Ctx: fLoc= " + context0.getLocation().toString() + ", fRnk= " + context0.getRank() + "] ");
 494     //          result.append("[Evt: " + event0.getTimestamp().toString() + "] ");
 495     //          result.append("[Loc: fLoc= " + location0.getLocation()[0].toString() + ", fRnk= " + location0.getRanks()[0] + "] ");
 496     //          result.append("[Rnk: " + rank0 + "], [Trc: " + trace + "]");
 497     //          Tracer.trace(result.toString());
 498     //  }
 499
 500     /**
 501      * Scan the next events from all traces and return the next one in
 502      * chronological order.
 503      *
 504      * @param context the trace context
 505      * @return the next event
 506      */
 507     @Override
 508     public synchronized ITmfEvent getNextEvent(final ITmfContext context) {
 509
 510         // Validate the context
 511         if (!(context instanceof TmfExperimentContext))
 512             return null; // Throw an exception?
 513
 514         if (!context.equals(fExperimentContext))
 515             // Tracer.trace("Ctx: Restoring context");
 516             fExperimentContext = seekLocation(context.getLocation());
 517
 518         final TmfExperimentContext expContext = (TmfExperimentContext) context;
 519
 520         //              dumpContext(expContext, true);
 521
 522         // If an event was consumed previously, get the next one from that trace
 523         final int lastTrace = expContext.getLastTrace();
 524         if (lastTrace != TmfExperimentContext.NO_TRACE) {
 525             final ITmfContext traceContext = expContext.getContexts()[lastTrace];
 526             expContext.getEvents()[lastTrace] = expContext.getTraces()[lastTrace].getNextEvent(traceContext);
 527             expContext.setLastTrace(TmfExperimentContext.NO_TRACE);
 528         }
 529
 530         // Scan the candidate events and identify the "next" trace to read from
 531         final ITmfEvent eventArray[] = expContext.getEvents();
 532         if (eventArray == null)
 533             return null;
 534         int trace = TmfExperimentContext.NO_TRACE;
 535         ITmfTimestamp timestamp = TmfTimestamp.BIG_CRUNCH;
 536         if (eventArray.length == 1) {
 537             if (eventArray[0] != null) {
 538                 timestamp = eventArray[0].getTimestamp();
 539                 trace = 0;
 540             }
 541         } else
 542             for (int i = 0; i < eventArray.length; i++) {
 543                 final ITmfEvent event = eventArray[i];
 544                 if (event != null && event.getTimestamp() != null) {
 545                     final ITmfTimestamp otherTS = event.getTimestamp();
 546                     if (otherTS.compareTo(timestamp, true) < 0) {
 547                         trace = i;
 548                         timestamp = otherTS;
 549                     }
 550                 }
 551             }
 552         // Update the experiment context and set the "next" event
 553         ITmfEvent event = null;
 554         if (trace != TmfExperimentContext.NO_TRACE) {
 555             updateIndex(expContext, timestamp);
 556
 557             final ITmfContext traceContext = expContext.getContexts()[trace];
 558             final TmfExperimentLocation expLocation = (TmfExperimentLocation) expContext.getLocation();
 559             //          expLocation.getLocation()[trace] = traceContext.getLocation().clone();
 560             expLocation.getLocation().locations[trace] = traceContext.getLocation().clone();
 561
 562             //          updateIndex(expContext, timestamp);
 563
 564             expLocation.getRanks()[trace] = traceContext.getRank();
 565             expContext.setLastTrace(trace);
 566             expContext.increaseRank();
 567             event = expContext.getEvents()[trace];
 568             fExperimentContext = expContext;
 569         }
 570
 571         //              if (event != null) {
 572         //              Tracer.trace("Exp: " + (expContext.getRank() - 1) + ": " + event.getTimestamp().toString());
 573         //              dumpContext(expContext, false);
 574         //              Tracer.trace("Ctx: Event returned= " + event.getTimestamp().toString());
 575         //              }
 576
 577         return event;
 578     }
 579
 580     public synchronized void updateIndex(final ITmfContext context, final ITmfTimestamp timestamp) {
 581         // Build the index as we go along
 582         final long rank = context.getRank();
 583         if (context.hasValidRank() && (rank % fIndexPageSize) == 0) {
 584             // Determine the table position
 585             final long position = rank / fIndexPageSize;
 586             // Add new entry at proper location (if empty)
 587             if (fCheckpoints.size() == position) {
 588                 final ITmfLocation<?> location = context.getLocation().clone();
 589                 fCheckpoints.add(new TmfCheckpoint(timestamp.clone(), location));
 590                 //                System.out.println(this + "[" + (fCheckpoints.size() - 1) + "] " + timestamp + ", "
 591                 //                        + location.toString());
 592             }
 593         }
 594     }
 595
 596     /* (non-Javadoc)
 597      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#parseEvent(org.eclipse.linuxtools .tmf.trace.TmfContext)
 598      */
 599     @Override
 600     public ITmfEvent parseEvent(final ITmfContext context) {
 601
 602         // Validate the context
 603         if (!(context instanceof TmfExperimentContext))
 604             return null; // Throw an exception?
 605
 606         if (!context.equals(fExperimentContext))
 607             // Tracer.trace("Ctx: Restoring context");
 608             seekLocation(context.getLocation());
 609
 610         final TmfExperimentContext expContext = (TmfExperimentContext) context;
 611
 612         // If an event was consumed previously, get the next one from that trace
 613         final int lastTrace = expContext.getLastTrace();
 614         if (lastTrace != TmfExperimentContext.NO_TRACE) {
 615             final ITmfContext traceContext = expContext.getContexts()[lastTrace];
 616             expContext.getEvents()[lastTrace] = expContext.getTraces()[lastTrace].getNextEvent(traceContext);
 617             expContext.setLastTrace(TmfExperimentContext.NO_TRACE);
 618             fExperimentContext = (TmfExperimentContext) context;
 619         }
 620
 621         // Scan the candidate events and identify the "next" trace to read from
 622         int trace = TmfExperimentContext.NO_TRACE;
 623         ITmfTimestamp timestamp = TmfTimestamp.BIG_CRUNCH;
 624         for (int i = 0; i < expContext.getTraces().length; i++) {
 625             final ITmfEvent event = expContext.getEvents()[i];
 626             if (event != null && event.getTimestamp() != null) {
 627                 final ITmfTimestamp otherTS = event.getTimestamp();
 628                 if (otherTS.compareTo(timestamp, true) < 0) {
 629                     trace = i;
 630                     timestamp = otherTS;
 631                 }
 632             }
 633         }
 634
 635         ITmfEvent event = null;
 636         if (trace != TmfExperimentContext.NO_TRACE)
 637             event = expContext.getEvents()[trace];
 638
 639         return event;
 640     }
 641
 642     /*
 643      * (non-Javadoc)
 644      *
 645      * @see java.lang.Object#toString()
 646      */
 647     @Override
 648     @SuppressWarnings("nls")
 649     public String toString() {
 650         return "[TmfExperiment (" + getName() + ")]";
 651     }
 652
 653     // ------------------------------------------------------------------------
 654     // Indexing
 655     // ------------------------------------------------------------------------
 656
 657     private synchronized void initializeStreamingMonitor() {
 658         if (fInitialized)
 659             return;
 660         fInitialized = true;
 661
 662         if (getStreamingInterval() == 0) {
 663             final TmfContext context = seekLocation(null);
 664             final ITmfEvent event = getNext(context);
 665             if (event == null)
 666                 return;
 667             final TmfTimeRange timeRange = new TmfTimeRange(event.getTimestamp().clone(), TmfTimestamp.BIG_CRUNCH);
 668             final TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(this, this, timeRange);
 669
 670             // Broadcast in separate thread to prevent deadlock
 671             new Thread() {
 672
 673                 @Override
 674                 public void run() {
 675                     broadcast(signal);
 676                 }
 677             }.start();
 678             return;
 679         }
 680
 681         final Thread thread = new Thread("Streaming Monitor for experiment " + getName()) { //$NON-NLS-1$
 682
 683             ITmfTimestamp safeTimestamp = null;
 684             TmfTimeRange timeRange = null;
 685
 686             @Override
 687             public void run() {
 688                 while (!fExecutor.isShutdown()) {
 689                     if (!isIndexingBusy()) {
 690                         ITmfTimestamp startTimestamp = TmfTimestamp.BIG_CRUNCH;
 691                         ITmfTimestamp endTimestamp = TmfTimestamp.BIG_BANG;
 692                         for (final ITmfTrace<T> trace : fTraces) {
 693                             if (trace.getStartTime().compareTo(startTimestamp) < 0)
 694                                 startTimestamp = trace.getStartTime();
 695                             if (trace.getStreamingInterval() != 0 && trace.getEndTime().compareTo(endTimestamp) > 0)
 696                                 endTimestamp = trace.getEndTime();
 697                         }
 698                         if (safeTimestamp != null && safeTimestamp.compareTo(getTimeRange().getEndTime(), false) > 0)
 699                             timeRange = new TmfTimeRange(startTimestamp, safeTimestamp);
 700                         else
 701                             timeRange = null;
 702                         safeTimestamp = endTimestamp;
 703                         if (timeRange != null) {
 704                             final TmfExperimentRangeUpdatedSignal signal =
 705                                     new TmfExperimentRangeUpdatedSignal(TmfExperiment.this, TmfExperiment.this,
 706                                             timeRange);
 707                             broadcast(signal);
 708                         }
 709                     }
 710                     try {
 711                         Thread.sleep(getStreamingInterval());
 712                     } catch (final InterruptedException e) {
 713                         e.printStackTrace();
 714                     }
 715                 }
 716             }
 717         };
 718         thread.start();
 719     }
 720
 721     /*
 722      * (non-Javadoc)
 723      *
 724      * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getStreamingInterval()
 725      */
 726     @Override
 727     public long getStreamingInterval() {
 728         long interval = 0;
 729         for (final ITmfTrace<T> trace : fTraces)
 730             interval = Math.max(interval, trace.getStreamingInterval());
 731         return interval;
 732     }
 733
 734     /*
 735      * The experiment holds the globally ordered events of its set of traces. It
 736      * is expected to provide access to each individual event by index i.e. it
 737      * must be possible to request the Nth event of the experiment.
 738      *
 739      * The purpose of the index is to keep the information needed to rapidly
 740      * restore the traces contexts at regular intervals (every INDEX_PAGE_SIZE
 741      * event).
 742      */
 743
 744     // The index page size
 745     private static final int DEFAULT_INDEX_PAGE_SIZE = 5000;
 746     protected int fIndexPageSize;
 747     protected boolean fIndexing = false;
 748     protected TmfTimeRange fIndexingPendingRange = TmfTimeRange.NULL_RANGE;
 749
 750     private Integer fEndSynchReference;
 751
 752     //  private static BufferedWriter fEventLog = null;
 753     //  private static BufferedWriter openLogFile(String filename) {
 754     //          BufferedWriter outfile = null;
 755     //          try {
 756     //                  outfile = new BufferedWriter(new FileWriter(filename));
 757     //          } catch (IOException e) {
 758     //                  e.printStackTrace();
 759     //          }
 760     //          return outfile;
 761     //  }
 762
 763     protected boolean isIndexingBusy() {
 764         synchronized (fCheckpoints) {
 765             return fIndexing;
 766         }
 767     }
 768
 769     @Override
 770     public void indexTrace(final boolean waitForCompletion) {
 771         if (waitForCompletion)
 772             initializeStreamingMonitor();
 773     }
 774
 775     @SuppressWarnings("unchecked")
 776     private void indexExperiment(final boolean waitForCompletion, final int index, final TmfTimeRange timeRange) {
 777
 778         synchronized (fCheckpoints) {
 779             if (fIndexing)
 780                 return;
 781             fIndexing = true;
 782         }
 783
 784         final Job job = new Job("Indexing " + getName() + "...") { //$NON-NLS-1$ //$NON-NLS-2$
 785
 786             @Override
 787             protected IStatus run(final IProgressMonitor monitor) {
 788                 while (!monitor.isCanceled())
 789                     try {
 790                         Thread.sleep(100);
 791                     } catch (final InterruptedException e) {
 792                         return Status.OK_STATUS;
 793                     }
 794                 monitor.done();
 795                 return Status.OK_STATUS;
 796             }
 797         };
 798         job.schedule();
 799
 800         //              fEventLog = openLogFile("TraceEvent.log");
 801         //        System.out.println(System.currentTimeMillis() + ": Experiment indexing started");
 802
 803         final ITmfEventRequest<ITmfEvent> request = new TmfEventRequest<ITmfEvent>(ITmfEvent.class, timeRange, index,
 804                 TmfDataRequest.ALL_DATA,
 805                 fIndexPageSize, ITmfDataRequest.ExecutionType.BACKGROUND) { // PATA
 806             // FOREGROUND
 807
 808             //            long indexingStart = System.nanoTime();
 809
 810             ITmfTimestamp startTime = (fTimeRange == TmfTimeRange.NULL_RANGE) ? null : fTimeRange.getStartTime();
 811             ITmfTimestamp lastTime = (fTimeRange == TmfTimeRange.NULL_RANGE) ? null : fTimeRange.getEndTime();
 812             long initialNbEvents = fNbEvents;
 813
 814             @Override
 815             public void handleStarted() {
 816                 super.handleStarted();
 817             }
 818
 819             @Override
 820             public void handleData(final ITmfEvent event) {
 821                 super.handleData(event);
 822                 if (event != null) {
 823                     final ITmfTimestamp ts = event.getTimestamp();
 824                     if (startTime == null)
 825                         startTime = ts.clone();
 826                     lastTime = ts.clone();
 827                     if ((getNbRead() % fIndexPageSize) == 1 && getNbRead() != 1)
 828                         updateExperiment();
 829                 }
 830             }
 831
 832             @Override
 833             public void handleSuccess() {
 834                 //                long indexingEnd = System.nanoTime();
 835
 836                 // if the end time is a real value then it is the streaming safe
 837                 // time stamp
 838                 // set the last time to the safe time stamp to prevent
 839                 // unnecessary indexing requests
 840                 if (getRange().getEndTime() != TmfTimestamp.BIG_CRUNCH)
 841                     lastTime = getRange().getEndTime();
 842                 updateExperiment();
 843                 //                System.out.println(System.currentTimeMillis() + ": Experiment indexing completed");
 844
 845                 //                long average = (indexingEnd - indexingStart) / fNbEvents;
 846                 //                System.out.println(getName() + ": start=" + startTime + ", end=" + lastTime + ", elapsed="
 847                 //                        + (indexingEnd * 1.0 - indexingStart) / 1000000000);
 848                 //                System.out.println(getName() + ": nbEvents=" + fNbEvents + " (" + (average / 1000) + "."
 849                 //                        + (average % 1000) + " us/evt)");
 850                 super.handleSuccess();
 851             }
 852
 853             @Override
 854             public void handleCompleted() {
 855                 job.cancel();
 856                 super.handleCompleted();
 857                 synchronized (fCheckpoints) {
 858                     fIndexing = false;
 859                     if (fIndexingPendingRange != TmfTimeRange.NULL_RANGE) {
 860                         indexExperiment(false, (int) fNbEvents, fIndexingPendingRange);
 861                         fIndexingPendingRange = TmfTimeRange.NULL_RANGE;
 862                     }
 863                 }
 864             }
 865
 866             private void updateExperiment() {
 867                 final int nbRead = getNbRead();
 868                 if (startTime != null)
 869                     fTimeRange = new TmfTimeRange(startTime, lastTime.clone());
 870                 if (nbRead != 0) {
 871                     //                                  updateTimeRange();
 872                     //                                  updateNbEvents();
 873                     fNbEvents = initialNbEvents + nbRead;
 874                     notifyListeners();
 875                 }
 876             }
 877         };
 878
 879         sendRequest((ITmfDataRequest<T>) request);
 880         if (waitForCompletion)
 881             try {
 882                 request.waitForCompletion();
 883             } catch (final InterruptedException e) {
 884                 e.printStackTrace();
 885             }
 886     }
 887
 888     protected void notifyListeners() {
 889         broadcast(new TmfExperimentUpdatedSignal(this, this)); // , null));
 890         // broadcast(new TmfExperimentRangeUpdatedSignal(this, this,
 891         // fTimeRange)); // , null));
 892     }
 893
 894     // ------------------------------------------------------------------------
 895     // Signal handlers
 896     // ------------------------------------------------------------------------
 897
 898     @TmfSignalHandler
 899     public void experimentSelected(final TmfExperimentSelectedSignal<T> signal) {
 900         final TmfExperiment<?> experiment = signal.getExperiment();
 901         if (experiment == this) {
 902             setCurrentExperiment(experiment);
 903             fEndSynchReference = Integer.valueOf(signal.getReference());
 904         }
 905     }
 906
 907     @TmfSignalHandler
 908     public void endSync(final TmfEndSynchSignal signal) {
 909         if (fEndSynchReference != null && fEndSynchReference.intValue() == signal.getReference()) {
 910             fEndSynchReference = null;
 911             initializeStreamingMonitor();
 912         }
 913     }
 914
 915     @TmfSignalHandler
 916     public void experimentUpdated(final TmfExperimentUpdatedSignal signal) {
 917     }
 918
 919     @TmfSignalHandler
 920     public void experimentRangeUpdated(final TmfExperimentRangeUpdatedSignal signal) {
 921         if (signal.getExperiment() == this)
 922             indexExperiment(false, (int) fNbEvents, signal.getRange());
 923     }
 924
 925     @TmfSignalHandler
 926     public void traceUpdated(final TmfTraceUpdatedSignal signal) {
 927         for (final ITmfTrace<T> trace : fTraces)
 928             if (trace == signal.getTrace()) {
 929                 synchronized (fCheckpoints) {
 930                     if (fIndexing) {
 931                         if (fIndexingPendingRange == TmfTimeRange.NULL_RANGE)
 932                             fIndexingPendingRange = signal.getRange();
 933                         else {
 934                             ITmfTimestamp startTime = fIndexingPendingRange.getStartTime();
 935                             ITmfTimestamp endTime = fIndexingPendingRange.getEndTime();
 936                             if (signal.getRange().getStartTime().compareTo(startTime) < 0)
 937                                 startTime = signal.getRange().getStartTime();
 938                             if (signal.getRange().getEndTime().compareTo(endTime) > 0)
 939                                 endTime = signal.getRange().getEndTime();
 940                             fIndexingPendingRange = new TmfTimeRange(startTime, endTime);
 941                         }
 942                         return;
 943                     }
 944                 }
 945                 indexExperiment(false, (int) fNbEvents, signal.getRange());
 946                 return;
 947             }
 948     }
 949
 950     @Override
 951     public String getPath() {
 952         // TODO Auto-generated method stub
 953         return null;
 954     }
 955
 956     /**
 957      * Set the file to be used for bookmarks on this experiment
 958      *
 959      * @param file the bookmarks file
 960      */
 961     public void setBookmarksFile(final IFile file) {
 962         fBookmarksFile = file;
 963     }
 964
 965     /**
 966      * Get the file used for bookmarks on this experiment
 967      *
 968      * @return the bookmarks file or null if none is set
 969      */
 970     public IFile getBookmarksFile() {
 971         return fBookmarksFile;
 972     }
 973
 974     /*
 975      * (non-Javadoc)
 976      *
 977      * @see
 978      * org.eclipse.linuxtools.tmf.core.trace.ITmfTrace#setResource(org.eclipse
 979      * .core.resources.IResource)
 980      */
 981     @Override
 982     public void setResource(final IResource resource) {
 983         fResource = resource;
 984     }
 985
 986     /*
 987      * (non-Javadoc)
 988      *
 989      * @see org.eclipse.linuxtools.tmf.core.trace.ITmfTrace#getResource()
 990      */
 991     @Override
 992     public IResource getResource() {
 993         return fResource;
 994     }
 995 }