org.eclipse.linuxtools.tmf.core/src/org/eclipse/linuxtools/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.linuxtools.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.linuxtools.statesystem.core.ITmfStateSystem;
  24 import org.eclipse.linuxtools.statesystem.core.backend.IStateHistoryBackend;
  25 import org.eclipse.linuxtools.statesystem.core.exceptions.AttributeNotFoundException;
  26 import org.eclipse.linuxtools.statesystem.core.exceptions.StateSystemDisposedException;
  27 import org.eclipse.linuxtools.statesystem.core.exceptions.TimeRangeException;
  28 import org.eclipse.linuxtools.statesystem.core.interval.ITmfStateInterval;
  29 import org.eclipse.linuxtools.statesystem.core.interval.TmfStateInterval;
  30 import org.eclipse.linuxtools.statesystem.core.statevalue.ITmfStateValue;
  31 import org.eclipse.linuxtools.tmf.core.event.ITmfEvent;
  32 import org.eclipse.linuxtools.tmf.core.request.ITmfEventRequest;
  33 import org.eclipse.linuxtools.tmf.core.request.TmfEventRequest;
  34 import org.eclipse.linuxtools.tmf.core.statesystem.AbstractTmfStateProvider;
  35 import org.eclipse.linuxtools.tmf.core.statesystem.ITmfStateProvider;
  36 import org.eclipse.linuxtools.tmf.core.timestamp.ITmfTimestamp;
  37 import org.eclipse.linuxtools.tmf.core.timestamp.TmfTimeRange;
  38 import org.eclipse.linuxtools.tmf.core.timestamp.TmfTimestamp;
  39 import org.eclipse.linuxtools.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 partialInput;
  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 partialSS;
  70
  71     /** Reference to the "real" state history that is used for storage */
  72     private final @NonNull IStateHistoryBackend innerHistory;
  73
  74     /** Checkpoints map, <Timestamp, Rank in the trace> */
  75     private final @NonNull TreeMap<Long, Long> checkpoints = new TreeMap<>();
  76
  77     /** Latch tracking if the initial checkpoint registration is done */
  78     private final @NonNull CountDownLatch checkpointsReady = new CountDownLatch(1);
  79
  80     private final long granularity;
  81
  82     private long latestTime;
  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         this.partialInput = partialInput;
 111         this.partialSS = pss;
 112
 113         this.innerHistory = realBackend;
 114         this.granularity = granularity;
 115
 116         latestTime = startTime;
 117
 118         registerCheckpoints();
 119     }
 120
 121     private void registerCheckpoints() {
 122         ITmfEventRequest request = new CheckpointsRequest(partialInput, checkpoints);
 123         partialInput.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 innerHistory.getStartTime();
 130     }
 131
 132     @Override
 133     public long getEndTime() {
 134         return latestTime;
 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 > latestTime) {
 144             latestTime = 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 <= checkpoints.floorKey(stateEndTime)) {
 156             innerHistory.insertPastState(stateStartTime, stateEndTime, quark, value);
 157         }
 158     }
 159
 160     @Override
 161     public void finishedBuilding(long endTime) throws TimeRangeException {
 162         innerHistory.finishedBuilding(endTime);
 163     }
 164
 165     @Override
 166     public FileInputStream supplyAttributeTreeReader() {
 167         return innerHistory.supplyAttributeTreeReader();
 168     }
 169
 170     @Override
 171     public File supplyAttributeTreeWriterFile() {
 172         return innerHistory.supplyAttributeTreeWriterFile();
 173     }
 174
 175     @Override
 176     public long supplyAttributeTreeWriterFilePosition() {
 177         return innerHistory.supplyAttributeTreeWriterFilePosition();
 178     }
 179
 180     @Override
 181     public void removeFiles() {
 182         innerHistory.removeFiles();
 183     }
 184
 185     @Override
 186     public void dispose() {
 187         partialInput.dispose();
 188         partialSS.dispose();
 189         innerHistory.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         partialSS.getUpstreamSS().waitUntilBuilt();
 198
 199         if (!checkValidTime(t)) {
 200             throw new TimeRangeException();
 201         }
 202
 203         /* Reload the previous checkpoint */
 204         long checkpointTime = checkpoints.floorKey(t);
 205         innerHistory.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         partialSS.takeQueryLock();
 212         partialSS.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(partialInput, range);
 224         partialInput.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) partialSS).getOngoingStartTime(i);
 242                 val = ((ITmfStateSystem) partialSS).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         partialSS.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     @Override
 265     public boolean checkValidTime(long t) {
 266         return (t >= getStartTime() && t <= getEndTime());
 267     }
 268
 269     @Override
 270     public void debugPrint(PrintWriter writer) {
 271         // TODO Auto-generated method stub
 272     }
 273
 274     private void waitForCheckpoints() {
 275         try {
 276             checkpointsReady.await();
 277         } catch (InterruptedException e) {
 278             e.printStackTrace();
 279         }
 280     }
 281
 282     // ------------------------------------------------------------------------
 283     // Event requests types
 284     // ------------------------------------------------------------------------
 285
 286     private class CheckpointsRequest extends TmfEventRequest {
 287         private final ITmfTrace trace;
 288         private final Map<Long, Long> checkpts;
 289         private long eventCount;
 290         private long lastCheckpointAt;
 291
 292         public CheckpointsRequest(ITmfStateProvider input, Map<Long, Long> checkpoints) {
 293             super(input.getExpectedEventType(),
 294                     TmfTimeRange.ETERNITY,
 295                     0,
 296                     ITmfEventRequest.ALL_DATA,
 297                     ITmfEventRequest.ExecutionType.FOREGROUND);
 298             checkpoints.clear();
 299             this.trace = input.getTrace();
 300             this.checkpts = checkpoints;
 301             eventCount = 0;
 302             lastCheckpointAt = 0;
 303
 304             /* Insert a checkpoint at the start of the trace */
 305             checkpoints.put(input.getStartTime(), 0L);
 306         }
 307
 308         @Override
 309         public void handleData(final ITmfEvent event) {
 310             super.handleData(event);
 311             if (event.getTrace() == trace) {
 312                 eventCount++;
 313
 314                 /* Check if we need to register a new checkpoint */
 315                 if (eventCount >= lastCheckpointAt + granularity) {
 316                     checkpts.put(event.getTimestamp().getValue(), eventCount);
 317                     lastCheckpointAt = eventCount;
 318                 }
 319             }
 320         }
 321
 322         @Override
 323         public void handleCompleted() {
 324             super.handleCompleted();
 325             checkpointsReady.countDown();
 326         }
 327     }
 328
 329     private class PartialStateSystemRequest extends TmfEventRequest {
 330         private final ITmfStateProvider sci;
 331         private final ITmfTrace trace;
 332
 333         PartialStateSystemRequest(ITmfStateProvider sci, TmfTimeRange range) {
 334             super(sci.getExpectedEventType(),
 335                     range,
 336                     0,
 337                     ITmfEventRequest.ALL_DATA,
 338                     ITmfEventRequest.ExecutionType.BACKGROUND);
 339             this.sci = sci;
 340             this.trace = sci.getTrace();
 341         }
 342
 343         @Override
 344         public void handleData(final ITmfEvent event) {
 345             super.handleData(event);
 346             if (event.getTrace() == trace) {
 347                 sci.processEvent(event);
 348             }
 349         }
 350
 351         @Override
 352         public void handleCompleted() {
 353             /*
 354              * If we're using a threaded state provider, we need to make sure
 355              * all events have been handled by the state system before doing
 356              * queries on it.
 357              */
 358             if (partialInput instanceof AbstractTmfStateProvider) {
 359                 ((AbstractTmfStateProvider) partialInput).waitForEmptyQueue();
 360             }
 361             super.handleCompleted();
 362         }
 363
 364     }
 365 }