org.jgroups.protocols

Class Executing

java.lang.Object

org.jgroups.stack.Protocol

org.jgroups.protocols.Executing

Direct Known Subclasses:

CENTRAL_EXECUTOR

public abstract class Executing
extends Protocol

This is the base protocol used for executions.

Author:

wburns

See Also:

CENTRAL_EXECUTOR

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	Executing.ExecutorHeader
static class	Executing.Owner
protected static class	Executing.Request
protected static class	Executing.RequestWithThread
protected static class	Executing.Type

Field Summary

Fields

Modifier and Type	Field and Description
protected java.util.Queue<java.lang.Runnable>	_awaitingConsumer This is a queue on the client side that holds all of the tasks that are awaiting a consumer to pick them up
protected java.util.Map<Executing.Owner,java.lang.Runnable>	_awaitingReturn This is a map on the client side that shows for which owner(consumer, request) the runnable they are currently using.
protected java.util.concurrent.ConcurrentMap<java.lang.Long,java.lang.Object>	_consumerId This is essentially a set on the consumer side of id's of all the threads currently running as consumers.
protected java.util.concurrent.locks.Lock	_consumerLock This lock is to protect the incoming run requests and the incoming consumer queues
protected java.util.Queue<Executing.Owner>	_consumersAvailable This is stored on the coordinator side.
protected java.util.concurrent.ConcurrentMap<java.lang.Runnable,java.lang.Long>	_requestId This is a map on the client side showing for all of the current pending requests
protected java.util.concurrent.ConcurrentMap<java.lang.Runnable,java.lang.Thread>	_runnableThreads This is a server side map to show which threads are running for a given runnable.
protected java.util.Map<java.lang.Runnable,Executing.Owner>	_running This is a map on the server side that shows which owner is currently tied to the runnable so we can return to them the results
protected java.util.Queue<Executing.Owner>	_runRequests This is stored on the coordinator side.
protected java.util.concurrent.ConcurrentMap<java.lang.Long,java.util.concurrent.CyclicBarrier>	_taskBarriers This is a server side store of all the barriers for respective tasks requests.
protected java.util.concurrent.ConcurrentMap<java.lang.Long,java.lang.Runnable>	_tasks This is a server side store of all the tasks that want to be ran on a given thread.
protected boolean	bypass_bundling
protected static java.util.concurrent.atomic.AtomicLong	counter This keeps track of all the requests we send.
protected Address	local_addr
protected java.util.concurrent.ConcurrentMap<java.util.concurrent.Future<?>,ExecutorNotification>	notifiers
protected static java.lang.Object	PRESENT This is a placeholder for a key value to make a concurrent hash map a concurrent hash set
protected View	view

Fields inherited from class org.jgroups.stack.Protocol

after_creation_hook, down_prot, ergonomics, id, log, name, stack, stats, up_prot

Constructor Summary

Constructors

Constructor and Description
Executing()

Method Summary

Modifier and Type	Method and Description
void	addExecutorListener(java.util.concurrent.Future<?> future, ExecutorNotification listener)
java.lang.Object	down(Event evt) An event is to be sent down the stack.
java.lang.String	getAddress()
boolean	getBypassBundling()
java.lang.String	getView()
protected void	handleConsumerFoundResponse(long threadId, Address address)
protected void	handleConsumerReadyRequest(long requestId, Address address)
protected void	handleConsumerUnreadyRequest(long requestId, Address address)
protected void	handleExceptionResponse(Address source, long requestId, java.lang.Throwable throwable)
protected void	handleInterruptRequest(Address source, long requestId)
protected void	handleNewConsumer(Executing.Owner sender)
protected void	handleNewRunRequest(Executing.Owner sender)
protected void	handleRemoveConsumer(Executing.Owner sender)
protected void	handleRemoveRunRequest(Executing.Owner sender)
protected void	handleTaskRejectedResponse(Address source, long requestId)
protected void	handleTaskRequest(long requestId, Address address)
protected void	handleTaskSubmittedRequest(java.lang.Runnable runnable, Address source, long requestId, long threadId)
protected void	handleValueResponse(Address source, long requestId, java.lang.Object value)
protected void	handleView(View view)
protected static <V,K> V	removeKeyForValue(java.util.Map<V,K> map, K value)
protected abstract void	sendNewConsumerRequest(Executing.Owner source)
protected abstract void	sendNewRunRequest(Executing.Owner source)
protected abstract void	sendRemoveConsumerRequest(Executing.Owner source)
protected abstract void	sendRemoveRunRequest(Executing.Owner source)
protected void	sendRequest(Address dest, Executing.Type type, long requestId, java.lang.Object object)
protected void	sendThreadRequest(Address dest, long threadId, Executing.Type type, long requestId, java.lang.Object object)
protected abstract void	sendToCoordinator(Executing.Type type, long requestId, Address address)
void	setBypassBundling(boolean bypass_bundling)
java.lang.Object	up(Event evt) An event was received from the layer below.

