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.internal.tmf.core.TmfCoreTracer;
  21 import org.eclipse.linuxtools.internal.tmf.core.component.TmfProviderManager;
  22 import org.eclipse.linuxtools.internal.tmf.core.component.TmfThread;
  23 import org.eclipse.linuxtools.internal.tmf.core.request.TmfCoalescedDataRequest;
  24 import org.eclipse.linuxtools.internal.tmf.core.request.TmfRequestExecutor;
  25 import org.eclipse.linuxtools.tmf.core.event.ITmfEvent;
  26 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest;
  27 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest.ExecutionType;
  28 import org.eclipse.linuxtools.tmf.core.request.TmfDataRequest;
  29 import org.eclipse.linuxtools.tmf.core.signal.TmfEndSynchSignal;
  30 import org.eclipse.linuxtools.tmf.core.signal.TmfSignalHandler;
  31 import org.eclipse.linuxtools.tmf.core.signal.TmfStartSynchSignal;
  32 import org.eclipse.linuxtools.tmf.core.trace.ITmfContext;
  33
  34 /**
  35  * An abstract base class that implements ITmfDataProvider.
  36  * <p>
  37  * This abstract class implements the housekeeping methods to register/
  38  * de-register the event provider and to handle generically the event requests.
  39  * <p>
  40  * The concrete class can either re-implement processRequest() entirely or just
  41  * implement the hooks (initializeContext() and getNext()).
  42  * <p>
  43  * TODO: Add support for providing multiple data types.
  44  *
  45  * @version 1.0
  46  * @author Francois Chouinard
  47  */
  48 public abstract class TmfDataProvider extends TmfComponent implements ITmfDataProvider {
  49
  50     // ------------------------------------------------------------------------
  51     // Constants
  52     // ------------------------------------------------------------------------
  53
  54     /** Default amount of events per request "chunk" */
  55     public static final int DEFAULT_BLOCK_SIZE = 50000;
  56
  57     /** Default size of the queue */
  58     public static final int DEFAULT_QUEUE_SIZE = 1000;
  59
  60     // ------------------------------------------------------------------------
  61     // Attributes
  62     // ------------------------------------------------------------------------
  63
  64     /** The type of event handled by this provider */
  65     protected Class<? extends ITmfEvent> fType;
  66
  67     /** Is there some data being logged? */
  68     protected boolean fLogData;
  69
  70     /** Are errors being logged? */
  71     protected boolean fLogError;
  72
  73     /** Queue of events */
  74     protected BlockingQueue<ITmfEvent> fDataQueue;
  75
  76     /** Size of the fDataQueue */
  77     protected int fQueueSize = DEFAULT_QUEUE_SIZE;
  78
  79     private TmfRequestExecutor fExecutor;
  80
  81     private int fSignalDepth = 0;
  82     private final Object fLock = new Object();
  83
  84     private int fRequestPendingCounter = 0;
  85
  86     // ------------------------------------------------------------------------
  87     // Constructors
  88     // ------------------------------------------------------------------------
  89
  90     /**
  91      * Default constructor
  92      */
  93     public TmfDataProvider() {
  94         super();
  95         fQueueSize = DEFAULT_QUEUE_SIZE;
  96         fDataQueue = new LinkedBlockingQueue<ITmfEvent>(fQueueSize);
  97         fExecutor = new TmfRequestExecutor();
  98     }
  99
 100     /**
 101      * Initialize this data provider
 102      *
 103      * @param name
 104      *            Name of the provider
 105      * @param type
 106      *            The type of events that will be handled
 107      */
 108     public void init(String name, Class<? extends ITmfEvent> type) {
 109         super.init(name);
 110         fType = type;
 111         fDataQueue = (fQueueSize > 1) ? new LinkedBlockingQueue<ITmfEvent>(fQueueSize) : new SynchronousQueue<ITmfEvent>();
 112
 113         fExecutor = new TmfRequestExecutor();
 114         fSignalDepth = 0;
 115
 116         fLogData = TmfCoreTracer.isEventTraced();
 117 //        fLogError = TmfCoreTracer.isErrorTraced();
 118
 119         TmfProviderManager.register(fType, this);
 120     }
 121
 122     /**
 123      * Constructor specifying the event type and the queue size.
 124      *
 125      * @param name
 126      *            Name of the provider
 127      * @param type
 128      *            Type of event that will be handled
 129      * @param queueSize
 130      *            Size of the event queue
 131      */
 132     protected TmfDataProvider(String name, Class<? extends ITmfEvent> type, int queueSize) {
 133         this();
 134         fQueueSize = queueSize;
 135         init(name, type);
 136     }
 137
 138     /**
 139      * Copy constructor
 140      *
 141      * @param other
 142      *            The other object to copy
 143      */
 144     public TmfDataProvider(TmfDataProvider other) {
 145         this();
 146         init(other.getName(), other.fType);
 147     }
 148
 149     /**
 150      * Standard constructor. Instantiate and initialize at the same time.
 151      *
 152      * @param name
 153      *            Name of the provider
 154      * @param type
 155      *            The type of events that will be handled
 156      */
 157     public TmfDataProvider(String name, Class<? extends ITmfEvent> type) {
 158         this(name, type, DEFAULT_QUEUE_SIZE);
 159     }
 160
 161     @Override
 162     public void dispose() {
 163         TmfProviderManager.deregister(fType, this);
 164         fExecutor.stop();
 165         super.dispose();
 166         // if (Tracer.isComponentTraced()) Tracer.traceComponent(this, "stopped");
 167     }
 168
 169     // ------------------------------------------------------------------------
 170     // Accessors
 171     // ------------------------------------------------------------------------
 172
 173     /**
 174      * Get the queue size of this provider
 175      *
 176      * @return The size of the queue
 177      */
 178     public int getQueueSize() {
 179         return fQueueSize;
 180     }
 181
 182     /**
 183      * Get the event type this provider handles
 184      *
 185      * @return The type of ITmfEvent
 186      */
 187     public Class<?> getType() {
 188         return fType;
 189     }
 190
 191     // ------------------------------------------------------------------------
 192     // ITmfRequestHandler
 193     // ------------------------------------------------------------------------
 194
 195     @Override
 196     public void sendRequest(final ITmfDataRequest request) {
 197         synchronized (fLock) {
 198             if (fSignalDepth > 0) {
 199                 coalesceDataRequest(request);
 200             } else {
 201                 dispatchRequest(request);
 202             }
 203         }
 204     }
 205
 206     @Override
 207     public void fireRequest() {
 208         synchronized (fLock) {
 209             if (fRequestPendingCounter > 0) {
 210                 return;
 211             }
 212             if (fPendingCoalescedRequests.size() > 0) {
 213                 for (TmfDataRequest request : fPendingCoalescedRequests) {
 214                     dispatchRequest(request);
 215                 }
 216                 fPendingCoalescedRequests.clear();
 217             }
 218         }
 219     }
 220
 221     /**
 222      * Increments/decrements the pending requests counters and fires the request
 223      * if necessary (counter == 0). Used for coalescing requests across multiple
 224      * TmfDataProvider's.
 225      *
 226      * @param isIncrement
 227      *            Should we increment (true) or decrement (false) the pending
 228      *            counter
 229      */
 230     @Override
 231     public void notifyPendingRequest(boolean isIncrement) {
 232         synchronized (fLock) {
 233             if (isIncrement) {
 234                 if (fSignalDepth > 0) {
 235                     fRequestPendingCounter++;
 236                 }
 237             } else {
 238                 if (fRequestPendingCounter > 0) {
 239                     fRequestPendingCounter--;
 240                 }
 241
 242                 // fire request if all pending requests are received
 243                 if (fRequestPendingCounter == 0) {
 244                     fireRequest();
 245                 }
 246             }
 247         }
 248     }
 249
 250     // ------------------------------------------------------------------------
 251     // Coalescing (primitive test...)
 252     // ------------------------------------------------------------------------
 253
 254     /** List of coalesced requests */
 255     protected Vector<TmfCoalescedDataRequest> fPendingCoalescedRequests = new Vector<TmfCoalescedDataRequest>();
 256
 257     /**
 258      * Create a new request from an existing one, and add it to the coalesced
 259      * requests
 260      *
 261      * @param request
 262      *            The request to copy
 263      */
 264     protected void newCoalescedDataRequest(ITmfDataRequest request) {
 265         synchronized (fLock) {
 266             TmfCoalescedDataRequest coalescedRequest = new TmfCoalescedDataRequest(request.getDataType(), request.getIndex(),
 267                     request.getNbRequested(), request.getBlockSize(), request.getExecType());
 268             coalescedRequest.addRequest(request);
 269             if (TmfCoreTracer.isRequestTraced()) {
 270                 TmfCoreTracer.traceRequest(request, "COALESCED with " + coalescedRequest.getRequestId()); //$NON-NLS-1$
 271                 TmfCoreTracer.traceRequest(coalescedRequest, "now contains " + coalescedRequest.getSubRequestIds()); //$NON-NLS-1$
 272             }
 273             fPendingCoalescedRequests.add(coalescedRequest);
 274         }
 275     }
 276
 277     /**
 278      * Add an existing requests to the list of coalesced ones
 279      *
 280      * @param request
 281      *            The request to add to the list
 282      */
 283     protected void coalesceDataRequest(ITmfDataRequest request) {
 284         synchronized (fLock) {
 285             for (TmfCoalescedDataRequest coalescedRequest : fPendingCoalescedRequests) {
 286                 if (coalescedRequest.isCompatible(request)) {
 287                     coalescedRequest.addRequest(request);
 288                     if (TmfCoreTracer.isRequestTraced()) {
 289                         TmfCoreTracer.traceRequest(request, "COALESCED with " + coalescedRequest.getRequestId()); //$NON-NLS-1$
 290                         TmfCoreTracer.traceRequest(coalescedRequest, "now contains " + coalescedRequest.getSubRequestIds()); //$NON-NLS-1$
 291                     }
 292                     return;
 293                 }
 294             }
 295             newCoalescedDataRequest(request);
 296         }
 297     }
 298
 299     // ------------------------------------------------------------------------
 300     // Request processing
 301     // ------------------------------------------------------------------------
 302
 303     private void dispatchRequest(final ITmfDataRequest request) {
 304         if (request.getExecType() == ExecutionType.FOREGROUND) {
 305             queueRequest(request);
 306         } else {
 307             queueBackgroundRequest(request, request.getBlockSize(), true);
 308         }
 309     }
 310
 311     /**
 312      * Queue a request.
 313      *
 314      * @param request
 315      *            The data request
 316      */
 317     protected void queueRequest(final ITmfDataRequest request) {
 318
 319         if (fExecutor.isShutdown()) {
 320             request.cancel();
 321             return;
 322         }
 323
 324         final TmfDataProvider provider = this;
 325
 326         // Process the request
 327         TmfThread thread = new TmfThread(request.getExecType()) {
 328
 329             @Override
 330             public void run() {
 331
 332                 if (TmfCoreTracer.isRequestTraced()) {
 333                     TmfCoreTracer.traceRequest(request, "is being serviced by " + provider.getName()); //$NON-NLS-1$
 334                 }
 335
 336                 // Extract the generic information
 337                 request.start();
 338                 int nbRequested = request.getNbRequested();
 339                 int nbRead = 0;
 340
 341                 // Initialize the execution
 342                 ITmfContext context = armRequest(request);
 343                 if (context == null) {
 344                     request.cancel();
 345                     return;
 346                 }
 347
 348                 try {
 349                     // Get the ordered events
 350                     ITmfEvent data = getNext(context);
 351                     if (TmfCoreTracer.isRequestTraced()) {
 352                         TmfCoreTracer.traceRequest(request, "read first event"); //$NON-NLS-1$
 353                     }
 354                     while (data != null && !isCompleted(request, data, nbRead)) {
 355                         if (fLogData) {
 356                             TmfCoreTracer.traceEvent(provider, request, data);
 357                         }
 358                         if (request.getDataType().isInstance(data)) {
 359                             request.handleData(data);
 360                         }
 361
 362                         // To avoid an unnecessary read passed the last data
 363                         // requested
 364                         if (++nbRead < nbRequested) {
 365                             data = getNext(context);
 366                         }
 367                     }
 368                     if (TmfCoreTracer.isRequestTraced()) {
 369                         TmfCoreTracer.traceRequest(request, "COMPLETED"); //$NON-NLS-1$
 370                     }
 371
 372                     if (request.isCancelled()) {
 373                         request.cancel();
 374                     } else {
 375                         request.done();
 376                     }
 377                 } catch (Exception e) {
 378                     request.fail();
 379                 }
 380
 381                 // Cleanup
 382                 context.dispose();
 383             }
 384
 385             @Override
 386             public void cancel() {
 387                 if (!request.isCompleted()) {
 388                     request.cancel();
 389                 }
 390             }
 391         };
 392
 393         if (TmfCoreTracer.isRequestTraced()) {
 394             TmfCoreTracer.traceRequest(request, "QUEUED"); //$NON-NLS-1$
 395         }
 396         fExecutor.execute(thread);
 397     }
 398
 399     /**
 400      * Queue a background request
 401      *
 402      * @param request
 403      *            The request
 404      * @param blockSize
 405      *            The request should be split in chunks of this size
 406      * @param indexing
 407      *            Should we index the chunks
 408      */
 409     protected void queueBackgroundRequest(final ITmfDataRequest request,
 410             final int blockSize, final boolean indexing) {
 411
 412         final TmfDataProvider provider = this;
 413
 414         Thread thread = new Thread() {
 415             @Override
 416             public void run() {
 417
 418                 if (TmfCoreTracer.isRequestTraced()) {
 419                     TmfCoreTracer.traceRequest(request, "is being serviced by " + provider.getName()); //$NON-NLS-1$
 420                 }
 421
 422                 request.start();
 423
 424                 final Integer[] CHUNK_SIZE = new Integer[1];
 425                 CHUNK_SIZE[0] = Math.min(request.getNbRequested(), blockSize + ((indexing) ? 1 : 0));
 426
 427                 final Integer[] nbRead = new Integer[1];
 428                 nbRead[0] = 0;
 429
 430                 final Boolean[] isFinished = new Boolean[1];
 431                 isFinished[0] = Boolean.FALSE;
 432
 433                 while (!isFinished[0]) {
 434
 435                     TmfDataRequest subRequest = new TmfDataRequest(request.getDataType(), request.getIndex()
 436                             + nbRead[0], CHUNK_SIZE[0], blockSize, ExecutionType.BACKGROUND) {
 437
 438                         @Override
 439                         public synchronized boolean isCompleted() {
 440                             return super.isCompleted() || request.isCompleted();
 441                         }
 442
 443                         @Override
 444                         public void handleData(ITmfEvent data) {
 445                             super.handleData(data);
 446                             if (request.getDataType().isInstance(data)) {
 447                                 request.handleData(data);
 448                             }
 449                             if (getNbRead() > CHUNK_SIZE[0]) {
 450                                 System.out.println("ERROR - Read too many events"); //$NON-NLS-1$
 451                             }
 452                         }
 453
 454                         @Override
 455                         public void handleCompleted() {
 456                             nbRead[0] += getNbRead();
 457                             if (nbRead[0] >= request.getNbRequested() || (getNbRead() < CHUNK_SIZE[0])) {
 458                                 if (this.isCancelled()) {
 459                                     request.cancel();
 460                                 } else if (this.isFailed()) {
 461                                     request.fail();
 462                                 } else {
 463                                     request.done();
 464                                 }
 465                                 isFinished[0] = Boolean.TRUE;
 466                             }
 467                             super.handleCompleted();
 468                         }
 469                     };
 470
 471                     if (!isFinished[0]) {
 472                         queueRequest(subRequest);
 473
 474                         try {
 475                             subRequest.waitForCompletion();
 476                             if (request.isCompleted()) {
 477                                 isFinished[0] = Boolean.TRUE;
 478                             }
 479                         } catch (InterruptedException e) {
 480                             e.printStackTrace();
 481                         }
 482
 483                         CHUNK_SIZE[0] = Math.min(request.getNbRequested() - nbRead[0], blockSize);
 484                     }
 485                 }
 486             }
 487         };
 488
 489         thread.start();
 490     }
 491
 492     /**
 493      * Initialize the provider based on the request. The context is provider
 494      * specific and will be updated by getNext().
 495      *
 496      * @param request
 497      *            The request
 498      * @return Sn application specific context; null if request can't be
 499      *         serviced
 500      */
 501     protected abstract ITmfContext armRequest(ITmfDataRequest request);
 502
 503 //    /**
 504 //     * Return the next event based on the context supplied. The context
 505 //     * will be updated for the subsequent read.
 506 //     *
 507 //     * @param context the trace read context (updated)
 508 //     * @return the event referred to by context
 509 //     */
 510 //    public abstract T getNext(ITmfContext context);
 511
 512     /**
 513      * Checks if the data meets the request completion criteria.
 514      *
 515      * @param request the request
 516      * @param data the data to verify
 517      * @param nbRead the number of events read so far
 518      * @return true if completion criteria is met
 519      */
 520     public boolean isCompleted(ITmfDataRequest request, ITmfEvent data, int nbRead) {
 521         return request.isCompleted() || nbRead >= request.getNbRequested();
 522     }
 523
 524     // ------------------------------------------------------------------------
 525     // Pass-through's to the request executor
 526     // ------------------------------------------------------------------------
 527
 528     /**
 529      * @return the shutdown state (i.e. if it is accepting new requests)
 530      * @since 2.0
 531      */
 532     protected boolean executorIsShutdown() {
 533         return fExecutor.isShutdown();
 534     }
 535
 536     /**
 537      * @return the termination state
 538      * @since 2.0
 539      */
 540     protected boolean executorIsTerminated() {
 541         return fExecutor.isTerminated();
 542     }
 543
 544     // ------------------------------------------------------------------------
 545     // Signal handlers
 546     // ------------------------------------------------------------------------
 547
 548     /**
 549      * Handler for the start synch signal
 550      *
 551      * @param signal
 552      *            Incoming signal
 553      */
 554     @TmfSignalHandler
 555     public void startSynch(TmfStartSynchSignal signal) {
 556         synchronized (fLock) {
 557             fSignalDepth++;
 558         }
 559     }
 560
 561     /**
 562      * Handler for the end synch signal
 563      *
 564      * @param signal
 565      *            Incoming signal
 566      */
 567     @TmfSignalHandler
 568     public void endSynch(TmfEndSynchSignal signal) {
 569         synchronized (fLock) {
 570             fSignalDepth--;
 571             if (fSignalDepth == 0) {
 572                 fireRequest();
 573             }
 574         }
 575     }
 576
 577 }