1 /*******************************************************************************
2 * Copyright (c) 2013 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
9 * Alexandre Montplaisir - Initial API and implementation
10 *******************************************************************************/
12 package org
.eclipse
.linuxtools
.internal
.tmf
.core
.statesystem
.backends
.partial
;
15 import java
.io
.FileInputStream
;
16 import java
.io
.PrintWriter
;
17 import java
.util
.List
;
19 import java
.util
.TreeMap
;
20 import java
.util
.concurrent
.CountDownLatch
;
22 import org
.eclipse
.linuxtools
.internal
.tmf
.core
.statesystem
.backends
.IStateHistoryBackend
;
23 import org
.eclipse
.linuxtools
.tmf
.core
.event
.ITmfEvent
;
24 import org
.eclipse
.linuxtools
.tmf
.core
.exceptions
.AttributeNotFoundException
;
25 import org
.eclipse
.linuxtools
.tmf
.core
.exceptions
.StateSystemDisposedException
;
26 import org
.eclipse
.linuxtools
.tmf
.core
.exceptions
.TimeRangeException
;
27 import org
.eclipse
.linuxtools
.tmf
.core
.interval
.ITmfStateInterval
;
28 import org
.eclipse
.linuxtools
.tmf
.core
.interval
.TmfStateInterval
;
29 import org
.eclipse
.linuxtools
.tmf
.core
.request
.ITmfDataRequest
;
30 import org
.eclipse
.linuxtools
.tmf
.core
.request
.ITmfEventRequest
;
31 import org
.eclipse
.linuxtools
.tmf
.core
.request
.TmfDataRequest
;
32 import org
.eclipse
.linuxtools
.tmf
.core
.request
.TmfEventRequest
;
33 import org
.eclipse
.linuxtools
.tmf
.core
.statesystem
.AbstractTmfStateProvider
;
34 import org
.eclipse
.linuxtools
.tmf
.core
.statesystem
.ITmfStateProvider
;
35 import org
.eclipse
.linuxtools
.tmf
.core
.statevalue
.ITmfStateValue
;
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
;
42 * Partial state history back-end.
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
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.
55 * @author Alexandre Montplaisir
57 public class PartialHistoryBackend
implements IStateHistoryBackend
{
60 * A partial history needs the state input plugin to re-generate state
61 * between checkpoints.
63 private final ITmfStateProvider partialInput
;
66 * Fake state system that is used for partially rebuilding the states (when
67 * going from a checkpoint to a target query timestamp).
69 private final PartialStateSystem partialSS
;
71 /** Reference to the "real" state history that is used for storage */
72 private final IStateHistoryBackend innerHistory
;
74 /** Checkpoints map, <Timestamp, Rank in the trace> */
75 private final TreeMap
<Long
, Long
> checkpoints
=
76 new TreeMap
<Long
, Long
>();
78 /** Latch tracking if the initial checkpoint registration is done */
79 private final CountDownLatch checkpointsReady
= new CountDownLatch(1);
81 private final long granularity
;
83 private long latestTime
;
89 * The state change input object that was used to build the
90 * upstream state system. This partial history will make its own
91 * copy (since they have different targets).
93 * The partial history's inner state system. It should already be
94 * assigned to partialInput.
96 * The real state history back-end to use. It's supposed to be
97 * modular, so it should be able to be of any type.
99 * Configuration parameter indicating how many trace events there
100 * should be between each checkpoint
102 public PartialHistoryBackend(ITmfStateProvider partialInput
, PartialStateSystem pss
,
103 IStateHistoryBackend realBackend
, long granularity
) {
104 if (granularity
<= 0 || partialInput
== null || pss
== null ||
105 partialInput
.getAssignedStateSystem() != pss
) {
106 throw new IllegalArgumentException();
109 final long startTime
= realBackend
.getStartTime();
111 this.partialInput
= partialInput
;
112 this.partialSS
= pss
;
114 this.innerHistory
= realBackend
;
115 this.granularity
= granularity
;
117 latestTime
= startTime
;
119 registerCheckpoints();
122 private void registerCheckpoints() {
123 ITmfEventRequest request
= new CheckpointsRequest(partialInput
, checkpoints
);
124 partialInput
.getTrace().sendRequest(request
);
125 /* The request will countDown the checkpoints latch once it's finished */
129 public long getStartTime() {
130 return innerHistory
.getStartTime();
134 public long getEndTime() {
139 public void insertPastState(long stateStartTime
, long stateEndTime
,
140 int quark
, ITmfStateValue value
) throws TimeRangeException
{
141 waitForCheckpoints();
143 /* Update the latest time */
144 if (stateEndTime
> latestTime
) {
145 latestTime
= stateEndTime
;
149 * Check if the interval intersects the previous checkpoint. If so,
150 * insert it in the real history back-end.
152 * FIXME since intervals are inserted in order of rank, we could avoid
153 * doing a map lookup every time here (just compare with the known
156 if (stateStartTime
<= checkpoints
.floorKey(stateEndTime
)) {
157 innerHistory
.insertPastState(stateStartTime
, stateEndTime
, quark
, value
);
162 public void finishedBuilding(long endTime
) throws TimeRangeException
{
163 innerHistory
.finishedBuilding(endTime
);
167 public FileInputStream
supplyAttributeTreeReader() {
168 return innerHistory
.supplyAttributeTreeReader();
172 public File
supplyAttributeTreeWriterFile() {
173 return innerHistory
.supplyAttributeTreeWriterFile();
177 public long supplyAttributeTreeWriterFilePosition() {
178 return innerHistory
.supplyAttributeTreeWriterFilePosition();
182 public void removeFiles() {
183 innerHistory
.removeFiles();
187 public void dispose() {
188 innerHistory
.dispose();
192 public void doQuery(List
<ITmfStateInterval
> currentStateInfo
, long t
)
193 throws TimeRangeException
, StateSystemDisposedException
{
194 /* Wait for required steps to be done */
195 waitForCheckpoints();
196 partialSS
.getUpstreamSS().waitUntilBuilt();
198 if (!checkValidTime(t
)) {
199 throw new TimeRangeException();
202 /* Reload the previous checkpoint */
203 long checkpointTime
= checkpoints
.floorKey(t
);
204 innerHistory
.doQuery(currentStateInfo
, checkpointTime
);
207 * Set the initial contents of the partial state system (which is the
208 * contents of the query at the checkpoint).
210 partialSS
.takeQueryLock();
211 partialSS
.replaceOngoingState(currentStateInfo
);
213 /* Send an event request to update the state system to the target time. */
214 TmfTimeRange range
= new TmfTimeRange(
216 * The state at the checkpoint already includes any state change
217 * caused by the event(s) happening exactly at 'checkpointTime',
218 * if any. We must not include those events in the query.
220 new TmfTimestamp(checkpointTime
+ 1, ITmfTimestamp
.NANOSECOND_SCALE
),
221 new TmfTimestamp(t
, ITmfTimestamp
.NANOSECOND_SCALE
));
222 ITmfEventRequest request
= new PartialStateSystemRequest(partialInput
, range
);
223 partialInput
.getTrace().sendRequest(request
);
226 request
.waitForCompletion();
227 } catch (InterruptedException e
) {
232 * Now the partial state system should have the ongoing time we are
233 * looking for. However, the method expects a List of *state intervals*,
234 * not state values, so we'll create intervals with a dummy end time.
237 for (int i
= 0; i
< currentStateInfo
.size(); i
++) {
239 ITmfStateValue val
= null;
240 start
= partialSS
.getOngoingStartTime(i
);
241 val
= partialSS
.queryOngoingState(i
);
243 ITmfStateInterval interval
= new TmfStateInterval(start
, t
, i
, val
);
244 currentStateInfo
.set(i
, interval
);
246 } catch (AttributeNotFoundException e
) {
247 /* Should not happen, we iterate over existing values. */
251 partialSS
.releaseQueryLock();
255 * Single queries are not supported in partial histories. To get the same
256 * result you can do a full query, then call fullState.get(attribute).
259 public ITmfStateInterval
doSingularQuery(long t
, int attributeQuark
) {
260 throw new UnsupportedOperationException();
264 public boolean checkValidTime(long t
) {
265 return (t
>= getStartTime() && t
<= getEndTime());
269 public void debugPrint(PrintWriter writer
) {
270 // TODO Auto-generated method stub
273 private void waitForCheckpoints() {
275 checkpointsReady
.await();
276 } catch (InterruptedException e
) {
281 // ------------------------------------------------------------------------
282 // Event requests types
283 // ------------------------------------------------------------------------
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
;
291 public CheckpointsRequest(ITmfStateProvider input
, Map
<Long
, Long
> checkpoints
) {
292 super(input
.getExpectedEventType(),
293 TmfTimeRange
.ETERNITY
,
295 TmfDataRequest
.ALL_DATA
,
296 ITmfDataRequest
.ExecutionType
.BACKGROUND
);
298 this.trace
= input
.getTrace();
299 this.checkpts
= checkpoints
;
301 lastCheckpointAt
= 0;
303 /* Insert a checkpoint at the start of the trace */
304 checkpoints
.put(input
.getStartTime(), 0L);
308 public void handleData(final ITmfEvent event
) {
309 super.handleData(event
);
310 if (event
!= null && event
.getTrace() == trace
) {
313 /* Check if we need to register a new checkpoint */
314 if (eventCount
>= lastCheckpointAt
+ granularity
) {
315 checkpts
.put(event
.getTimestamp().getValue(), eventCount
);
316 lastCheckpointAt
= eventCount
;
322 public void handleCompleted() {
323 super.handleCompleted();
324 checkpointsReady
.countDown();
328 private class PartialStateSystemRequest
extends TmfEventRequest
{
329 private final ITmfStateProvider sci
;
330 private final ITmfTrace trace
;
332 PartialStateSystemRequest(ITmfStateProvider sci
, TmfTimeRange range
) {
333 super(sci
.getExpectedEventType(),
336 TmfDataRequest
.ALL_DATA
,
337 ITmfDataRequest
.ExecutionType
.BACKGROUND
);
339 this.trace
= sci
.getTrace();
343 public void handleData(final ITmfEvent event
) {
344 super.handleData(event
);
345 if (event
!= null && event
.getTrace() == trace
) {
346 sci
.processEvent(event
);
351 public void handleCompleted() {
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
357 if (partialInput
instanceof AbstractTmfStateProvider
) {
358 ((AbstractTmfStateProvider
) partialInput
).waitForEmptyQueue();
360 super.handleCompleted();