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