1 /*******************************************************************************
2 * Copyright (c) 2009, 2010 Ericsson
4 * All rights reserved. This program and the accompanying materials are
5 * made available under the terms of the Eclipse Public License v1.0 which
6 * accompanies this distribution, and is available at
7 * http://www.eclipse.org/legal/epl-v10.html
10 * Francois Chouinard - Initial API and implementation
11 *******************************************************************************/
13 package org
.eclipse
.linuxtools
.tmf
.experiment
;
15 import java
.util
.Collections
;
16 import java
.util
.Vector
;
18 import org
.eclipse
.core
.runtime
.IProgressMonitor
;
19 import org
.eclipse
.core
.runtime
.IStatus
;
20 import org
.eclipse
.core
.runtime
.Status
;
21 import org
.eclipse
.core
.runtime
.jobs
.Job
;
22 import org
.eclipse
.linuxtools
.tmf
.component
.TmfEventProvider
;
23 import org
.eclipse
.linuxtools
.tmf
.event
.TmfEvent
;
24 import org
.eclipse
.linuxtools
.tmf
.event
.TmfTimeRange
;
25 import org
.eclipse
.linuxtools
.tmf
.event
.TmfTimestamp
;
26 import org
.eclipse
.linuxtools
.tmf
.request
.ITmfDataRequest
;
27 import org
.eclipse
.linuxtools
.tmf
.request
.ITmfEventRequest
;
28 import org
.eclipse
.linuxtools
.tmf
.signal
.TmfExperimentSelectedSignal
;
29 import org
.eclipse
.linuxtools
.tmf
.signal
.TmfExperimentUpdatedSignal
;
30 import org
.eclipse
.linuxtools
.tmf
.signal
.TmfRangeSynchSignal
;
31 import org
.eclipse
.linuxtools
.tmf
.signal
.TmfSignalHandler
;
32 import org
.eclipse
.linuxtools
.tmf
.signal
.TmfTraceUpdatedSignal
;
33 import org
.eclipse
.linuxtools
.tmf
.trace
.ITmfContext
;
34 import org
.eclipse
.linuxtools
.tmf
.trace
.ITmfLocation
;
35 import org
.eclipse
.linuxtools
.tmf
.trace
.ITmfTrace
;
36 import org
.eclipse
.linuxtools
.tmf
.trace
.TmfCheckpoint
;
37 import org
.eclipse
.linuxtools
.tmf
.trace
.TmfContext
;
40 * <b><u>TmfExperiment</u></b>
42 * TmfExperiment presents a time-ordered, unified view of a set of TmfTraces
43 * that are part of a tracing experiment.
46 public class TmfExperiment
<T
extends TmfEvent
> extends TmfEventProvider
<T
> implements ITmfTrace
{
48 // ------------------------------------------------------------------------
50 // ------------------------------------------------------------------------
52 // The currently selected experiment
53 private static TmfExperiment
<?
> fCurrentExperiment
= null;
55 // The set of traces that constitute the experiment
56 private ITmfTrace
[] fTraces
;
58 // The total number of events
59 private long fNbEvents
;
61 // The experiment time range
62 private TmfTimeRange fTimeRange
;
64 // The experiment reference timestamp (default: BigBang)
65 private TmfTimestamp fEpoch
;
67 // The experiment index
68 private Vector
<TmfCheckpoint
> fCheckpoints
= new Vector
<TmfCheckpoint
>();
70 // ------------------------------------------------------------------------
72 // ------------------------------------------------------------------------
79 * @param indexPageSize
81 public TmfExperiment(Class
<T
> type
, String id
, ITmfTrace
[] traces
, TmfTimestamp epoch
, int indexPageSize
) {
86 fIndexPageSize
= indexPageSize
;
97 public TmfExperiment(Class
<T
> type
, String id
, ITmfTrace
[] traces
) {
98 this(type
, id
, traces
, TmfTimestamp
.Zero
, DEFAULT_INDEX_PAGE_SIZE
);
105 * @param indexPageSize
107 public TmfExperiment(Class
<T
> type
, String id
, ITmfTrace
[] traces
, int indexPageSize
) {
108 this(type
, id
, traces
, TmfTimestamp
.Zero
, indexPageSize
);
112 public TmfExperiment(TmfExperiment
<T
> other
) {
113 super(other
.getName(), other
.fType
);
115 fEpoch
= other
.fEpoch
;
116 fIndexPageSize
= other
.fIndexPageSize
;
118 fTraces
= new ITmfTrace
[other
.fTraces
.length
];
119 for (int trace
= 0; trace
< other
.fTraces
.length
; trace
++) {
120 fTraces
[trace
] = other
.fTraces
[trace
].createTraceCopy();
123 fNbEvents
= other
.fNbEvents
;
124 fTimeRange
= other
.fTimeRange
;
127 public TmfExperiment
<T
> createTraceCopy() {
128 return new TmfExperiment
<T
>(this);
132 * Clears the experiment
135 public void dispose() {
136 for (ITmfTrace trace
: fTraces
) {
140 fCheckpoints
.clear();
144 private static void setCurrentExperiment(TmfExperiment
<?
> experiment
) {
145 fCurrentExperiment
= experiment
;
148 // ------------------------------------------------------------------------
150 // ------------------------------------------------------------------------
152 public String
getPath() {
156 public long getNbEvents() {
160 public int getCacheSize() {
161 return fIndexPageSize
;
164 public TmfTimeRange
getTimeRange() {
168 public TmfTimestamp
getStartTime() {
169 return fTimeRange
.getStartTime();
172 public TmfTimestamp
getEndTime() {
173 return fTimeRange
.getEndTime();
176 public Vector
<TmfCheckpoint
> getCheckpoints() {
180 // ------------------------------------------------------------------------
182 // ------------------------------------------------------------------------
184 public static TmfExperiment
<?
> getCurrentExperiment() {
185 return fCurrentExperiment
;
188 public TmfTimestamp
getEpoch() {
192 public ITmfTrace
[] getTraces() {
197 * Returns the rank of the first event with the requested timestamp.
198 * If none, returns the index of the next event (if any).
203 public long getRank(TmfTimestamp timestamp
) {
204 TmfExperimentContext context
= seekEvent(timestamp
);
205 return context
.getRank();
209 * Returns the timestamp of the event at the requested index.
210 * If none, returns null.
215 public TmfTimestamp
getTimestamp(int index
) {
216 TmfExperimentContext context
= seekEvent(index
);
217 TmfEvent event
= getNextEvent(context
);
218 return (event
!= null) ? event
.getTimestamp() : null;
221 // ------------------------------------------------------------------------
223 // ------------------------------------------------------------------------
226 * Update the total number of events
228 private void updateNbEvents() {
230 for (ITmfTrace trace
: fTraces
) {
231 nbEvents
+= trace
.getNbEvents();
233 fNbEvents
= nbEvents
;
237 * Update the global time range
239 private void updateTimeRange() {
240 TmfTimestamp startTime
= fTimeRange
!= null ? fTimeRange
.getStartTime() : TmfTimestamp
.BigCrunch
;
241 TmfTimestamp endTime
= fTimeRange
!= null ? fTimeRange
.getEndTime() : TmfTimestamp
.BigBang
;
243 for (ITmfTrace trace
: fTraces
) {
244 if (trace
.getNbEvents() > 0) {
245 TmfTimestamp traceStartTime
= trace
.getStartTime();
246 if (traceStartTime
.compareTo(startTime
, true) < 0)
247 startTime
= traceStartTime
;
249 TmfTimestamp traceEndTime
= trace
.getEndTime();
250 if (traceEndTime
.compareTo(endTime
, true) > 0)
251 endTime
= traceEndTime
;
254 fTimeRange
= new TmfTimeRange(startTime
, endTime
);
257 // ------------------------------------------------------------------------
259 // ------------------------------------------------------------------------
262 public ITmfContext
armRequest(ITmfDataRequest
<T
> request
) {
263 TmfTimestamp timestamp
= (request
instanceof ITmfEventRequest
<?
>) ?
264 ((ITmfEventRequest
<T
>) request
).getRange().getStartTime() : null;
265 TmfExperimentContext context
= (timestamp
!= null) ?
266 seekEvent(timestamp
) : seekEvent(request
.getIndex());
270 @SuppressWarnings("unchecked")
272 public T
getNext(ITmfContext context
) {
273 if (context
instanceof TmfExperimentContext
) {
274 return (T
) getNextEvent((TmfExperimentContext
) context
);
279 // ------------------------------------------------------------------------
280 // ITmfTrace trace positioning
281 // ------------------------------------------------------------------------
283 // Returns a brand new context based on the location provided
284 // and initializes the event queues
285 public TmfExperimentContext
seekLocation(ITmfLocation
<?
> location
) {
287 // Validate the location
288 if (location
!= null && !(location
instanceof TmfExperimentLocation
)) {
289 return null; // Throw an exception?
292 // Instantiate the location
293 TmfExperimentLocation expLocation
= (location
== null)
294 ?
new TmfExperimentLocation(new ITmfLocation
<?
>[fTraces
.length
], new long[fTraces
.length
])
295 : (TmfExperimentLocation
) location
.clone();
297 // Create and populate the context's traces contexts
298 TmfExperimentContext context
= new TmfExperimentContext(fTraces
, new TmfContext
[fTraces
.length
]);
300 for (int i
= 0; i
< fTraces
.length
; i
++) {
301 // Get the relevant trace attributes
302 ITmfLocation
<?
> traceLocation
= expLocation
.getLocation()[i
];
303 long traceRank
= expLocation
.getRanks()[i
];
305 // Set the corresponding sub-context
306 context
.getContexts()[i
] = fTraces
[i
].seekLocation(traceLocation
);
307 context
.getContexts()[i
].setRank(traceRank
);
310 // Set the trace location and read the corresponding event
311 expLocation
.getLocation()[i
] = context
.getContexts()[i
].getLocation();
312 context
.getEvents()[i
] = fTraces
[i
].getNextEvent(context
.getContexts()[i
]);
316 context
.setLocation(expLocation
);
317 context
.setRank(rank
);
318 context
.setLastTrace(TmfExperimentContext
.NO_TRACE
);
323 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(org.eclipse.linuxtools.tmf.event.TmfTimestamp)
325 public TmfExperimentContext
seekEvent(TmfTimestamp timestamp
) {
327 if (timestamp
== null) {
328 timestamp
= TmfTimestamp
.BigBang
;
331 // First, find the right checkpoint
332 int index
= Collections
.binarySearch(fCheckpoints
, new TmfCheckpoint(timestamp
, null));
334 // In the very likely case that the checkpoint was not found, bsearch
335 // returns its negated would-be location (not an offset...). From that
336 // index, we can then position the stream and get the event.
338 index
= Math
.max(0, -(index
+ 2));
341 // Position the experiment at the checkpoint
342 ITmfLocation
<?
> location
;
343 synchronized (fCheckpoints
) {
344 if (fCheckpoints
.size() > 0) {
345 if (index
>= fCheckpoints
.size()) {
346 index
= fCheckpoints
.size() - 1;
348 location
= fCheckpoints
.elementAt(index
).getLocation();
355 TmfExperimentContext context
= seekLocation(location
);
356 context
.setRank((long) index
* fIndexPageSize
);
358 // And locate the event
359 TmfExperimentContext nextEventContext
= new TmfExperimentContext(context
);
360 TmfEvent event
= getNextEvent(nextEventContext
);
361 while (event
!= null && event
.getTimestamp().compareTo(timestamp
, false) < 0) {
362 context
= new TmfExperimentContext(nextEventContext
);
363 event
= getNextEvent(nextEventContext
);
365 context
.setLastTrace(TmfExperimentContext
.NO_TRACE
);
371 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(long)
373 public TmfExperimentContext
seekEvent(long rank
) {
375 // Position the stream at the previous checkpoint
376 int index
= (int) rank
/ fIndexPageSize
;
377 ITmfLocation
<?
> location
;
378 synchronized (fCheckpoints
) {
379 if (fCheckpoints
.size() == 0) {
383 if (index
>= fCheckpoints
.size()) {
384 index
= fCheckpoints
.size() - 1;
386 location
= fCheckpoints
.elementAt(index
).getLocation();
390 TmfExperimentContext context
= seekLocation(location
);
391 long pos
= (long) index
* fIndexPageSize
;
392 context
.setRank(pos
);
394 // And locate the event
395 TmfExperimentContext nextEventContext
= new TmfExperimentContext(context
);
396 TmfEvent event
= getNextEvent(nextEventContext
);
397 while (event
!= null && pos
++ < rank
) {
398 event
= getNextEvent(nextEventContext
);
399 context
= new TmfExperimentContext(nextEventContext
);
400 if (event
!= null) context
.updateRank(-1);
402 context
.setLastTrace(TmfExperimentContext
.NO_TRACE
);
408 * Scan the next events from all traces and return the next one
409 * in chronological order.
414 public synchronized TmfEvent
getNextEvent(TmfContext context
) {
416 // Validate the context
417 if (!(context
instanceof TmfExperimentContext
)) {
418 return null; // Throw an exception?
421 TmfExperimentContext expContext
= (TmfExperimentContext
) context
;
423 // If an event was consumed previously, get the next one from that trace
424 int lastTrace
= expContext
.getLastTrace();
425 if (lastTrace
!= TmfExperimentContext
.NO_TRACE
) {
426 TmfContext traceContext
= expContext
.getContexts()[lastTrace
];
427 expContext
.getEvents()[lastTrace
] = expContext
.getTraces()[lastTrace
].getNextEvent(traceContext
);
430 // Scan the candidate events and identify the "next" trace to read from
431 int trace
= TmfExperimentContext
.NO_TRACE
;
432 TmfTimestamp timestamp
= TmfTimestamp
.BigCrunch
;
433 for (int i
= 0; i
< expContext
.getTraces().length
; i
++) {
434 TmfEvent event
= expContext
.getEvents()[i
];
435 if (event
!= null && event
.getTimestamp() != null) {
436 TmfTimestamp otherTS
= event
.getTimestamp();
437 if (otherTS
.compareTo(timestamp
, true) < 0) {
444 // Update the experiment context and set the "next" event
445 TmfEvent event
= null;
447 expContext
.setLastTrace(trace
);
448 expContext
.updateRank(1);
449 TmfExperimentLocation expLocation
= (TmfExperimentLocation
) expContext
.getLocation();
450 TmfContext traceContext
= expContext
.getContexts()[trace
];
451 expLocation
.getLocation()[trace
] = traceContext
.getLocation().clone();
452 expLocation
.getRanks()[trace
] = traceContext
.getRank();
453 event
= expContext
.getEvents()[trace
];
460 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#parseEvent(org.eclipse.linuxtools.tmf.trace.TmfContext)
462 public TmfEvent
parseEvent(TmfContext context
) {
464 if (context
instanceof TmfExperimentContext
) {
465 TmfExperimentContext expContext
= (TmfExperimentContext
) context
;
466 int lastTrace
= expContext
.getLastTrace();
467 if (lastTrace
!= -1) {
468 TmfContext traceContext
= expContext
.getContexts()[lastTrace
];
469 expContext
.getEvents()[lastTrace
] = expContext
.getTraces()[lastTrace
].getNextEvent(traceContext
);
470 expContext
.updateRank(1);
471 TmfExperimentLocation expLocation
= (TmfExperimentLocation
) expContext
.getLocation();
472 expLocation
.getLocation()[lastTrace
] = traceContext
.getLocation().clone();
476 TmfTimestamp timestamp
= TmfTimestamp
.BigCrunch
;
477 for (int i
= 0; i
< expContext
.getTraces().length
; i
++) {
478 if (expContext
.getEvents()[i
] != null) {
479 if (expContext
.getEvents()[i
].getTimestamp() != null) {
480 TmfTimestamp otherTS
= expContext
.getEvents()[i
].getTimestamp();
481 if (otherTS
.compareTo(timestamp
, true) < 0) {
489 expContext
.setLastTrace(TmfExperimentContext
.NO_TRACE
);
490 return expContext
.getEvents()[trace
];
498 * @see java.lang.Object#toString()
501 public String
toString() {
502 return "[TmfExperiment (" + getName() + ")]";
505 // ------------------------------------------------------------------------
507 // ------------------------------------------------------------------------
510 * The experiment holds the globally ordered events of its set of traces.
511 * It is expected to provide access to each individual event by index i.e.
512 * it must be possible to request the Nth event of the experiment.
514 * The purpose of the index is to keep the information needed to rapidly
515 * restore the traces contexts at regular intervals (every INDEX_PAGE_SIZE
519 // The index page size
520 private static final int DEFAULT_INDEX_PAGE_SIZE
= 1000;
521 private final int fIndexPageSize
;
523 // Indicates that an indexing job is already running
524 private Boolean fIndexing
= false;
525 private Boolean fIndexed
= false;
528 private IndexingJob job
;
533 * Creates the experiment index.
535 public void indexExperiment(boolean waitForCompletion
) {
538 if (fIndexed
|| fIndexing
) {
539 // An indexing job is already running but a new request came
540 // in (probably due to a change in the trace set). The index
541 // being currently built is therefore already invalid.
542 // TODO: Cancel and restart the job
543 // TODO: Add support for dynamically adding/removing traces
549 job
= new IndexingJob(new TmfExperiment
<T
>(this));
552 if (waitForCompletion
) {
555 } catch (InterruptedException e
) {
561 private class IndexingJob
extends Job
{
563 private final String experimentId
;
564 private final ITmfTrace
[] traces
;
565 private Vector
<TmfCheckpoint
> checkpoints
;
567 public IndexingJob(TmfExperiment
<T
> experiment
) {
568 super(experiment
.getName());
569 experimentId
= experiment
.getName();
570 traces
= experiment
.getTraces();
574 * @see org.eclipse.core.runtime.jobs.Job#run(org.eclipse.core.runtime.IProgressMonitor)
577 protected IStatus
run(IProgressMonitor monitor
) {
580 if (traces
.length
== 0) {
582 return Status
.OK_STATUS
;
585 monitor
.beginTask("Indexing " + experimentId
, IProgressMonitor
.UNKNOWN
);
588 TmfTimestamp startTime
= null;
589 TmfTimestamp lastTime
= null;
592 checkpoints
= new Vector
<TmfCheckpoint
>();
595 // Position the trace at the beginning
596 TmfExperimentContext context
= seekLocation(null);
597 TmfExperimentLocation location
= (TmfExperimentLocation
) context
.getLocation().clone();
599 // Get the first event
600 TmfEvent event
= getNextEvent(context
);
602 startTime
= new TmfTimestamp(event
.getTimestamp());
605 // Index the experiment
606 while (event
!= null) {
607 lastTime
= event
.getTimestamp();
608 if ((nbEvents
++ % fIndexPageSize
) == 0) {
609 checkpoints
.add(new TmfCheckpoint(lastTime
, location
));
613 // Check monitor *after* fCheckpoints has been updated
614 if (monitor
.isCanceled()) {
616 return Status
.CANCEL_STATUS
;
620 // We will need the contexts at the next iteration
621 if ((nbEvents
% fIndexPageSize
) == 0) {
622 location
= (TmfExperimentLocation
) context
.getLocation().clone();
625 event
= getNextEvent(context
);
631 fCheckpoints
= checkpoints
;
632 fNbEvents
= nbEvents
;
633 fTimeRange
= new TmfTimeRange(startTime
, lastTime
);
637 // notifyListeners(fTimeRange);
641 return Status
.OK_STATUS
;
645 protected void notifyListeners(TmfTimeRange range
) {
646 broadcast(new TmfRangeSynchSignal(this, range
, range
.getStartTime()));
649 // ------------------------------------------------------------------------
651 // ------------------------------------------------------------------------
654 public void experimentSelected(TmfExperimentSelectedSignal
<T
> signal
) {
655 TmfExperiment
<?
> experiment
= signal
.getExperiment();
656 if (experiment
== this) {
657 setCurrentExperiment(experiment
);
662 // if (signal.getExperiment() == this) {
663 // indexExperiment(true);
668 public void experimentUpdated(TmfExperimentUpdatedSignal signal
) {
669 // indexExperiment(true);
673 public void traceUpdated(TmfTraceUpdatedSignal signal
) {
674 // TODO: Incremental index update
679 broadcast(new TmfExperimentUpdatedSignal(this, this, signal
.getTrace()));