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

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