Methods inherited from class org.jgroups.stack.Protocol

accept, afterCreationHook, destroy, dumpStats, enableStats, getConfigurableObjects, getDownProtocol, getDownServices, getId, getIdsAbove, getLevel, getLog, getName, getProtocolStack, getSocketFactory, getThreadFactory, getTransport, getUpProtocol, getUpServices, getValue, init, isErgonomics, level, parse, printStats, providedDownServices, providedUpServices, requiredDownServices, requiredUpServices, resetStatistics, resetStats, setDownProtocol, setErgonomics, setId, setLevel, setProtocolStack, setSocketFactory, setUpProtocol, setValue, setValues, start, statsEnabled, stop, up

Methods inherited from class java.lang.Object

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

Field Detail

bypass_bundling

protected boolean bypass_bundling

local_addr

protected Address local_addr

view

protected View view

_awaitingConsumer

protected final java.util.Queue<java.lang.Runnable> _awaitingConsumer

This is a queue on the client side that holds all of the tasks that are awaiting a consumer to pick them up

_requestId

protected final java.util.concurrent.ConcurrentMap<java.lang.Runnable,java.lang.Long> _requestId

This is a map on the client side showing for all of the current pending requests

_consumerId

protected final java.util.concurrent.ConcurrentMap<java.lang.Long,java.lang.Object> _consumerId

This is essentially a set on the consumer side of id's of all the threads currently running as consumers. This is basically a set, but since there is no ConcurrentHashSet we use a phoney value

notifiers

protected final java.util.concurrent.ConcurrentMap<java.util.concurrent.Future<?>,ExecutorNotification> notifiers

_running

protected final java.util.Map<java.lang.Runnable,Executing.Owner> _running

This is a map on the server side that shows which owner is currently tied to the runnable so we can return to them the results

_awaitingReturn

protected final java.util.Map<Executing.Owner,java.lang.Runnable> _awaitingReturn

This is a map on the client side that shows for which owner(consumer, request) the runnable they are currently using. This also allows us to set the values on a future when finished.

_tasks

protected java.util.concurrent.ConcurrentMap<java.lang.Long,java.lang.Runnable> _tasks

This is a server side store of all the tasks that want to be ran on a given thread. This map should be updated by an incoming request before awaking the task with the latch. This map should only be retrieved after first waiting on the latch for a consumer

_taskBarriers

protected java.util.concurrent.ConcurrentMap<java.lang.Long,java.util.concurrent.CyclicBarrier> _taskBarriers

This is a server side store of all the barriers for respective tasks requests. When a consumer is starting up they should create a latch place in map with its id and wait on it until a request comes in to wake it up it would only then touch the _tasks map. A requestor should first place in the _tasks map and then create a latch and notify the consumer

_runnableThreads

protected final java.util.concurrent.ConcurrentMap<java.lang.Runnable,java.lang.Thread> _runnableThreads

This is a server side map to show which threads are running for a given runnable. This is used to interrupt those threads if needed.

_consumerLock

protected java.util.concurrent.locks.Lock _consumerLock

This lock is to protect the incoming run requests and the incoming consumer queues

_runRequests

protected java.util.Queue<Executing.Owner> _runRequests

This is stored on the coordinator side. This queue holds all of the addresses that currently want to run something. If this queue has elements the consumer queue must be empty.

_consumersAvailable

protected java.util.Queue<Executing.Owner> _consumersAvailable

