org.eclipse.tracecompass.tmf.core/src/org/eclipse/tracecompass/internal/tmf/core/statesystem/backends/partial/PartialHistoryBackend.java

   1 /*******************************************************************************
   2  * Copyright (c) 2013, 2014 Ericsson
   3  * All rights reserved. This program and the accompanying materials are
   4  * made available under the terms of the Eclipse Public License v1.0 which
   5  * accompanies this distribution, and is available at
   6  * http://www.eclipse.org/legal/epl-v10.html
   7  *
   8  * Contributors:
   9  *   Alexandre Montplaisir - Initial API and implementation
  10  *******************************************************************************/
  11
  12 package org.eclipse.tracecompass.internal.tmf.core.statesystem.backends.partial;
  13
  14 import java.io.File;
  15 import java.io.FileInputStream;
  16 import java.io.PrintWriter;
  17 import java.util.List;
  18 import java.util.Map;
  19 import java.util.TreeMap;
  20 import java.util.concurrent.CountDownLatch;
  21
  22 import org.eclipse.jdt.annotation.NonNull;
  23 import org.eclipse.tracecompass.statesystem.core.ITmfStateSystem;
  24 import org.eclipse.tracecompass.statesystem.core.backend.IStateHistoryBackend;
  25 import org.eclipse.tracecompass.statesystem.core.exceptions.AttributeNotFoundException;
  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.interval.TmfStateInterval;
  30 import org.eclipse.tracecompass.statesystem.core.statevalue.ITmfStateValue;
  31 import org.eclipse.tracecompass.tmf.core.event.ITmfEvent;
  32 import org.eclipse.tracecompass.tmf.core.request.ITmfEventRequest;
  33 import org.eclipse.tracecompass.tmf.core.request.TmfEventRequest;
  34 import org.eclipse.tracecompass.tmf.core.statesystem.AbstractTmfStateProvider;
  35 import org.eclipse.tracecompass.tmf.core.statesystem.ITmfStateProvider;
  36 import org.eclipse.tracecompass.tmf.core.timestamp.ITmfTimestamp;
  37 import org.eclipse.tracecompass.tmf.core.timestamp.TmfTimeRange;
  38 import org.eclipse.tracecompass.tmf.core.timestamp.TmfTimestamp;
  39 import org.eclipse.tracecompass.tmf.core.trace.ITmfTrace;
  40
  41 /**
  42  * Partial state history back-end.
  43  *
  44  * This is a shim inserted between the real state system and a "real" history
  45  * back-end. It will keep checkpoints, every n trace events (where n is called
  46  * the granularity) and will only forward to the real state history the state
  47  * intervals that crosses at least one checkpoint. Every other interval will
  48  * be discarded.
  49  *
  50  * This would mean that it can only answer queries exactly at the checkpoints.
  51  * For any other timestamps (ie, most of the time), it will load the closest
  52  * earlier checkpoint, and will re-feed the state-change-input with events from
  53  * the trace, to restore the real state at the time that was requested.
  54  *
  55  * @author Alexandre Montplaisir
  56  */
  57 public class PartialHistoryBackend implements IStateHistoryBackend {
  58
  59     /**
  60      * A partial history needs the state input plugin to re-generate state
  61      * between checkpoints.
  62      */
  63     private final @NonNull ITmfStateProvider fPartialInput;
  64
  65     /**
  66      * Fake state system that is used for partially rebuilding the states (when
  67      * going from a checkpoint to a target query timestamp).
  68      */
  69     private final @NonNull PartialStateSystem fPartialSS;
  70
  71     /** Reference to the "real" state history that is used for storage */
  72     private final @NonNull IStateHistoryBackend fInnerHistory;
  73
  74     /** Checkpoints map, <Timestamp, Rank in the trace> */
  75     private final @NonNull TreeMap<Long, Long> fCheckpoints = new TreeMap<>();
  76
  77     /** Latch tracking if the initial checkpoint registration is done */
  78     private final @NonNull CountDownLatch fCheckpointsReady = new CountDownLatch(1);
  79
  80     private final long fGranularity;
  81
  82     private long fLatestTime;
  83
  84     /**
  85      * Constructor
  86      *
  87      * @param partialInput
  88      *            The state change input object that was used to build the
  89      *            upstream state system. This partial history will make its own
  90      *            copy (since they have different targets).
  91      * @param pss
  92      *            The partial history's inner state system. It should already be
  93      *            assigned to partialInput.
  94      * @param realBackend
  95      *            The real state history back-end to use. It's supposed to be
  96      *            modular, so it should be able to be of any type.
  97      * @param granularity
  98      *            Configuration parameter indicating how many trace events there
  99      *            should be between each checkpoint
 100      */
 101     public PartialHistoryBackend(ITmfStateProvider partialInput, PartialStateSystem pss,
 102             IStateHistoryBackend realBackend, long granularity) {
 103         if (granularity <= 0 || partialInput == null || pss == null ||
 104                 partialInput.getAssignedStateSystem() != pss) {
 105             throw new IllegalArgumentException();
 106         }
 107
 108         final long startTime = realBackend.getStartTime();
 109
 110         fPartialInput = partialInput;
 111         fPartialSS = pss;
 112
 113         fInnerHistory = realBackend;
 114         fGranularity = granularity;
 115
 116         fLatestTime = startTime;
 117
 118         registerCheckpoints();
 119     }
 120
 121     private void registerCheckpoints() {
 122         ITmfEventRequest request = new CheckpointsRequest(fPartialInput, fCheckpoints);
 123         fPartialInput.getTrace().sendRequest(request);
 124         /* The request will countDown the checkpoints latch once it's finished */
 125     }
 126
 127     @Override
 128     public long getStartTime() {
 129         return fInnerHistory.getStartTime();
 130     }
 131
 132     @Override
 133     public long getEndTime() {
 134         return fLatestTime;
 135     }
 136
 137     @Override
 138     public void insertPastState(long stateStartTime, long stateEndTime,
 139             int quark, ITmfStateValue value) throws TimeRangeException {
 140         waitForCheckpoints();
 141
 142         /* Update the latest time */
 143         if (stateEndTime > fLatestTime) {
 144             fLatestTime = stateEndTime;
 145         }
 146
 147         /*
 148          * Check if the interval intersects the previous checkpoint. If so,
 149          * insert it in the real history back-end.
 150          *
 151          * FIXME since intervals are inserted in order of rank, we could avoid
 152          * doing a map lookup every time here (just compare with the known
 153          * previous one).
 154          */
 155         if (stateStartTime <= fCheckpoints.floorKey(stateEndTime)) {
 156             fInnerHistory.insertPastState(stateStartTime, stateEndTime, quark, value);
 157         }
 158     }
 159
 160     @Override
 161     public void finishedBuilding(long endTime) throws TimeRangeException {
 162         fInnerHistory.finishedBuilding(endTime);
 163     }
 164
 165     @Override
 166     public FileInputStream supplyAttributeTreeReader() {
 167         return fInnerHistory.supplyAttributeTreeReader();
 168     }
 169
 170     @Override
 171     public File supplyAttributeTreeWriterFile() {
 172         return fInnerHistory.supplyAttributeTreeWriterFile();
 173     }
 174
 175     @Override
 176     public long supplyAttributeTreeWriterFilePosition() {
 177         return fInnerHistory.supplyAttributeTreeWriterFilePosition();
 178     }
 179
 180     @Override
 181     public void removeFiles() {
 182         fInnerHistory.removeFiles();
 183     }
 184
 185     @Override
 186     public void dispose() {
 187         fPartialInput.dispose();
 188         fPartialSS.dispose();
 189         fInnerHistory.dispose();
 190     }
 191
 192     @Override
 193     public void doQuery(List<ITmfStateInterval> currentStateInfo, long t)
 194             throws TimeRangeException, StateSystemDisposedException {
 195         /* Wait for required steps to be done */
 196         waitForCheckpoints();
 197         fPartialSS.getUpstreamSS().waitUntilBuilt();
 198
 199         if (!checkValidTime(t)) {
 200             throw new TimeRangeException();
 201         }
 202
 203         /* Reload the previous checkpoint */
 204         long checkpointTime = fCheckpoints.floorKey(t);
 205         fInnerHistory.doQuery(currentStateInfo, checkpointTime);
 206
 207         /*
 208          * Set the initial contents of the partial state system (which is the
 209          * contents of the query at the checkpoint).
 210          */
 211         fPartialSS.takeQueryLock();
 212         fPartialSS.replaceOngoingState(currentStateInfo);
 213
 214         /* Send an event request to update the state system to the target time. */
 215         TmfTimeRange range = new TmfTimeRange(
 216                 /*
 217                  * The state at the checkpoint already includes any state change
 218                  * caused by the event(s) happening exactly at 'checkpointTime',
 219                  * if any. We must not include those events in the query.
 220                  */
 221                 new TmfTimestamp(checkpointTime + 1, ITmfTimestamp.NANOSECOND_SCALE),
 222                 new TmfTimestamp(t, ITmfTimestamp.NANOSECOND_SCALE));
 223         ITmfEventRequest request = new PartialStateSystemRequest(fPartialInput, range);
 224         fPartialInput.getTrace().sendRequest(request);
 225
 226         try {
 227             request.waitForCompletion();
 228         } catch (InterruptedException e) {
 229             e.printStackTrace();
 230         }
 231
 232         /*
 233          * Now the partial state system should have the ongoing time we are
 234          * looking for. However, the method expects a List of *state intervals*,
 235          * not state values, so we'll create intervals with a dummy end time.
 236          */
 237         try {
 238             for (int i = 0; i < currentStateInfo.size(); i++) {
 239                 long start = 0;
 240                 ITmfStateValue val = null;
 241                 start = ((ITmfStateSystem) fPartialSS).getOngoingStartTime(i);
 242                 val = ((ITmfStateSystem) fPartialSS).queryOngoingState(i);
 243
 244                 ITmfStateInterval interval = new TmfStateInterval(start, t, i, val);
 245                 currentStateInfo.set(i, interval);
 246             }
 247         } catch (AttributeNotFoundException e) {
 248             /* Should not happen, we iterate over existing values. */
 249             e.printStackTrace();
 250         }
 251
 252         fPartialSS.releaseQueryLock();
 253     }
 254
 255     /**
 256      * Single queries are not supported in partial histories. To get the same
 257      * result you can do a full query, then call fullState.get(attribute).
 258      */
 259     @Override
 260     public ITmfStateInterval doSingularQuery(long t, int attributeQuark) {
 261         throw new UnsupportedOperationException();
 262     }
 263
 264     private boolean checkValidTime(long t) {
 265         return (t >= getStartTime() && t <= getEndTime());
 266     }
 267
 268     @Override
 269     public void debugPrint(PrintWriter writer) {
 270         // TODO Auto-generated method stub
 271     }
 272
 273     private void waitForCheckpoints() {
 274         try {
 275             fCheckpointsReady.await();
 276         } catch (InterruptedException e) {
 277             e.printStackTrace();
 278         }
 279     }
 280
 281     // ------------------------------------------------------------------------
 282     // Event requests types
 283     // ------------------------------------------------------------------------
 284
 285     private class CheckpointsRequest extends TmfEventRequest {
 286         private final ITmfTrace trace;
 287         private final Map<Long, Long> checkpts;
 288         private long eventCount;
 289         private long lastCheckpointAt;
 290
 291         public CheckpointsRequest(ITmfStateProvider input, Map<Long, Long> checkpoints) {
 292             super(ITmfEvent.class,
 293                     TmfTimeRange.ETERNITY,
 294                     0,
 295                     ITmfEventRequest.ALL_DATA,
 296                     ITmfEventRequest.ExecutionType.FOREGROUND);
 297             checkpoints.clear();
 298             this.trace = input.getTrace();
 299             this.checkpts = checkpoints;
 300             eventCount = 0;
 301             lastCheckpointAt = 0;
 302
 303             /* Insert a checkpoint at the start of the trace */
 304             checkpoints.put(input.getStartTime(), 0L);
 305         }
 306
 307         @Override
 308         public void handleData(final ITmfEvent event) {
 309             super.handleData(event);
 310             if (event.getTrace() == trace) {
 311                 eventCount++;
 312
 313                 /* Check if we need to register a new checkpoint */
 314                 if (eventCount >= lastCheckpointAt + fGranularity) {
 315                     checkpts.put(event.getTimestamp().getValue(), eventCount);
 316                     lastCheckpointAt = eventCount;
 317                 }
 318             }
 319         }
 320
 321         @Override
 322         public void handleCompleted() {
 323             super.handleCompleted();
 324             fCheckpointsReady.countDown();
 325         }
 326     }
 327
 328     private class PartialStateSystemRequest extends TmfEventRequest {
 329         private final ITmfStateProvider sci;
 330         private final ITmfTrace trace;
 331
 332         PartialStateSystemRequest(ITmfStateProvider sci, TmfTimeRange range) {
 333             super(ITmfEvent.class,
 334                     range,
 335                     0,
 336                     ITmfEventRequest.ALL_DATA,
 337                     ITmfEventRequest.ExecutionType.BACKGROUND);
 338             this.sci = sci;
 339             this.trace = sci.getTrace();
 340         }
 341
 342         @Override
 343         public void handleData(final ITmfEvent event) {
 344             super.handleData(event);
 345             if (event.getTrace() == trace) {
 346                 sci.processEvent(event);
 347             }
 348         }
 349
 350         @Override
 351         public void handleCompleted() {
 352             /*
 353              * If we're using a threaded state provider, we need to make sure
 354              * all events have been handled by the state system before doing
 355              * queries on it.
 356              */
 357             if (fPartialInput instanceof AbstractTmfStateProvider) {
 358                 ((AbstractTmfStateProvider) fPartialInput).waitForEmptyQueue();
 359             }
 360             super.handleCompleted();
 361         }
 362
 363     }
 364 }