Class ActionDistributor

java.lang.Object
org.simpleframework.common.thread.Daemon
org.simpleframework.transport.reactor.ActionDistributor
All Implemented Interfaces:
Runnable, OperationDistributor

class ActionDistributor extends Daemon implements OperationDistributor
The ActionDistributor 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 associated SelectableChannel 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:
  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    private boolean
    This is used to determine the mode the distributor uses.
    This is used to determine the operations that need cancelling.
    private Executor
    This is used to execute the operations that are ready.
    private long
    This is the duration in milliseconds the operation expires in.
    private Queue<Channel>
    This is the queue that is used to invalidate channels.
    private Latch
    This is used to signal when the distributor has closed.
    private Queue<Action>
    This is the queue that is used to provide the operations.
    This is used to keep track of actions currently in selection.
    This is the selector used to select for interested events.
    private long
    This is time in milliseconds when the next expiry will occur.
  • Constructor Summary

    Constructors
    Constructor
    Description
    Constructor for the ActionDistributor object.
    ActionDistributor(Executor executor, boolean cancel)
    Constructor for the ActionDistributor object.
    ActionDistributor(Executor executor, boolean cancel, long expiry)
    Constructor for the ActionDistributor object.
  • Method Summary

    Modifier and Type
    Method
    Description
    private void
    This is used to cancel any selection keys that have previously been selected with an interested I/O event.
    private void
    Here we perform an expire which will take all of the registered sockets and expire it.
    void
    This is used to close the distributor such that it cancels all of the registered channels and closes down the selector.
    private void
    This method is used to perform the select and if required queue the operations that are ready for execution.
    private void
    Performs the execution of the distributor.
    private void
    execute(Action action)
    This is where the action is handed off to the executor.
    private void
    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
    This method is used to remove the channel from the selecting registry.
    private void
    This will iterate over the set of selection keys and process each of them.
    private void
    This will use the specified action set to acquire the channel and Operation associated with it to hand to the executor to perform the channel operation.
    void
    process(Operation task, int require)
    This is used to process the Operation object.
    private void
    This will purge all the actions from the distributor when the distributor ends.
    private void
    Here all the enqueued Operation objects will be registered for selection.
    private void
    register(Action action)
    Here the specified Operation object is registered with the selector.
    private void
    This method ensures that references to the actions and channel are cleared from this instance.
    private void
    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
    This returns the number of channels that are currently selecting with this distributor.
    private void
    update(Action action, ActionSet set)
    Here the specified Operation 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
    This method is used to perform simple validation.

    Methods inherited from class org.simpleframework.common.thread.Daemon

    interrupt, isActive, start, stop

    Methods inherited from class java.lang.Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
  • Field Details

    • executing

      private Map<Channel,ActionSet> executing
      This is used to determine the operations that need cancelling.
    • selecting

      private Map<Channel,ActionSet> selecting
      This is used to keep track of actions currently in selection.
    • invalid

      private Queue<Channel> invalid
      This is the queue that is used to invalidate channels.
    • pending

      private 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 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 Details

    • ActionDistributor

      public ActionDistributor(Executor executor) throws IOException
      Constructor for the ActionDistributor 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 the SelectionKey. Distribution of the operations is passed to the provided executor object.
      Parameters:
      executor - this is the executor used to execute operations
      Throws:
      IOException
    • ActionDistributor

      public ActionDistributor(Executor executor, boolean cancel) throws IOException
      Constructor for the ActionDistributor 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 the SelectionKey. Distribution of the operations is passed to the provided executor object.
      Parameters:
      executor - this is the executor used to execute operations
      cancel - should the channel be removed from selection
      Throws:
      IOException
    • ActionDistributor

      public ActionDistributor(Executor executor, boolean cancel, long expiry) throws IOException
      Constructor for the ActionDistributor 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 the SelectionKey. Distribution of the operations is passed to the provided executor object.
      Parameters:
      executor - this is the executor used to execute operations
      cancel - should the channel be removed from selection
      expiry - this the maximum idle time for an operation
      Throws:
      IOException
  • Method Details

    • process

      public void process(Operation task, int require) throws IOException
      This is used to process the Operation 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 associated SelectableChannel for selection. The operation will then be executed when the channel is ready for the interested I/O events.
      Specified by:
      process in interface OperationDistributor
      Parameters:
      task - this is the task that is scheduled for distribution
      require - this is the bit-mask value for interested events
      Throws:
      IOException
    • close

      public void close() throws 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 interface OperationDistributor
      Throws:
      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 the Reactor 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 interface Runnable
    • execute

      private void execute()
      Performs the execution of the distributor. Each distributor runs on an asynchronous thread to the Reactor 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(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 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:
      IOException
    • expire

      private void expire() throws 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:
      IOException
    • expire

      private void expire(ActionSet set, long time) throws 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 actions
      time - this is the time to check the expiry against
      Throws:
      IOException
    • update

      private void update(ActionSet set, int interest) throws 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 updated
      interest - this is the bitmask containing the operations
      Throws:
      IOException
    • expire

      private void expire(ActionSet set, Action action) throws 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 actions
      action - this is the actual action to be cancelled
      Throws:
      IOException
    • validate

      private void validate() throws 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:
      IOException
    • invalidate

      private void invalidate(Channel channel) throws 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:
      IOException
    • cancel

      private void cancel() throws 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:
      IOException
    • register

      private void register() throws IOException
      Here all the enqueued Operation 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:
      IOException
    • register

      private void register(Action action) throws IOException
      Here the specified Operation 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:
      IOException
    • update

      private void update(Action action, ActionSet set) throws IOException
      Here the specified Operation 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
      set - this is the action set to register the action with
      Throws:
      IOException
    • select

      private void select(Action action) throws 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:
      IOException
    • distribute

      private void distribute() throws 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:
      IOException
    • process

      private void process() throws IOException
      This will iterate over the set of selection keys and process each of them. The Operation 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:
      IOException
    • process

      private void process(ActionSet set) throws IOException
      This will use the specified action set to acquire the channel and Operation 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:
      IOException
    • remove

      private void remove(ActionSet set) throws 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:
      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