org.eclipse.tracecompass.statesystem.core/src/org/eclipse/tracecompass/statesystem/core/backend/historytree/ThreadedHistoryTreeBackend.java

   1 /*******************************************************************************
   2  * Copyright (c) 2012, 2015 Ericsson
   3  * Copyright (c) 2010, 2011 École Polytechnique de Montréal
   4  * Copyright (c) 2010, 2011 Alexandre Montplaisir <alexandre.montplaisir@gmail.com>
   5  *
   6  * All rights reserved. This program and the accompanying materials are
   7  * made available under the terms of the Eclipse Public License v1.0 which
   8  * accompanies this distribution, and is available at
   9  * http://www.eclipse.org/legal/epl-v10.html
  10  *
  11  * Contributors:
  12  *   Alexandre Montplaisir - Initial API and implementation
  13  *******************************************************************************/
  14
  15 package org.eclipse.tracecompass.statesystem.core.backend.historytree;
  16
  17 import java.io.File;
  18 import java.io.IOException;
  19 import java.util.List;
  20 import java.util.concurrent.ArrayBlockingQueue;
  21 import java.util.concurrent.BlockingQueue;
  22
  23 import org.eclipse.jdt.annotation.NonNull;
  24 import org.eclipse.tracecompass.internal.statesystem.core.Activator;
  25 import org.eclipse.tracecompass.internal.statesystem.core.backend.historytree.HTInterval;
  26 import org.eclipse.tracecompass.statesystem.core.exceptions.StateSystemDisposedException;
  27 import org.eclipse.tracecompass.statesystem.core.exceptions.TimeRangeException;
  28 import org.eclipse.tracecompass.statesystem.core.interval.ITmfStateInterval;
  29 import org.eclipse.tracecompass.statesystem.core.statevalue.ITmfStateValue;
  30 import org.eclipse.tracecompass.statesystem.core.statevalue.TmfStateValue;
  31
  32 /**
  33  * Variant of the HistoryTreeBackend which runs all the interval-insertion logic
  34  * in a separate thread.
  35  *
  36  * @author Alexandre Montplaisir
  37  * @since 3.0
  38  */
  39 public final class ThreadedHistoryTreeBackend extends HistoryTreeBackend
  40         implements Runnable {
  41
  42     private final @NonNull BlockingQueue<HTInterval> intervalQueue;
  43     private final @NonNull Thread shtThread;
  44
  45     /**
  46      * New state history constructor
  47      *
  48      * Note that it usually doesn't make sense to use a Threaded HT if you're
  49      * opening an existing state-file, but you know what you're doing...
  50      *
  51      * @param ssid
  52      *            The state system's id
  53      * @param newStateFile
  54      *            The name of the history file that will be created. Should end
  55      *            in ".ht"
  56      * @param blockSize
  57      *            The size of the blocks in the file
  58      * @param maxChildren
  59      *            The maximum number of children allowed for each core node
  60      * @param startTime
  61      *            The earliest timestamp stored in the history
  62      * @param providerVersion
  63      *            Version of of the state provider. We will only try to reopen
  64      *            existing files if this version matches the one in the
  65      *            framework.
  66      * @param queueSize
  67      *            The size of the interval insertion queue. 2000 - 10000 usually
  68      *            works well
  69      * @throws IOException
  70      *             If there was a problem opening the history file for writing
  71      */
  72     public ThreadedHistoryTreeBackend(@NonNull String ssid, File newStateFile, int blockSize,
  73             int maxChildren, long startTime, int providerVersion, int queueSize)
  74             throws IOException {
  75         super(ssid, newStateFile, blockSize, maxChildren, providerVersion, startTime);
  76
  77         intervalQueue = new ArrayBlockingQueue<>(queueSize);
  78         shtThread = new Thread(this, "History Tree Thread"); //$NON-NLS-1$
  79         shtThread.start();
  80     }
  81
  82     /**
  83      * New State History constructor. This version provides default values for
  84      * blockSize and maxChildren.
  85      *
  86      * @param ssid
  87      *            The state system's id
  88      * @param newStateFile
  89      *            The name of the history file that will be created. Should end
  90      *            in ".ht"
  91      * @param startTime
  92      *            The earliest timestamp stored in the history
  93      * @param providerVersion
  94      *            Version of of the state provider. We will only try to reopen
  95      *            existing files if this version matches the one in the
  96      *            framework.
  97      * @param queueSize
  98      *            The size of the interval insertion queue. 2000 - 10000 usually
  99      *            works well
 100      * @throws IOException
 101      *             If there was a problem opening the history file for writing
 102      */
 103     public ThreadedHistoryTreeBackend(@NonNull String ssid, File newStateFile, long startTime,
 104             int providerVersion, int queueSize) throws IOException {
 105         super(ssid, newStateFile, providerVersion, startTime);
 106
 107         intervalQueue = new ArrayBlockingQueue<>(queueSize);
 108         shtThread = new Thread(this, "History Tree Thread"); //$NON-NLS-1$
 109         shtThread.start();
 110     }
 111
 112     /*
 113      * The Threaded version does not specify an "existing file" constructor,
 114      * since the history is already built (and we only use the other thread
 115      * during building). Just use a plain HistoryTreeProvider in this case.
 116      *
 117      * TODO but what about streaming??
 118      */
 119
 120     @Override
 121     public void insertPastState(long stateStartTime, long stateEndTime,
 122             int quark, ITmfStateValue value) throws TimeRangeException {
 123         /*
 124          * Here, instead of directly inserting the elements in the History Tree
 125          * underneath, we'll put them in the Queue. They will then be taken and
 126          * processed by the other thread executing the run() method.
 127          */
 128         HTInterval interval = new HTInterval(stateStartTime, stateEndTime,
 129                 quark, (TmfStateValue) value);
 130         try {
 131             intervalQueue.put(interval);
 132         } catch (InterruptedException e) {
 133             Activator.getDefault().logError("State system interrupted", e); //$NON-NLS-1$
 134         }
 135     }
 136
 137     @Override
 138     public void finishedBuilding(long endTime) {
 139         /*
 140          * We need to commit everything in the History Tree and stop the
 141          * standalone thread before returning to the StateHistorySystem. (SHS
 142          * will then write the Attribute Tree to the file, that must not happen
 143          * at the same time we are writing the last nodes!)
 144          */
 145
 146         stopRunningThread(endTime);
 147         setFinishedBuilding(true);
 148         return;
 149     }
 150
 151     @Override
 152     public void dispose() {
 153         if (!isFinishedBuilding()) {
 154             stopRunningThread(Long.MAX_VALUE);
 155         }
 156         /*
 157          * isFinishedBuilding remains false, so the superclass will ask the
 158          * back-end to delete the file.
 159          */
 160         super.dispose();
 161     }
 162
 163     private void stopRunningThread(long endTime) {
 164         if (!shtThread.isAlive()) {
 165             return;
 166         }
 167
 168         /*
 169          * Send a "poison pill" in the queue, then wait for the HT to finish its
 170          * closeTree()
 171          */
 172         try {
 173             HTInterval pill = new HTInterval(-1, endTime, -1, TmfStateValue.nullValue());
 174             intervalQueue.put(pill);
 175             shtThread.join();
 176         } catch (TimeRangeException e) {
 177             Activator.getDefault().logError("Error closing state system", e); //$NON-NLS-1$
 178         } catch (InterruptedException e) {
 179             Activator.getDefault().logError("State system interrupted", e); //$NON-NLS-1$
 180         }
 181     }
 182
 183     @Override
 184     public void run() {
 185         HTInterval currentInterval;
 186         try {
 187             currentInterval = intervalQueue.take();
 188             while (currentInterval.getStartTime() != -1) {
 189                 /* Send the interval to the History Tree */
 190                 getSHT().insertInterval(currentInterval);
 191                 currentInterval = intervalQueue.take();
 192             }
 193             if (currentInterval.getAttribute() != -1) {
 194                 /* Make sure this is the "poison pill" we are waiting for */
 195                 throw new IllegalStateException();
 196             }
 197             /*
 198              * We've been told we're done, let's write down everything and quit.
 199              * The end time of this "signal interval" is actually correct.
 200              */
 201             getSHT().closeTree(currentInterval.getEndTime());
 202             return;
 203         } catch (InterruptedException e) {
 204             /* We've been interrupted abnormally */
 205             Activator.getDefault().logError("State History Tree interrupted!", e); //$NON-NLS-1$
 206         } catch (TimeRangeException e) {
 207             /* This also should not happen */
 208             Activator.getDefault().logError("Error starting the state system", e); //$NON-NLS-1$
 209         }
 210     }
 211
 212     // ------------------------------------------------------------------------
 213     // Query methods
 214     // ------------------------------------------------------------------------
 215
 216     @Override
 217     public void doQuery(List<ITmfStateInterval> currentStateInfo, long t)
 218             throws TimeRangeException, StateSystemDisposedException {
 219         super.doQuery(currentStateInfo, t);
 220
 221         if (isFinishedBuilding()) {
 222             /*
 223              * The history tree is the only place to look for intervals once
 224              * construction is finished.
 225              */
 226             return;
 227         }
 228
 229         /*
 230          * It is possible we may have missed some intervals due to them being in
 231          * the queue while the query was ongoing. Go over the results to see if
 232          * we missed any.
 233          */
 234         for (int i = 0; i < currentStateInfo.size(); i++) {
 235             if (currentStateInfo.get(i) == null) {
 236                 /* Query the missing interval via "unicast" */
 237                 ITmfStateInterval interval = doSingularQuery(t, i);
 238                 currentStateInfo.set(i, interval);
 239             }
 240         }
 241     }
 242
 243     @Override
 244     public ITmfStateInterval doSingularQuery(long t, int attributeQuark)
 245             throws TimeRangeException, StateSystemDisposedException {
 246         ITmfStateInterval ret = super.doSingularQuery(t, attributeQuark);
 247         if (ret != null) {
 248             return ret;
 249         }
 250
 251         /*
 252          * We couldn't find the interval in the history tree. It's possible that
 253          * it is currently in the intervalQueue. Look for it there. Note that
 254          * ArrayBlockingQueue's iterator() is thread-safe (no need to lock the
 255          * queue).
 256          */
 257         for (ITmfStateInterval interval : intervalQueue) {
 258             if (interval.getAttribute() == attributeQuark && interval.intersects(t)) {
 259                 return interval;
 260             }
 261         }
 262
 263         /*
 264          * If we missed it again, it's because it got inserted in the tree
 265          * *while we were iterating* on the queue. One last pass in the tree
 266          * should find it.
 267          *
 268          * This case is really rare, which is why we do a second pass at the end
 269          * if needed, instead of systematically checking in the queue first
 270          * (which is slow).
 271          */
 272         return super.doSingularQuery(t, attributeQuark);
 273     }
 274
 275 }