com.bigdata.bop.fed

Class FederatedRunningQuery

java.lang.Object

com.bigdata.bop.engine.AbstractRunningQuery

com.bigdata.bop.engine.ChunkedRunningQuery

com.bigdata.bop.fed.FederatedRunningQuery

All Implemented Interfaces:

IRunningQuery, IQueryContext, IHaltable<Void>, Future<Void>

public class FederatedRunningQuery
extends ChunkedRunningQuery

Extends ChunkedRunningQuery to provide additional state and logic required to support distributed query evaluation against an IBigdataFederation .

Author:

Bryan Thompson

Field Summary

Fields inherited from class com.bigdata.bop.engine.AbstractRunningQuery

ERR_DUPLICATE_IDENTIFIER, ERR_NO_SUCH_BOP, ERR_NOT_CONTROLLER, ERR_NOT_PIPELINE_START, ERR_QUERY_DONE, lock

Constructor Summary

Constructors

Constructor and Description
FederatedRunningQuery(FederatedQueryEngine queryEngine, UUID queryId, boolean controller, IQueryClient clientProxy, UUID queryControllerId, PipelineOp query, IChunkMessage<IBindingSet> realSource)

Method Summary

Methods

Modifier and Type	Method and Description
protected boolean	acceptChunk(IChunkMessage<IBindingSet> msg) Overridden to make this visible to the FederatedQueryEngine.
protected boolean	cancelQueryOnPeers(Throwable cause, Set<UUID> startedOn) Overridden to notify each peer on which the query was started.
protected void	doLastPass(int bopId, Set doneOn) Overridden to broadcast to all nodes and/or shards on which the operator has run in scale-out.
DirectBufferPoolAllocator.IAllocationContext	getAllocationContext(AllocationContextKey key) Return the DirectBufferPoolAllocator.IAllocationContext for the given key.
protected IChunkHandler	getChunkHandler(QueryEngine queryEngine, PipelineOp query) Factory returns the effective IChunkHandler for this query.
FederatedQueryEngine	getQueryEngine() The query engine class executing the query on this node.
IQueryPeer	getQueryPeer(UUID serviceUUID) Resolve the proxy for an IQueryPeer.
protected void	releaseNativeMemoryForOperator(int bopId) Extended to release all allocations associated with the specified operator.
protected void	releaseNativeMemoryForQuery() Extended to release all DirectBufferPoolAllocator.IAllocationContexts associated with the query when it terminates.

Methods inherited from class com.bigdata.bop.engine.ChunkedRunningQuery

cancelRunningOperators, consumeChunk, getChunkHandler, getQueueStats, haltOp, releaseAcceptedMessages

Methods inherited from class com.bigdata.bop.engine.AbstractRunningQuery

addChild, cancel, checkDeadline, get, get, getAsThrownCause, getAttributes, getBOp, getBOpIndex, getCause, getChildren, getDeadline, getDoneTime, getElapsed, getFederation, getFuture, getLocalIndexManager, getMemoryManager, getQuery, getQueryBuffer, getQueryController, getQueryId, getRunningCount, getRunState, getStartedOnCount, getStartTime, getStaticAnalysisStats, getStats, getStats, halt, halt, isAtOnceReady, isCancelled, isController, isDone, isRootCauseInterrupt, iterator, newQueryBuffer, runStateString, setDeadline, setStaticAnalysisStats, startOp, startQuery, toString, tryGetRunState

Methods inherited from class java.lang.Object

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

Constructor Detail

FederatedRunningQuery

public FederatedRunningQuery(FederatedQueryEngine queryEngine,
                     UUID queryId,
                     boolean controller,
                     IQueryClient clientProxy,
                     UUID queryControllerId,
                     PipelineOp query,
                     IChunkMessage<IBindingSet> realSource)

Method Detail

releaseNativeMemoryForOperator

protected void releaseNativeMemoryForOperator(int bopId)

Extended to release all allocations associated with the specified operator.

Release native memory associated with this operator, if any (NOP, but overridden in scale-out to release NIO buffers used to move solutions around in the cluster).

Note: Operators are responsible for releasing their child IMemoryManager context, if any, when they terminate and should specify the PipelineOp.Annotations#LAST_PASS annotation to receive notice in the form of a final evaluation pass over an empty IChunkMessage. If they do NOT release an IMemoryManager context which is a child of the {query's context, then their child IMemoryManager context will be retained until the termination of the query, at which point the query's IMemoryManager context will be release, and all child contexts will be released automatically along with it.

Overrides:

releaseNativeMemoryForOperator in class AbstractRunningQuery

See Also:

AbstractRunningQuery.releaseNativeMemoryForQuery()

releaseNativeMemoryForQuery

protected void releaseNativeMemoryForQuery()

Extended to release all DirectBufferPoolAllocator.IAllocationContexts associated with the query when it terminates.

