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