org.eclipse.linuxtools.lttng/src/org/eclipse/linuxtools/lttng/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  *******************************************************************************/
  12
  13 package org.eclipse.linuxtools.lttng.control;
  14
  15 import java.util.HashMap;
  16 import java.util.Map;
  17
  18 import org.eclipse.linuxtools.lttng.TraceDebug;
  19 import org.eclipse.linuxtools.lttng.event.LttngEvent;
  20 import org.eclipse.linuxtools.lttng.event.LttngEventType;
  21 import org.eclipse.linuxtools.lttng.event.LttngSyntheticEvent;
  22 import org.eclipse.linuxtools.lttng.event.LttngSyntheticEvent.SequenceInd;
  23 import org.eclipse.linuxtools.lttng.event.LttngTimestamp;
  24 import org.eclipse.linuxtools.lttng.model.LTTngTreeNode;
  25 import org.eclipse.linuxtools.lttng.request.LttngBaseEventRequest;
  26 import org.eclipse.linuxtools.lttng.state.evProcessor.ITransEventProcessor;
  27 import org.eclipse.linuxtools.lttng.state.evProcessor.state.StateEventToHandlerFactory;
  28 import org.eclipse.linuxtools.lttng.state.model.LttngTraceState;
  29 import org.eclipse.linuxtools.lttng.state.trace.IStateTraceManager;
  30 import org.eclipse.linuxtools.tmf.component.TmfEventProvider;
  31 import org.eclipse.linuxtools.tmf.event.TmfEventSource;
  32 import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  33 import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
  34 import org.eclipse.linuxtools.tmf.experiment.TmfExperiment;
  35 import org.eclipse.linuxtools.tmf.request.ITmfDataRequest;
  36 import org.eclipse.linuxtools.tmf.request.ITmfEventRequest;
  37 import org.eclipse.linuxtools.tmf.request.TmfEventRequest;
  38 import org.eclipse.linuxtools.tmf.signal.TmfEndSynchSignal;
  39 import org.eclipse.linuxtools.tmf.signal.TmfStartSynchSignal;
  40 import org.eclipse.linuxtools.tmf.trace.ITmfContext;
  41 import org.eclipse.linuxtools.tmf.trace.ITmfTrace;
  42 import org.eclipse.linuxtools.tmf.trace.TmfContext;
  43 import org.eclipse.linuxtools.tmf.trace.TmfTrace;
  44
  45 /**
  46  * @author alvaro
  47  *
  48  */
  49 public class LttngSyntheticEventProvider extends TmfEventProvider<LttngSyntheticEvent> {
  50
  51         // ========================================================================
  52         // Data
  53         // ========================================================================
  54
  55         public static final int BLOCK_SIZE = 1;
  56         public static final int NB_EVENTS  = 1;
  57         public static final int QUEUE_SIZE = 1; // lttng specific, one event at a time
  58
  59         // TmfDataProvider<LttngEvent> fExtProvider = null;
  60         private ITmfDataRequest<LttngSyntheticEvent> fmainRequest = null;
  61         private final Map<IStateTraceManager, LttngBaseEventRequest> fEventProviderRequests = new HashMap<IStateTraceManager, LttngBaseEventRequest>();
  62         private final LttngSyntheticEvent fStatusEvent;
  63         private final LttngSyntheticEvent fStatusEventAck;
  64         private int fMainReqEventCount = 0;
  65         volatile boolean startIndSent = false;
  66         private LTTngTreeNode fExperiment = null;
  67         private ITransEventProcessor fstateUpdateProcessor = StateEventToHandlerFactory
  68                         .getInstance();
  69         private boolean waitForRequest = false;
  70
  71         // ========================================================================
  72         // Constructor
  73         // ========================================================================
  74         /**
  75          * Accessibility to package - use factory instead of this constructor
  76          *
  77          * @param type
  78          */
  79         LttngSyntheticEventProvider(Class<LttngSyntheticEvent> type) {
  80                 super("LttngSyntheticEventProvider", type, QUEUE_SIZE);
  81
  82                 // prepare empty instance status indicators and allow them to travel via
  83                 // the framework
  84                 TmfEventSource source = new TmfEventSource(this);
  85                 LttngEventType dtype = new LttngEventType();
  86                 LttngTimestamp statusTimeStamp = new LttngTimestamp(
  87                                 TmfTimestamp.Zero);
  88
  89                 fStatusEvent = new LttngSyntheticEvent(null, statusTimeStamp, source,
  90                                 dtype, null, null, null);
  91                 fStatusEvent.setSequenceInd(SequenceInd.STARTREQ);
  92
  93                 fStatusEventAck = new LttngSyntheticEvent(null, statusTimeStamp,
  94                                 source, dtype, null, null, null);
  95                 fStatusEventAck.setSequenceInd(SequenceInd.ACK);
  96         }
  97
  98         // ========================================================================
  99         // Methods
 100         // ========================================================================
 101
 102         @SuppressWarnings("unchecked")
 103         @Override
 104         public ITmfContext armRequest(
 105                         final ITmfDataRequest<LttngSyntheticEvent> request) {
 106                 // validate
 107                 // make sure we have the right type of request
 108                 if (!(request instanceof ITmfEventRequest<?>)) {
 109                         request.cancel();
 110                         TraceDebug.debug("Request is not an instance of ITmfEventRequest");
 111                         return null;
 112                 }
 113
 114                 if (fExperiment == null) {
 115                         TraceDebug.debug("Experiment is null");
 116                         request.cancel();
 117                         return null;
 118                 }
 119
 120                 // get ready to start processing
 121                 reset(fExperiment);
 122
 123                 // At least one base provider shall be available
 124                 if (fEventProviderRequests.size() < 1) {
 125                         request.cancel();
 126                         TraceDebug.debug("No Base event providers available");
 127                         return null;
 128                 }
 129
 130                 fmainRequest = request;
 131                 // define event data handling
 132                 ITmfEventRequest<LttngSyntheticEvent> eventRequest = (ITmfEventRequest<LttngSyntheticEvent>) fmainRequest;
 133                 TmfTimeRange reqWindow = eventRequest.getRange();
 134
 135                 TraceDebug.debug("Main Synthethic event request started on thread:  " + Thread.currentThread().getName());
 136
 137                 // loop for every traceManager in current experiment
 138                 boolean subRequestQueued = false;
 139                 TmfExperiment<LttngEvent> experiment = (TmfExperiment<LttngEvent>) fExperiment.getValue();
 140                 experiment.startSynch(new TmfStartSynchSignal(0));
 141                 for (IStateTraceManager traceManager : fEventProviderRequests.keySet()) {
 142
 143                         // restore trace state system to nearest check point
 144                         TmfTimestamp checkPoint = traceManager
 145                                         .restoreCheckPointByTimestamp(reqWindow.getStartTime());
 146
 147                         // adjust start time bound to check point
 148
 149                         // validate so checkpoint restore is within requested bounds
 150                         TmfTimeRange traceRange = traceManager.getTrace().getTimeRange();
 151                         if ((checkPoint != null) && !(
 152                                         checkPoint.getValue() >= traceRange.getStartTime().getValue() &&
 153                                         checkPoint.getValue() <= traceRange.getEndTime().getValue() &&
 154                                         checkPoint.getValue() < reqWindow.getEndTime().getValue())
 155                                         ) {
 156                                 // checkpoint is out of trace bounds
 157                                 continue;
 158                         }
 159                         TmfTimeRange adjustedRange = reqWindow;
 160                         if (checkPoint != null) {
 161                                 adjustedRange = new TmfTimeRange(checkPoint, reqWindow.getEndTime());
 162                         }
 163
 164                         LttngTraceState traceModel = traceManager.getStateModel();
 165                         // String key = (traceManager.getTrace().getPath() +
 166                         // traceManager.getTrace().getName()).hashCode();
 167                         ITmfTrace trace = traceManager.getTrace();
 168                         // create sub-request for one trace within experiment
 169                         final LttngBaseEventRequest subRequest = new LttngBaseEventRequest(adjustedRange, reqWindow.getStartTime(),
 170                                         0, TmfEventRequest.ALL_DATA, BLOCK_SIZE, traceModel, ITmfDataRequest.ExecutionType.FOREGROUND, trace) {
 171
 172                                 private LttngSyntheticEvent syntheticEvent = null;
 173                                 private LttngSyntheticEvent syntheticAckIndicator = null;
 174                                 long subEventCount = 0L;
 175
 176                                 private final long fDispatchTime = getDispatchTime().getValue();
 177                                 private final LttngTraceState fTraceModel = getTraceModel();
 178
 179                                 /*
 180                                  * (non-Javadoc)
 181                                  *
 182                                  * @see org.eclipse.linuxtools.lttng.control.LttngEventRequest#handleData()
 183                                  */
 184                                 @Override
 185                                 public void handleData(LttngEvent event) {
 186                                         super.handleData(event);
 187                                         if (event != null) {
 188                                                 handleIncomingData(event);
 189                                         } else {
 190                                                 TraceDebug.debug("handle data received with no data");
 191                                         }
 192                                 }
 193                                 /*
 194                                  * (non-Javadoc)
 195                                  *
 196                                  * @see org.eclipse.linuxtools.tmf.request.TmfDataRequest#done()
 197                                  */
 198                                 @Override
 199                                 public void done() {
 200                                         // mark this sub-request as completed
 201                                         super.done();
 202                                         handleProviderDone(getTraceModel());
 203                                 }
 204
 205                                 /**
 206                                  * Trigger the Analysis and sequential control of the events.
 207                                  *
 208                                  * @param e
 209                                  */
 210                                 private void handleIncomingData(LttngEvent e) {
 211                                         long eventTime = e.getTimestamp().getValue();
 212
 213                                         TmfTrace<LttngEvent> inTrace =  e.getParentTrace();
 214                                         if (!(inTrace == getTrace())) {
 215                                                 return;
 216                                         }
 217
 218                                         // queue the new event data and an ACK
 219                                         updateSynEvent(e);
 220
 221                                         // If time at or above requested time, update application
 222                                         if (eventTime >= fDispatchTime) {
 223                                                 // Before update
 224                                                 syntheticEvent.setSequenceInd(SequenceInd.BEFORE);
 225                                                 fmainRequest.handleData(syntheticEvent);
 226                                                 fmainRequest.handleData(syntheticAckIndicator);
 227
 228                                                 // Update state locally
 229                                                 syntheticEvent.setSequenceInd(SequenceInd.UPDATE);
 230                                                 fstateUpdateProcessor.process(syntheticEvent, fTraceModel);
 231
 232                                                 // After Update
 233                                                 syntheticEvent.setSequenceInd(SequenceInd.AFTER);
 234                                                 fmainRequest.handleData(syntheticEvent);
 235                                                 fmainRequest.handleData(syntheticAckIndicator);
 236
 237                                                 // increment once per dispatch
 238                                                 incrementSynEvenCount();
 239                                                 subEventCount++;
 240                                         } else {
 241                                                 // event time is between checkpoint adjusted time and
 242                                                 // requested time i.e. application does not expect the
 243                                                 // event, however the state system needs to be re-built
 244                                                 // to the dispatch point
 245                                                 syntheticEvent.setSequenceInd(SequenceInd.UPDATE);
 246                                                 fstateUpdateProcessor.process(syntheticEvent, fTraceModel);
 247                                         }
 248                                 }
 249
 250                                 /**
 251                                  * Create a synthetic event from the received new reference, if
 252                                  * the reference is the same there is no need for a new instance
 253                                  *
 254                                  * if this is the first event for this request, call start
 255                                  * handler
 256                                  *
 257                                  * @param e
 258                                  * @return
 259                                  */
 260                                 private LttngSyntheticEvent updateSynEvent(LttngEvent e) {
 261                                         if (syntheticEvent == null
 262                                                         || syntheticEvent.getBaseEvent() != e) {
 263                                                 syntheticEvent = new LttngSyntheticEvent(e);
 264                                                 syntheticAckIndicator = new LttngSyntheticEvent(e);
 265                                                 syntheticAckIndicator.setSequenceInd(SequenceInd.ACK);
 266                                         }
 267
 268                                         // Trace model needed by application handlers
 269                                         syntheticEvent.setTraceModel(fTraceModel);
 270
 271                                         // send the start request indication once per request thread
 272                                         if (!startIndSent) {
 273                                                 TraceDebug.debug("Thread started: " + Thread.currentThread().getName());
 274                                                 handleProviderStarted(getTraceModel());
 275                                                 startIndSent = true;
 276                                         }
 277
 278                                         return syntheticEvent;
 279                                 }
 280                         };
 281
 282                         // preserve the associated sub request to control it e.g.
 283                         // cancellation
 284                         fEventProviderRequests.put(traceManager, subRequest);
 285
 286                         // start request
 287                         TmfExperiment<LttngEvent> provider = (TmfExperiment<LttngEvent>) fExperiment.getValue();
 288                         provider.sendRequest(subRequest);
 289
 290                         // provider.sendRequest(subRequest, ExecutionType.LONG);
 291                         subRequestQueued = true;
 292                 }
 293
 294                 experiment.endSynch(new TmfEndSynchSignal(0));
 295
 296                 // Return a dummy context, not used for relay provider
 297                 return (subRequestQueued) ? new TmfContext() : null;
 298         }
 299
 300         /**
 301          * Notify listeners to prepare to receive data e.g. clean previous data etc.
 302          */
 303         public void handleProviderStarted(LttngTraceState traceModel) {
 304                 LttngSyntheticEvent startIndEvent = new LttngSyntheticEvent(fStatusEvent);
 305                 startIndEvent.setSequenceInd(SequenceInd.STARTREQ);
 306
 307                 // Notify application
 308                 fmainRequest.handleData(startIndEvent);
 309                 fmainRequest.handleData(fStatusEventAck);
 310
 311                 // Notify state event processor
 312                 fstateUpdateProcessor.process(startIndEvent, null);
 313         }
 314
 315         /**
 316          * Notify listeners, no more events for the current request will be
 317          * distributed e.g. update view.
 318          */
 319         public synchronized void handleProviderDone(LttngTraceState traceModel) {
 320                 // TODO: The use of a thread per main request and thread per sub-request
 321                 // requires
 322                 // to make sure the proper main request is marked completed. So a
 323                 // relationship of sub-requests to parent needs to be established to
 324                 // handle completion and cancellations properly
 325
 326                 // Close the main request when all sub-requests are marked completed
 327                 for (LttngBaseEventRequest subRequest : fEventProviderRequests.values()) {
 328                         if (subRequest != null) {
 329                                 if (!subRequest.isCompleted()) {
 330                                         // Not ready to complete main request
 331                                         return;
 332                                 }
 333                         }
 334                 }
 335
 336                 // All sub-requests are marked completed so the main request can be
 337                 // completed as well
 338                 // Notify application,
 339                 LttngSyntheticEvent finishEvent = new LttngSyntheticEvent(fStatusEvent);
 340                 finishEvent.setSequenceInd(SequenceInd.ENDREQ);
 341                 finishEvent.setTraceModel(traceModel);
 342
 343                 fmainRequest.handleData(finishEvent);
 344                 fmainRequest.handleData(fStatusEventAck);
 345                 fmainRequest.done();
 346         }
 347
 348         /**
 349          * Increment the global event counter i.e. events from any sub requests
 350          */
 351         private synchronized void incrementSynEvenCount() {
 352                 fMainReqEventCount++;
 353         }
 354
 355         /**
 356          * @return
 357          */
 358         public synchronized int getSynEvenCount() {
 359                 return fMainReqEventCount;
 360         }
 361
 362         /**
 363          * Reset provider to a state ready to begin thread execution
 364          *
 365          * @param experimentNode
 366          */
 367         public synchronized void reset(LTTngTreeNode experimentNode) {
 368
 369                 fmainRequest = null;
 370
 371                 // Make sure previous request are terminated
 372                 for (LttngBaseEventRequest tmpRequest : fEventProviderRequests.values()) {
 373                         if (tmpRequest != null && !tmpRequest.isCompleted()) {
 374                                 tmpRequest.cancel();
 375                         }
 376                 }
 377
 378                 fEventProviderRequests.clear();
 379                 fMainReqEventCount = 0;
 380                 startIndSent = false;
 381
 382                 // set of base event providers
 383                 if (fExperiment != null) {
 384                         LTTngTreeNode[] traces = fExperiment.getChildren();
 385                         for (LTTngTreeNode trace : traces) {
 386                                 IStateTraceManager traceBaseEventProvider = (IStateTraceManager) trace;
 387                                 fEventProviderRequests.put(traceBaseEventProvider, null);
 388                         }
 389                 }
 390
 391                 if (fExperiment != experimentNode) {
 392                         updateExperimentNode(experimentNode);
 393                 }
 394         }
 395
 396         /**
 397          * Point to a new experiment reference
 398          *
 399          * @param experiment
 400          */
 401         private synchronized void updateExperimentNode(LTTngTreeNode experiment) {
 402                 if (experiment != null
 403                                 && experiment.getValue() instanceof TmfExperiment<?>) {
 404                         fExperiment = experiment;
 405                 } else {
 406                         TraceDebug
 407                                         .debug("Experiment received is not instance of TmfExperiment: "
 408                                                         + experiment.getClass().getName());
 409                 }
 410         }
 411
 412         /*
 413          * (non-Javadoc)
 414          *
 415          * @see
 416          * org.eclipse.linuxtools.tmf.component.TmfDataProvider#sendRequest(org.
 417          * eclipse.linuxtools.tmf.request.TmfDataRequest)
 418          */
 419         @Override
 420         public void sendRequest(final ITmfDataRequest<LttngSyntheticEvent> request) {
 421                 super.sendRequest(request);
 422                 if (waitForRequest) {
 423                         try {
 424                                 request.waitForCompletion();
 425                         } catch (InterruptedException e) {
 426                                 e.printStackTrace();
 427                         }
 428                 }
 429         }
 430
 431         /**
 432          * @return the waitForRequest
 433          */
 434         public boolean isWaitForRequest() {
 435                 return waitForRequest;
 436         }
 437
 438         /**
 439          * @param waitForRequest
 440          *            configures the provider to wait for the request completion
 441          */
 442         public void setWaitForRequest(boolean waitForRequest) {
 443                 this.waitForRequest = waitForRequest;
 444         }
 445
 446         @Override
 447         public LttngSyntheticEvent getNext(ITmfContext context) {
 448                 try {
 449                         fmainRequest.waitForCompletion();
 450                 } catch (InterruptedException e) {
 451                         e.printStackTrace();
 452                 }
 453                 return null;
 454         }
 455
 456 }