org.eclipse.linuxtools.tmf/src/org/eclipse/linuxtools/tmf/trace/TmfTrace.java

   1 /*******************************************************************************
   2  * Copyright (c) 2009 Ericsson
   3  *
   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
   8  *
   9  * Contributors:
  10  *   Francois Chouinard - Initial API and implementation
  11  *******************************************************************************/
  12
  13 package org.eclipse.linuxtools.tmf.trace;
  14
  15 import java.io.FileNotFoundException;
  16 import java.io.IOException;
  17 import java.util.Collections;
  18 import java.util.Vector;
  19
  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.event.TmfEvent;
  25 import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  26 import org.eclipse.linuxtools.tmf.event.TmfTimestamp;
  27 import org.eclipse.linuxtools.tmf.signal.TmfSignalManager;
  28
  29 /**
  30  * <b><u>TmfEventStream</u></b>
  31  * <p>
  32  * TODO: Implement me. Please.
  33  */
  34 public abstract class TmfTrace implements ITmfTrace {
  35
  36     // ========================================================================
  37     // Constants
  38     // ========================================================================
  39
  40     // The default number of events to cache
  41     public static final int DEFAULT_CACHE_SIZE = 1000;
  42
  43     // ========================================================================
  44     // Attributes
  45     // ========================================================================
  46
  47     // The stream name
  48     private final String fName;
  49
  50     // The stream parser
  51     private final ITmfEventParser fParser;
  52
  53     // The cache size
  54     private final int fCacheSize;
  55
  56     // The set of event stream checkpoints (for random access)
  57     private Vector<TmfStreamCheckpoint> fCheckpoints = new Vector<TmfStreamCheckpoint>();
  58
  59     // The number of events collected
  60     private int fNbEvents = 0;
  61
  62     // The time span of the event stream
  63     private TmfTimeRange fTimeRange = new TmfTimeRange(TmfTimestamp.BigBang, TmfTimestamp.BigBang);
  64
  65     // ========================================================================
  66     // Constructors
  67     // ========================================================================
  68
  69     /**
  70      * @param filename
  71      * @param parser
  72      * @param cacheSize
  73      * @throws FileNotFoundException
  74      */
  75     protected TmfTrace(String filename, ITmfEventParser parser, int cacheSize) throws FileNotFoundException {
  76         fName = filename;
  77         fParser = parser;
  78         fCacheSize = cacheSize;
  79     }
  80
  81     /**
  82      * @param filename
  83      * @param parser
  84      * @throws FileNotFoundException
  85      */
  86     protected TmfTrace(String filename, ITmfEventParser parser) throws FileNotFoundException {
  87         this(filename, parser, DEFAULT_CACHE_SIZE);
  88     }
  89
  90     // ========================================================================
  91     // Accessors
  92     // ========================================================================
  93
  94     /**
  95      * @return
  96      */
  97     public int getCacheSize() {
  98         return fCacheSize;
  99     }
 100
 101     /**
 102      * @return
 103      */
 104     public String getName() {
 105         return fName;
 106     }
 107
 108     /* (non-Javadoc)
 109      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getNbEvents()
 110      */
 111     public int getNbEvents() {
 112         return fNbEvents;
 113     }
 114
 115     /* (non-Javadoc)
 116      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getTimeRange()
 117      */
 118     public TmfTimeRange getTimeRange() {
 119         return fTimeRange;
 120     }
 121
 122     /* (non-Javadoc)
 123      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getIndex(org.eclipse.linuxtools.tmf.event.TmfTimestamp)
 124      */
 125     public int getIndex(TmfTimestamp timestamp) {
 126         StreamContext context = seekEvent(timestamp);
 127         return context.index;
 128     }
 129
 130     /* (non-Javadoc)
 131      * @see org.eclipse.linuxtools.tmf.stream.ITmfEventStream#getTimestamp(int)
 132      */
 133     public TmfTimestamp getTimestamp(int index) {
 134         StreamContext context = seekEvent(index);
 135         TmfEvent event = peekEvent(context);
 136         return event.getTimestamp();
 137     }
 138
 139     // ========================================================================
 140     // Operators
 141     // ========================================================================
 142
 143     public StreamContext seekEvent(TmfTimestamp timestamp) {
 144
 145         // First, find the right checkpoint
 146         int index = Collections.binarySearch(fCheckpoints, new TmfStreamCheckpoint(timestamp, 0));
 147
 148         // In the very likely event that the checkpoint was not found, bsearch
 149         // returns its negated would-be location (not an offset...). From that
 150         // index, we can then position the stream and get the event.
 151         if (index < 0) {
 152             index = Math.max(0, -(index + 2));
 153         }
 154
 155         // Position the stream at the checkpoint
 156         Object location = (index < fCheckpoints.size()) ? fCheckpoints.elementAt(index).getLocation() : null;
 157         StreamContext nextEventContext;
 158         synchronized(this) {
 159                 nextEventContext = seekLocation(location);
 160         }
 161         StreamContext currentEventContext = new StreamContext(nextEventContext.location, index * fCacheSize);
 162
 163         // And get the event
 164         TmfEvent event = getNextEvent(nextEventContext);
 165         while (event != null && event.getTimestamp().compareTo(timestamp, false) < 0) {
 166                 currentEventContext.location = nextEventContext.location;
 167                 currentEventContext.index++;
 168                 event = getNextEvent(nextEventContext);
 169         }
 170
 171         return currentEventContext;
 172     }
 173
 174     public StreamContext seekEvent(int position) {
 175
 176         // Position the stream at the previous checkpoint
 177         int index = position / fCacheSize;
 178         Object location = (index < fCheckpoints.size()) ? fCheckpoints.elementAt(index).getLocation() : null;
 179         StreamContext nextEventContext;
 180         synchronized(this) {
 181                 nextEventContext = seekLocation(location);
 182         }
 183         StreamContext currentEventContext = new StreamContext(nextEventContext);
 184
 185         // And locate the event (if it exists)
 186         int current = index * fCacheSize;
 187         TmfEvent event = getNextEvent(nextEventContext);
 188         while (event != null && current < position) {
 189                 currentEventContext.location = nextEventContext.location;
 190                 event = getNextEvent(nextEventContext);
 191             current++;
 192         }
 193
 194         return currentEventContext;
 195     }
 196
 197     public TmfEvent peekEvent(StreamContext context) {
 198         StreamContext ctx = new StreamContext(context);
 199         return getNextEvent(ctx);
 200     }
 201
 202     public TmfEvent getEvent(StreamContext context, TmfTimestamp timestamp) {
 203
 204         // Position the stream and update the context object
 205         StreamContext ctx = seekEvent(timestamp);
 206         context.location = ctx.location;
 207
 208         return getNextEvent(context);
 209     }
 210
 211     public TmfEvent getEvent(StreamContext context, int position) {
 212
 213         // Position the stream and update the context object
 214         StreamContext ctx = seekEvent(position);
 215         context.location = ctx.location;
 216
 217         return getNextEvent(context);
 218     }
 219
 220     public synchronized TmfEvent getNextEvent(StreamContext context) {
 221         try {
 222                 seekLocation(context.location);
 223                 TmfEvent event = fParser.getNextEvent(this);
 224                 context.location = getCurrentLocation();
 225                 return event;
 226         } catch (IOException e) {
 227             e.printStackTrace();
 228         }
 229         return null;
 230     }
 231
 232         private void notifyListeners(TmfTimeRange range) {
 233                 TmfSignalManager.dispatchSignal(new TmfStreamUpdatedSignal(this, this, range));
 234         }
 235
 236     /* (non-Javadoc)
 237      * @see org.eclipse.linuxtools.tmf.eventlog.ITmfEventStream#indexStream()
 238      */
 239     public void indexStream(boolean waitForCompletion) {
 240         IndexingJob job = new IndexingJob(fName);
 241         job.schedule();
 242         if (waitForCompletion) {
 243             try {
 244                                 job.join();
 245                         } catch (InterruptedException e) {
 246                                 // TODO Auto-generated catch block
 247                                 e.printStackTrace();
 248                         }
 249         }
 250     }
 251
 252     private class IndexingJob extends Job {
 253
 254                 public IndexingJob(String name) {
 255                         super(name);
 256                 }
 257
 258                 /* (non-Javadoc)
 259                  * @see org.eclipse.core.runtime.jobs.Job#run(org.eclipse.core.runtime.IProgressMonitor)
 260                  */
 261                 @Override
 262                 protected IStatus run(IProgressMonitor monitor) {
 263
 264             int nbEvents = 0;
 265             TmfTimestamp startTime = new TmfTimestamp();
 266             TmfTimestamp lastTime  = new TmfTimestamp();
 267             TmfTimestamp rangeStartTime = new TmfTimestamp();
 268
 269             monitor.beginTask("Indexing " + fName, IProgressMonitor.UNKNOWN);
 270
 271             try {
 272                 StreamContext nextEventContext;
 273                 synchronized(this) {
 274                         nextEventContext = seekLocation(null);
 275                 }
 276                 StreamContext currentEventContext = new StreamContext(nextEventContext);
 277                 TmfEvent event = getNextEvent(nextEventContext);
 278                 if (event != null) {
 279                         startTime = event.getTimestamp();
 280                         lastTime  = event.getTimestamp();
 281                 }
 282
 283                 rangeStartTime = startTime;
 284                 while (event != null) {
 285                     lastTime = event.getTimestamp();
 286                     if ((nbEvents++ % fCacheSize) == 0) {
 287                         TmfStreamCheckpoint bookmark = new TmfStreamCheckpoint(lastTime, currentEventContext.location);
 288                         synchronized(this) {
 289                                 fCheckpoints.add(bookmark);
 290                                 fNbEvents = nbEvents;
 291                             fTimeRange = new TmfTimeRange(startTime, lastTime);
 292                         }
 293                         notifyListeners(new TmfTimeRange(rangeStartTime, lastTime));
 294                         monitor.worked(1);
 295                         // Check monitor *after* fCheckpoints has been updated
 296                         if (monitor.isCanceled()) {
 297                                 return Status.CANCEL_STATUS;
 298                         }
 299                     }
 300
 301                         currentEventContext.location = nextEventContext.location;
 302                                         event = getNextEvent(nextEventContext);
 303                 }
 304             }
 305             finally {
 306                 synchronized(this) {
 307                         fNbEvents = nbEvents;
 308                         fTimeRange = new TmfTimeRange(startTime, lastTime);
 309                 }
 310                 notifyListeners(new TmfTimeRange(rangeStartTime, lastTime));
 311                 monitor.done();
 312             }
 313
 314                         return Status.OK_STATUS;
 315                 }
 316     }
 317
 318 }