com.bigdata.service.ndx.pipeline

Class AbstractSubtask<HS extends AbstractSubtaskStats,M extends AbstractMasterTask<? extends AbstractMasterStats<L,HS>,E,? extends AbstractSubtask,L>,E,L>

java.lang.Object

com.bigdata.service.ndx.pipeline.AbstractSubtask<HS,M,E,L>

Type Parameters:

HS - The generic type of the value returned by Callable.call() for the subtask.

M - The generic type of the master task implementation class.

E - The generic type of the elements in the chunks stored in the BlockingBuffer.

L - The generic type of the key used to lookup a subtask in the internal map (must be unique and must implement hashCode() and equals() per their contracts).

All Implemented Interfaces:

Callable<HS>

Direct Known Subclasses:

AbstractPendingSetSubtask, IndexPartitionWriteTask, MockSubtask

public abstract class AbstractSubtask<HS extends AbstractSubtaskStats,M extends AbstractMasterTask<? extends AbstractMasterStats<L,HS>,E,? extends AbstractSubtask,L>,E,L>
extends Object
implements Callable<HS>

Abstract implementation of a subtask for the AbstractMasterTask handles the protocol for startup and termination of the subtask. A concrete implementation must handle the chunks of elements being drained from the subtask's buffer via handleChunk(Object[]).

Version:

$Id$

Author:

Bryan Thompson

Field Summary

Fields

Modifier and Type	Field and Description
protected BlockingBuffer<E[]>	buffer The buffer on which the master is writing.
protected long	lastChunkAvailableNanos The timestamp at IAsynchronousIterator.hasNext(long, TimeUnit) last returned true when queried with a timeout of AbstractMasterTask.sinkPollTimeoutNanos nanoseconds.
protected long	lastChunkNanos The timestamp at which a chunk was last written on the buffer for this sink by the master.
protected L	locator The unique key for the subtask.
protected static org.apache.log4j.Logger	log
protected M	master The master.
protected IAsynchronousIterator<E[]>	src The iterator draining the buffer.
protected HS	stats The statistics used by this task.

Constructor Summary

Constructors

Constructor and Description
AbstractSubtask(M master, L locator, BlockingBuffer<E[]> buffer)

Method Summary

Methods

Modifier and Type	Method and Description
protected void	awaitPending() Wait until any asynchronous processing for the subtask is done.
HS	call()
protected void	cancelRemoteTask(boolean mayInterruptIfRunning) Cancel the remote task.
protected abstract boolean	handleChunk(E[] chunk) Process a chunk from the buffer.
protected void	handleRedirect(E[] chunk, Throwable cause) This method MUST be invoked if a sink receives a "stale locator exception" within handleChunk(Object[]).
protected abstract void	notifyClientOfRedirect(L locator, Throwable cause) Notify the client that the locator is stale.
String	toString()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Field Detail

log

protected static final transient org.apache.log4j.Logger log

master

protected final M extends AbstractMasterTask<? extends AbstractMasterStats<L,HS>,E,? extends AbstractSubtask,L> master

The master.

locator

protected final L locator

The unique key for the subtask.

buffer

protected final BlockingBuffer<E[]> buffer

The buffer on which the master is writing.

src

protected final IAsynchronousIterator<E[]> src

The iterator draining the buffer.

Note: DO NOT close this iterator from within call() as that would cause this task to interrupt itself!

stats

protected final HS extends AbstractSubtaskStats stats

The statistics used by this task.

lastChunkNanos

protected volatile long lastChunkNanos

The timestamp at which a chunk was last written on the buffer for this sink by the master. This is used to help determine whether or not a sink has become idle. A sink on which a master has recently written a chunk is not idle even if the sink has not read any chunks from its buffer.

lastChunkAvailableNanos

protected volatile long lastChunkAvailableNanos

The timestamp at IAsynchronousIterator.hasNext(long, TimeUnit) last returned true when queried with a timeout of AbstractMasterTask.sinkPollTimeoutNanos nanoseconds. This tests whether or not a chunk is available and is used to help decide if the sink has become idle. (A sink with an available chunk is never idle.)

Constructor Detail

AbstractSubtask

public AbstractSubtask(M master,
               L locator,
               BlockingBuffer<E[]> buffer)

Method Detail

toString

public String toString()

Overrides:

toString in class Object

call

public HS call()
                                     throws Exception

Specified by:

call in interface Callable<HS extends AbstractSubtaskStats>

Throws:

Exception

awaitPending

protected void awaitPending()
                     throws InterruptedException

Wait until any asynchronous processing for the subtask is done. This is an extension hook which is used if the remote task accepts chunks for processing and uses an asynchronous notification mechanism to indicate the success or failure of elements. The default implementation is a NOP.

Throws:

InterruptedException

cancelRemoteTask

protected void cancelRemoteTask(boolean mayInterruptIfRunning)
                         throws InterruptedException

Cancel the remote task. This is an extension hook which is used if the remote task accepts chunks for processing and uses an asynchronous notification mechanism to indicate the success or failure of elements. The default implementation is a NOP.

Throws:

InterruptedException

handleRedirect

protected void handleRedirect(E[] chunk,
                  Throwable cause)
                       throws InterruptedException

This method MUST be invoked if a sink receives a "stale locator exception" within handleChunk(Object[]).

This method asynchronously closes the sink, so that no further data may be written on it by setting the cause on BlockingBuffer.abort(Throwable). Next, the current chunk is placed onto the master's redirectQueue and the sink's buffer is drained, transferring all chunks which can be read from that buffer onto the master's redirectQueue.

Parameters:

chunk - The chunk which the sink was processing when it discovered that it need to redirect its outputs to a different sink (that is, a chunk which it had already read from its buffer and hence which needs to be redirected now).

cause - The stale locator exception.

Throws:

InterruptedException

handleChunk

protected abstract boolean handleChunk(E[] chunk)
                                throws Exception

Process a chunk from the buffer.

Returns:

true iff the task should exit immediately.

Throws:

Exception

notifyClientOfRedirect

protected abstract void notifyClientOfRedirect(L locator,
                          Throwable cause)

Notify the client that the locator is stale.

Parameters:

locator - The locator.

cause - The cause.