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