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