org.eclipse.linuxtools.tmf.core/src/org/eclipse/linuxtools/tmf/core/statesystem/AbstractStateChangeInput.java

   1 /*******************************************************************************
   2  * Copyright (c) 2012, 2013 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.linuxtools.tmf.core.statesystem;
  14
  15 import java.util.concurrent.ArrayBlockingQueue;
  16 import java.util.concurrent.BlockingQueue;
  17
  18 import org.eclipse.linuxtools.tmf.core.ctfadaptor.CtfTmfEventFactory;
  19 import org.eclipse.linuxtools.tmf.core.event.ITmfEvent;
  20 import org.eclipse.linuxtools.tmf.core.exceptions.TimeRangeException;
  21 import org.eclipse.linuxtools.tmf.core.timestamp.ITmfTimestamp;
  22 import org.eclipse.linuxtools.tmf.core.trace.ITmfTrace;
  23
  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 AbstractStateChangeInput implements IStateChangeInput {
  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     private ITmfEvent currentEvent;
  49
  50     /** State system in which to insert the state changes */
  51     protected ITmfStateSystemBuilder ss;
  52
  53     /**
  54      * Instantiate a new state provider plugin.
  55      *
  56      * @param trace
  57      *            The LTTng 2.0 kernel trace directory
  58      * @param eventType
  59      *            The specific class for the event type that will be used within
  60      *            the subclass
  61      * @param id
  62      *            Name given to this state change input. Only used internally.
  63      */
  64     public AbstractStateChangeInput(ITmfTrace trace,
  65             Class<? extends ITmfEvent> eventType, String id) {
  66         this.trace = trace;
  67         this.eventType = eventType;
  68         eventsQueue = new ArrayBlockingQueue<ITmfEvent>(DEFAULT_EVENTS_QUEUE_SIZE);
  69         ssAssigned = false;
  70
  71         String id2 = (id == null ? "Unamed" : id); //$NON-NLS-1$
  72         eventHandlerThread = new Thread(new EventProcessor(), id2 + " Event Handler"); //$NON-NLS-1$
  73
  74     }
  75
  76     @Override
  77     public ITmfTrace getTrace() {
  78         return trace;
  79     }
  80
  81     @Override
  82     public long getStartTime() {
  83         return trace.getStartTime().normalize(0, ITmfTimestamp.NANOSECOND_SCALE).getValue();
  84     }
  85
  86     @Override
  87     public void assignTargetStateSystem(ITmfStateSystemBuilder ssb) {
  88         ss = ssb;
  89         ssAssigned = true;
  90         eventHandlerThread.start();
  91     }
  92
  93     @Override
  94     public ITmfStateSystem getAssignedStateSystem() {
  95         return ss;
  96     }
  97
  98     @Override
  99     public void dispose() {
 100         /* Insert a null event in the queue to stop the event handler's thread. */
 101         try {
 102             eventsQueue.put(CtfTmfEventFactory.getNullEvent());
 103             eventHandlerThread.join();
 104         } catch (InterruptedException e) {
 105             e.printStackTrace();
 106         }
 107         ssAssigned = false;
 108         ss = null;
 109     }
 110
 111     @Override
 112     public final Class<? extends ITmfEvent> getExpectedEventType() {
 113         return eventType;
 114     }
 115
 116     @Override
 117     public final void processEvent(ITmfEvent event) {
 118         /* Make sure the target state system has been assigned */
 119         if (!ssAssigned) {
 120             System.err.println("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         try {
 127             eventsQueue.put(curEvent);
 128         } catch (InterruptedException e) {
 129             e.printStackTrace();
 130         }
 131     }
 132
 133     /**
 134      * This is the runner class for the second thread, which will take the
 135      * events from the queue and pass them through the state system.
 136      */
 137     private class EventProcessor implements Runnable {
 138
 139         @Override
 140         public void run() {
 141             if (ss == null) {
 142                 System.err.println("Cannot run event manager without assigning a target state system first!"); //$NON-NLS-1$
 143                 return;
 144             }
 145             ITmfEvent event;
 146
 147             try {
 148                 event = eventsQueue.take();
 149                 while (event.getTimestamp().getValue() != -1) {
 150                     currentEvent = event;
 151
 152                     /* Make sure this is an event the sub-class can process */
 153                     if (eventType.isInstance(event)) {
 154                         eventHandle(event);
 155                     }
 156                     event = eventsQueue.take();
 157                 }
 158                 /* We've received the last event, clean up */
 159                 closeStateSystem();
 160                 return;
 161             } catch (InterruptedException e) {
 162                 /* We've been interrupted abnormally */
 163                 System.out.println("Event handler interrupted!"); //$NON-NLS-1$
 164                 e.printStackTrace();
 165             }
 166         }
 167
 168         private void closeStateSystem() {
 169             /* Close the History system, if there is one */
 170             if (currentEvent == null) {
 171                 return;
 172             }
 173             try {
 174                 ss.closeHistory(currentEvent.getTimestamp().normalize(0, ITmfTimestamp.NANOSECOND_SCALE).getValue());
 175             } catch (TimeRangeException e) {
 176                 /*
 177                  * Since we're using currentEvent.getTimestamp, this shouldn't
 178                  * cause any problem
 179                  */
 180                 e.printStackTrace();
 181             }
 182         }
 183     }
 184
 185     // ------------------------------------------------------------------------
 186     // Abstract methods
 187     // ------------------------------------------------------------------------
 188
 189     /**
 190      * Handle the given event and send the appropriate state transitions into
 191      * the the state system.
 192      *
 193      * This is basically the same thing as IStateChangeInput.processEvent(),
 194      * except here processEvent() and eventHandle() are run in two different
 195      * threads (and the AbstractStateChangeInput takes care of processEvent()
 196      * already).
 197      *
 198      * @param event
 199      *            The event to process. If you need a specific event type, you
 200      *            should check for its instance right at the beginning.
 201      */
 202     protected abstract void eventHandle(ITmfEvent event);
 203 }