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       int                   fNbRead;        // The number of reads so far
  96
  97     private CountDownLatch startedLatch   = new CountDownLatch(1);
  98     private CountDownLatch completedLatch = new CountDownLatch(1);
  99     private boolean fRequestRunning   = false;
 100     private boolean fRequestCompleted = false;
 101     private boolean fRequestFailed    = false;
 102     private boolean fRequestCanceled  = false;
 103
 104     // ------------------------------------------------------------------------
 105     // Constructors
 106     // ------------------------------------------------------------------------
 107
 108     /**
 109      * Resets the request counter (used for testing)
 110      */
 111     public static void reset() {
 112         fRequestNumber = 0;
 113     }
 114
 115     /**
 116      * Default constructor
 117          *
 118      * @param dataType the requested data type
 119      */
 120     public TmfDataRequest(Class<T> dataType) {
 121         this(dataType, 0, ALL_DATA, DEFAULT_BLOCK_SIZE, ExecutionType.FOREGROUND);
 122     }
 123
 124     public TmfDataRequest(Class<T> dataType, ExecutionType execType) {
 125         this(dataType, 0, ALL_DATA, DEFAULT_BLOCK_SIZE, execType);
 126     }
 127
 128     /**
 129      * @param dataType the requested data type
 130      * @param nbRequested the number of data items requested
 131      */
 132     public TmfDataRequest(Class<T> dataType, int index) {
 133         this(dataType, index, ALL_DATA, DEFAULT_BLOCK_SIZE, ExecutionType.FOREGROUND);
 134     }
 135
 136     public TmfDataRequest(Class<T> dataType, int index, ExecutionType execType) {
 137         this(dataType, index, ALL_DATA, DEFAULT_BLOCK_SIZE, execType);
 138     }
 139
 140     /**
 141      * @param dataType the requested data type
 142      * @param index the index (rank) of the first event requested
 143      * @param blockSize the number of data items per block
 144      */
 145     public TmfDataRequest(Class<T> dataType, int index, int nbRequested) {
 146         this(dataType, index, nbRequested, DEFAULT_BLOCK_SIZE, ExecutionType.FOREGROUND);
 147     }
 148
 149     public TmfDataRequest(Class<T> dataType, int index, int nbRequested, ExecutionType execType) {
 150         this(dataType, index, nbRequested, DEFAULT_BLOCK_SIZE, execType);
 151     }
 152
 153     /**
 154      * @param dataType the requested data type
 155      * @param index the index (rank) of the first event requested
 156      * @param nbRequested the number of data items requested
 157      * @param blockSize the number of data items per block
 158      */
 159     public TmfDataRequest(Class<T> dataType, int index, int nbRequested, int blockSize) {
 160         this(dataType, index, nbRequested, blockSize, ExecutionType.FOREGROUND);
 161     }
 162
 163     public TmfDataRequest(Class<T> dataType, int index, int nbRequested, int blockSize, ExecutionType execType) {
 164         fRequestId   = fRequestNumber++;
 165         fDataType    = dataType;
 166         fIndex       = index;
 167         fNbRequested = nbRequested;
 168         fExecType    = execType;
 169         fNbRead      = 0;
 170         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "created");
 171     }
 172
 173     /**
 174      * Copy constructor
 175      */
 176     @SuppressWarnings("unused")
 177         private TmfDataRequest(TmfDataRequest<T> other) {
 178         this(null, 0, ALL_DATA, DEFAULT_BLOCK_SIZE);
 179     }
 180
 181     // ------------------------------------------------------------------------
 182     // Accessors
 183     // ------------------------------------------------------------------------
 184
 185         /**
 186          * @return the request ID
 187          */
 188         @Override
 189         public int getRequestId() {
 190                 return fRequestId;
 191         }
 192
 193         /**
 194          * @return the index of the first event requested
 195          */
 196         @Override
 197         public int getIndex() {
 198                 return fIndex;
 199         }
 200
 201         /**
 202          * @return the index of the first event requested
 203          */
 204         @Override
 205         public ExecutionType getExecType() {
 206                 return fExecType;
 207         }
 208
 209     /**
 210      * @return the number of requested events (ALL_DATA = all)
 211      */
 212     @Override
 213         public int getNbRequested() {
 214         return fNbRequested;
 215     }
 216
 217     /**
 218      * @return the number of events read so far
 219      */
 220     @Override
 221         public synchronized int getNbRead() {
 222         return fNbRead;
 223     }
 224
 225     /**
 226      * @return indicates if the request is completed
 227      */
 228     @Override
 229         public boolean isRunning() {
 230         return fRequestRunning;
 231     }
 232
 233     /**
 234      * @return indicates if the request is completed
 235      */
 236     @Override
 237         public boolean isCompleted() {
 238         return fRequestCompleted;
 239     }
 240
 241     /**
 242      * @return indicates if the request is canceled
 243      */
 244     @Override
 245         public boolean isFailed() {
 246         return fRequestFailed;
 247     }
 248
 249     /**
 250      * @return indicates if the request is canceled
 251      */
 252     @Override
 253         public boolean isCancelled() {
 254         return fRequestCanceled;
 255     }
 256
 257     /**
 258      * @return the requested data type
 259      */
 260     @Override
 261         public Class<T> getDataType() {
 262         return fDataType;
 263     }
 264
 265     // ------------------------------------------------------------------------
 266     // Operators
 267     // ------------------------------------------------------------------------
 268
 269     /**
 270      * Sets the data object to specified value. To be called by the
 271      * asynchronous method implementor.
 272      *
 273      * @param data Data value to set.
 274      */
 275     /**
 276      * Handle a block of incoming data. This method is called every time
 277      * a block of data becomes available.
 278      *
 279      * - Data items are received in the order they appear in the stream.
 280      * - Called by the request processor, in its execution thread, every time a
 281      *   block of data becomes available.
 282      * - Request processor performs a synchronous call to handlePartialResult()
 283      *   i.e. its execution threads holds until handlePartialData() returns.
 284      * - Original data items are disposed of on return i.e. keep a reference
 285      *   (or a copy) if some persistence is needed between invocations.
 286      * - When there is no more data, done() is called.
 287      *
 288      * @param events - an events
 289      */
 290     @Override
 291         public void handleData(T data) {
 292         if (data != null) {
 293                 fNbRead++;
 294         }
 295     }
 296
 297     @Override
 298         public void handleStarted() {
 299     }
 300
 301     /**
 302      * Handle the completion of the request. It is called when there is no more
 303      * data available either because:
 304      * - the request completed normally
 305      * - the request failed
 306      * - the request was canceled
 307      *
 308      * As a convenience, handleXXXX methods are provided. They are meant to be
 309      * overridden by the application if it needs to handle these conditions.
 310      */
 311     @Override
 312         public void handleCompleted() {
 313         if (fRequestFailed) {
 314             if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "failed");
 315                 handleFailure();
 316         }
 317         else if (fRequestCanceled) {
 318             if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "cancelled");
 319                 handleCancel();
 320         }
 321         else {
 322             if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "succeeded");
 323                         handleSuccess();
 324         }
 325     }
 326
 327     @Override
 328         public void handleSuccess() {
 329     }
 330
 331     @Override
 332         public void handleFailure() {
 333     }
 334
 335     @Override
 336         public void handleCancel() {
 337     }
 338
 339     /**
 340      * To suspend the client thread until the request starts (or is
 341      * canceled).
 342      *
 343      * @throws InterruptedException
 344      */
 345     public void waitForStart() throws InterruptedException {
 346                 while (!fRequestRunning) {
 347                         startedLatch.await();
 348                 }
 349     }
 350
 351     /**
 352      * To suspend the client thread until the request completes (or is
 353      * canceled).
 354      *
 355      * @throws InterruptedException
 356      */
 357     @Override
 358         public void waitForCompletion() throws InterruptedException {
 359                 while (!fRequestCompleted) {
 360                         completedLatch.await();
 361                 }
 362     }
 363
 364     /**
 365      * Called by the request processor upon starting to service the request.
 366      */
 367     @Override
 368         public void start() {
 369         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "starting");
 370         synchronized(this) {
 371             fRequestRunning = true;
 372         }
 373         handleStarted();
 374         startedLatch.countDown();
 375         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "started");
 376     }
 377
 378     /**
 379      * Called by the request processor upon completion.
 380      */
 381     @Override
 382         public void done() {
 383         if (Tracer.isRequestTraced()) Tracer.traceRequest(this, "completing");
 384         synchronized(this) {
 385                 if (!fRequestCompleted) {
 386                 fRequestRunning   = false;
 387                 fRequestCompleted = true;
 388             }
 389         }
 390                 handleCompleted();
 391                 completedLatch.countDown();
 392     }
 393
 394     /**
 395      * Called by the request processor upon failure.
 396      */
 397     @Override
 398         public synchronized void fail() {
 399         fRequestFailed = true;
 400         done();
 401     }
 402
 403     /**
 404      * Called by the request processor upon cancellation.
 405      */
 406     @Override
 407         public synchronized void cancel() {
 408                 fRequestCanceled = true;
 409                 done();
 410     }
 411
 412     // ------------------------------------------------------------------------
 413     // Object
 414     // ------------------------------------------------------------------------
 415
 416     @Override
 417     // All requests have a unique id
 418     public int hashCode() {
 419         return getRequestId();
 420     }
 421
 422     @Override
 423     public boolean equals(Object other) {
 424         if (other instanceof TmfDataRequest<?>) {
 425                 TmfDataRequest<?> request = (TmfDataRequest<?>) other;
 426                 return  (request.fDataType    == fDataType) &&
 427                                 (request.fIndex       == fIndex)    &&
 428                                 (request.fNbRequested == fNbRequested);
 429         }
 430         return false;
 431     }
 432
 433     @Override
 434     public String toString() {
 435                 return "[TmfDataRequest(" + fRequestId + "," + fDataType.getSimpleName() +
 436                         "," + fIndex + "," + fNbRequested + ")]";
 437     }
 438 }