org.eclipse.linuxtools.lttng.core/src/org/eclipse/linuxtools/lttng/core/control/LttngSyntheticEventProvider.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  *   Alvaro Sanchez-Leon (alvsan09@gmail.com) - Initial API and implementation
  11  *   Marc Dumais (marc.dumais@ericsson.com) - Fix for 316455 (first part)
  12  *******************************************************************************/
  13
  14 package org.eclipse.linuxtools.lttng.core.control;
  15
  16 import java.util.HashMap;
  17 import java.util.Iterator;
  18 import java.util.List;
  19 import java.util.Map;
  20 import java.util.Vector;
  21
  22 import org.eclipse.linuxtools.lttng.core.TraceDebug;
  23 import org.eclipse.linuxtools.lttng.core.event.LttngEvent;
  24 import org.eclipse.linuxtools.lttng.core.event.LttngEventType;
  25 import org.eclipse.linuxtools.lttng.core.event.LttngSyntheticEvent;
  26 import org.eclipse.linuxtools.lttng.core.event.LttngTimestamp;
  27 import org.eclipse.linuxtools.lttng.core.event.LttngSyntheticEvent.SequenceInd;
  28 import org.eclipse.linuxtools.lttng.core.model.LTTngTreeNode;
  29 import org.eclipse.linuxtools.lttng.core.request.LttngBaseEventRequest;
  30 import org.eclipse.linuxtools.lttng.core.state.evProcessor.ITransEventProcessor;
  31 import org.eclipse.linuxtools.lttng.core.state.evProcessor.state.StateEventToHandlerFactory;
  32 import org.eclipse.linuxtools.lttng.core.state.model.LttngTraceState;
  33 import org.eclipse.linuxtools.lttng.core.state.trace.IStateTraceManager;
  34 import org.eclipse.linuxtools.tmf.core.component.TmfEventProvider;
  35 import org.eclipse.linuxtools.tmf.core.event.TmfEventSource;
  36 import org.eclipse.linuxtools.tmf.core.event.TmfTimeRange;
  37 import org.eclipse.linuxtools.tmf.core.event.TmfTimestamp;
  38 import org.eclipse.linuxtools.tmf.core.experiment.TmfExperiment;
  39 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest;
  40 import org.eclipse.linuxtools.tmf.core.request.ITmfEventRequest;
  41 import org.eclipse.linuxtools.tmf.core.request.TmfDataRequest;
  42 import org.eclipse.linuxtools.tmf.core.signal.TmfEndSynchSignal;
  43 import org.eclipse.linuxtools.tmf.core.signal.TmfStartSynchSignal;
  44 import org.eclipse.linuxtools.tmf.core.trace.ITmfContext;
  45 import org.eclipse.linuxtools.tmf.core.trace.ITmfTrace;
  46 import org.eclipse.linuxtools.tmf.core.trace.TmfCheckpoint;
  47 import org.eclipse.linuxtools.tmf.core.trace.TmfContext;
  48
  49 /**
  50  * @author alvaro
  51  *
  52  */
  53 public class LttngSyntheticEventProvider extends TmfEventProvider<LttngSyntheticEvent> {
  54
  55         // ========================================================================
  56         // Data
  57         // ========================================================================
  58         public static final int BLOCK_SIZE = 50000;
  59         public static final int NB_EVENTS  = 1;
  60         public static final int QUEUE_SIZE = 1; // lttng specific, one event at a time
  61
  62         private ITmfDataRequest<LttngSyntheticEvent> fmainRequest = null;
  63     private LttngBaseEventRequest fSubRequest = null;
  64
  65         private final List<IStateTraceManager> fEventProviderRequests = new Vector<IStateTraceManager>();
  66
  67         private final LttngSyntheticEvent fStatusEvent;
  68         volatile boolean startIndSent = false;
  69         private LTTngTreeNode fExperiment = null;
  70         private ITransEventProcessor fstateUpdateProcessor = StateEventToHandlerFactory.getInstance();
  71         private boolean waitForRequest = false;
  72         long dispatchTime = 0L;
  73         long dispatchIndex = 0L;
  74         long eventIndex;
  75         private final Map<ITmfTrace, LttngTraceState> traceToTraceStateModel = new HashMap<ITmfTrace, LttngTraceState>();
  76
  77         private boolean fIsExperimentNotified = false;
  78
  79         // ========================================================================
  80         // Constructor
  81         // ========================================================================
  82         /**
  83          * Accessibility to package - use factory instead of this constructor
  84          *
  85          * @param type
  86          */
  87         LttngSyntheticEventProvider(Class<LttngSyntheticEvent> type) {
  88                 super("LttngSyntheticEventProvider", type, QUEUE_SIZE); //$NON-NLS-1$
  89
  90                 // prepare empty instance status indicators and allow them to travel via
  91                 // the framework
  92                 TmfEventSource source = new TmfEventSource(this);
  93                 LttngEventType dtype = new LttngEventType();
  94                 LttngTimestamp statusTimeStamp = new LttngTimestamp(
  95                                 TmfTimestamp.Zero);
  96
  97                 fStatusEvent = new LttngSyntheticEvent(null, statusTimeStamp, source,
  98                                 dtype, null, null, null);
  99                 fStatusEvent.setSequenceInd(SequenceInd.STARTREQ);
 100         }
 101
 102         // ========================================================================
 103         // Methods
 104         // ========================================================================
 105
 106         @SuppressWarnings("unchecked")
 107         @Override
 108         public ITmfContext armRequest(final ITmfDataRequest<LttngSyntheticEvent> request) {
 109                 // validate
 110                 // make sure we have the right type of request
 111                 if (!(request instanceof ITmfEventRequest<?>)) {
 112                         request.cancel();
 113                         TraceDebug.debug("Request is not an instance of ITmfEventRequest"); //$NON-NLS-1$
 114                         return null;
 115                 }
 116
 117                 if (fExperiment == null) {
 118                         TraceDebug.debug("Experiment is null"); //$NON-NLS-1$
 119                         request.cancel();
 120                         return null;
 121                 }
 122
 123                 // get ready to start processing
 124                 reset(fExperiment);
 125
 126                 // At least one base provider shall be available
 127                 if (fEventProviderRequests.size() < 1) {
 128                         request.cancel();
 129                         TraceDebug.debug("No Base event providers available"); //$NON-NLS-1$
 130                         return null;
 131                 }
 132
 133                 fmainRequest = request;
 134
 135                 // define event data handling
 136                 ITmfEventRequest<LttngSyntheticEvent> eventRequest = (ITmfEventRequest<LttngSyntheticEvent>) fmainRequest;
 137                 TmfTimeRange reqWindow = eventRequest.getRange();
 138
 139                 TraceDebug.debug("Main Synthethic event request started on thread:  " + Thread.currentThread().getName()); //$NON-NLS-1$
 140
 141                 TmfExperiment<LttngEvent> experiment = (TmfExperiment<LttngEvent>) fExperiment.getValue();
 142                 experiment.startSynch(new TmfStartSynchSignal(0));
 143
 144                 TmfTimeRange adjustedRange = reqWindow;
 145                 long adjustedIndex = eventRequest.getIndex();
 146                 int nbRequested = eventRequest.getNbRequested();
 147
 148                 // Figure-out if we need to increase the range of the request:  if some
 149                 // checkpoints are before the beginning of the range, increase the
 150                 // range to catch them.   We will then exercise the state system of
 151                 // those traces until the requested beginning time range, discarding
 152                 // the unrequested data.
 153                 IStateTraceManager traceManager;
 154                 Iterator<IStateTraceManager> iter = fEventProviderRequests.iterator();
 155                 // For each traceManager in the current experiment...
 156                 while(iter.hasNext()) {
 157                         traceManager = iter.next();
 158                         // restore trace state system to nearest check point
 159                         TmfCheckpoint checkPoint = null;
 160                         if (eventRequest.getIndex() > 0) {
 161                                 checkPoint = traceManager.restoreCheckPointByIndex(eventRequest.getIndex());
 162                         } else {
 163                                 checkPoint = traceManager.restoreCheckPointByTimestamp(reqWindow.getStartTime());
 164                         }
 165
 166                         // validate that the checkpoint restored is within requested bounds
 167                         // (not outside the current trace's range or after the end of requested range)
 168                         TmfTimeRange traceRange = traceManager.getTrace().getTimeRange();
 169                         if ((checkPoint == null) ||
 170                                         checkPoint.getTimestamp().getValue() < traceRange.getStartTime().getValue() ||
 171                                         checkPoint.getTimestamp().getValue() > traceRange.getEndTime().getValue() ||
 172                                         checkPoint.getTimestamp().getValue() >= reqWindow.getEndTime().getValue()
 173                                         ) {
 174                                 // checkpoint is out of trace bounds; no need to adjust request for this
 175                                 // trace
 176                         }
 177                         else {
 178                                 // use checkpoint time as new startTime for request if it's earlier than
 179                                 // current startTime
 180                                 if (adjustedRange.getStartTime().getValue() > checkPoint.getTimestamp().getValue() || adjustedIndex > (Long) checkPoint.getLocation().getLocation()) {
 181                                         adjustedRange = new TmfTimeRange(checkPoint.getTimestamp(), reqWindow.getEndTime());
 182                                         adjustedIndex = (Long) checkPoint.getLocation().getLocation();
 183                                         if (nbRequested < TmfDataRequest.ALL_DATA) {
 184                                                 nbRequested += (eventRequest.getIndex() - adjustedIndex);
 185                                         }
 186                                 }
 187                         }
 188                         // Save which trace state model corresponds to current trace
 189                         traceToTraceStateModel.put(traceManager.getTrace(), traceManager.getStateModel());
 190                 }
 191
 192                 dispatchTime = reqWindow.getStartTime().getValue();
 193                 dispatchIndex = eventRequest.getIndex();
 194                 eventIndex = adjustedIndex;
 195
 196                 // Create a single request for all traces in the experiment, with coalesced time range.
 197                 fSubRequest = new LttngBaseEventRequest(adjustedRange, reqWindow.getStartTime(),
 198                                 adjustedIndex, nbRequested, BLOCK_SIZE, eventRequest.getExecType() /*ITmfDataRequest.ExecutionType.FOREGROUND*/) {
 199
 200                         private LttngSyntheticEvent syntheticEvent = null;
 201
 202                         /*
 203                          * (non-Javadoc)
 204                          *
 205                          * @see org.eclipse.linuxtools.lttng.control.LttngEventRequest#handleData()
 206                          */
 207                         @Override
 208                         public void handleData(LttngEvent event) {
 209                                 super.handleData(event);
 210                                 if (event != null) {
 211                                     synchronized (LttngSyntheticEventProvider.this) {
 212                                         // Check if request was canceled
 213                                         if ((fmainRequest == null) || (fmainRequest.isCompleted()) ) {
 214                                             TraceDebug.debug("fmainRequest was canceled. Ignoring event " + event); //$NON-NLS-1$
 215                                             return;
 216                                         }
 217
 218                                         handleIncomingData(event);
 219                                     }
 220                                 } else {
 221                                         TraceDebug.debug("handle data received with no data"); //$NON-NLS-1$
 222                                 }
 223                         }
 224
 225                         /*
 226                          * (non-Javadoc)
 227                          *
 228                          * @see org.eclipse.linuxtools.tmf.request.TmfDataRequest#handleCompleted()
 229                          */
 230                         @Override
 231                  public void handleCompleted() {
 232                                 // mark this sub-request as completed
 233                 handleProviderDone(!isCancelled() && !isFailed());
 234                 super.handleCompleted();
 235                         }
 236
 237                         /**
 238                          * Trigger the Analysis and sequential control of the events.
 239                          *
 240                          * @param e
 241                          */
 242                         private void handleIncomingData(LttngEvent e) {
 243                                 long eventTime = e.getTimestamp().getValue();
 244
 245                                 ITmfTrace inTrace = e.getParentTrace();
 246                                 LttngTraceState traceModel = traceToTraceStateModel.get(inTrace);
 247
 248                                 // queue the new event data
 249                                 updateSynEvent(e);
 250
 251                                 // If time at or above requested time, update application
 252                                 if (eventTime >= dispatchTime && eventIndex >= dispatchIndex) {
 253                                         // Before update
 254                                         syntheticEvent.setSequenceInd(SequenceInd.BEFORE);
 255                                         fmainRequest.handleData(syntheticEvent);
 256
 257                                         // Update state locally
 258                                         syntheticEvent.setSequenceInd(SequenceInd.UPDATE);
 259                                         fstateUpdateProcessor.process(syntheticEvent, traceModel);
 260
 261                                         // After Update
 262                                         syntheticEvent.setSequenceInd(SequenceInd.AFTER);
 263                                         fmainRequest.handleData(syntheticEvent);
 264
 265                                 } else {
 266                                         // event time is between checkpoint adjusted time and
 267                                         // requested time i.e. application does not expect the
 268                                         // event, however the state system needs to be re-built
 269                                         // to the dispatch point
 270                                         syntheticEvent.setSequenceInd(SequenceInd.UPDATE);
 271                                         fstateUpdateProcessor.process(syntheticEvent, traceModel);
 272                                 }
 273                                 eventIndex++;
 274                         }
 275
 276                         /**
 277                          * Create a synthetic event from the received new reference, if
 278                          * the reference is the same there is no need for a new instance
 279                          *
 280                          * if this is the first event for this request, call start
 281                          * handler
 282                          *
 283                          * @param e
 284                          * @return
 285                          */
 286                         private LttngSyntheticEvent updateSynEvent(LttngEvent e) {
 287                                 if ((syntheticEvent == null) || (syntheticEvent.getBaseEvent() != e)) {
 288                                         syntheticEvent = new LttngSyntheticEvent(e);
 289                                 }
 290
 291                                 ITmfTrace inTrace = e.getParentTrace();
 292                                 LttngTraceState traceModel = traceToTraceStateModel.get(inTrace);
 293
 294                                 // Trace model needed by application handlers
 295                                 syntheticEvent.setTraceModel(traceModel);
 296
 297                                 // send the start request indication once per request thread
 298                                 if (!startIndSent) {
 299                                         TraceDebug.debug("Thread started: " + Thread.currentThread().getName()); //$NON-NLS-1$
 300                                         handleProviderStarted(traceModel);
 301                                         startIndSent = true;
 302                                 }
 303
 304                                 return syntheticEvent;
 305                         }
 306                 };
 307
 308                 // start request
 309                 TmfExperiment<LttngEvent> provider = (TmfExperiment<LttngEvent>) fExperiment.getValue();
 310                 provider.sendRequest(fSubRequest);
 311
 312                 // notify LTTngEvent provider that all requests were sent
 313                 synchronized (this) {
 314                     TmfExperiment.getCurrentExperiment().notifyPendingRequest(false);
 315                     fIsExperimentNotified = false;
 316                 }
 317
 318                 experiment.endSynch(new TmfEndSynchSignal(0));
 319
 320                 // Return a dummy context, not used for relay provider
 321                 return new TmfContext();
 322         }
 323
 324         /**
 325          * Notify listeners to prepare to receive data e.g. clean previous data etc.
 326          */
 327         public synchronized void handleProviderStarted(LttngTraceState traceModel) {
 328                 LttngSyntheticEvent startIndEvent = new LttngSyntheticEvent(fStatusEvent);
 329                 startIndEvent.setSequenceInd(SequenceInd.STARTREQ);
 330
 331                 // Notify application
 332                 fmainRequest.handleData(startIndEvent);
 333
 334                 // Notify state event processor
 335                 fstateUpdateProcessor.process(startIndEvent, null);
 336         }
 337
 338         /**
 339          * Notify listeners, no more events for the current request will be
 340          * distributed e.g. update view.
 341          */
 342         public synchronized void handleProviderDone(boolean isSuccess) {
 343                 // Notify application. One notification per trace so the last state of each trace can be
 344                 // drawn
 345             for (LttngTraceState traceModel : traceToTraceStateModel.values()) {
 346                 // Take the trace model from traceToTraceStateModel list since it has a copy
 347                 // of the state
 348                 LttngSyntheticEvent finishEvent = new LttngSyntheticEvent(fStatusEvent);
 349             finishEvent.setSequenceInd(SequenceInd.ENDREQ);
 350             finishEvent.setTraceModel(traceModel);
 351
 352             fmainRequest.handleData(finishEvent);
 353             }
 354
 355         if(isSuccess) {
 356             // Finish main request
 357             fmainRequest.done();
 358         }
 359         else {
 360             // Cancel main request
 361             fmainRequest.cancel();
 362
 363         }
 364         }
 365
 366         /**
 367          * Reset provider to a state ready to begin thread execution
 368          *
 369          * @param experimentNode
 370          */
 371         public synchronized void reset(LTTngTreeNode experimentNode) {
 372
 373             conditionallyCancelRequests();
 374
 375                 fEventProviderRequests.clear();
 376                 startIndSent = false;
 377
 378                 // set of base event providers
 379                 if (fExperiment != null) {
 380                         LTTngTreeNode[] traces = fExperiment.getChildren();
 381                         for (LTTngTreeNode trace : traces) {
 382                                 IStateTraceManager traceBaseEventProvider = (IStateTraceManager) trace;
 383                                 fEventProviderRequests.add(traceBaseEventProvider);
 384                         }
 385                 }
 386
 387                 if (fExperiment != experimentNode) {
 388                         updateExperimentNode(experimentNode);
 389                 }
 390         }
 391
 392         /**
 393          * Point to a new experiment reference
 394          *
 395          * @param experiment
 396          */
 397         private synchronized void updateExperimentNode(LTTngTreeNode experiment) {
 398                 if (experiment != null
 399                                 && experiment.getValue() instanceof TmfExperiment<?>) {
 400                         fExperiment = experiment;
 401                 } else {
 402                         TraceDebug
 403                                         .debug("Experiment received is not instance of TmfExperiment: " //$NON-NLS-1$
 404                                                         + experiment.getClass().getName());
 405                 }
 406         }
 407
 408         /*
 409          * (non-Javadoc)
 410          *
 411          * @see
 412          * org.eclipse.linuxtools.tmf.component.TmfDataProvider#sendRequest(org.
 413          * eclipse.linuxtools.tmf.request.TmfDataRequest)
 414          */
 415         @Override
 416         public void sendRequest(final ITmfDataRequest<LttngSyntheticEvent> request) {
 417             synchronized (this) {
 418                 if (!fIsExperimentNotified) {
 419                     @SuppressWarnings("unchecked")
 420                     TmfExperiment<LttngSyntheticEvent> experiment = (TmfExperiment<LttngSyntheticEvent>) TmfExperiment.getCurrentExperiment();
 421                     if (experiment != null) {
 422                         experiment.notifyPendingRequest(true);
 423                         fIsExperimentNotified = true;
 424                     }
 425                 }
 426             }
 427
 428                 super.sendRequest(request);
 429                 if (waitForRequest) {
 430                         try {
 431                                 request.waitForCompletion();
 432                         } catch (InterruptedException e) {
 433                                 e.printStackTrace();
 434                         }
 435                 }
 436         }
 437
 438         /**
 439          * @return the waitForRequest
 440          */
 441         public boolean isWaitForRequest() {
 442                 return waitForRequest;
 443         }
 444
 445         /**
 446          * @param waitForRequest
 447          *            configures the provider to wait for the request completion
 448          */
 449         public void setWaitForRequest(boolean waitForRequest) {
 450                 this.waitForRequest = waitForRequest;
 451         }
 452
 453         @Override
 454         public LttngSyntheticEvent getNext(ITmfContext context) {
 455                 try {
 456                         fmainRequest.waitForCompletion();
 457                 } catch (InterruptedException e) {
 458                         e.printStackTrace();
 459                 }
 460                 return null;
 461         }
 462
 463         /**
 464          * Cancels the ongoing requests for this data provider if necessary
 465          */
 466         public synchronized void conditionallyCancelRequests() {
 467             if ((fSubRequest != null) && (!fSubRequest.isCompleted())) {
 468
 469                 TraceDebug.debug("Canceling synthethic event request!"); //$NON-NLS-1$
 470
 471                 // This will also cancel the fmainRequest
 472                 fSubRequest.cancel();
 473                 // Reset the request references
 474                 fSubRequest = null;
 475                 fmainRequest = null;
 476             }
 477         }
 478
 479         @Override
 480         protected void queueBackgroundRequest(ITmfDataRequest<LttngSyntheticEvent> request, int blockSize, boolean indexing) {
 481                 // do not split background synthetic requests
 482                 queueRequest(request);
 483         }
 484 }