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