This is stored on the coordinator side. This queue holds all of the addresses that currently are able to run something. If this queue has elements the run request queue must be empty.

counter

protected static final java.util.concurrent.atomic.AtomicLong counter

This keeps track of all the requests we send. This is used so that the response doesn't have to send back the future but instead the counter We just let this roll over

PRESENT

protected static final java.lang.Object PRESENT

This is a placeholder for a key value to make a concurrent hash map a concurrent hash set

Constructor Detail

Executing

public Executing()

Method Detail

getBypassBundling

public boolean getBypassBundling()

setBypassBundling

public void setBypassBundling(boolean bypass_bundling)

addExecutorListener

public void addExecutorListener(java.util.concurrent.Future<?> future,
                                ExecutorNotification listener)

getAddress

public java.lang.String getAddress()

getView

public java.lang.String getView()

down

public java.lang.Object down(Event evt)

Description copied from class: Protocol

An event is to be sent down the stack. The layer may want to examine its type and perform some action on it, depending on the event's type. If the event is a message MSG, then the layer may need to add a header to it (or do nothing at all) before sending it down the stack using down_prot.down(). In case of a GET_ADDRESS event (which tries to retrieve the stack's address from one of the bottom layers), the layer may need to send a new response event back up the stack using up_prot.up().

Overrides:

down in class Protocol

removeKeyForValue

protected static <V,K> V removeKeyForValue(java.util.Map<V,K> map,
                                           K value)

up

public java.lang.Object up(Event evt)

Description copied from class: Protocol

An event was received from the layer below. Usually the current layer will want to examine the event type and - depending on its type - perform some computation (e.g. removing headers from a MSG event type, or updating the internal membership list when receiving a VIEW_CHANGE event). Finally the event is either a) discarded, or b) an event is sent down the stack using down_prot.down() or c) the event (or another event) is sent up the stack using up_prot.up().

Overrides:

up in class Protocol

handleView

protected void handleView(View view)

sendToCoordinator

protected abstract void sendToCoordinator(Executing.Type type,
                                          long requestId,
                                          Address address)

sendNewRunRequest

protected abstract void sendNewRunRequest(Executing.Owner source)

sendRemoveRunRequest

protected abstract void sendRemoveRunRequest(Executing.Owner source)

sendNewConsumerRequest

protected abstract void sendNewConsumerRequest(Executing.Owner source)

sendRemoveConsumerRequest

protected abstract void sendRemoveConsumerRequest(Executing.Owner source)

handleTaskRequest

protected void handleTaskRequest(long requestId,
                                 Address address)

handleConsumerReadyRequest

protected void handleConsumerReadyRequest(long requestId,
                                          Address address)

handleConsumerUnreadyRequest

protected void handleConsumerUnreadyRequest(long requestId,
                                            Address address)

handleConsumerFoundResponse

protected void handleConsumerFoundResponse(long threadId,
                                           Address address)

handleTaskSubmittedRequest

protected void handleTaskSubmittedRequest(java.lang.Runnable runnable,
                                          Address source,
                                          long requestId,
                                          long threadId)

handleTaskRejectedResponse

protected void handleTaskRejectedResponse(Address source,
                                          long requestId)

handleValueResponse

protected void handleValueResponse(Address source,
                                   long requestId,
                                   java.lang.Object value)

handleExceptionResponse

protected void handleExceptionResponse(Address source,
                                       long requestId,
                                       java.lang.Throwable throwable)

handleInterruptRequest

protected void handleInterruptRequest(Address source,
                                      long requestId)

handleNewRunRequest

protected void handleNewRunRequest(Executing.Owner sender)

handleRemoveRunRequest

protected void handleRemoveRunRequest(Executing.Owner sender)

handleNewConsumer

protected void handleNewConsumer(Executing.Owner sender)

handleRemoveConsumer

protected void handleRemoveConsumer(Executing.Owner sender)

sendRequest

protected void sendRequest(Address dest,
                           Executing.Type type,
                           long requestId,
                           java.lang.Object object)

sendThreadRequest

protected void sendThreadRequest(Address dest,
                                 long threadId,
                                 Executing.Type type,
                                 long requestId,
                                 java.lang.Object object)