org.eclipse.linuxtools.lttng.core/src/org/eclipse/linuxtools/lttng/trace/LTTngExperiment.java

   1 /*******************************************************************************
   2  * Copyright (c) 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.lttng.trace;
  14
  15 import org.eclipse.linuxtools.lttng.event.LttngEvent;
  16 import org.eclipse.linuxtools.lttng.event.LttngTimestamp;
  17 import org.eclipse.linuxtools.lttng.jni.JniTrace;
  18 import org.eclipse.linuxtools.tmf.event.TmfEvent;
  19 import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  20 import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
  21 import org.eclipse.linuxtools.tmf.experiment.TmfExperiment;
  22 import org.eclipse.linuxtools.tmf.experiment.TmfExperimentContext;
  23 import org.eclipse.linuxtools.tmf.experiment.TmfExperimentLocation;
  24 import org.eclipse.linuxtools.tmf.request.ITmfDataRequest;
  25 import org.eclipse.linuxtools.tmf.request.ITmfDataRequest.ExecutionType;
  26 import org.eclipse.linuxtools.tmf.request.TmfEventRequest;
  27 import org.eclipse.linuxtools.tmf.signal.TmfExperimentRangeUpdatedSignal;
  28 import org.eclipse.linuxtools.tmf.signal.TmfSignalHandler;
  29 import org.eclipse.linuxtools.tmf.signal.TmfSignalManager;
  30 import org.eclipse.linuxtools.tmf.trace.ITmfTrace;
  31 import org.eclipse.linuxtools.tmf.trace.TmfContext;
  32
  33 /**
  34  * <b><u>LTTngExperiment</u></b>
  35  * <p>
  36  * Temporary class to resolve a basic incompatibility between TMF and LTTng.
  37  * <p>
  38  */
  39 public class LTTngExperiment<T extends TmfEvent> extends TmfExperiment<T> {
  40
  41     private static final int DEFAULT_INDEX_PAGE_SIZE = 50000;
  42
  43     // ------------------------------------------------------------------------
  44     // Constructors
  45     // ------------------------------------------------------------------------
  46
  47     /**
  48      * @param type
  49      * @param id
  50      * @param traces
  51      * @param epoch
  52      * @param indexPageSize
  53      */
  54     public LTTngExperiment(Class<T> type, String id, ITmfTrace<T>[] traces, TmfTimestamp epoch, int indexPageSize) {
  55         this(type, id, traces, TmfTimestamp.Zero, indexPageSize, false);
  56     }
  57
  58     public LTTngExperiment(Class<T> type, String id, ITmfTrace<T>[] traces, TmfTimestamp epoch, int indexPageSize, boolean preIndexExperiment) {
  59         super(type, id, traces, epoch, indexPageSize, preIndexExperiment);
  60     }
  61
  62     /**
  63      * @param type
  64      * @param id
  65      * @param traces
  66      */
  67     public LTTngExperiment(Class<T> type, String id, ITmfTrace<T>[] traces) {
  68         this(type, id, traces, TmfTimestamp.Zero, DEFAULT_INDEX_PAGE_SIZE);
  69     }
  70
  71     /**
  72      * @param type
  73      * @param id
  74      * @param traces
  75      * @param indexPageSize
  76      */
  77     public LTTngExperiment(Class<T> type, String id, ITmfTrace<T>[] traces, int indexPageSize) {
  78         this(type, id, traces, TmfTimestamp.Zero, indexPageSize);
  79     }
  80
  81     @SuppressWarnings("unchecked")
  82     public LTTngExperiment(LTTngExperiment<T> other) {
  83         super(other.getName() + "(clone)", other.fType); //$NON-NLS-1$
  84
  85         fEpoch = other.fEpoch;
  86         fIndexPageSize = other.fIndexPageSize;
  87
  88         fTraces = new ITmfTrace[other.fTraces.length];
  89         for (int trace = 0; trace < other.fTraces.length; trace++) {
  90                 fTraces[trace] = other.fTraces[trace].copy();
  91         }
  92
  93         fNbEvents = other.fNbEvents;
  94         fTimeRange = other.fTimeRange;
  95     }
  96
  97     @Override
  98         public LTTngExperiment<T> copy() {
  99         LTTngExperiment<T> experiment = new LTTngExperiment<T>(this);
 100         TmfSignalManager.deregister(experiment);
 101         return experiment;
 102     }
 103
 104     // ------------------------------------------------------------------------
 105     // ITmfTrace trace positioning
 106     // ------------------------------------------------------------------------
 107
 108     @Override
 109     public synchronized TmfEvent getNextEvent(TmfContext context) {
 110
 111         // Validate the context
 112         if (!(context instanceof TmfExperimentContext)) {
 113             return null; // Throw an exception?
 114         }
 115
 116         if (!context.equals(fExperimentContext)) {
 117 //              Tracer.trace("Ctx: Restoring context");
 118             fExperimentContext = seekLocation(context.getLocation());
 119         }
 120
 121         TmfExperimentContext expContext = (TmfExperimentContext) context;
 122
 123 //              dumpContext(expContext, true);
 124
 125         // If an event was consumed previously, get the next one from that trace
 126         int lastTrace = expContext.getLastTrace();
 127         if (lastTrace != TmfExperimentContext.NO_TRACE) {
 128             TmfContext traceContext = expContext.getContexts()[lastTrace];
 129             expContext.getEvents()[lastTrace] = expContext.getTraces()[lastTrace].getNextEvent(traceContext);
 130             expContext.setLastTrace(TmfExperimentContext.NO_TRACE);
 131         }
 132
 133         // Scan the candidate events and identify the "next" trace to read from
 134         TmfEvent eventArray[] = expContext.getEvents();
 135         if (eventArray == null) {
 136             return null;
 137         }
 138         int trace = TmfExperimentContext.NO_TRACE;
 139         TmfTimestamp timestamp = TmfTimestamp.BigCrunch;
 140         if (eventArray.length == 1) {
 141             if (eventArray[0] != null) {
 142                 timestamp = eventArray[0].getTimestamp();
 143                 trace = 0;
 144             }
 145         } else {
 146             for (int i = 0; i < eventArray.length; i++) {
 147                 TmfEvent event = eventArray[i];
 148                 if (event != null && event.getTimestamp() != null) {
 149                     TmfTimestamp otherTS = event.getTimestamp();
 150                     if (otherTS.compareTo(timestamp, true) < 0) {
 151                         trace = i;
 152                         timestamp = otherTS;
 153                     }
 154                 }
 155             }
 156         }
 157
 158         // Update the experiment context and set the "next" event
 159         TmfEvent event = null;
 160         if (trace != TmfExperimentContext.NO_TRACE) {
 161 //              updateIndex(expContext, timestamp);
 162
 163             TmfContext traceContext = expContext.getContexts()[trace];
 164             TmfExperimentLocation expLocation = (TmfExperimentLocation) expContext.getLocation();
 165             expLocation.getLocation().locations[trace] = traceContext.getLocation();
 166
 167             updateIndex(expContext, timestamp);
 168
 169             expLocation.getRanks()[trace] = traceContext.getRank();
 170             expContext.setLastTrace(trace);
 171             expContext.updateRank(1);
 172             event = expContext.getEvents()[trace];
 173             fExperimentContext = expContext;
 174         }
 175
 176 //              if (event != null) {
 177 //              Tracer.trace("Exp: " + (expContext.getRank() - 1) + ": " + event.getTimestamp().toString());
 178 //              dumpContext(expContext, false);
 179 //              Tracer.trace("Ctx: Event returned= " + event.getTimestamp().toString());
 180 //              }
 181
 182         return event;
 183     }
 184
 185     @SuppressWarnings("unchecked")
 186     @Override
 187     protected void indexExperiment(final boolean waitForCompletion) {
 188         if (waitForCompletion) {
 189             TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
 190                     TmfTimeRange.Eternity);
 191             broadcast(signal);
 192             while (isIndexingBusy()) {
 193                 try {
 194                     Thread.sleep(100);
 195                 } catch (InterruptedException e) {
 196                     e.printStackTrace();
 197                 }
 198             }
 199             ;
 200             return;
 201         }
 202         for (ITmfTrace trace : fTraces) {
 203             if (trace instanceof LTTngTrace) {
 204                 JniTrace jniTrace = ((LTTngTrace) trace).getCurrentJniTrace();
 205                 if (jniTrace != null && !jniTrace.isLiveTraceSupported()) {
 206                     updateTimeRange();
 207                     TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
 208                             getTimeRange());
 209                     broadcast(signal);
 210                     return;
 211                 }
 212             }
 213         }
 214         final Thread thread = new Thread("Streaming Monitor for " + getName()) { //$NON-NLS-1$
 215             LttngTimestamp safeTimestamp = null;
 216             TmfTimeRange timeRange = null;
 217
 218             @Override
 219             public void run() {
 220                 while (!fExecutor.isShutdown()) {
 221                     final TmfEventRequest<LttngEvent> request = new TmfEventRequest<LttngEvent>(LttngEvent.class, TmfTimeRange.Eternity, 0,
 222                             ExecutionType.FOREGROUND) {
 223                         @Override
 224                         public void handleCompleted() {
 225                             super.handleCompleted();
 226                             if (isIndexingBusy()) {
 227                                 timeRange = null;
 228                                 return;
 229                             }
 230                             long startTime = Long.MAX_VALUE;
 231                             long endTime = Long.MIN_VALUE;
 232                             for (ITmfTrace trace : getTraces()) {
 233                                 if (trace instanceof LTTngTrace) {
 234                                     LTTngTrace lttngTrace = (LTTngTrace) trace;
 235                                     JniTrace jniTrace = lttngTrace.getCurrentJniTrace();
 236                                     jniTrace.updateTrace();
 237                                     startTime = Math.min(startTime, lttngTrace.getStartTime().getValue());
 238                                     endTime = Math.max(endTime, jniTrace.getEndTime().getTime());
 239                                 }
 240                             }
 241                             LttngTimestamp startTimestamp = new LttngTimestamp(startTime);
 242                             LttngTimestamp endTimestamp = new LttngTimestamp(endTime);
 243                             if (safeTimestamp != null && safeTimestamp.compareTo(getTimeRange().getEndTime(), false) > 0) {
 244                                 timeRange = new TmfTimeRange(startTimestamp, safeTimestamp);
 245                             } else {
 246                                 timeRange = null;
 247                             }
 248                             safeTimestamp = endTimestamp;
 249                         }
 250                     };
 251                     try {
 252                         sendRequest((ITmfDataRequest<T>) request);
 253                         request.waitForCompletion();
 254                         if (timeRange != null && !timeRange.equals(TmfTimeRange.Null)) {
 255                             TmfExperimentRangeUpdatedSignal signal = new TmfExperimentRangeUpdatedSignal(LTTngExperiment.this, LTTngExperiment.this,
 256                                     timeRange);
 257                             broadcast(signal);
 258                         }
 259                         Thread.sleep(2000);
 260                     } catch (InterruptedException e) {
 261                         e.printStackTrace();
 262                     }
 263                 }
 264             }
 265         };
 266         thread.start();
 267     }
 268
 269     @TmfSignalHandler
 270     public void experimentRangeUpdated(TmfExperimentRangeUpdatedSignal signal) {
 271         indexExperiment(false, (int) fNbEvents, signal.getRange());
 272     }
 273
 274     /*
 275      * (non-Javadoc)
 276      *
 277      * @see java.lang.Object#toString()
 278      */
 279     @Override
 280     @SuppressWarnings("nls")
 281     public String toString() {
 282         return "[LTTngExperiment (" + getName() + ")]";
 283     }
 284
 285 }