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