org.eclipse.linuxtools.tmf/src/org/eclipse/linuxtools/tmf/trace/TmfExperiment.java

   1 /*******************************************************************************
   2  * Copyright (c) 2009 Ericsson
   3  *
   4  * All rights reserved. This program and the accompanying materials are
   5  * made available under the terms of the Eclipse Public License v1.0 which
   6  * accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *   Francois Chouinard - Initial API and implementation
  11  *******************************************************************************/
  12
  13 package org.eclipse.linuxtools.tmf.trace;
  14
  15 import java.util.Vector;
  16
  17 import org.eclipse.core.runtime.IProgressMonitor;
  18 import org.eclipse.core.runtime.IStatus;
  19 import org.eclipse.core.runtime.Status;
  20 import org.eclipse.core.runtime.jobs.Job;
  21 import org.eclipse.linuxtools.tmf.component.TmfComponent;
  22 import org.eclipse.linuxtools.tmf.event.TmfEvent;
  23 import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  24 import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
  25 import org.eclipse.linuxtools.tmf.request.ITmfRequestHandler;
  26 import org.eclipse.linuxtools.tmf.request.TmfDataRequest;
  27 import org.eclipse.linuxtools.tmf.signal.TmfSignalHandler;
  28
  29 /**
  30  * <b><u>TmfExperiment</u></b>
  31  * <p>
  32  * TmfExperiment presents a time-ordered, unified view of a set of TmfTraces
  33  * that are part of a tracing experiment.
  34  * <p>
  35  */
  36 public class TmfExperiment extends TmfComponent implements ITmfRequestHandler<TmfEvent> {
  37
  38 // TODO: Complete multi-trace experiment
  39 // TODO: Add support for dynamic addition/removal of traces
  40 // TODO: Add support for live streaming (notifications, incremental indexing, ...)
  41
  42         // ========================================================================
  43     // Attributes
  44     // ========================================================================
  45
  46         // The currently selected experiment
  47     private static TmfExperiment fCurrentExperiment;
  48
  49         // The experiment ID
  50     private String fExperimentId;
  51
  52     // The set of trace sthat constitute the experiment
  53     private Vector<ITmfTrace> fTraces;
  54
  55     // The total number of events
  56     private int fNbEvents;
  57
  58     // The experiment time range
  59     private TmfTimeRange fTimeRange;
  60
  61     // The experiment reference timestamp (default: BigBang)
  62     private TmfTimestamp fEpoch;
  63
  64 //    // Indicates if the stream should be indexed synchronously (default: false)
  65 //    private final boolean fWaitForIndexCompletion;
  66
  67     // ========================================================================
  68     // Constructors/Destructor
  69     // ========================================================================
  70
  71     /**
  72      * @param id
  73      * @param traces
  74      */
  75     public TmfExperiment(String id, ITmfTrace[] traces) {
  76         this(id, traces, TmfTimestamp.BigBang, DEFAULT_INDEX_PAGE_SIZE, false);
  77     }
  78
  79     /**
  80      * @param id
  81      * @param traces
  82      * @param waitForIndexCompletion
  83      */
  84     public TmfExperiment(String id, ITmfTrace[] traces, boolean waitForIndexCompletion) {
  85         this(id, traces, TmfTimestamp.BigBang, DEFAULT_INDEX_PAGE_SIZE, waitForIndexCompletion);
  86     }
  87
  88     /**
  89      * @param id
  90      * @param traces
  91      * @param epoch
  92      * @param waitForIndexCompletion
  93      */
  94     public TmfExperiment(String id, ITmfTrace[] traces, TmfTimestamp epoch, int indexPageSize, boolean waitForIndexCompletion) {
  95         super();
  96
  97         fExperimentId = id;
  98         fTraces = new Vector<ITmfTrace>();
  99         for (ITmfTrace trace : traces) {
 100                 fTraces.add(trace);
 101         }
 102         fEpoch = epoch;
 103         fIndexPageSize = indexPageSize;
 104 //        fWaitForIndexCompletion = waitForIndexCompletion;
 105
 106                 updateNbEvents();
 107                 updateTimeRange();
 108 //              indexExperiment();
 109     }
 110
 111     /**
 112      *
 113      */
 114     @Override
 115         public void dispose() {
 116         super.dispose();
 117         fTraces.clear();
 118         fCurrentExperiment= null;
 119     }
 120
 121     // ========================================================================
 122     // Accessors
 123     // ========================================================================
 124
 125     public static TmfExperiment getCurrentExperiment() {
 126         return fCurrentExperiment;
 127     }
 128
 129     public String getExperimentId() {
 130         return fExperimentId;
 131     }
 132
 133     public ITmfTrace[] getTraces() {
 134         ITmfTrace[] result = new ITmfTrace[fTraces.size()];
 135         return fTraces.toArray(result);
 136     }
 137
 138     public TmfTimestamp getEpoch() {
 139         return fEpoch;
 140     }
 141
 142     public TmfTimeRange getTimeRange() {
 143         return fTimeRange;
 144     }
 145
 146     public int getNbEvents() {
 147         return fNbEvents;
 148     }
 149
 150     /**
 151      * Returns the index of the first event with the requested timestamp.
 152      * If none, returns the index of the next event (if any).
 153      *
 154      * @param ts
 155      * @return
 156      */
 157     public long getIndex(TmfTimestamp ts) {
 158         // TODO: Go over all the traces
 159         ITmfTrace trace = fTraces.firstElement();
 160         TmfTraceContext context = trace.seekEvent(ts);
 161         return context.getIndex();
 162     }
 163
 164     /**
 165      * Returns the timestamp of the event at the requested index.
 166      * If none, returns null.
 167      *
 168      * @param index
 169      * @return
 170      */
 171     public TmfTimestamp getTimestamp(int index) {
 172         // TODO: Go over all the traces
 173         ITmfTrace trace = fTraces.firstElement();
 174         TmfTraceContext context = trace.seekEvent(index);
 175         return context.getTimestamp();
 176     }
 177
 178     // ========================================================================
 179     // Operators
 180     // ========================================================================
 181
 182     /**
 183      * Add a trace to the experiment trace set
 184      *
 185      * @param trace
 186      */
 187     public void addTrace(ITmfTrace trace) {
 188                 fTraces.add(trace);
 189                 synchronized(this) {
 190                         updateNbEvents();
 191                         updateTimeRange();
 192                 }
 193     }
 194
 195     /**
 196      * Update the total number of events
 197      */
 198     private void updateNbEvents() {
 199         int nbEvents = 0;
 200         for (ITmfTrace trace : fTraces) {
 201                 nbEvents += trace.getNbEvents();
 202         }
 203         fNbEvents = nbEvents;
 204     }
 205
 206     /**
 207      * Update the global time range
 208      */
 209     private void updateTimeRange() {
 210                 TmfTimestamp startTime = fTimeRange != null ? fTimeRange.getStartTime() : TmfTimestamp.BigCrunch;
 211                 TmfTimestamp endTime   = fTimeRange != null ? fTimeRange.getEndTime()   : TmfTimestamp.BigBang;
 212
 213                 for (ITmfTrace trace : fTraces) {
 214                 TmfTimestamp traceStartTime = trace.getTimeRange().getStartTime();
 215                 if (traceStartTime.compareTo(startTime, true) < 0)
 216                         startTime = traceStartTime;
 217
 218                 TmfTimestamp traceEndTime = trace.getTimeRange().getEndTime();
 219                 if (traceEndTime.compareTo(endTime, true) > 0)
 220                         endTime = traceEndTime;
 221         }
 222                 fTimeRange = new TmfTimeRange(startTime, endTime);
 223     }
 224
 225     // ========================================================================
 226     // ITmfRequestHandler
 227     // ========================================================================
 228
 229         /* (non-Javadoc)
 230          * @see org.eclipse.linuxtools.tmf.request.ITmfRequestHandler#processRequest(org.eclipse.linuxtools.tmf.request.TmfDataRequest, boolean)
 231          */
 232         public void processRequest(TmfDataRequest<TmfEvent> request, boolean waitForCompletion) {
 233
 234                 // Process the request
 235                 processDataRequest(request);
 236
 237                 // Wait for completion if needed
 238         if (waitForCompletion) {
 239                         request.waitForCompletion();
 240                 }
 241         }
 242
 243         /**
 244          * Process a data request
 245          *
 246          * @param request
 247          */
 248         private void processDataRequest(final TmfDataRequest<TmfEvent> request) {
 249
 250                 // General request parameters
 251                 final TmfTimestamp endTime;
 252                 final long index;
 253
 254                 // Initialize request params depending on request type
 255                 if (request.getRange() != null) {
 256                         index = getIndex(request.getRange().getStartTime());
 257                         endTime = request.getRange().getEndTime();
 258         } else {
 259                 index = request.getIndex();
 260                 endTime = TmfTimestamp.BigCrunch;
 261         }
 262
 263                 // Process the request
 264                 Thread thread = new Thread() {
 265
 266                         @Override
 267                         public void run() {
 268
 269                                 // Key variables
 270                                 ITmfTrace[] traces = new ITmfTrace[0];  // The set of traces
 271                                 TmfTraceContext[] contexts;                             // The set of trace contexts
 272
 273                                 // Extract the general request information
 274                                 int blockSize = request.getBlockize();
 275                                 int nbRequestedEvents = request.getNbRequestedEvents();
 276                                 if (nbRequestedEvents == -1) {
 277                                         nbRequestedEvents = Integer.MAX_VALUE;
 278                                 }
 279
 280                                 // Create the result buffer
 281                                 Vector<TmfEvent> events = new Vector<TmfEvent>();
 282                                 int nbEvents = 0;
 283
 284                                 // Initialize the traces array and position the traces
 285                                 // at the first requested event
 286                                 traces = fTraces.toArray(traces);
 287                                 contexts = new TmfTraceContext[traces.length];
 288                                 positionTraces(index, traces, contexts);
 289
 290                                 // Get the ordered events
 291                                 TmfEvent event = getNextEvent(traces, contexts);
 292                                 while (!request.isCancelled() && nbEvents < nbRequestedEvents && event != null
 293                                                 && event.getTimestamp().compareTo(endTime, false) < 0)
 294                                 {
 295                                         events.add(event);
 296                                         if (++nbEvents % blockSize == 0) {
 297                                                 pushData(request, events);
 298                                         }
 299                                         // Avoid an unnecessary read passed the last event requested
 300                                         if (nbEvents < nbRequestedEvents)
 301                                                 event = getNextEvent(traces, contexts);
 302                                 }
 303
 304                                 if (!request.isCancelled() && !request.isFailed()) {
 305                                         pushData(request, events);
 306                                         request.done();
 307                                 }
 308                         }
 309                 };
 310                 thread.start();
 311         }
 312
 313         /**
 314          * Given an experiment event index, position the set of traces so a call
 315          * to getNextEvent() will retrieve the corresponding event.
 316          *
 317          * @param index
 318          * @param traces
 319          * @param contexts
 320          * @param nextEvents
 321          */
 322         private synchronized void positionTraces(long index, ITmfTrace[] traces, TmfTraceContext[] contexts) {
 323
 324                 // Compute the index page and corresponding index
 325                 int page = (int) index / fIndexPageSize;
 326                 int current = page * fIndexPageSize;
 327
 328                 // Retrieve the checkpoint and set the contexts (make copies)
 329                 TmfTraceContext[] saveContexts = new TmfTraceContext[contexts.length];
 330                 if (page < fExperimentIndex.size()) {
 331                         saveContexts = fExperimentIndex.elementAt(page);
 332                         for (int i = 0; i < contexts.length; i++) {
 333                                 contexts[i] = new TmfTraceContext(saveContexts[i]);
 334                         }
 335                 } else {
 336                         // If the page entry doesn't exist (e.g. indexing not completed),
 337                         // set contexts at the the last entry (if it exists)
 338                         page = fExperimentIndex.size() - 1;
 339                         if (page >= 0) {
 340                                 saveContexts = fExperimentIndex.elementAt(page);
 341                                 for (int i = 0; i < contexts.length; i++) {
 342                                         contexts[i] = new TmfTraceContext(saveContexts[i]);
 343                                 }
 344                                 current = page * fIndexPageSize;
 345                         }
 346                         // Index is empty... position traces at their beginning
 347                         else {
 348                                 for (int i = 0; i < contexts.length; i++) {
 349                                         contexts[i] = new TmfTraceContext(traces[i].seekLocation(null));
 350                                 }
 351                                 current = 0;
 352                         }
 353                 }
 354
 355                 // Position the traces at the requested index
 356                 while (current++ < index) {
 357                         getNextEvent(traces, contexts);
 358                 }
 359         }
 360
 361         /**
 362          * Scan the next events from all traces and return the next one
 363          * in chronological order.
 364          *
 365          * @param traces
 366          * @param contexts
 367          * @param nextEvents
 368          * @return
 369          */
 370         private TmfEvent getNextEvent(ITmfTrace[] traces, TmfTraceContext[] contexts) {
 371                 // TODO: Consider the time adjustment
 372                 int trace = 0;
 373                 TmfTimestamp timestamp = contexts[trace].getTimestamp();
 374                 if (timestamp == null) {
 375                         timestamp = TmfTimestamp.BigCrunch;
 376                 }
 377                 for (int i = 1; i < traces.length; i++) {
 378                         if (contexts[i].getTimestamp() != null) {
 379                                 TmfTimestamp otherTS = contexts[i].getTimestamp();
 380                                 if (otherTS.compareTo(timestamp, true) < 0) {
 381                                         trace = i;
 382                                         timestamp = otherTS;
 383                                 }
 384                         }
 385                 }
 386                 TmfEvent event = traces[trace].getNextEvent(contexts[trace]);
 387                 return event;
 388         }
 389
 390         /**
 391          * Format the result data and notify the requester.
 392          * Note: after handling, the data is *removed*.
 393          *
 394          * @param request
 395          * @param events
 396          */
 397         private void pushData(TmfDataRequest<TmfEvent> request, Vector<TmfEvent> events) {
 398                 TmfEvent[] result = new TmfEvent[events.size()];
 399                 events.toArray(result);
 400                 request.setData(result);
 401                 request.handleData();
 402                 events.removeAllElements();
 403         }
 404
 405         /* (non-Javadoc)
 406          * @see java.lang.Object#toString()
 407          */
 408         @Override
 409         public String toString() {
 410                 return "[TmfExperiment (" + fExperimentId + ")]";
 411         }
 412
 413     // ========================================================================
 414     // Indexing
 415     // ========================================================================
 416
 417         /*
 418          * The experiment holds the globally ordered events of its set of traces.
 419          * It is expected to provide access to each individual event by index i.e.
 420          * it must be possible to request the Nth event of the experiment.
 421          *
 422          * The purpose of the index is to keep the information needed to rapidly
 423          * restore the traces contexts at regular intervals (every INDEX_PAGE_SIZE
 424          * event).
 425          */
 426
 427         // The index page size
 428         private static final int DEFAULT_INDEX_PAGE_SIZE = 1000;
 429         private final int fIndexPageSize;
 430
 431         // The experiment index
 432         private Vector<TmfTraceContext[]> fExperimentIndex = new Vector<TmfTraceContext[]>();
 433
 434         // Indicates that an indexing job is already running
 435         private Boolean fIndexing = false;
 436
 437         // The indexing job
 438         private IndexingJob job;
 439
 440         /**
 441          * indexExperiment
 442          *
 443          * Creates the experiment index.
 444          */
 445         private void indexExperiment() {
 446
 447                 synchronized(fIndexing) {
 448                         if (fIndexing) {
 449                                 // An indexing job is already running but a new request came
 450                                 // in (probably due to a change in the trace set). The index
 451                                 // being currently built is therefore already invalid.
 452                                 // TODO: Cancel and restart the job
 453                                 // TODO: Add support for dynamically adding/removing traces
 454                                 return;
 455                         }
 456                         fIndexing = true;
 457                 }
 458
 459                 job = new IndexingJob(fExperimentId);
 460                 job.schedule();
 461                 try {
 462                         job.join();
 463                 } catch (InterruptedException e) {
 464                         // TODO Auto-generated catch block
 465                         e.printStackTrace();
 466                 }
 467 //              if (fWaitForIndexCompletion) {
 468 //              ProgressMonitorDialog dialog = new ProgressMonitorDialog(null);
 469 //              try {
 470 //                              // TODO: Handle cancel!
 471 //                  dialog.run(true, true, new IRunnableWithProgress() {
 472 //                      public void run(IProgressMonitor monitor) throws InvocationTargetException, InterruptedException {
 473 //                          monitor.beginTask("Indexing " + fExperimentId, IProgressMonitor.UNKNOWN);
 474 //                                      job.join();
 475 //                          monitor.done();
 476 //                      }
 477 //                  });
 478 //              } catch (InvocationTargetException e) {
 479 //                  e.printStackTrace();
 480 //              } catch (InterruptedException e) {
 481 //                  e.printStackTrace();
 482 //              }
 483 //              }
 484         }
 485
 486         private class IndexingJob extends Job {
 487
 488                 public IndexingJob(String name) {
 489                         super(name);
 490                 }
 491
 492                 /* (non-Javadoc)
 493                  * @see org.eclipse.core.runtime.jobs.Job#run(org.eclipse.core.runtime.IProgressMonitor)
 494                  */
 495                 @Override
 496                 protected IStatus run(IProgressMonitor monitor) {
 497
 498                         Vector<TmfTraceContext[]> indices = new Vector<TmfTraceContext[]>();
 499
 500                         // Minimal check
 501                         if (fTraces.size() == 0) {
 502                     fIndexing = false;
 503                                 return Status.OK_STATUS;
 504                         }
 505
 506                         monitor.beginTask("Indexing " + fExperimentId, IProgressMonitor.UNKNOWN);
 507
 508                 ITmfTrace[] traces = new ITmfTrace[0];
 509                 TmfTraceContext[] contexts;
 510                 int nbTraces = fTraces.size();
 511
 512                 // Initialize the traces and contexts arrays
 513                 traces = fTraces.toArray(traces);
 514                 contexts = new TmfTraceContext[nbTraces];
 515                 TmfTraceContext[] savedContexts = new TmfTraceContext[nbTraces];
 516                         int nullEvents = 0;
 517                 for (int i = 0; i < nbTraces; i++) {
 518                         // Context of the first event of the trace
 519                         contexts[i] = traces[i].seekLocation(null);
 520                         savedContexts[i] = new TmfTraceContext(contexts[i].getLocation(), contexts[i].getTimestamp(), 0);
 521                         if (contexts[i].getTimestamp() == null)
 522                                 nullEvents++;
 523                 }
 524                 // Check if there is anything to index
 525                 if (nullEvents >= nbTraces) {
 526                     fIndexing = false;
 527                                 return Status.OK_STATUS;
 528                 }
 529                 // FIXME: LTTng hack - start
 530 //                      indices.add(savedContexts);     // TMF
 531                 // FIXME: LTTng hack - end
 532
 533                 // Get the ordered events and populate the indices
 534                 // FIXME: LTTng hack - start
 535 //              int nbEvents = 0; // TMF
 536                 int nbEvents = -1; // LTTng
 537                 // FIXME: LTTng hack - end
 538                 while ((getNextEvent(traces, contexts)) != null)
 539                 {
 540                         if (++nbEvents % fIndexPageSize == 0) {
 541                                 // Special case: if the total number of events is a multiple of the
 542                                 // DEFAULT_PAGE_SIZE then all the pending events are null. In that
 543                                 // case, we don't store an additional entry in the index array.
 544                                 nullEvents = 0;
 545                                 savedContexts = new TmfTraceContext[nbTraces];
 546                         for (int i = 0; i < nbTraces; i++) {
 547                                 savedContexts[i] = new TmfTraceContext(contexts[i]);
 548                                 if (contexts[i].getTimestamp() ==  null)
 549                                         nullEvents++;
 550                         }
 551                         if (nullEvents < nbTraces) {
 552                                 indices.add(savedContexts);
 553                         }
 554                         }
 555
 556                         monitor.worked(1);
 557                 if (monitor.isCanceled()) {
 558                     monitor.done();
 559                         return Status.CANCEL_STATUS;
 560                 }
 561                 }
 562
 563             monitor.done();
 564             fExperimentIndex = indices;
 565
 566 //            dumpIndex();
 567
 568             fIndexing = false;
 569             return Status.OK_STATUS;
 570                 }
 571     }
 572
 573 //      /**
 574 //       * Dump the experiment index
 575 //       */
 576 //      private void dumpIndex() {
 577 //              System.out.println("-----");
 578 //              System.out.println("Index of " + fExperimentId);
 579 //              for (int i = 0; i < fExperimentIndex.size(); i++) {
 580 //              System.out.println("Entry:" + i);
 581 //              TmfTraceContext[] contexts = fExperimentIndex.get(i);
 582 //              int nbEvents = 0;
 583 //              for (int j = 0; j < contexts.length; j++) {
 584 //                      ITmfTrace trace = fTraces.get(j);
 585 //              TmfTraceContext context = trace.seekLocation(contexts[j].getLocation());
 586 //              TmfEvent event = fTraces.get(j).getNextEvent(new TmfTraceContext(context));
 587 //              nbEvents += contexts[j].getIndex();
 588 //              System.out.println("  ["  + trace.getName() + "]"
 589 //                              +  " index: "     + contexts[j].getIndex()
 590 //                              + ", timestamp: " + contexts[j].getTimestamp()
 591 //                              + ", event: "     + event.getTimestamp());
 592 //              assert (contexts[j].getTimestamp().compareTo(event.getTimestamp(), false) == 0);
 593 //              }
 594 //              assert ((i+1) * fIndexPageSize == nbEvents);
 595 //
 596 //        }
 597 //      }
 598
 599     // ========================================================================
 600     // Signal handlers
 601     // ========================================================================
 602
 603     @TmfSignalHandler
 604     public void experimentSelected(TmfExperimentSelectedSignal signal) {
 605                 fCurrentExperiment = this;
 606         indexExperiment();
 607     }
 608
 609     @TmfSignalHandler
 610     public void experimentUpdated(TmfExperimentUpdatedSignal signal) {
 611 //      indexExperiment();
 612     }
 613
 614     @TmfSignalHandler
 615     public void traceUpdated(TmfTraceUpdatedSignal signal) {
 616         // TODO: Incremental index update
 617         synchronized(this) {
 618                 updateNbEvents();
 619                 updateTimeRange();
 620         }
 621                 broadcastSignal(new TmfExperimentUpdatedSignal(this, this, signal.getTrace()));
 622     }
 623 }