org.eclipse.linuxtools.tmf/src/org/eclipse/linuxtools/tmf/request/TmfDataRequest.java

   1 /*******************************************************************************
   2  * Copyright (c) 2009, 2010 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.request;
  14
  15 import java.util.concurrent.CountDownLatch;
  16
  17 import org.eclipse.linuxtools.tmf.Tracer;
  18 import org.eclipse.linuxtools.tmf.event.TmfData;
  19
  20 /**
  21  * <b><u>TmfDataRequest</u></b>
  22  * <p>
  23  * TmfDataRequests are used to obtain blocks of contiguous data from a data
  24  * provider. Open ranges can be used, especially for continuous streaming.
  25  * <p>
  26  * The request is processed asynchronously by a TmfProvider and, as blocks
  27  * of data become available, handleData() is invoked synchronously for each
  28  * block. Upon return, the data instances go out of scope and become eligible
  29  * for gc. It is is thus the responsibility of the requester to either clone
  30  * or keep a reference to the data it wishes to track specifically.
  31  * <p>
  32  * This data block approach is used to avoid  busting the heap for very
  33  * large trace files. The block size is configurable.
  34  * <p>
  35  * The TmfProvider indicates that the request is completed by calling done().
  36  * The request can be canceled at any time with cancel().
  37  * <p>
  38  * Typical usage:
  39  *<pre><code><i>TmfTimeWindow range = new TmfTimewindow(...);
  40  *TmfDataRequest&lt;DataType[]&gt; request = new TmfDataRequest&lt;DataType[]&gt;(DataType.class, 0, NB_EVENTS, BLOCK_SIZE) {
  41  *    public void handleData() {
  42  *         DataType[] data = request.getData();
  43  *         for (DataType e : data) {
  44  *             // do something
  45  *         }
  46  *    }
  47  *    public void handleSuccess() {
  48  *         // do something
  49  *         }
  50  *    }
  51  *    public void handleFailure() {
  52  *         // do something
  53  *         }
  54  *    }
  55  *    public void handleCancel() {
  56  *         // do something
  57  *         }
  58  *    }
  59  *};
  60  *fProcessor.process(request, true);
  61  *</i></code></pre>
  62  *
  63  * TODO: Consider decoupling from "time range", "rank", etc and for the more
  64  * generic notion of "criteria". This would allow to extend for "time range", etc
  65  * instead of providing specialized constructors. This also means removing the
  66  * criteria info from the data structure (with the possible exception of fNbRequested).
  67  * The nice thing about it is that it would prepare us well for the coming generation
  68  * of analysis tools.
  69  *
  70  * TODO: Implement request failures (codes, etc...)
  71  */
  72 public abstract class TmfDataRequest<T extends TmfData> implements ITmfDataRequest<T> {
  73
  74     // ------------------------------------------------------------------------
  75     // Constants
  76     // ------------------------------------------------------------------------
  77
  78     // The default maximum number of events per chunk
  79     public static final int DEFAULT_BLOCK_SIZE = 1000;
  80
  81     // The request count for all the events
  82     public static final int ALL_DATA = Integer.MAX_VALUE;
  83
  84     private static int fRequestNumber = 0;
  85
  86     // ------------------------------------------------------------------------
  87     // Attributes
  88     // ------------------------------------------------------------------------
  89
  90     private final Class<T>      fDataType;
  91     private final ExecutionType fExecType;
  92     private final int                   fRequestId;     // A unique request ID
  93     private final int                   fIndex;         // The index (rank) of the requested event
  94     private final int                   fNbRequested;   // The number of requested events (ALL_DATA for all)
  95     private final int                   fBlockSize;             // The block size (for BG requests)
  96     private       int                   fNbRead;        // The number of reads so far
  97
  98     private CountDownLatch startedLatch   = new CountDownLatch(1);
  99     private CountDownLatch completedLatch = new CountDownLatch(1);
 100     private boolean fRequestRunning   = false;
 101     private boolean fRequestCompleted = false;
 102     private boolean fRequestFailed    = false;
 103     private boolean fRequestCanceled  = false;
 104
 105     // ------------------------------------------------------------------------
 106     // Constructors
 107     // ------------------------------------------------------------------------
 108
 109     /**
 110      * Resets the request counter (used for testing)
 111      */
 112     public static void reset() {
 113         fRequestNumber = 0;
 114     }
 115
 116     /**
 117      * Default constructor
 118          *
 119      * @param dataType the requested data type
 120      */
 121     public TmfDataRequest(Class<T> dataType) {
 122         this(dataType, 0, ALL_DATA, DEFAULT_BLOCK_SIZE, ExecutionType.FOREGROUND);
 123     }
 124
 125     public TmfDataRequest(Class<T> dataType, ExecutionType execType) {
 126         this(dataType, 0, ALL_DATA, DEFAULT_BLOCK_SIZE, execType);
 127     }
 128
 129     /**
 130      * @param dataType the requested data type
 131      * @param nbRequested the number of data items requested
 132      */
 133     public TmfDataRequest(Class<T> dataType, int index) {
 134         this(dataType, index, ALL_DATA, DEFAULT_BLOCK_SIZE, ExecutionType.FOREGROUND);
 135     }
 136
 137     public TmfDataRequest(Class<T> dataType, int index, ExecutionType execType) {
 138         this(dataType, index, ALL_DATA, DEFAULT_BLOCK_SIZE, execType);
 139     }
 140
 141     /**
 142      * @param dataType the requested data type
 143      * @param index the index (rank) of the first event requested
 144      * @param blockSize the number of data items per block
 145      */
 146     public TmfDataRequest(Class<T> dataType, int index, int nbRequested) {
 147         this(dataType, index, nbRequested, DEFAULT_BLOCK_SIZE, ExecutionType.FOREGROUND);
 148     }
 149
 150     public TmfDataRequest(Class<T> dataType, int index, int nbRequested, ExecutionType execType) {
 151         this(dataType, index, nbRequested, DEFAULT_BLOCK_SIZE, execType);
 152     }
 153
 154     /**
 155      * @param dataType the requested data type
 156      * @param index the index (rank) of the first event requested
 157      * @param nbRequested the number of data items requested
 158      * @param blockSize the number of data items per block
 159      */
 160     public TmfDataRequest(Class<T> dataType, int index, int nbRequested, int blockSize) {
 161         this(dataType, index, nbRequested, blockSize, ExecutionType.FOREGROUND);
 162     }
 163
 164     public TmfDataRequest(Class<T> dataType, int index, int nbRequested, int blockSize, ExecutionType execType) {
 165         fRequestId   = fRequestNumber++;
 166         fDataType    = dataType;
 167         fIndex       = index;
 168         fNbRequested = nbRequested;
 169         fBlockSize   = blockSize;
 170         fExecType    = execType;
 171         fNbRead      = 0;
 172         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "created"); //$NON-NLS-1$
 173     }
 174
 175     /**
 176      * Copy constructor
 177      */
 178     @SuppressWarnings("unused")
 179         private TmfDataRequest(TmfDataRequest<T> other) {
 180         this(null, 0, ALL_DATA, DEFAULT_BLOCK_SIZE);
 181     }
 182
 183     // ------------------------------------------------------------------------
 184     // Accessors
 185     // ------------------------------------------------------------------------
 186
 187         /**
 188          * @return the request ID
 189          */
 190         @Override
 191         public int getRequestId() {
 192                 return fRequestId;
 193         }
 194
 195         /**
 196          * @return the index of the first event requested
 197          */
 198         @Override
 199         public int getIndex() {
 200                 return fIndex;
 201         }
 202
 203         /**
 204          * @return the index of the first event requested
 205          */
 206         @Override
 207         public ExecutionType getExecType() {
 208                 return fExecType;
 209         }
 210
 211     /**
 212      * @return the number of requested events (ALL_DATA = all)
 213      */
 214     @Override
 215         public int getNbRequested() {
 216         return fNbRequested;
 217     }
 218
 219     /**
 220      * @return the block size (for BG requests)
 221      */
 222     @Override
 223         public int getBlockSize() {
 224         return fBlockSize;
 225     }
 226
 227     /**
 228      * @return the number of events read so far
 229      */
 230     @Override
 231         public synchronized int getNbRead() {
 232         return fNbRead;
 233     }
 234
 235     /**
 236      * @return indicates if the request is completed
 237      */
 238     @Override
 239     public synchronized boolean isRunning() {
 240         return fRequestRunning;
 241     }
 242
 243     /**
 244      * @return indicates if the request is completed
 245      */
 246     @Override
 247     public synchronized boolean isCompleted() {
 248         return fRequestCompleted;
 249     }
 250
 251     /**
 252      * @return indicates if the request is canceled
 253      */
 254     @Override
 255     public synchronized boolean isFailed() {
 256         return fRequestFailed;
 257     }
 258
 259     /**
 260      * @return indicates if the request is canceled
 261      */
 262     @Override
 263     public synchronized boolean isCancelled() {
 264         return fRequestCanceled;
 265     }
 266
 267     /**
 268      * @return the requested data type
 269      */
 270     @Override
 271         public Class<T> getDataType() {
 272         return fDataType;
 273     }
 274
 275     // ------------------------------------------------------------------------
 276     // Operators
 277     // ------------------------------------------------------------------------
 278
 279     /**
 280      * Sets the data object to specified value. To be called by the
 281      * asynchronous method implementor.
 282      *
 283      * @param data Data value to set.
 284      */
 285
 286     /**
 287      * Handle a block of incoming data. This method is called every time
 288      * a block of data becomes available.
 289      *
 290      * - Data items are received in the order they appear in the stream.
 291      * - Called by the request processor, in its execution thread, every time a
 292      *   block of data becomes available.
 293      * - Request processor performs a synchronous call to handlePartialResult()
 294      *   i.e. its execution threads holds until handlePartialData() returns.
 295      * - Original data items are disposed of on return i.e. keep a reference
 296      *   (or a copy) if some persistence is needed between invocations.
 297      * - When there is no more data, done() is called.
 298      *
 299      * @param events - an events
 300      */
 301     @Override
 302         public void handleData(T data) {
 303         if (data != null) {
 304                 fNbRead++;
 305         }
 306     }
 307
 308     @Override
 309         public void handleStarted() {
 310         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "started"); //$NON-NLS-1$
 311     }
 312
 313     /**
 314      * Handle the completion of the request. It is called when there is no more
 315      * data available either because:
 316      * - the request completed normally
 317      * - the request failed
 318      * - the request was canceled
 319      *
 320      * As a convenience, handleXXXX methods are provided. They are meant to be
 321      * overridden by the application if it needs to handle these conditions.
 322      */
 323     @Override
 324         public void handleCompleted() {
 325         if (fRequestFailed) {
 326                 handleFailure();
 327         }
 328         else if (fRequestCanceled) {
 329                 handleCancel();
 330         }
 331         else {
 332                         handleSuccess();
 333         }
 334         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "completed"); //$NON-NLS-1$
 335     }
 336
 337     @Override
 338         public void handleSuccess() {
 339         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "succeeded"); //$NON-NLS-1$
 340     }
 341
 342     @Override
 343         public void handleFailure() {
 344         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "failed"); //$NON-NLS-1$
 345     }
 346
 347     @Override
 348         public void handleCancel() {
 349         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "cancelled"); //$NON-NLS-1$
 350     }
 351
 352     /**
 353      * To suspend the client thread until the request starts (or is
 354      * canceled).
 355      *
 356      * @throws InterruptedException
 357      */
 358     public void waitForStart() throws InterruptedException {
 359                 while (!fRequestRunning) {
 360                         startedLatch.await();
 361                 }
 362     }
 363
 364     /**
 365      * To suspend the client thread until the request completes (or is
 366      * canceled).
 367      *
 368      * @throws InterruptedException
 369      */
 370     @Override
 371         public void waitForCompletion() throws InterruptedException {
 372                 while (!fRequestCompleted) {
 373                         completedLatch.await();
 374                 }
 375     }
 376
 377     /**
 378      * Called by the request processor upon starting to service the request.
 379      */
 380     @Override
 381         public void start() {
 382         synchronized(this) {
 383             fRequestRunning = true;
 384         }
 385         handleStarted();
 386         startedLatch.countDown();
 387     }
 388
 389     /**
 390      * Called by the request processor upon completion.
 391      */
 392     @Override
 393         public void done() {
 394         synchronized(this) {
 395                 if (!fRequestCompleted) {
 396                 fRequestRunning   = false;
 397                 fRequestCompleted = true;
 398             }
 399         }
 400                 handleCompleted();
 401                 completedLatch.countDown();
 402     }
 403
 404     /**
 405      * Called by the request processor upon failure.
 406      */
 407     @Override
 408     public void fail() {
 409         synchronized(this) {
 410             fRequestFailed = true;
 411         }
 412         done();
 413     }
 414
 415     /**
 416      * Called by the request processor upon cancellation.
 417      */
 418     @Override
 419     public void cancel() {
 420         synchronized(this) {
 421             fRequestCanceled = true;
 422         }
 423                 done();
 424     }
 425
 426     // ------------------------------------------------------------------------
 427     // Object
 428     // ------------------------------------------------------------------------
 429
 430     @Override
 431     // All requests have a unique id
 432     public int hashCode() {
 433         return getRequestId();
 434     }
 435
 436     @Override
 437     public boolean equals(Object other) {
 438         if (other instanceof TmfDataRequest<?>) {
 439                 TmfDataRequest<?> request = (TmfDataRequest<?>) other;
 440                 return  (request.fDataType    == fDataType) &&
 441                                 (request.fIndex       == fIndex)    &&
 442                                 (request.fNbRequested == fNbRequested);
 443         }
 444         return false;
 445     }
 446
 447     @Override
 448     @SuppressWarnings("nls")
 449     public String toString() {
 450                 return "[TmfDataRequest(" + fRequestId + "," + fDataType.getSimpleName() +
 451                         "," + fIndex + "," + fNbRequested + "," + getBlockSize() + ")]";
 452     }
 453 }