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
.trace
;
16 import java
.io
.FileNotFoundException
;
17 import java
.util
.Collections
;
18 import java
.util
.Vector
;
20 import org
.eclipse
.core
.runtime
.IProgressMonitor
;
21 import org
.eclipse
.core
.runtime
.IStatus
;
22 import org
.eclipse
.core
.runtime
.Status
;
23 import org
.eclipse
.core
.runtime
.jobs
.Job
;
24 import org
.eclipse
.linuxtools
.tmf
.component
.TmfEventProvider
;
25 import org
.eclipse
.linuxtools
.tmf
.event
.TmfEvent
;
26 import org
.eclipse
.linuxtools
.tmf
.event
.TmfTimeRange
;
27 import org
.eclipse
.linuxtools
.tmf
.event
.TmfTimestamp
;
28 import org
.eclipse
.linuxtools
.tmf
.request
.ITmfDataRequest
;
29 import org
.eclipse
.linuxtools
.tmf
.request
.ITmfDataRequest
.ExecutionType
;
30 import org
.eclipse
.linuxtools
.tmf
.request
.ITmfEventRequest
;
31 import org
.eclipse
.linuxtools
.tmf
.request
.TmfDataRequest
;
32 import org
.eclipse
.linuxtools
.tmf
.request
.TmfEventRequest
;
33 import org
.eclipse
.linuxtools
.tmf
.signal
.TmfTraceUpdatedSignal
;
36 * <b><u>TmfTrace</u></b>
38 * Abstract implementation of ITmfTrace. It should be sufficient to extend this
39 * class and provide implementation for <code>getCurrentLocation()</code> and
40 * <code>seekLocation()</code>, as well as a proper parser, to have a working
41 * concrete implementation.
43 * Note: The notion of event rank is still under heavy discussion. Although
44 * used by the Events View and probably useful in the general case, there
45 * is no easy way to implement it for LTTng (actually a strong case is being
46 * made that this is useless).
48 * That it is not supported by LTTng does by no mean indicate that it is not
49 * useful for (just about) every other tracing tool. Therefore, this class
50 * provides a minimal (and partial) implementation of rank. However, the current
51 * implementation should not be relied on in the general case.
53 * TODO: Add support for live streaming (notifications, incremental indexing, ...)
55 public abstract class TmfTrace
<T
extends TmfEvent
> extends TmfEventProvider
<T
> implements ITmfTrace
, Cloneable
{
57 // ------------------------------------------------------------------------
59 // ------------------------------------------------------------------------
61 // The default number of events to cache
62 // TODO: Make the DEFAULT_CACHE_SIZE a preference
63 public static final int DEFAULT_INDEX_PAGE_SIZE
= 50000;
65 // ------------------------------------------------------------------------
67 // ------------------------------------------------------------------------
70 private final String fPath
;
72 // The cache page size AND checkpoints interval
73 protected int fIndexPageSize
;
75 // The set of event stream checkpoints (for random access)
76 protected Vector
<TmfCheckpoint
> fCheckpoints
= new Vector
<TmfCheckpoint
>();
78 // The number of events collected
79 protected long fNbEvents
= 0;
81 // The time span of the event stream
82 private TmfTimestamp fStartTime
= TmfTimestamp
.BigCrunch
;
83 private TmfTimestamp fEndTime
= TmfTimestamp
.BigBang
;
85 // ------------------------------------------------------------------------
87 // ------------------------------------------------------------------------
91 * @throws FileNotFoundException
93 protected TmfTrace(String name
, Class
<T
> type
, String path
) throws FileNotFoundException
{
94 this(name
, type
, path
, DEFAULT_INDEX_PAGE_SIZE
, true);
100 * @throws FileNotFoundException
102 protected TmfTrace(String name
, Class
<T
> type
, String path
, int cacheSize
) throws FileNotFoundException
{
103 this(name
, type
, path
, cacheSize
, true);
109 * @throws FileNotFoundException
111 protected TmfTrace(String name
, Class
<T
> type
, String path
, boolean indexTrace
) throws FileNotFoundException
{
112 this(name
, type
, path
, DEFAULT_INDEX_PAGE_SIZE
, indexTrace
);
119 * @throws FileNotFoundException
121 protected TmfTrace(String name
, Class
<T
> type
, String path
, int cacheSize
, boolean indexTrace
) throws FileNotFoundException
{
123 int sep
= path
.lastIndexOf(File
.separator
);
124 String simpleName
= (sep
>= 0) ? path
.substring(sep
+ 1) : path
;
127 fIndexPageSize
= (cacheSize
> 0) ? cacheSize
: DEFAULT_INDEX_PAGE_SIZE
;
129 // indexTrace(false);
134 * @see java.lang.Object#clone()
136 @SuppressWarnings("unchecked")
138 public TmfTrace
<T
> clone() throws CloneNotSupportedException
{
139 TmfTrace
<T
> clone
= (TmfTrace
<T
>) super.clone();
140 clone
.fCheckpoints
= (Vector
<TmfCheckpoint
>) fCheckpoints
;
141 clone
.fStartTime
= new TmfTimestamp(fStartTime
);
142 clone
.fEndTime
= new TmfTimestamp(fEndTime
);
146 // ------------------------------------------------------------------------
148 // ------------------------------------------------------------------------
151 * @return the trace path
154 public String
getPath() {
159 * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getNbEvents()
162 public long getNbEvents() {
167 * @return the size of the cache
170 public int getCacheSize() {
171 return fIndexPageSize
;
175 * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getTimeRange()
178 public TmfTimeRange
getTimeRange() {
179 return new TmfTimeRange(fStartTime
, fEndTime
);
183 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getStartTime()
186 public TmfTimestamp
getStartTime() {
191 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getEndTime()
194 public TmfTimestamp
getEndTime() {
198 @SuppressWarnings("unchecked")
199 public Vector
<TmfCheckpoint
> getCheckpoints() {
200 return (Vector
<TmfCheckpoint
>) fCheckpoints
.clone();
204 * Returns the rank of the first event with the requested timestamp.
205 * If none, returns the index of the next event (if any).
211 public long getRank(TmfTimestamp timestamp
) {
212 TmfContext context
= seekEvent(timestamp
);
213 return context
.getRank();
216 // ------------------------------------------------------------------------
218 // ------------------------------------------------------------------------
220 protected void setTimeRange(TmfTimeRange range
) {
221 fStartTime
= range
.getStartTime();
222 fEndTime
= range
.getEndTime();
225 protected void setStartTime(TmfTimestamp startTime
) {
226 fStartTime
= startTime
;
229 protected void setEndTime(TmfTimestamp endTime
) {
233 // ------------------------------------------------------------------------
235 // ------------------------------------------------------------------------
238 public ITmfContext
armRequest(ITmfDataRequest
<T
> request
) {
239 if (request
instanceof ITmfEventRequest
<?
>) {
240 return seekEvent(((ITmfEventRequest
<T
>) request
).getRange().getStartTime());
242 return seekEvent(request
.getIndex());
246 * Return the next piece of data based on the context supplied. The context
247 * would typically be updated for the subsequent read.
252 @SuppressWarnings("unchecked")
254 public T
getNext(ITmfContext context
) {
255 if (context
instanceof TmfContext
) {
256 return (T
) getNextEvent((TmfContext
) context
);
261 // ------------------------------------------------------------------------
263 // ------------------------------------------------------------------------
266 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(org.eclipse.linuxtools.tmf.event.TmfTimestamp)
269 public TmfContext
seekEvent(TmfTimestamp timestamp
) {
271 if (timestamp
== null) {
272 timestamp
= TmfTimestamp
.BigBang
;
275 // First, find the right checkpoint
276 int index
= Collections
.binarySearch(fCheckpoints
, new TmfCheckpoint(timestamp
, null));
278 // In the very likely case that the checkpoint was not found, bsearch
279 // returns its negated would-be location (not an offset...). From that
280 // index, we can then position the stream and get the event.
282 index
= Math
.max(0, -(index
+ 2));
285 // Position the stream at the checkpoint
286 ITmfLocation
<?
> location
;
287 synchronized (fCheckpoints
) {
288 if (fCheckpoints
.size() > 0) {
289 if (index
>= fCheckpoints
.size()) {
290 index
= fCheckpoints
.size() - 1;
292 location
= fCheckpoints
.elementAt(index
).getLocation();
298 TmfContext context
= seekLocation(location
);
299 context
.setRank(index
* fIndexPageSize
);
301 // And locate the event
302 TmfContext nextEventContext
= context
.clone(); // Must use clone() to get the right subtype...
303 TmfEvent event
= getNextEvent(nextEventContext
);
304 while (event
!= null && event
.getTimestamp().compareTo(timestamp
, false) < 0) {
305 context
.setLocation(nextEventContext
.getLocation().clone());
306 context
.updateRank(1);
307 event
= getNextEvent(nextEventContext
);
314 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#seekEvent(int)
317 public TmfContext
seekEvent(long rank
) {
319 // Position the stream at the previous checkpoint
320 int index
= (int) rank
/ fIndexPageSize
;
321 ITmfLocation
<?
> location
;
322 synchronized (fCheckpoints
) {
323 if (fCheckpoints
.size() == 0) {
327 if (index
>= fCheckpoints
.size()) {
328 index
= fCheckpoints
.size() - 1;
330 location
= fCheckpoints
.elementAt(index
).getLocation();
334 TmfContext context
= seekLocation(location
);
335 long pos
= index
* fIndexPageSize
;
336 context
.setRank(pos
);
339 TmfEvent event
= getNextEvent(context
);
340 while (event
!= null && ++pos
< rank
) {
341 event
= getNextEvent(context
);
349 * @see org.eclipse.linuxtools.tmf.trace.ITmfTrace#getNextEvent(org.eclipse.linuxtools.tmf.trace.ITmfTrace.TraceContext)
352 public synchronized TmfEvent
getNextEvent(TmfContext context
) {
353 // parseEvent() does not update the context
354 TmfEvent event
= parseEvent(context
);
356 updateIndex(context
, context
.getRank(), event
.getTimestamp());
357 context
.setLocation(getCurrentLocation());
358 context
.updateRank(1);
364 protected synchronized void updateIndex(ITmfContext context
, long rank
, TmfTimestamp timestamp
) {
365 if (fStartTime
.compareTo(timestamp
, false) > 0) fStartTime
= timestamp
;
366 if (fEndTime
.compareTo(timestamp
, false) < 0) fEndTime
= timestamp
;
367 if (context
.isValidRank()) {
368 if (fNbEvents
<= rank
)
369 fNbEvents
= rank
+ 1;
370 // Build the index as we go along
371 if ((rank
% fIndexPageSize
) == 0) {
372 // Determine the table position
373 long position
= rank
/ fIndexPageSize
;
374 // Add new entry at proper location (if empty)
375 if (fCheckpoints
.size() == position
) {
376 ITmfLocation
<?
> location
= context
.getLocation().clone();
377 fCheckpoints
.add(new TmfCheckpoint(timestamp
.clone(), location
));
378 // System.out.println(getName() + "[" + (fCheckpoints.size() - 1) + "] " + timestamp + ", " + location.toString());
385 * Hook for "special" processing by the concrete class
386 * (called by getNextEvent())
390 protected void processEvent(TmfEvent event
) {
391 // Do nothing by default
395 * To be implemented by the concrete class
398 public abstract TmfContext
seekLocation(ITmfLocation
<?
> location
);
399 public abstract ITmfLocation
<?
> getCurrentLocation();
401 public abstract TmfEvent
parseEvent(TmfContext context
);
403 // ------------------------------------------------------------------------
405 // ------------------------------------------------------------------------
408 * @see java.lang.Object#toString()
411 @SuppressWarnings("nls")
412 public String
toString() {
413 return "[TmfTrace (" + getName() + ")]";
416 // ------------------------------------------------------------------------
418 // ------------------------------------------------------------------------
421 * The purpose of the index is to keep the information needed to rapidly
422 * restore the traces contexts at regular intervals (every INDEX_PAGE_SIZE
426 @SuppressWarnings({ "unchecked" })
427 private void indexTrace(boolean waitForCompletion
) {
429 final Job job
= new Job("Indexing " + getName() + "...") { //$NON-NLS-1$ //$NON-NLS-2$
431 protected IStatus
run(IProgressMonitor monitor
) {
432 while (!monitor
.isCanceled()) {
435 } catch (InterruptedException e
) {
436 return Status
.OK_STATUS
;
440 return Status
.OK_STATUS
;
445 fCheckpoints
.clear();
446 ITmfEventRequest
<TmfEvent
> request
= new TmfEventRequest
<TmfEvent
>(TmfEvent
.class, TmfTimeRange
.Eternity
,
447 TmfDataRequest
.ALL_DATA
, 1, ITmfDataRequest
.ExecutionType
.BACKGROUND
) {
449 TmfTimestamp startTime
= null;
450 TmfTimestamp lastTime
= null;
453 public void handleData(TmfEvent event
) {
454 super.handleData(event
);
456 TmfTimestamp ts
= event
.getTimestamp();
457 if (startTime
== null)
458 startTime
= new TmfTimestamp(ts
);
459 lastTime
= new TmfTimestamp(ts
);
461 if ((getNbRead() % fIndexPageSize
) == 0) {
468 public void handleSuccess() {
473 public void handleCompleted() {
475 super.handleCompleted();
478 private void updateTrace() {
479 int nbRead
= getNbRead();
481 fStartTime
= startTime
;
489 sendRequest((ITmfDataRequest
<T
>) request
);
490 if (waitForCompletion
)
492 request
.waitForCompletion();
493 } catch (InterruptedException e
) {
498 protected void notifyListeners() {
499 broadcast(new TmfTraceUpdatedSignal(this, this, new TmfTimeRange(fStartTime
, fEndTime
)));
502 // ------------------------------------------------------------------------
504 // ------------------------------------------------------------------------
507 protected void queueBackgroundRequest(final ITmfDataRequest
<T
> request
, final int blockSize
, final boolean indexing
) {
509 // TODO: Handle the data requests also...
510 if (!(request
instanceof ITmfEventRequest
<?
>)) {
511 super.queueRequest(request
);
514 final ITmfEventRequest
<T
> eventRequest
= (ITmfEventRequest
<T
>) request
;
516 Thread thread
= new Thread() {
520 // final long requestStart = System.nanoTime();
522 final Integer
[] CHUNK_SIZE
= new Integer
[1];
523 CHUNK_SIZE
[0] = blockSize
+ ((indexing
) ?
1 : 0);
525 final Integer
[] nbRead
= new Integer
[1];
528 // final TmfTimestamp[] timestamp = new TmfTimestamp[1];
529 // timestamp[0] = new TmfTimestamp(eventRequest.getRange().getStartTime());
530 // final TmfTimestamp endTS = eventRequest.getRange().getEndTime();
532 final Boolean
[] isFinished
= new Boolean
[1];
533 isFinished
[0] = Boolean
.FALSE
;
535 while (!isFinished
[0]) {
537 // TmfEventRequest<T> subRequest = new TmfEventRequest<T>(eventRequest.getDataType(), new TmfTimeRange(timestamp[0], endTS), CHUNK_SIZE[0], eventRequest.getBlockize(), ExecutionType.BACKGROUND)
538 // TmfDataRequest<T> subRequest = new TmfDataRequest<T>(eventRequest.getDataType(), nbRead[0], CHUNK_SIZE[0], eventRequest.getBlockize(), ExecutionType.BACKGROUND)
539 TmfDataRequest
<T
> subRequest
= new TmfDataRequest
<T
>(eventRequest
.getDataType(), nbRead
[0], CHUNK_SIZE
[0], ExecutionType
.BACKGROUND
)
542 public void handleData(T data
) {
543 super.handleData(data
);
544 eventRequest
.handleData(data
);
545 if (getNbRead() == CHUNK_SIZE
[0]) {
546 nbRead
[0] += getNbRead();
548 if (getNbRead() > CHUNK_SIZE
[0]) {
549 System
.out
.println("ERROR - Read too many events"); //$NON-NLS-1$
554 public void handleCompleted() {
555 // System.out.println("Request completed at: " + timestamp[0]);
556 if (getNbRead() < CHUNK_SIZE
[0]) {
558 eventRequest
.cancel();
563 isFinished
[0] = Boolean
.TRUE
;
564 nbRead
[0] += getNbRead();
565 // System.out.println("fNbRead=" + getNbRead() + " total=" + nbRead[0]);
567 super.handleCompleted();
571 if (!isFinished
[0]) {
572 queueRequest(subRequest
);
575 subRequest
.waitForCompletion();
576 // System.out.println("Finished at " + timestamp[0]);
577 } catch (InterruptedException e
) {
581 // TmfTimestamp newTS = new TmfTimestamp(timestamp[0].getValue() + 1, timestamp[0].getScale(), timestamp[0].getPrecision());
582 // timestamp[0] = newTS;
583 CHUNK_SIZE
[0] = blockSize
;
584 // System.out.println("New timestamp: " + timestamp[0]);
587 // final long requestEnded = System.nanoTime();
588 // System.out.println("Background request completed. Elapsed= " + (requestEnded * 1.0 - requestStart) / 1000000000);
596 // protected void queueBackgroundRequest(final ITmfDataRequest<T> request, final int blockSize, final boolean adjust) {
597 // super.queueBackgroundRequest(request, fIndexPageSize, true);