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