1 /*******************************************************************************
2 * Copyright (c) 2009, 2010 Ericsson
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
10 * Francois Chouinard - Initial API and implementation
11 *******************************************************************************/
13 package org
.eclipse
.linuxtools
.tmf
.core
.component
;
15 import java
.util
.Vector
;
16 import java
.util
.concurrent
.BlockingQueue
;
17 import java
.util
.concurrent
.LinkedBlockingQueue
;
18 import java
.util
.concurrent
.SynchronousQueue
;
20 import org
.eclipse
.linuxtools
.internal
.tmf
.core
.Tracer
;
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
;
35 * An abstract base class that implements ITmfDataProvider.
37 * This abstract class implements the housekeeping methods to register/
38 * de-register the event provider and to handle generically the event requests.
40 * The concrete class can either re-implement processRequest() entirely or just
41 * implement the hooks (initializeContext() and getNext()).
43 * TODO: Add support for providing multiple data types.
46 * @author Francois Chouinard
48 public abstract class TmfDataProvider
extends TmfComponent
implements ITmfDataProvider
{
50 // ------------------------------------------------------------------------
52 // ------------------------------------------------------------------------
54 /** Default amount of events per request "chunk" */
55 public static final int DEFAULT_BLOCK_SIZE
= 50000;
57 /** Default size of the queue */
58 public static final int DEFAULT_QUEUE_SIZE
= 1000;
60 // ------------------------------------------------------------------------
62 // ------------------------------------------------------------------------
64 protected Class
<?
extends ITmfEvent
> fType
;
65 protected boolean fLogData
;
66 protected boolean fLogError
;
68 protected int fQueueSize
= DEFAULT_QUEUE_SIZE
;
69 protected BlockingQueue
<ITmfEvent
> fDataQueue
;
70 private TmfRequestExecutor fExecutor
;
72 private int fSignalDepth
= 0;
73 private final Object fLock
= new Object();
75 private int fRequestPendingCounter
= 0;
77 // ------------------------------------------------------------------------
79 // ------------------------------------------------------------------------
84 public TmfDataProvider() {
86 fQueueSize
= DEFAULT_QUEUE_SIZE
;
87 fDataQueue
= new LinkedBlockingQueue
<ITmfEvent
>(fQueueSize
);
88 fExecutor
= new TmfRequestExecutor();
92 * Initialize this data provider
95 * Name of the provider
97 * The type of events that will be handled
99 public void init(String name
, Class
<?
extends ITmfEvent
> type
) {
102 fDataQueue
= (fQueueSize
> 1) ?
new LinkedBlockingQueue
<ITmfEvent
>(fQueueSize
) : new SynchronousQueue
<ITmfEvent
>();
104 fExecutor
= new TmfRequestExecutor();
107 fLogData
= Tracer
.isEventTraced();
108 fLogError
= Tracer
.isErrorTraced();
110 TmfProviderManager
.register(fType
, this);
113 protected TmfDataProvider(String name
, Class
<?
extends ITmfEvent
> type
, int queueSize
) {
115 fQueueSize
= queueSize
;
123 * The other object to copy
125 public TmfDataProvider(TmfDataProvider other
) {
127 init(other
.getName(), other
.fType
);
131 * Standard constructor. Instantiate and initialize at the same time.
134 * Name of the provider
136 * The type of events that will be handled
138 public TmfDataProvider(String name
, Class
<?
extends ITmfEvent
> type
) {
139 this(name
, type
, DEFAULT_QUEUE_SIZE
);
143 public void dispose() {
144 TmfProviderManager
.deregister(fType
, this);
147 // if (Tracer.isComponentTraced()) Tracer.traceComponent(this, "stopped");
150 // ------------------------------------------------------------------------
152 // ------------------------------------------------------------------------
155 * Get the queue size of this provider
157 * @return The size of the queue
159 public int getQueueSize() {
164 * Get the event type this provider handles
166 * @return The type of ITmfEvent
168 public Class
<?
> getType() {
172 // ------------------------------------------------------------------------
173 // ITmfRequestHandler
174 // ------------------------------------------------------------------------
177 public void sendRequest(final ITmfDataRequest request
) {
178 synchronized (fLock
) {
179 if (fSignalDepth
> 0) {
180 coalesceDataRequest(request
);
182 dispatchRequest(request
);
188 public void fireRequest() {
189 synchronized (fLock
) {
190 if (fRequestPendingCounter
> 0) {
193 if (fPendingCoalescedRequests
.size() > 0) {
194 for (TmfDataRequest request
: fPendingCoalescedRequests
) {
195 dispatchRequest(request
);
197 fPendingCoalescedRequests
.clear();
203 * Increments/decrements the pending requests counters and fires the request
204 * if necessary (counter == 0). Used for coalescing requests across multiple
208 * Should we increment (true) or decrement (false) the pending
212 public void notifyPendingRequest(boolean isIncrement
) {
213 synchronized (fLock
) {
215 if (fSignalDepth
> 0) {
216 fRequestPendingCounter
++;
219 if (fRequestPendingCounter
> 0) {
220 fRequestPendingCounter
--;
223 // fire request if all pending requests are received
224 if (fRequestPendingCounter
== 0) {
231 // ------------------------------------------------------------------------
232 // Coalescing (primitive test...)
233 // ------------------------------------------------------------------------
235 protected Vector
<TmfCoalescedDataRequest
> fPendingCoalescedRequests
= new Vector
<TmfCoalescedDataRequest
>();
237 protected void newCoalescedDataRequest(ITmfDataRequest request
) {
238 synchronized (fLock
) {
239 TmfCoalescedDataRequest coalescedRequest
= new TmfCoalescedDataRequest(request
.getDataType(), request
.getIndex(),
240 request
.getNbRequested(), request
.getBlockSize(), request
.getExecType());
241 coalescedRequest
.addRequest(request
);
242 if (Tracer
.isRequestTraced()) {
243 Tracer
.traceRequest(request
, "COALESCED with " + coalescedRequest
.getRequestId()); //$NON-NLS-1$
244 Tracer
.traceRequest(coalescedRequest
, "now contains " + coalescedRequest
.getSubRequestIds()); //$NON-NLS-1$
246 fPendingCoalescedRequests
.add(coalescedRequest
);
250 protected void coalesceDataRequest(ITmfDataRequest request
) {
251 synchronized (fLock
) {
252 for (TmfCoalescedDataRequest coalescedRequest
: fPendingCoalescedRequests
) {
253 if (coalescedRequest
.isCompatible(request
)) {
254 coalescedRequest
.addRequest(request
);
255 if (Tracer
.isRequestTraced()) {
256 Tracer
.traceRequest(request
, "COALESCED with " + coalescedRequest
.getRequestId()); //$NON-NLS-1$
257 Tracer
.traceRequest(coalescedRequest
, "now contains " + coalescedRequest
.getSubRequestIds()); //$NON-NLS-1$
262 newCoalescedDataRequest(request
);
266 // ------------------------------------------------------------------------
267 // Request processing
268 // ------------------------------------------------------------------------
270 private void dispatchRequest(final ITmfDataRequest request
) {
271 if (request
.getExecType() == ExecutionType
.FOREGROUND
) {
272 queueRequest(request
);
274 queueBackgroundRequest(request
, request
.getBlockSize(), true);
278 protected void queueRequest(final ITmfDataRequest request
) {
280 if (fExecutor
.isShutdown()) {
285 final TmfDataProvider provider
= this;
287 // Process the request
288 TmfThread thread
= new TmfThread(request
.getExecType()) {
293 if (Tracer
.isRequestTraced()) {
294 Tracer
.traceRequest(request
, "is being serviced by " + provider
.getName()); //$NON-NLS-1$
297 // Extract the generic information
299 int nbRequested
= request
.getNbRequested();
302 // Initialize the execution
303 ITmfContext context
= armRequest(request
);
304 if (context
== null) {
310 // Get the ordered events
311 ITmfEvent data
= getNext(context
);
312 if (Tracer
.isRequestTraced()) {
313 Tracer
.traceRequest(request
, "read first event"); //$NON-NLS-1$
315 while (data
!= null && !isCompleted(request
, data
, nbRead
)) {
317 Tracer
.traceEvent(provider
, request
, data
);
319 if (request
.getDataType().isInstance(data
)) {
320 request
.handleData(data
);
323 // To avoid an unnecessary read passed the last data
325 if (++nbRead
< nbRequested
) {
326 data
= getNext(context
);
329 if (Tracer
.isRequestTraced()) {
330 Tracer
.traceRequest(request
, "COMPLETED"); //$NON-NLS-1$
333 if (request
.isCancelled()) {
338 } catch (Exception e
) {
347 public void cancel() {
348 if (!request
.isCompleted()) {
354 if (Tracer
.isRequestTraced()) {
355 Tracer
.traceRequest(request
, "QUEUED"); //$NON-NLS-1$
357 fExecutor
.execute(thread
);
361 protected void queueBackgroundRequest(final ITmfDataRequest request
, final int blockSize
, final boolean indexing
) {
363 final TmfDataProvider provider
= this;
365 Thread thread
= new Thread() {
369 if (Tracer
.isRequestTraced()) {
370 Tracer
.traceRequest(request
, "is being serviced by " + provider
.getName()); //$NON-NLS-1$
375 final Integer
[] CHUNK_SIZE
= new Integer
[1];
376 CHUNK_SIZE
[0] = Math
.min(request
.getNbRequested(), blockSize
+ ((indexing
) ?
1 : 0));
378 final Integer
[] nbRead
= new Integer
[1];
381 final Boolean
[] isFinished
= new Boolean
[1];
382 isFinished
[0] = Boolean
.FALSE
;
384 while (!isFinished
[0]) {
386 TmfDataRequest subRequest
= new TmfDataRequest(request
.getDataType(), request
.getIndex()
387 + nbRead
[0], CHUNK_SIZE
[0], blockSize
, ExecutionType
.BACKGROUND
) {
390 public synchronized boolean isCompleted() {
391 return super.isCompleted() || request
.isCompleted();
395 public void handleData(ITmfEvent data
) {
396 super.handleData(data
);
397 if (request
.getDataType().isInstance(data
)) {
398 request
.handleData(data
);
400 if (getNbRead() > CHUNK_SIZE
[0]) {
401 System
.out
.println("ERROR - Read too many events"); //$NON-NLS-1$
406 public void handleCompleted() {
407 nbRead
[0] += getNbRead();
408 if (nbRead
[0] >= request
.getNbRequested() || (getNbRead() < CHUNK_SIZE
[0])) {
409 if (this.isCancelled()) {
411 } else if (this.isFailed()) {
416 isFinished
[0] = Boolean
.TRUE
;
418 super.handleCompleted();
422 if (!isFinished
[0]) {
423 queueRequest(subRequest
);
426 subRequest
.waitForCompletion();
427 if (request
.isCompleted()) {
428 isFinished
[0] = Boolean
.TRUE
;
430 } catch (InterruptedException e
) {
434 CHUNK_SIZE
[0] = Math
.min(request
.getNbRequested() - nbRead
[0], blockSize
);
444 * Initialize the provider based on the request. The context is provider
445 * specific and will be updated by getNext().
448 * @return an application specific context; null if request can't be serviced
450 protected abstract ITmfContext
armRequest(ITmfDataRequest request
);
453 // * Return the next event based on the context supplied. The context
454 // * will be updated for the subsequent read.
456 // * @param context the trace read context (updated)
457 // * @return the event referred to by context
459 // public abstract T getNext(ITmfContext context);
462 * Checks if the data meets the request completion criteria.
464 * @param request the request
465 * @param data the data to verify
466 * @param nbRead the number of events read so far
467 * @return true if completion criteria is met
469 public boolean isCompleted(ITmfDataRequest request
, ITmfEvent data
, int nbRead
) {
470 return request
.isCompleted() || nbRead
>= request
.getNbRequested();
473 // ------------------------------------------------------------------------
474 // Pass-through's to the request executor
475 // ------------------------------------------------------------------------
478 * @return the shutdown state (i.e. if it is accepting new requests)
481 protected boolean executorIsShutdown() {
482 return fExecutor
.isShutdown();
486 * @return the termination state
489 protected boolean executorIsTerminated() {
490 return fExecutor
.isTerminated();
493 // ------------------------------------------------------------------------
495 // ------------------------------------------------------------------------
498 * Handler for the start synch signal
504 public void startSynch(TmfStartSynchSignal signal
) {
505 synchronized (fLock
) {
511 * Handler for the end synch signal
517 public void endSynch(TmfEndSynchSignal signal
) {
518 synchronized (fLock
) {
520 if (fSignalDepth
== 0) {