org.eclipse.linuxtools.tmf.core/src/org/eclipse/linuxtools/tmf/core/component/TmfDataProvider.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.component;
  14
  15 import java.util.Vector;
  16 import java.util.concurrent.BlockingQueue;
  17 import java.util.concurrent.LinkedBlockingQueue;
  18 import java.util.concurrent.SynchronousQueue;
  19
  20 import org.eclipse.linuxtools.tmf.core.Tracer;
  21 import org.eclipse.linuxtools.tmf.core.event.TmfData;
  22 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest;
  23 import org.eclipse.linuxtools.tmf.core.request.TmfCoalescedDataRequest;
  24 import org.eclipse.linuxtools.tmf.core.request.TmfDataRequest;
  25 import org.eclipse.linuxtools.tmf.core.request.TmfRequestExecutor;
  26 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest.ExecutionType;
  27 import org.eclipse.linuxtools.tmf.core.signal.TmfEndSynchSignal;
  28 import org.eclipse.linuxtools.tmf.core.signal.TmfSignalHandler;
  29 import org.eclipse.linuxtools.tmf.core.signal.TmfStartSynchSignal;
  30 import org.eclipse.linuxtools.tmf.core.trace.ITmfContext;
  31
  32 /**
  33  * <b><u>TmfProvider</u></b>
  34  * <p>
  35  * The TmfProvider<T> is a provider for a data of type <T>.
  36  * <p>
  37  * This abstract class implements the housekeeking methods to register/ deregister the event provider and to handle
  38  * generically the event requests.
  39  * <p>
  40  * The concrete class can either re-implement processRequest() entirely or just implement the hooks (initializeContext()
  41  * and getNext()).
  42  * <p>
  43  * TODO: Add support for providing multiple data types.
  44  */
  45 public abstract class TmfDataProvider<T extends TmfData> extends TmfComponent implements ITmfDataProvider<T> {
  46
  47     // ------------------------------------------------------------------------
  48     // Constants
  49     // ------------------------------------------------------------------------
  50
  51     public static final int DEFAULT_BLOCK_SIZE = 50000;
  52     public static final int DEFAULT_QUEUE_SIZE = 1000;
  53
  54     // ------------------------------------------------------------------------
  55     // Attributes
  56     // ------------------------------------------------------------------------
  57
  58     protected Class<T> fType;
  59     protected boolean fLogData;
  60     protected boolean fLogError;
  61
  62     protected int fQueueSize = DEFAULT_QUEUE_SIZE;
  63     protected BlockingQueue<T> fDataQueue;
  64     protected TmfRequestExecutor fExecutor;
  65
  66     private int fSignalDepth = 0;
  67     private final Object fLock = new Object();
  68
  69     private int fRequestPendingCounter = 0;
  70
  71     // ------------------------------------------------------------------------
  72     // Constructors
  73     // ------------------------------------------------------------------------
  74
  75     public TmfDataProvider() {
  76         super();
  77         fQueueSize = DEFAULT_QUEUE_SIZE;
  78         fDataQueue = new LinkedBlockingQueue<T>(fQueueSize);
  79         fExecutor = new TmfRequestExecutor();
  80     }
  81
  82     public void init(String name, Class<T> dataType) {
  83         super.init(name);
  84         fType = dataType;
  85         fDataQueue = (fQueueSize > 1) ? new LinkedBlockingQueue<T>(fQueueSize) : new SynchronousQueue<T>();
  86
  87         fExecutor = new TmfRequestExecutor();
  88         fSignalDepth = 0;
  89
  90         fLogData = Tracer.isEventTraced();
  91         fLogError = Tracer.isErrorTraced();
  92
  93         TmfProviderManager.register(fType, this);
  94     }
  95
  96     protected TmfDataProvider(String name, Class<T> type, int queueSize) {
  97         this();
  98         fQueueSize = queueSize;
  99         init(name, type);
 100     }
 101
 102     public TmfDataProvider(TmfDataProvider<T> other) {
 103         this();
 104         init(other.getName(), other.fType);
 105     }
 106
 107     public TmfDataProvider(String name, Class<T> type) {
 108         this(name, type, DEFAULT_QUEUE_SIZE);
 109     }
 110
 111     @Override
 112     public void dispose() {
 113         TmfProviderManager.deregister(fType, this);
 114         fExecutor.stop();
 115         super.dispose();
 116         // if (Tracer.isComponentTraced()) Tracer.traceComponent(this, "stopped");
 117     }
 118
 119     // ------------------------------------------------------------------------
 120     // Accessors
 121     // ------------------------------------------------------------------------
 122
 123     public int getQueueSize() {
 124         return fQueueSize;
 125     }
 126
 127     public Class<?> getType() {
 128         return fType;
 129     }
 130
 131     // ------------------------------------------------------------------------
 132     // ITmfRequestHandler
 133     // ------------------------------------------------------------------------
 134
 135     @Override
 136     public void sendRequest(final ITmfDataRequest<T> request) {
 137         synchronized (fLock) {
 138             if (fSignalDepth > 0) {
 139                 coalesceDataRequest(request);
 140             } else {
 141                 dispatchRequest(request);
 142             }
 143         }
 144     }
 145
 146     /**
 147      * This method queues the coalesced requests.
 148      *
 149      * @param thread
 150      */
 151     @Override
 152     public void fireRequest() {
 153         synchronized (fLock) {
 154             if (fRequestPendingCounter > 0) {
 155                 return;
 156             }
 157             if (fPendingCoalescedRequests.size() > 0) {
 158                 for (TmfDataRequest<T> request : fPendingCoalescedRequests) {
 159                     dispatchRequest(request);
 160                 }
 161                 fPendingCoalescedRequests.clear();
 162             }
 163         }
 164     }
 165
 166     /**
 167      * Increments/decrements the pending requests counters and fires the request if necessary (counter == 0). Used for
 168      * coalescing requests accross multiple TmfDataProvider.
 169      *
 170      * @param isIncrement
 171      */
 172     @Override
 173     public void notifyPendingRequest(boolean isIncrement) {
 174         synchronized (fLock) {
 175             if (isIncrement) {
 176                 if (fSignalDepth > 0) {
 177                     fRequestPendingCounter++;
 178                 }
 179             } else {
 180                 if (fRequestPendingCounter > 0) {
 181                     fRequestPendingCounter--;
 182                 }
 183
 184                 // fire request if all pending requests are received
 185                 if (fRequestPendingCounter == 0) {
 186                     fireRequest();
 187                 }
 188             }
 189         }
 190     }
 191
 192     // ------------------------------------------------------------------------
 193     // Coalescing (primitive test...)
 194     // ------------------------------------------------------------------------
 195
 196     protected Vector<TmfCoalescedDataRequest<T>> fPendingCoalescedRequests = new Vector<TmfCoalescedDataRequest<T>>();
 197
 198     protected void newCoalescedDataRequest(ITmfDataRequest<T> request) {
 199         synchronized (fLock) {
 200             TmfCoalescedDataRequest<T> coalescedRequest = new TmfCoalescedDataRequest<T>(fType, request.getIndex(),
 201                     request.getNbRequested(), request.getBlockSize(), request.getExecType());
 202             coalescedRequest.addRequest(request);
 203             if (Tracer.isRequestTraced()) {
 204                 Tracer.traceRequest(request, "coalesced with " + coalescedRequest.getRequestId()); //$NON-NLS-1$
 205             }
 206             fPendingCoalescedRequests.add(coalescedRequest);
 207         }
 208     }
 209
 210     protected void coalesceDataRequest(ITmfDataRequest<T> request) {
 211         synchronized (fLock) {
 212             for (TmfCoalescedDataRequest<T> coalescedRequest : fPendingCoalescedRequests) {
 213                 if (coalescedRequest.isCompatible(request)) {
 214                     coalescedRequest.addRequest(request);
 215                     if (Tracer.isRequestTraced()) {
 216                         Tracer.traceRequest(request, "coalesced with " + coalescedRequest.getRequestId()); //$NON-NLS-1$
 217                     }
 218                     return;
 219                 }
 220             }
 221             newCoalescedDataRequest(request);
 222         }
 223     }
 224
 225     // ------------------------------------------------------------------------
 226     // Request processing
 227     // ------------------------------------------------------------------------
 228
 229     private void dispatchRequest(final ITmfDataRequest<T> request) {
 230         if (request.getExecType() == ExecutionType.FOREGROUND)
 231             queueRequest(request);
 232         else
 233             queueBackgroundRequest(request, request.getBlockSize(), true);
 234     }
 235
 236     protected void queueRequest(final ITmfDataRequest<T> request) {
 237
 238         if (fExecutor.isShutdown()) {
 239             request.cancel();
 240             return;
 241         }
 242
 243         final TmfDataProvider<T> provider = this;
 244
 245         // Process the request
 246         TmfThread thread = new TmfThread(request.getExecType()) {
 247
 248             @Override
 249             public void run() {
 250
 251                 if (Tracer.isRequestTraced())
 252                     Tracer.trace("Request #" + request.getRequestId() + " is being serviced by " + provider.getName()); //$NON-NLS-1$//$NON-NLS-2$
 253
 254                 // Extract the generic information
 255                 request.start();
 256                 int nbRequested = request.getNbRequested();
 257                 int nbRead = 0;
 258
 259                 // Initialize the execution
 260                 ITmfContext context = armRequest(request);
 261                 if (context == null) {
 262                     request.cancel();
 263                     return;
 264                 }
 265
 266                 try {
 267                     // Get the ordered events
 268                     T data = getNext(context);
 269                     if (Tracer.isRequestTraced())
 270                         Tracer.trace("Request #" + request.getRequestId() + " read first event"); //$NON-NLS-1$ //$NON-NLS-2$
 271                     while (data != null && !isCompleted(request, data, nbRead)) {
 272                         if (fLogData)
 273                             Tracer.traceEvent(provider, request, data);
 274                         request.handleData(data);
 275
 276                         // To avoid an unnecessary read passed the last data
 277                         // requested
 278                         if (++nbRead < nbRequested) {
 279                             data = getNext(context);
 280                         }
 281                     }
 282                     if (Tracer.isRequestTraced())
 283                         Tracer.trace("Request #" + request.getRequestId() + " finished"); //$NON-NLS-1$//$NON-NLS-2$
 284
 285                     if (request.isCancelled()) {
 286                         request.cancel();
 287                     } else {
 288                         request.done();
 289                     }
 290                 } catch (Exception e) {
 291                     request.fail();
 292                 }
 293
 294                 // Cleanup
 295                 context.dispose();
 296             }
 297
 298             @Override
 299             public void cancel() {
 300                 if (!request.isCompleted()) {
 301                     request.cancel();
 302                 }
 303             }
 304         };
 305
 306         if (Tracer.isRequestTraced())
 307             Tracer.traceRequest(request, "queued"); //$NON-NLS-1$
 308         fExecutor.execute(thread);
 309
 310     }
 311
 312     protected void queueBackgroundRequest(final ITmfDataRequest<T> request, final int blockSize, final boolean indexing) {
 313
 314         Thread thread = new Thread() {
 315             @Override
 316             public void run() {
 317                 request.start();
 318
 319                 final Integer[] CHUNK_SIZE = new Integer[1];
 320                 CHUNK_SIZE[0] = Math.min(request.getNbRequested(), blockSize + ((indexing) ? 1 : 0));
 321
 322                 final Integer[] nbRead = new Integer[1];
 323                 nbRead[0] = 0;
 324
 325                 final Boolean[] isFinished = new Boolean[1];
 326                 isFinished[0] = Boolean.FALSE;
 327
 328                 while (!isFinished[0]) {
 329
 330                     TmfDataRequest<T> subRequest = new TmfDataRequest<T>(request.getDataType(), request.getIndex()
 331                             + nbRead[0], CHUNK_SIZE[0], blockSize, ExecutionType.BACKGROUND) {
 332                         @Override
 333                         public void handleData(T data) {
 334                             super.handleData(data);
 335                             request.handleData(data);
 336                             if (getNbRead() > CHUNK_SIZE[0]) {
 337                                 System.out.println("ERROR - Read too many events"); //$NON-NLS-1$
 338                             }
 339                         }
 340
 341                         @Override
 342                         public void handleCompleted() {
 343                             nbRead[0] += getNbRead();
 344                             if (nbRead[0] >= request.getNbRequested() || (getNbRead() < CHUNK_SIZE[0])) {
 345                                 if (this.isCancelled()) {
 346                                     request.cancel();
 347                                 } else if (this.isFailed()) {
 348                                     request.fail();
 349                                 } else {
 350                                     request.done();
 351                                 }
 352                                 isFinished[0] = Boolean.TRUE;
 353                             }
 354                             super.handleCompleted();
 355                         }
 356                     };
 357
 358                     if (!isFinished[0]) {
 359                         queueRequest(subRequest);
 360
 361                         try {
 362                             subRequest.waitForCompletion();
 363                         } catch (InterruptedException e) {
 364                             e.printStackTrace();
 365                         }
 366
 367                         CHUNK_SIZE[0] = Math.min(request.getNbRequested() - nbRead[0], blockSize);
 368                     }
 369                 }
 370             }
 371         };
 372
 373         thread.start();
 374     }
 375
 376     /**
 377      * Initialize the provider based on the request. The context is provider specific and will be updated by getNext().
 378      *
 379      * @param request
 380      * @return an application specific context; null if request can't be serviced
 381      */
 382     public abstract ITmfContext armRequest(ITmfDataRequest<T> request);
 383
 384     public abstract T getNext(ITmfContext context);
 385
 386     /**
 387      * Checks if the data meets the request completion criteria.
 388      *
 389      * @param request
 390      * @param data
 391      * @return
 392      */
 393     public boolean isCompleted(ITmfDataRequest<T> request, T data, int nbRead) {
 394         return request.isCompleted() || nbRead >= request.getNbRequested() || data.isNullRef();
 395     }
 396
 397     // ------------------------------------------------------------------------
 398     // Signal handlers
 399     // ------------------------------------------------------------------------
 400
 401     @TmfSignalHandler
 402     public void startSynch(TmfStartSynchSignal signal) {
 403         synchronized (fLock) {
 404             fSignalDepth++;
 405         }
 406     }
 407
 408     @TmfSignalHandler
 409     public void endSynch(TmfEndSynchSignal signal) {
 410         synchronized (fLock) {
 411             fSignalDepth--;
 412             if (fSignalDepth == 0) {
 413                 fireRequest();
 414             }
 415         }
 416     }
 417
 418 }