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