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