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

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