Class ActionDistributor
- java.lang.Object
-
- org.simpleframework.common.thread.Daemon
-
- org.simpleframework.transport.reactor.ActionDistributor
-
- All Implemented Interfaces:
java.lang.Runnable
,OperationDistributor
class ActionDistributor extends Daemon implements OperationDistributor
TheActionDistributor
is used to execute operations that have an interested I/O event ready. This acts much like a scheduler would in that it delays the execution of the operations until such time as the associatedSelectableChannel
has an interested I/O event ready.This distributor has two modes, one mode is used to cancel the channel once an I/O event has occurred. This means that the channel is removed from the
Selector
so that the selector does not break when asked to select again. cancelling the channel is useful when the operation execution may not fully read the payload or when the operation takes a significant amount of time.- See Also:
ExecutorReactor
-
-
Field Summary
Fields Modifier and Type Field Description private boolean
cancel
This is used to determine the mode the distributor uses.private java.util.Map<java.nio.channels.Channel,ActionSet>
executing
This is used to determine the operations that need cancelling.private java.util.concurrent.Executor
executor
This is used to execute the operations that are ready.private long
expiry
This is the duration in milliseconds the operation expires in.private java.util.Queue<java.nio.channels.Channel>
invalid
This is the queue that is used to invalidate channels.private Latch
latch
This is used to signal when the distributor has closed.private java.util.Queue<Action>
pending
This is the queue that is used to provide the operations.private java.util.Map<java.nio.channels.Channel,ActionSet>
selecting
This is used to keep track of actions currently in selection.private ActionSelector
selector
This is the selector used to select for interested events.private long
update
This is time in milliseconds when the next expiry will occur.
-
Constructor Summary
Constructors Constructor Description ActionDistributor(java.util.concurrent.Executor executor)
Constructor for theActionDistributor
object.ActionDistributor(java.util.concurrent.Executor executor, boolean cancel)
Constructor for theActionDistributor
object.ActionDistributor(java.util.concurrent.Executor executor, boolean cancel, long expiry)
Constructor for theActionDistributor
object.
-
Method Summary
All Methods Instance Methods Concrete Methods Modifier and Type Method Description private void
cancel()
This is used to cancel any selection keys that have previously been selected with an interested I/O event.private void
clear()
Here we perform an expire which will take all of the registered sockets and expire it.void
close()
This is used to close the distributor such that it cancels all of the registered channels and closes down the selector.private void
distribute()
This method is used to perform the select and if required queue the operations that are ready for execution.private void
execute()
Performs the execution of the distributor.private void
execute(Action action)
This is where the action is handed off to the executor.private void
expire()
This method is used to expire registered operations that remain idle within the selector.private void
expire(ActionSet set, long time)
This method is used to expire registered operations that remain idle within the selector.private void
expire(ActionSet set, Action action)
This method is used to expire registered operations that remain idle within the selector.private void
invalidate(java.nio.channels.Channel channel)
This method is used to remove the channel from the selecting registry.private void
process()
This will iterate over the set of selection keys and process each of them.private void
process(ActionSet set)
This will use the specified action set to acquire the channel andOperation
associated with it to hand to the executor to perform the channel operation.void
process(Operation task, int require)
This is used to process theOperation
object.private void
purge()
This will purge all the actions from the distributor when the distributor ends.private void
register()
Here all the enqueuedOperation
objects will be registered for selection.private void
register(Action action)
Here the specifiedOperation
object is registered with the selector.private void
remove(ActionSet set)
This method ensures that references to the actions and channel are cleared from this instance.private void
report(java.lang.Exception cause)
This method is called to ensure that if there is a global error that each action will know about it.void
run()
Performs the execution of the distributor.private void
select(Action action)
This method is used to perform an actual select on a channel.int
size()
This returns the number of channels that are currently selecting with this distributor.private void
update(Action action, ActionSet set)
Here the specifiedOperation
object is registered with the selector.private void
update(ActionSet set, int interest)
This is used to update the interested operations of a set of actions.private void
validate()
This method is used to perform simple validation.
-
-
-
Field Detail
-
executing
private java.util.Map<java.nio.channels.Channel,ActionSet> executing
This is used to determine the operations that need cancelling.
-
selecting
private java.util.Map<java.nio.channels.Channel,ActionSet> selecting
This is used to keep track of actions currently in selection.
-
invalid
private java.util.Queue<java.nio.channels.Channel> invalid
This is the queue that is used to invalidate channels.
-
pending
private java.util.Queue<Action> pending
This is the queue that is used to provide the operations.
-
selector
private ActionSelector selector
This is the selector used to select for interested events.
-
executor
private java.util.concurrent.Executor executor
This is used to execute the operations that are ready.
-
latch
private Latch latch
This is used to signal when the distributor has closed.
-
expiry
private long expiry
This is the duration in milliseconds the operation expires in.
-
update
private long update
This is time in milliseconds when the next expiry will occur.
-
cancel
private boolean cancel
This is used to determine the mode the distributor uses.
-
-
Constructor Detail
-
ActionDistributor
public ActionDistributor(java.util.concurrent.Executor executor) throws java.io.IOException
Constructor for theActionDistributor
object. This will create a distributor that distributes operations when those operations show that they are ready for a given I/O event. The interested I/O events are provided as a bitmask taken from the actions of theSelectionKey
. Distribution of the operations is passed to the provided executor object.- Parameters:
executor
- this is the executor used to execute operations- Throws:
java.io.IOException
-
ActionDistributor
public ActionDistributor(java.util.concurrent.Executor executor, boolean cancel) throws java.io.IOException
Constructor for theActionDistributor
object. This will create a distributor that distributes operations when those operations show that they are ready for a given I/O event. The interested I/O events are provided as a bitmask taken from the actions of theSelectionKey
. Distribution of the operations is passed to the provided executor object.- Parameters:
executor
- this is the executor used to execute operationscancel
- should the channel be removed from selection- Throws:
java.io.IOException
-
ActionDistributor
public ActionDistributor(java.util.concurrent.Executor executor, boolean cancel, long expiry) throws java.io.IOException
Constructor for theActionDistributor
object. This will create a distributor that distributes operations when those operations show that they are ready for a given I/O event. The interested I/O events are provided as a bitmask taken from the actions of theSelectionKey
. Distribution of the operations is passed to the provided executor object.- Parameters:
executor
- this is the executor used to execute operationscancel
- should the channel be removed from selectionexpiry
- this the maximum idle time for an operation- Throws:
java.io.IOException
-
-
Method Detail
-
process
public void process(Operation task, int require) throws java.io.IOException
This is used to process theOperation
object. This will wake up the selector if it is currently blocked selecting and register the operations associated channel. Once the selector is awake it will acquire the operation from the queue and register the associatedSelectableChannel
for selection. The operation will then be executed when the channel is ready for the interested I/O events.- Specified by:
process
in interfaceOperationDistributor
- Parameters:
task
- this is the task that is scheduled for distributionrequire
- this is the bit-mask value for interested events- Throws:
java.io.IOException
-
close
public void close() throws java.io.IOException
This is used to close the distributor such that it cancels all of the registered channels and closes down the selector. This is used when the distributor is no longer required, after the close further attempts to process operations will fail.- Specified by:
close
in interfaceOperationDistributor
- Throws:
java.io.IOException
-
size
public int size()
This returns the number of channels that are currently selecting with this distributor. When busy this can get quite high, however it must return to zero as soon as all tasks have completed.- Returns:
- return the number of channels currently selecting
-
run
public void run()
Performs the execution of the distributor. Each distributor runs on an asynchronous thread to theReactor
which is used to perform the selection on a set of channels. Each time there is a new operation to be processed this will take the operation from the ready queue, cancel all outstanding channels, and register the operations associated channel for selection.- Specified by:
run
in interfacejava.lang.Runnable
-
execute
private void execute()
Performs the execution of the distributor. Each distributor runs on an asynchronous thread to theReactor
which is used to perform the selection on a set of channels. Each time there is a new operation to be processed this will take the operation from the ready queue, cancel all outstanding channels, and register the operations associated channel for selection.
-
purge
private void purge()
This will purge all the actions from the distributor when the distributor ends. If there are any threads waiting on the close to finish they are signalled when all operations are purged. This will allow them to return ensuring no operations linger.
-
report
private void report(java.lang.Exception cause)
This method is called to ensure that if there is a global error that each action will know about it. Such an issue could be file handle exhaustion or an out of memory error. It is also possible that a poorly behaving action could cause an issue which should be know the the entire system.- Parameters:
cause
- this is the exception to report
-
clear
private void clear() throws java.io.IOException
Here we perform an expire which will take all of the registered sockets and expire it. This ensures that the operations can be executed within the executor and the cancellation of the sockets can be performed. Once this method has finished then all of the operations will have been scheduled for execution.- Throws:
java.io.IOException
-
expire
private void expire() throws java.io.IOException
This method is used to expire registered operations that remain idle within the selector. Operations specify a time at which point they wish to be cancelled if the I/O event they wait on has not arisen. This will enables the cancelled operation to be cancelled so that the resources it occupies can be released.- Throws:
java.io.IOException
-
expire
private void expire(ActionSet set, long time) throws java.io.IOException
This method is used to expire registered operations that remain idle within the selector. Operations specify a time at which point they wish to be if the I/O event they wait on has not arisen. This will enables the cancelled operation to be cancelled so that the resources it occupies can be released.- Parameters:
set
- this is the selection set check for expired actionstime
- this is the time to check the expiry against- Throws:
java.io.IOException
-
update
private void update(ActionSet set, int interest) throws java.io.IOException
This is used to update the interested operations of a set of actions. If there are no interested operations the set will be cancelled, otherwise the selection key will be updated with the new operations provided by the bitmask.- Parameters:
set
- this is the action set that is to be updatedinterest
- this is the bitmask containing the operations- Throws:
java.io.IOException
-
expire
private void expire(ActionSet set, Action action) throws java.io.IOException
This method is used to expire registered operations that remain idle within the selector. Operations specify a time at which point they wish to be cancelled if the I/O event they wait on has not arisen. This will enables the cancelled operation to be cancelled so that the resources it occupies can be released.- Parameters:
set
- this is the action set containing the actionsaction
- this is the actual action to be cancelled- Throws:
java.io.IOException
-
validate
private void validate() throws java.io.IOException
This method is used to perform simple validation. It ensures that directly after the processing loop any channels that are registered that have been cancelled or are closed will be removed from the selecting map and rejected.- Throws:
java.io.IOException
-
invalidate
private void invalidate(java.nio.channels.Channel channel) throws java.io.IOException
This method is used to remove the channel from the selecting registry. It is rare that this will every happen, however it is important that tasks are cleared out in this manner as it could lead to a memory leak if left for a long time.- Parameters:
channel
- this is the channel being validated- Throws:
java.io.IOException
-
cancel
private void cancel() throws java.io.IOException
This is used to cancel any selection keys that have previously been selected with an interested I/O event. Performing a cancel here ensures that on a the next select the associated channel is not considered, this ensures the select does not break.- Throws:
java.io.IOException
-
register
private void register() throws java.io.IOException
Here all the enqueuedOperation
objects will be registered for selection. Each operations channel is used for selection on the interested I/O events. Once the I/O event occurs for the channel the operation is scheduled for execution.- Throws:
java.io.IOException
-
register
private void register(Action action) throws java.io.IOException
Here the specifiedOperation
object is registered with the selector. If the associated channel had previously been cancelled it is removed from the cancel map to ensure it is not removed from the selector when cancellation is done.- Parameters:
action
- this is the operation that is to be registered- Throws:
java.io.IOException
-
update
private void update(Action action, ActionSet set) throws java.io.IOException
Here the specifiedOperation
object is registered with the selector. If the associated channel had previously been cancelled it is removed from the cancel map to ensure it is not removed from the selector when cancellation is done.- Parameters:
action
- this is the operation that is to be registeredset
- this is the action set to register the action with- Throws:
java.io.IOException
-
select
private void select(Action action) throws java.io.IOException
This method is used to perform an actual select on a channel. It will register the channel with the internal selector using the required I/O event bit mask. In order to ensure that selection is performed correctly the provided channel must be connected.- Parameters:
action
- this is the operation that is to be registered- Throws:
java.io.IOException
-
distribute
private void distribute() throws java.io.IOException
This method is used to perform the select and if required queue the operations that are ready for execution. If the selector is woken up without any ready channels then this will return quietly. If however there are a number of channels ready to be processed then they are handed to the executor object and marked as ready for cancellation.- Throws:
java.io.IOException
-
process
private void process() throws java.io.IOException
This will iterate over the set of selection keys and process each of them. TheOperation
associated with the selection key is handed to the executor to perform the channel operation. Also, if configured to cancel, this method will add the channel and the associated selection key to the cancellation map.- Throws:
java.io.IOException
-
process
private void process(ActionSet set) throws java.io.IOException
This will use the specified action set to acquire the channel andOperation
associated with it to hand to the executor to perform the channel operation.- Parameters:
set
- this is the set of actions that are to be processed- Throws:
java.io.IOException
-
remove
private void remove(ActionSet set) throws java.io.IOException
This method ensures that references to the actions and channel are cleared from this instance. To ensure there are no memory leaks it is important to clear out all actions and channels. Also, if configured to cancel executing actions this will register the channel and actions to cancel on the next loop.- Parameters:
set
- this is the set of actions that are to be removed- Throws:
java.io.IOException
-
execute
private void execute(Action action)
This is where the action is handed off to the executor. Before the action is executed a trace event is generated, this will ensure that the entry and exit points can be tracked. It is also useful in debugging performance issues and memory leaks.- Parameters:
action
- this is the action to execute
-
-