org.eclipse.tracecompass.tmf.core/src/org/eclipse/tracecompass/tmf/core/component/TmfEventProvider.java

   1 /*******************************************************************************
   2  * Copyright (c) 2009, 2014 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, replace background
  11  *       requests by preemptable requests
  12  *   Alexandre Montplaisir - Merge with TmfDataProvider
  13  *   Bernd Hufmann - Add timer based coalescing for background requests
  14  *******************************************************************************/
  15
  16 package org.eclipse.tracecompass.tmf.core.component;
  17
  18 import java.util.ArrayList;
  19 import java.util.Collections;
  20 import java.util.Iterator;
  21 import java.util.LinkedList;
  22 import java.util.List;
  23 import java.util.Timer;
  24 import java.util.TimerTask;
  25
  26 import org.eclipse.jdt.annotation.NonNull;
  27 import org.eclipse.tracecompass.common.core.NonNullUtils;
  28 import org.eclipse.tracecompass.internal.tmf.core.TmfCoreTracer;
  29 import org.eclipse.tracecompass.internal.tmf.core.component.TmfEventThread;
  30 import org.eclipse.tracecompass.internal.tmf.core.component.TmfProviderManager;
  31 import org.eclipse.tracecompass.internal.tmf.core.request.TmfCoalescedEventRequest;
  32 import org.eclipse.tracecompass.internal.tmf.core.request.TmfRequestExecutor;
  33 import org.eclipse.tracecompass.tmf.core.event.ITmfEvent;
  34 import org.eclipse.tracecompass.tmf.core.request.ITmfEventRequest;
  35 import org.eclipse.tracecompass.tmf.core.request.ITmfEventRequest.ExecutionType;
  36 import org.eclipse.tracecompass.tmf.core.signal.TmfEndSynchSignal;
  37 import org.eclipse.tracecompass.tmf.core.signal.TmfSignalHandler;
  38 import org.eclipse.tracecompass.tmf.core.signal.TmfStartSynchSignal;
  39 import org.eclipse.tracecompass.tmf.core.timestamp.ITmfTimestamp;
  40 import org.eclipse.tracecompass.tmf.core.trace.ITmfContext;
  41
  42 /**
  43  * An abstract base class that implements ITmfEventProvider.
  44  * <p>
  45  * This abstract class implements the housekeeping methods to register/
  46  * de-register the event provider and to handle generically the event requests.
  47  * </p>
  48  *
  49  * @author Francois Chouinard
  50  * @since 3.0
  51  */
  52 public abstract class TmfEventProvider extends TmfComponent implements ITmfEventProvider {
  53
  54     // ------------------------------------------------------------------------
  55     // Constants
  56     // ------------------------------------------------------------------------
  57
  58     /** Default amount of events per request "chunk"
  59      * @since 3.0 */
  60     public static final int DEFAULT_BLOCK_SIZE = 50000;
  61
  62     /** Delay for coalescing background requests (in milli-seconds) */
  63     private static final long DELAY = 1000;
  64
  65     // ------------------------------------------------------------------------
  66     // Attributes
  67     // ------------------------------------------------------------------------
  68
  69     /** List of coalesced requests */
  70     private final List<TmfCoalescedEventRequest> fPendingCoalescedRequests = new LinkedList<>();
  71
  72     /** The type of event handled by this provider */
  73     private Class<? extends ITmfEvent> fType;
  74
  75     private final TmfRequestExecutor fExecutor;
  76
  77     private final Object fLock = new Object();
  78
  79     private int fSignalDepth = 0;
  80
  81     private int fRequestPendingCounter = 0;
  82
  83     private Timer fTimer;
  84
  85     /** Current timer task */
  86     @NonNull private TimerTask fCurrentTask = new TimerTask() { @Override public void run() {} };
  87
  88     private boolean fIsTimerEnabled;
  89
  90     /**
  91      * The parent event provider.
  92      */
  93     private TmfEventProvider fParent = null;
  94     /**
  95      * The list if children event provider.
  96      */
  97     private final List<TmfEventProvider> fChildren = Collections.synchronizedList(new ArrayList<TmfEventProvider>());
  98
  99     // ------------------------------------------------------------------------
 100     // Constructors
 101     // ------------------------------------------------------------------------
 102
 103     /**
 104      * Default constructor
 105      */
 106     public TmfEventProvider() {
 107         super();
 108         setTimerEnabled(true);
 109         fExecutor = new TmfRequestExecutor();
 110     }
 111
 112     /**
 113      * Standard constructor. Instantiate and initialize at the same time.
 114      *
 115      * @param name
 116      *            Name of the provider
 117      * @param type
 118      *            The type of events that will be handled
 119      */
 120     public TmfEventProvider(String name, Class<? extends ITmfEvent> type) {
 121         this();
 122         init(name, type);
 123     }
 124
 125     /**
 126      * Initialize this data provider
 127      *
 128      * @param name
 129      *            Name of the provider
 130      * @param type
 131      *            The type of events that will be handled
 132      */
 133     public void init(String name, Class<? extends ITmfEvent> type) {
 134         super.init(name);
 135         fType = type;
 136         fExecutor.init();
 137
 138         fSignalDepth = 0;
 139
 140         synchronized (fLock) {
 141              fTimer = new Timer();
 142         }
 143
 144         TmfProviderManager.register(fType, this);
 145     }
 146
 147     @Override
 148     public void dispose() {
 149         TmfProviderManager.deregister(fType, this);
 150         fExecutor.stop();
 151         synchronized (fLock) {
 152             if (fTimer != null) {
 153                 fTimer.cancel();
 154             }
 155             fTimer = null;
 156         }
 157
 158         synchronized (fChildren) {
 159             for (TmfEventProvider child : fChildren) {
 160                 child.dispose();
 161             }
 162             fChildren.clear();
 163         }
 164         clearPendingRequests();
 165         super.dispose();
 166     }
 167
 168     // ------------------------------------------------------------------------
 169     // Accessors
 170     // ------------------------------------------------------------------------
 171
 172     /**
 173      * Get the event type this provider handles
 174      *
 175      * @return The type of ITmfEvent
 176      */
 177     public Class<? extends ITmfEvent> getType() {
 178         return fType;
 179     }
 180
 181     // ------------------------------------------------------------------------
 182     // ITmfRequestHandler
 183     // ------------------------------------------------------------------------
 184
 185     /**
 186      * @since 3.0
 187      */
 188     @Override
 189     public void sendRequest(final ITmfEventRequest request) {
 190         synchronized (fLock) {
 191
 192             if (TmfCoreTracer.isRequestTraced()) {
 193                 TmfCoreTracer.traceRequest(request.getRequestId(), "SENT to provider " + getName()); //$NON-NLS-1$
 194             }
 195
 196             if (request.getEventProvider() == null) {
 197                 request.setEventProvider(this);
 198             }
 199
 200             if (sendWithParent(request)) {
 201                 return;
 202             }
 203
 204             if (request.getExecType() == ExecutionType.FOREGROUND) {
 205                 if ((fSignalDepth > 0) || (fRequestPendingCounter > 0)) {
 206                     coalesceEventRequest(request);
 207                 } else {
 208                     queueRequest(request);
 209                 }
 210                 return;
 211             }
 212
 213             /*
 214              * Dispatch request in case timer is not running.
 215              */
 216             if (fTimer == null) {
 217                 queueRequest(request);
 218                 return;
 219             }
 220
 221             coalesceEventRequest(request);
 222
 223             if (fIsTimerEnabled) {
 224                 fCurrentTask.cancel();
 225                 fCurrentTask = new TimerTask() {
 226                     @Override
 227                     public void run() {
 228                         synchronized (fLock) {
 229                             fireRequest(true);
 230                         }
 231                     }
 232                 };
 233                 fTimer.schedule(fCurrentTask, DELAY);
 234             }
 235         }
 236     }
 237
 238     private void fireRequest(boolean isTimeout) {
 239         synchronized (fLock) {
 240             if (fRequestPendingCounter > 0) {
 241                 return;
 242             }
 243
 244             if (fPendingCoalescedRequests.size() > 0) {
 245                 Iterator<TmfCoalescedEventRequest> iter = fPendingCoalescedRequests.iterator();
 246                 while (iter.hasNext()) {
 247                     ExecutionType type = (isTimeout ? ExecutionType.BACKGROUND : ExecutionType.FOREGROUND);
 248                     ITmfEventRequest request = iter.next();
 249                     if (type == request.getExecType()) {
 250                         queueRequest(request);
 251                         iter.remove();
 252                     }
 253                 }
 254             }
 255         }
 256     }
 257
 258     /**
 259      * Increments/decrements the pending requests counters and fires the request
 260      * if necessary (counter == 0). Used for coalescing requests across multiple
 261      * TmfDataProvider's.
 262      *
 263      * @param isIncrement
 264      *            Should we increment (true) or decrement (false) the pending
 265      *            counter
 266      */
 267     @Override
 268     public void notifyPendingRequest(boolean isIncrement) {
 269         synchronized (fLock) {
 270             if (isIncrement) {
 271                 fRequestPendingCounter++;
 272             } else {
 273                 if (fRequestPendingCounter > 0) {
 274                     fRequestPendingCounter--;
 275                 }
 276
 277                 // fire request if all pending requests are received
 278                 if (fRequestPendingCounter == 0) {
 279                     fireRequest(false);
 280                     fireRequest(true);
 281                 }
 282             }
 283         }
 284     }
 285
 286     // ------------------------------------------------------------------------
 287     // Coalescing
 288     // ------------------------------------------------------------------------
 289
 290     /**
 291      * Create a new request from an existing one, and add it to the coalesced
 292      * requests
 293      *
 294      * @param request
 295      *            The request to copy
 296      * @since 3.0
 297      */
 298     protected void newCoalescedEventRequest(ITmfEventRequest request) {
 299         synchronized (fLock) {
 300             TmfCoalescedEventRequest coalescedRequest = new TmfCoalescedEventRequest(
 301                     request.getDataType(),
 302                     request.getRange(),
 303                     request.getIndex(),
 304                     request.getNbRequested(),
 305                     request.getExecType());
 306             coalescedRequest.addRequest(request);
 307             coalescedRequest.setEventProvider(this);
 308             if (TmfCoreTracer.isRequestTraced()) {
 309                 TmfCoreTracer.traceRequest(request.getRequestId(), "COALESCED with " + coalescedRequest.getRequestId()); //$NON-NLS-1$
 310                 TmfCoreTracer.traceRequest(coalescedRequest.getRequestId(), "now contains " + coalescedRequest.getSubRequestIds()); //$NON-NLS-1$
 311             }
 312             coalesceChildrenRequests(coalescedRequest);
 313             fPendingCoalescedRequests.add(coalescedRequest);
 314         }
 315     }
 316
 317     /**
 318      * Add an existing requests to the list of coalesced ones
 319      *
 320      * @param request
 321      *            The request to add to the list
 322      * @since 3.0
 323      */
 324     protected void coalesceEventRequest(ITmfEventRequest request) {
 325         synchronized (fLock) {
 326             for (TmfCoalescedEventRequest coalescedRequest : getPendingRequests()) {
 327                 if (coalescedRequest.isCompatible(request)) {
 328                     coalescedRequest.addRequest(request);
 329                     if (TmfCoreTracer.isRequestTraced()) {
 330                         TmfCoreTracer.traceRequest(request.getRequestId(), "COALESCED with " + coalescedRequest.getRequestId()); //$NON-NLS-1$
 331                         TmfCoreTracer.traceRequest(coalescedRequest.getRequestId(), "now contains " + coalescedRequest.getSubRequestIds()); //$NON-NLS-1$
 332                     }
 333                     coalesceChildrenRequests(coalescedRequest);
 334                     return;
 335                 }
 336             }
 337             newCoalescedEventRequest(request);
 338         }
 339     }
 340
 341     /*
 342      * Sends a request with the parent if compatible.
 343      */
 344     private boolean sendWithParent(final ITmfEventRequest request) {
 345         ITmfEventProvider parent = getParent();
 346         if (parent instanceof TmfEventProvider) {
 347             return ((TmfEventProvider) parent).sendIfCompatible(request);
 348         }
 349         return false;
 350     }
 351
 352     /*
 353      * Sends a request if compatible with a pending coalesced request.
 354      */
 355     private boolean sendIfCompatible(ITmfEventRequest request) {
 356         synchronized (fLock) {
 357             for (TmfCoalescedEventRequest coalescedRequest : getPendingRequests()) {
 358                 if (coalescedRequest.isCompatible(request)) {
 359                     // Send so it can be coalesced with the parent(s)
 360                     sendRequest(request);
 361                     return true;
 362                 }
 363             }
 364         }
 365         return sendWithParent(request);
 366     }
 367
 368     /*
 369      * Coalesces children requests with given request if compatible.
 370      */
 371     private void coalesceChildrenRequests(final TmfCoalescedEventRequest request) {
 372         synchronized (fChildren) {
 373             for (TmfEventProvider child : fChildren) {
 374                 child.coalesceCompatibleRequests(request);
 375             }
 376         }
 377     }
 378
 379
 380     /*
 381      * Coalesces all pending requests that are compatible with coalesced request.
 382      */
 383     private void coalesceCompatibleRequests(TmfCoalescedEventRequest request) {
 384         Iterator<TmfCoalescedEventRequest> iter = getPendingRequests().iterator();
 385         while (iter.hasNext()) {
 386             TmfCoalescedEventRequest pendingRequest = iter.next();
 387             if (request.isCompatible(pendingRequest)) {
 388                 request.addRequest(pendingRequest);
 389                 if (TmfCoreTracer.isRequestTraced()) {
 390                     TmfCoreTracer.traceRequest(pendingRequest.getRequestId(), "COALESCED with " + request.getRequestId()); //$NON-NLS-1$
 391                     TmfCoreTracer.traceRequest(request.getRequestId(), "now contains " + request.getSubRequestIds()); //$NON-NLS-1$
 392                 }
 393                 iter.remove();
 394             }
 395         }
 396     }
 397
 398     // ------------------------------------------------------------------------
 399     // Request processing
 400     // ------------------------------------------------------------------------
 401
 402     /**
 403      * Queue a request.
 404      *
 405      * @param request
 406      *            The data request
 407      * @since 3.0
 408      */
 409     protected void queueRequest(final ITmfEventRequest request) {
 410
 411         if (fExecutor.isShutdown()) {
 412             request.cancel();
 413             return;
 414         }
 415
 416         TmfEventThread thread = new TmfEventThread(this, request);
 417
 418         if (TmfCoreTracer.isRequestTraced()) {
 419             TmfCoreTracer.traceRequest(request.getRequestId(), "QUEUED"); //$NON-NLS-1$
 420         }
 421
 422         fExecutor.execute(thread);
 423     }
 424
 425     /**
 426      * Initialize the provider based on the request. The context is provider
 427      * specific and will be updated by getNext().
 428      *
 429      * @param request
 430      *            The request
 431      * @return An application specific context; null if request can't be
 432      *         serviced
 433      * @since 3.0
 434      */
 435     public abstract ITmfContext armRequest(ITmfEventRequest request);
 436
 437     /**
 438      * Checks if the data meets the request completion criteria.
 439      *
 440      * @param request
 441      *            The request
 442      * @param event
 443      *            The data to verify
 444      * @param nbRead
 445      *            The number of events read so far
 446      * @return true if completion criteria is met
 447      * @since 3.0
 448      */
 449     public boolean isCompleted(ITmfEventRequest request, ITmfEvent event, int nbRead) {
 450         boolean requestCompleted = isCompleted2(request, nbRead);
 451         if (!requestCompleted) {
 452             ITmfTimestamp endTime = request.getRange().getEndTime();
 453             return event.getTimestamp().compareTo(endTime) > 0;
 454         }
 455         return requestCompleted;
 456     }
 457
 458     private static boolean isCompleted2(ITmfEventRequest request,int nbRead) {
 459         return request.isCompleted() || nbRead >= request.getNbRequested();
 460     }
 461
 462     // ------------------------------------------------------------------------
 463     // Pass-through's to the request executor
 464     // ------------------------------------------------------------------------
 465
 466     /**
 467      * @return the shutdown state (i.e. if it is accepting new requests)
 468      * @since 2.0
 469      */
 470     protected boolean executorIsShutdown() {
 471         return fExecutor.isShutdown();
 472     }
 473
 474     /**
 475      * @return the termination state
 476      * @since 2.0
 477      */
 478     protected boolean executorIsTerminated() {
 479         return fExecutor.isTerminated();
 480     }
 481
 482     // ------------------------------------------------------------------------
 483     // Signal handlers
 484     // ------------------------------------------------------------------------
 485
 486     /**
 487      * Handler for the start synch signal
 488      *
 489      * @param signal
 490      *            Incoming signal
 491      */
 492     @TmfSignalHandler
 493     public void startSynch(TmfStartSynchSignal signal) {
 494         synchronized (fLock) {
 495             fSignalDepth++;
 496         }
 497     }
 498
 499     /**
 500      * Handler for the end synch signal
 501      *
 502      * @param signal
 503      *            Incoming signal
 504      */
 505     @TmfSignalHandler
 506     public void endSynch(TmfEndSynchSignal signal) {
 507         synchronized (fLock) {
 508             fSignalDepth--;
 509             if (fSignalDepth == 0) {
 510                 fireRequest(false);
 511             }
 512         }
 513     }
 514
 515     @Override
 516     public ITmfEventProvider getParent() {
 517         synchronized (fLock) {
 518             return fParent;
 519         }
 520     }
 521
 522     @Override
 523     public void setParent(ITmfEventProvider parent) {
 524         if (!(parent instanceof TmfEventProvider)) {
 525             throw new IllegalArgumentException();
 526         }
 527
 528         synchronized (fLock) {
 529             fParent = (TmfEventProvider) parent;
 530         }
 531     }
 532
 533     @Override
 534     public List<ITmfEventProvider> getChildren() {
 535         synchronized (fChildren) {
 536             List<ITmfEventProvider> list = new ArrayList<>();
 537             list.addAll(fChildren);
 538             return list;
 539         }
 540     }
 541
 542     @Override
 543     public <T extends ITmfEventProvider> List<T> getChildren(Class<T> clazz) {
 544        List<T> list = new ArrayList<>();
 545        synchronized (fChildren) {
 546            for (TmfEventProvider child : fChildren) {
 547                if (clazz.isAssignableFrom(child.getClass())) {
 548                    list.add(clazz.cast(child));
 549                }
 550            }
 551        }
 552        return list;
 553     }
 554
 555     @Override
 556     public ITmfEventProvider getChild(String name) {
 557         synchronized (fChildren) {
 558             for (TmfEventProvider child : fChildren) {
 559                 if (child.getName().equals(name)) {
 560                     return child;
 561                 }
 562             }
 563         }
 564         return null;
 565     }
 566
 567     @Override
 568     public ITmfEventProvider getChild(int index) {
 569         return NonNullUtils.checkNotNull(fChildren.get(index));
 570     }
 571
 572     @Override
 573     public void addChild(ITmfEventProvider child) {
 574         if (!(child instanceof TmfEventProvider)) {
 575             throw new IllegalArgumentException();
 576         }
 577         child.setParent(this);
 578         fChildren.add((TmfEventProvider)child);
 579     }
 580
 581     @Override
 582     public int getNbChildren() {
 583         return fChildren.size();
 584     }
 585
 586     @Override
 587     public boolean providesEvent(ITmfEvent event) {
 588         if ((event.getTrace() == this)) {
 589             return true;
 590         }
 591         if (fChildren.size() > 0) {
 592             synchronized (fLock) {
 593                 List <TmfEventProvider> children = getChildren(TmfEventProvider.class);
 594                 for (TmfEventProvider child : children) {
 595                     if (child.providesEvent(event)) {
 596                         return true;
 597                     }
 598                 }
 599             }
 600         }
 601         return false;
 602     }
 603
 604     // ------------------------------------------------------------------------
 605     // Debug code (will also used in tests using reflection)
 606     // ------------------------------------------------------------------------
 607
 608     /**
 609      * Gets a list of all pending requests. Debug code.
 610      *
 611      * @return a list of all pending requests
 612      */
 613     private List<TmfCoalescedEventRequest> getPendingRequests() {
 614         return fPendingCoalescedRequests;
 615     }
 616
 617     /**
 618      * Clears all pending requests. Debug code.
 619      */
 620     private void  clearPendingRequests() {
 621         fPendingCoalescedRequests.clear();
 622     }
 623
 624     /**
 625      * Enables/disables the timer. Debug code.
 626      *
 627      * @param enabled
 628      *            the enable flag to set
 629      */
 630     private void setTimerEnabled(Boolean enabled) {
 631         fIsTimerEnabled = enabled;
 632     }
 633 }