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