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