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

   1 /*******************************************************************************
   2  * Copyright (c) 2009 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 org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  16
  17 /**
  18  * <b><u>TmfDataRequest</u></b>
  19  * <p>
  20  * TmfDataRequests are used to obtain blocks of contiguous data from a data
  21  * provider, either all the data within a given time window or n elements
  22  * starting at a specific timestamp. Open ranges can be used, especially for
  23  * continuous streaming.
  24  * <p>
  25  * The request is processed asynchronously by an ITmfRequestProcessor and,
  26  * as blocks of data become available, the callback handlePartialData() is
  27  * invoked, synchronously, for each block. When returning from the callback,
  28  * the data instances go out of scope and become eligible for gc. It is
  29  * is thus the responsibility of the requester to either copy or keep a
  30  * reference to the data it wishes to track specifically.
  31  * <p>
  32  * This data block approach is necessary to avoid  busting the heap for very
  33  * large trace files. The block size is configurable.
  34  * <p>
  35  * The ITmfRequestProcessor indicates that the request is completed by
  36  * calling done(). 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;(range, 0, NB_EVENTS, BLOCK_SIZE) {
  41  *    &#64;Override
  42  *    public void handlePartialResult() {
  43  *         DataType[] data = request.getData();
  44  *         for (DataType e : data) {
  45  *             // do something
  46  *         }
  47  *    }
  48  *};
  49  *fProcessor.process(request, true);
  50  *</i></code></pre>
  51  *
  52  * TODO: Consider extending DataRequestMonitor from DSF concurrency plugin.
  53  * The main issue is the slicing of the result in blocks and continuous
  54  * streams. This would require using a thread executor and to carefully
  55  * look at setData() and getData().
  56  */
  57 public class TmfDataRequest<V> {
  58
  59     // ========================================================================
  60     // Constants
  61     // ========================================================================
  62
  63     // The default maximum number of events per chunk
  64     public static final int DEFAULT_BLOCK_SIZE = 1000;
  65
  66     // The request count for all the events
  67     public static final int ALL_EVENTS = -1;
  68
  69     // ========================================================================
  70     // Attributes
  71     // ========================================================================
  72
  73     private final TmfTimeRange fRange;     // The requested events timestamp range
  74     private final int  fIndex;              // The event index to get
  75     private final long fOffset;             // The synchronization offset to apply
  76     private final int  fNbRequestedItems;   // The number of items to read (-1 == the whole range)
  77     private final int  fBlockSize;          // The maximum number of events per chunk
  78
  79     private Object lock = new Object();
  80     private boolean fRequestCompleted = false;
  81     private boolean fRequestCanceled  = false;
  82
  83     private V[] fData;  // Data object
  84
  85     // ========================================================================
  86     // Constructors
  87     // ========================================================================
  88
  89     /**
  90      * @param index
  91      * @param nbEvents
  92      */
  93     public TmfDataRequest(int index, long offset, int nbEvents) {
  94         this(null, index, offset, nbEvents, DEFAULT_BLOCK_SIZE);
  95     }
  96
  97     /**
  98      * @param range
  99      * @param offset
 100      * @param nbEvents
 101      */
 102     public TmfDataRequest(TmfTimeRange range, long offset, int nbEvents) {
 103         this(range, 0, offset, nbEvents, DEFAULT_BLOCK_SIZE);
 104     }
 105
 106     /**
 107      * @param range
 108      * @param offset
 109      * @param nbItems
 110      * @param maxBlockSize Size of the largest blocks expected
 111      */
 112     public TmfDataRequest(TmfTimeRange range, long offset, int nbEvents, int maxBlockSize) {
 113         this(range, 0, offset, nbEvents, maxBlockSize);
 114     }
 115
 116     /**
 117      * @param range
 118      * @param index
 119      * @param offset
 120      * @param nbItems
 121      * @param maxBlockSize Size of the largest blocks expected
 122      */
 123     public TmfDataRequest(TmfTimeRange range, int index, long offset, int nbEvents, int maxBlockSize) {
 124         fRange = range;
 125         fIndex = index;
 126         fOffset = offset;
 127         fNbRequestedItems = nbEvents;
 128         fBlockSize = maxBlockSize;
 129     }
 130
 131     // ========================================================================
 132     // Accessors
 133     // ========================================================================
 134
 135     /**
 136      * @return the requested time range
 137      */
 138     public TmfTimeRange getRange() {
 139         return fRange;
 140     }
 141
 142     /**
 143      * @return the index
 144      */
 145     public int getIndex() {
 146         return fIndex;
 147     }
 148
 149     /**
 150      * @return the offset
 151      */
 152     public long getOffset() {
 153         return fOffset;
 154     }
 155
 156     /**
 157      * @return the number of requested events (-1 = all)
 158      */
 159     public int getNbRequestedItems() {
 160         return fNbRequestedItems;
 161     }
 162
 163     /**
 164      * @return the block size
 165      */
 166     public int getBlockize() {
 167         return fBlockSize;
 168     }
 169
 170     /**
 171      * @return indicates if the request is completed
 172      */
 173     public boolean isCompleted() {
 174         return fRequestCompleted;
 175     }
 176
 177     /**
 178      * @return indicates if the request is canceled
 179      */
 180     public boolean isCancelled() {
 181         return fRequestCanceled;
 182     }
 183
 184     // ========================================================================
 185     // Operators
 186     // ========================================================================
 187
 188     /**
 189      * Sets the data object to specified value.  To be called by the
 190      * asynchronous method implementor.
 191      * @param data Data value to set.
 192      */
 193     public synchronized void setData(V[] data) {
 194         fData = data;
 195     }
 196
 197     /**
 198      * Returns the data value, null if not set.
 199      */
 200     public synchronized V[] getData() {
 201         return fData;
 202     }
 203
 204     /**
 205      * handlePartialResult()
 206      *
 207      * - Data items are received in the order they appear in the stream.
 208      * - Called by the request processor, in its execution thread, every time a
 209      *   block of data becomes available.
 210      * - Request processor performs a synchronous call to handlePartialResult()
 211      *   i.e. its execution threads holds until handlePartialData() returns.
 212      * - Original data items are disposed of on return i.e. keep a reference
 213      *   (or a copy) if some persistence is needed between invocations.
 214      * - When there is no more data, done() is called.
 215      *
 216      * @param events - an array of events
 217      */
 218     public void handlePartialResult() {
 219     }
 220
 221     public void handleCompleted() {
 222     }
 223
 224     /**
 225      * To suspend the client thread until the request completes (or is
 226      * canceled).
 227      *
 228      * @throws InterruptedException
 229      */
 230     public void waitForCompletion() {
 231         synchronized (lock) {
 232             while (!fRequestCompleted)
 233                 try {
 234                     lock.wait();
 235                 } catch (InterruptedException e) {
 236                     e.printStackTrace();
 237                 }
 238         }
 239     }
 240
 241     /**
 242      * Complete the request. Called by the request processor upon completion.
 243      */
 244     public void done() {
 245         synchronized(lock) {
 246             fRequestCompleted = true;
 247             lock.notify();
 248         }
 249         handleCompleted();
 250     }
 251
 252     /**
 253      * Cancel the request.
 254      */
 255     public void cancel() {
 256         synchronized(lock) {
 257             fRequestCanceled = true;
 258             fRequestCompleted = true;
 259             lock.notify();
 260         }
 261     }
 262
 263 }