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