Release native memory associated with this query, if any. FIXME This could cause direct buffers to be released back to the pool before the operator tasks have terminated. That is NOT safe as the buffers could then be reissued to other threads while existing threads still have references to the buffers. Really, the same problem exists with the allocation contexts used for NIO transfers of IBindingSet[]s.

We will have to be very careful to wait until each operator's Future isDone() before calling clear() on the IMemoryManager to release the native buffers back to the pool. If we release a buffer while an operator is still running, then we will get data corruption arising from the recycling of the buffer to another native buffer user.

AbstractRunningQuery.cancel(...) is where we need to handle this, more specifically cancelRunningOperators(). Right now it is not waiting for those operators to terminate.

Making this work is tricky. AbstractRunningQuery is holding a lock. The operator tasks do not actually require that lock to terminate, but they are wrapped by a ChunkWrapperTask, which handles reporting back to the AbstractRunningQuery and *does* need the lock, and also by a ChunkFutureTask. Since we actually do do ChunkFutureTask.get(), we are going to deadlock if we invoke that while holding the AbstractRunningQuery's lock.

The alternative is to handle the tear down of the native buffers for a query asynchronously after the query has been cancelled, deferring the release of the native buffers back to the direct buffer pool until all tasks for the query are known to be done.

Overrides:

releaseNativeMemoryForQuery in class AbstractRunningQuery

See Also:

MemoryManager should know when it has been closed FIXME We need to have distinct events for the query evaluation life cycle and the query results life cycle. Really, this means that temporary solution sets are scoped to the parent query. This is a matter of the scope of the allocation context for the {@link DirectBufferPoolAllocator} and releasing that scope when the parent query is done (in cancel()). [Also consider scoping the temporary solution sets to a transaction or an HTTP session, e.g., by an integration with the NSS using traditional session concepts.]

getQueryEngine

public FederatedQueryEngine getQueryEngine()

Description copied from interface: IRunningQuery

The query engine class executing the query on this node.

Specified by:

getQueryEngine in interface IRunningQuery

Overrides:

getQueryEngine in class AbstractRunningQuery

acceptChunk

protected boolean acceptChunk(IChunkMessage<IBindingSet> msg)

Overridden to make this visible to the FederatedQueryEngine.

Make a chunk of binding sets available for consumption by the query.

Note: this is invoked by QueryEngine.acceptChunk(IChunkMessage)

Overrides:

acceptChunk in class ChunkedRunningQuery

Parameters:

msg - The chunk.

Returns:

true if the message was accepted.

getQueryPeer

public IQueryPeer getQueryPeer(UUID serviceUUID)

Resolve the proxy for an IQueryPeer. This is special cased for both this service (the actual reference is returned) and the query controller (we use an alternative path to discover the query controller since it might not be registered against a lookup service if it is not a data service).

Parameters:

serviceUUID - The service identifier for the peer.

Returns:

The proxy for the service, the actual QueryEngine reference if the identified service is this service, or null if the service could not be discovered.

getAllocationContext

public DirectBufferPoolAllocator.IAllocationContext getAllocationContext(AllocationContextKey key)

Return the DirectBufferPoolAllocator.IAllocationContext for the given key.

Parameters:

key - The key.

Returns:

The allocation context.

cancelQueryOnPeers

protected boolean cancelQueryOnPeers(Throwable cause,
                         Set<UUID> startedOn)

Overridden to notify each peer on which the query was started.

Cancel the query on each node where it is known to be running.

Note: The default implementation verifies that the caller is holding the AbstractRunningQuery.lock but is otherwise a NOP. This is overridden for scale-out.

Overrides:

cancelQueryOnPeers in class AbstractRunningQuery

Parameters:

cause - When non-null, the cause.

Returns:

true iff something was cancelled.

getChunkHandler

protected IChunkHandler getChunkHandler(QueryEngine queryEngine,
                            PipelineOp query)

Description copied from class: ChunkedRunningQuery

Factory returns the effective IChunkHandler for this query.

Overrides:

getChunkHandler in class ChunkedRunningQuery

Returns:

The effective IChunkHandler policy.

See Also:

QueryHints.QUERY_ENGINE_CHUNK_HANDLER, Vector query engine on native heap.

doLastPass

protected void doLastPass(int bopId,
              Set doneOn)

Overridden to broadcast to all nodes and/or shards on which the operator has run in scale-out. Broadcast will be to nodes if the operator is hash partitioned. Broadcast will be to shards if the operator is key-range partitioned.

Queue empty IChunkMessages to trigger the last evaluation pass for an operator which can not be re-triggered by any upstream operator or by IChunkMessages which have already been buffered.

Note: If the queue for accepting new chunks could block then this could deadlock. We work around that by using the same lock for the AbstractRunningQuery and the queue of accepted messages. If the queue blocks, this thread will be yield the lock and another thread may make progress.

Overrides:

doLastPass in class AbstractRunningQuery

doneOn - The collection of shards or services on which the operator need to receive a last evaluation pass message.

See Also:

BOpEvaluationContext