tmf/org.eclipse.tracecompass.tmf.core/src/org/eclipse/tracecompass/tmf/core/statesystem/AbstractTmfStateProvider.java

   1 /*******************************************************************************
   2  * Copyright (c) 2012, 2015 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  *     Alexandre Montplaisir - Initial API and implementation
  11  *******************************************************************************/
  12
  13 package org.eclipse.tracecompass.tmf.core.statesystem;
  14
  15 import org.eclipse.jdt.annotation.NonNull;
  16 import org.eclipse.jdt.annotation.Nullable;
  17 import org.eclipse.tracecompass.common.core.collect.BufferedBlockingQueue;
  18 import org.eclipse.tracecompass.internal.tmf.core.Activator;
  19 import org.eclipse.tracecompass.statesystem.core.ITmfStateSystem;
  20 import org.eclipse.tracecompass.statesystem.core.ITmfStateSystemBuilder;
  21 import org.eclipse.tracecompass.tmf.core.event.ITmfEvent;
  22 import org.eclipse.tracecompass.tmf.core.event.TmfEvent;
  23 import org.eclipse.tracecompass.tmf.core.trace.ITmfContext;
  24 import org.eclipse.tracecompass.tmf.core.trace.ITmfTrace;
  25
  26 /**
  27  * Instead of using IStateChangeInput directly, one can extend this class, which
  28  * defines a lot of the common functions of the state change input plugin.
  29  *
  30  * It will handle the state-system-processing in a separate thread, which is
  31  * normally not a bad idea for traces of some size.
  32  *
  33  * processEvent() is replaced with eventHandle(), so that all the multi-thread
  34  * logic is abstracted away.
  35  *
  36  * @author Alexandre Montplaisir
  37  */
  38 public abstract class AbstractTmfStateProvider implements ITmfStateProvider {
  39
  40     private static final int DEFAULT_EVENTS_QUEUE_SIZE = 127;
  41     private static final int DEFAULT_EVENTS_CHUNK_SIZE = 127;
  42
  43     private final ITmfTrace fTrace;
  44     private final BufferedBlockingQueue<ITmfEvent> fEventsQueue;
  45     private final Thread fEventHandlerThread;
  46
  47     private boolean fStateSystemAssigned;
  48     /** State system in which to insert the state changes */
  49     private @Nullable ITmfStateSystemBuilder fSS = null;
  50
  51     /* The last safe time at which this state provider can be queried */
  52     private volatile long fSafeTime;
  53
  54     /**
  55      * Instantiate a new state provider plugin.
  56      *
  57      * @param trace
  58      *            The LTTng 2.0 kernel trace directory
  59      * @param id
  60      *            Name given to this state change input. Only used internally.
  61      */
  62     public AbstractTmfStateProvider(ITmfTrace trace, String id) {
  63         fTrace = trace;
  64         fEventsQueue = new BufferedBlockingQueue<>(DEFAULT_EVENTS_QUEUE_SIZE, DEFAULT_EVENTS_CHUNK_SIZE);
  65         fStateSystemAssigned = false;
  66         // set the safe time to before the trace start, the analysis has not yet started
  67         fSafeTime = trace.getStartTime().toNanos() - 1;
  68
  69         fEventHandlerThread = new Thread(new EventProcessor(), id + " Event Handler"); //$NON-NLS-1$
  70     }
  71
  72     /**
  73      * Get the state system builder of this provider (to insert states in).
  74      *
  75      * @return The state system object to be filled
  76      */
  77     protected @Nullable ITmfStateSystemBuilder getStateSystemBuilder() {
  78         return fSS;
  79     }
  80
  81     @Override
  82     public ITmfTrace getTrace() {
  83         return fTrace;
  84     }
  85
  86     @Override
  87     public long getStartTime() {
  88         return fTrace.getStartTime().toNanos();
  89     }
  90
  91     /**
  92      * @since 2.0
  93      */
  94     @Override
  95     public long getLatestSafeTime() {
  96         return fSafeTime;
  97     }
  98
  99     @Override
 100     public void assignTargetStateSystem(ITmfStateSystemBuilder ssb) {
 101         fSS = ssb;
 102         fStateSystemAssigned = true;
 103         fEventHandlerThread.start();
 104     }
 105
 106     @Override
 107     public @Nullable ITmfStateSystem getAssignedStateSystem() {
 108         return fSS;
 109     }
 110
 111     @Override
 112     public void dispose() {
 113         /* Insert a null event in the queue to stop the event handler's thread. */
 114         try {
 115             fEventsQueue.put(END_EVENT);
 116             fEventsQueue.flushInputBuffer();
 117             fEventHandlerThread.join();
 118         } catch (InterruptedException e) {
 119             e.printStackTrace();
 120         }
 121         fStateSystemAssigned = false;
 122         fSS = null;
 123     }
 124
 125     @Override
 126     public void processEvent(ITmfEvent event) {
 127         /* Make sure the target state system has been assigned */
 128         if (!fStateSystemAssigned) {
 129             throw new IllegalStateException("Cannot process event without a target state system. ID: " + getClass().getSimpleName()); //$NON-NLS-1$
 130         }
 131
 132         /* Insert the event we're received into the events queue */
 133         ITmfEvent curEvent = event;
 134         fEventsQueue.put(curEvent);
 135     }
 136
 137     /**
 138      * Block the caller until the events queue is empty.
 139      */
 140     public void waitForEmptyQueue() {
 141         /*
 142          * We will first insert a dummy event that is guaranteed to not modify
 143          * the state. That way, when that event leaves the queue, we will know
 144          * for sure that the state system processed the preceding real event.
 145          */
 146         try {
 147             fEventsQueue.put(EMPTY_QUEUE_EVENT);
 148             fEventsQueue.flushInputBuffer();
 149             while (!fEventsQueue.isEmpty()) {
 150                 Thread.sleep(100);
 151             }
 152         } catch (InterruptedException e) {
 153             e.printStackTrace();
 154         }
 155     }
 156
 157     // ------------------------------------------------------------------------
 158     // Special event types
 159     // ------------------------------------------------------------------------
 160
 161     /** Fake event indicating the build is over, and the provider should close */
 162     private static class EndEvent extends TmfEvent {
 163         public EndEvent() {
 164             super(null, ITmfContext.UNKNOWN_RANK, null, null, null);
 165         }
 166     }
 167
 168     /** Fake event indicating we want to clear the current queue */
 169     private static class EmptyQueueEvent extends TmfEvent {
 170         public EmptyQueueEvent() {
 171             super(null, ITmfContext.UNKNOWN_RANK, null, null, null);
 172         }
 173     }
 174
 175     private static final EndEvent END_EVENT = new EndEvent();
 176     private static final EmptyQueueEvent EMPTY_QUEUE_EVENT = new EmptyQueueEvent();
 177
 178     // ------------------------------------------------------------------------
 179     // Inner classes
 180     // ------------------------------------------------------------------------
 181
 182     /**
 183      * This is the runner class for the second thread, which will take the
 184      * events from the queue and pass them through the state system.
 185      */
 186     private class EventProcessor implements Runnable {
 187
 188         private @Nullable ITmfEvent currentEvent;
 189
 190         @Override
 191         public void run() {
 192             if (!fStateSystemAssigned) {
 193                 Activator.logError("Cannot run event manager without assigning a target state system first!"); //$NON-NLS-1$
 194                 return;
 195             }
 196
 197
 198             /*
 199              * We never insert null in the queue. Cannot be checked at
 200              * compile-time until Java 8 annotations...
 201              */
 202             @NonNull ITmfEvent event = fEventsQueue.take();
 203             /* This is a singleton, we want to do != instead of !x.equals */
 204             while (event != END_EVENT) {
 205                 if (event == EMPTY_QUEUE_EVENT) {
 206                     /* Synchronization event, should be ignored */
 207                     event = fEventsQueue.take();
 208                     continue;
 209                 }
 210                 currentEvent = event;
 211                 fSafeTime = event.getTimestamp().toNanos() - 1;
 212                 eventHandle(event);
 213                 event = fEventsQueue.take();
 214             }
 215             /* We've received the last event, clean up */
 216             done();
 217             closeStateSystem();
 218         }
 219
 220         private void closeStateSystem() {
 221             ITmfEvent event = currentEvent;
 222             final long endTime = (event == null) ? 0 :
 223                     event.getTimestamp().toNanos();
 224
 225             if (fSS != null) {
 226                 fSS.closeHistory(endTime);
 227             }
 228         }
 229     }
 230
 231     // ------------------------------------------------------------------------
 232     // Abstract methods
 233     // ------------------------------------------------------------------------
 234
 235     /**
 236      * Handle the given event and send the appropriate state transitions into
 237      * the the state system.
 238      *
 239      * This is basically the same thing as IStateChangeInput.processEvent(),
 240      * except here processEvent() and eventHandle() are run in two different
 241      * threads (and the AbstractStateChangeInput takes care of processEvent()
 242      * already).
 243      *
 244      * @param event
 245      *            The event to process. If you need a specific event type, you
 246      *            should check for its instance right at the beginning.
 247      */
 248     protected abstract void eventHandle(ITmfEvent event);
 249
 250 }