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

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