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

   1 /*******************************************************************************
   2  * Copyright (c) 2012 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.event.ITmfEvent;
  19 import org.eclipse.linuxtools.tmf.core.event.ITmfTimestamp;
  20 import org.eclipse.linuxtools.tmf.core.exceptions.TimeRangeException;
  21 import org.eclipse.linuxtools.tmf.core.trace.ITmfTrace;
  22
  23
  24 /**
  25  * Instead of using IStateChangeInput directly, one can extend this class, which
  26  * defines a lot of the common functions of the state change input plugin.
  27  *
  28  * It will handle the state-system-processing in a separate thread, which is
  29  * normally not a bad idea for traces of some size.
  30  *
  31  * processEvent() is replaced with eventHandle(), so that all the multi-thread
  32  * logic is abstracted away.
  33  *
  34  * @author Alexandre Montplaisir
  35  * @since 2.0
  36  */
  37 public abstract class AbstractStateChangeInput implements IStateChangeInput {
  38
  39     private static final int DEFAULT_EVENTS_QUEUE_SIZE = 10000;
  40
  41     private final ITmfTrace trace;
  42     private final Class<? extends ITmfEvent> eventType;
  43     private final BlockingQueue<ITmfEvent> eventsQueue;
  44     private final Thread eventHandlerThread;
  45
  46     private boolean ssAssigned;
  47     private ITmfEvent currentEvent;
  48
  49     /** State system in which to insert the state changes */
  50     protected ITmfStateSystemBuilder ss;
  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 AbstractStateChangeInput(ITmfTrace trace,
  64             Class<? extends ITmfEvent> eventType, String id) {
  65         this.trace = trace;
  66         this.eventType = eventType;
  67         eventsQueue = new ArrayBlockingQueue<ITmfEvent>(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
  75     @Override
  76     public ITmfTrace getTrace() {
  77         return trace;
  78     }
  79
  80     @Override
  81     public long getStartTime() {
  82         return trace.getStartTime().normalize(0, ITmfTimestamp.NANOSECOND_SCALE).getValue();
  83     }
  84
  85     @Override
  86     public void assignTargetStateSystem(ITmfStateSystemBuilder ssb) {
  87         ss = ssb;
  88         ssAssigned = true;
  89         eventHandlerThread.start();
  90     }
  91
  92     @Override
  93     public void dispose() {
  94         /* Insert a null event in the queue to stop the event handler's thread. */
  95         try {
  96             eventsQueue.put(org.eclipse.linuxtools.tmf.core.ctfadaptor.CtfTmfEvent.getNullEvent());
  97             eventHandlerThread.join();
  98         } catch (InterruptedException e) {
  99             e.printStackTrace();
 100         }
 101         ssAssigned = false;
 102         ss = null;
 103     }
 104
 105     @Override
 106     public final Class<? extends ITmfEvent> getExpectedEventType() {
 107         return eventType;
 108     }
 109
 110     @Override
 111     public final void processEvent(ITmfEvent event) {
 112         /* Make sure the target state system has been assigned */
 113         if (!ssAssigned) {
 114             System.err.println("Cannot process event without a target state system"); //$NON-NLS-1$
 115             return;
 116         }
 117
 118         /* Insert the event we're received into the events queue */
 119         ITmfEvent curEvent = event;
 120         try {
 121             eventsQueue.put(curEvent);
 122         } catch (InterruptedException e) {
 123             e.printStackTrace();
 124         }
 125     }
 126
 127     /**
 128      * This is the runner class for the second thread, which will take the
 129      * events from the queue and pass them through the state system.
 130      */
 131     private class EventProcessor implements Runnable {
 132
 133         @Override
 134         public void run() {
 135             if (ss == null) {
 136                 System.err.println("Cannot run event manager without assigning a target state system first!"); //$NON-NLS-1$
 137                 return;
 138             }
 139             ITmfEvent event;
 140
 141             try {
 142                 event = eventsQueue.take();
 143                 while (event.getTimestamp().getValue() != -1) {
 144                     currentEvent = event;
 145
 146                     /* Make sure this is an event the sub-class can process */
 147                     if (eventType.isInstance(event)) {
 148                         eventHandle(event);
 149                     }
 150                     event = eventsQueue.take();
 151                 }
 152                 /* We've received the last event, clean up */
 153                 closeStateSystem();
 154                 return;
 155             } catch (InterruptedException e) {
 156                 /* We've been interrupted abnormally */
 157                 System.out.println("Event handler interrupted!"); //$NON-NLS-1$
 158                 e.printStackTrace();
 159             }
 160         }
 161
 162         private void closeStateSystem() {
 163             /* Close the History system, if there is one */
 164             if (currentEvent == null) {
 165                 return;
 166             }
 167             try {
 168                 ss.closeHistory(currentEvent.getTimestamp().normalize(0, ITmfTimestamp.NANOSECOND_SCALE).getValue());
 169             } catch (TimeRangeException e) {
 170                 /*
 171                  * Since we're using currentEvent.getTimestamp, this shouldn't
 172                  * cause any problem
 173                  */
 174                 e.printStackTrace();
 175             }
 176         }
 177     }
 178
 179     // ------------------------------------------------------------------------
 180     // Abstract methods
 181     // ------------------------------------------------------------------------
 182
 183     /**
 184      * Handle the given event and send the appropriate state transitions into
 185      * the the state system.
 186      *
 187      * This is basically the same thing as IStateChangeInput.processEvent(),
 188      * except here processEvent() and eventHandle() are run in two different
 189      * threads (and the AbstractStateChangeInput takes care of processEvent()
 190      * already).
 191      *
 192      * @param event
 193      *            The event to process. If you need a specific event type, you
 194      *            should check for its instance right at the beginning.
 195      */
 196     protected abstract void eventHandle(ITmfEvent event);
 197 }