org.jumpi.impl.controller.ordering
Class OrderingController

java.lang.Object
  |
  +--org.jumpi.impl.AbstractController
        |
        +--org.jumpi.impl.controller.ordering.OrderingController

All Implemented Interfaces:

Component, Configurable, Controller, Manageable, java.lang.Runnable, Status, Status, Task, TimerListener

public class OrderingController

extends AbstractController

implements Task, TimerListener, Status

The OrderingController enforces single send and single receive operation concurrency for each Destination. Submitting multiple sends or receives to a single Destination will cause the operations to be queued and processed in order.

Any number of Connectors are supported. A message is sent to or received from a single Connector, which is the first to support the Destination's Url, determined in configuration order. Supports any Destination Url syntax which the Connectors support.

The OrderingController maintains a cache of state per Destination. This is cleaned up at configurable timer intervals. The state per Destination contains a send and a receive pending queue for ordering operations. The cache size is also configurable.

The Destination Urls supported are those of the configured Connectors, without alteration.

Compatible with Connectors which have either single operation per Destination or multiple operation per Destination concurrency. Care must be taken with Connectors which only support a single send or receive operation irrespective of Destination, since this is not enforced by the OrderingController.

Field Summary

Fields inherited from class org.jumpi.impl.AbstractController

connectors_, joinFailInterval_, joinTestInterval_, jumpi_, name_, recvTransformer_, sendTransformer_, started_, stopping_

Fields inherited from interface org.jumpi.spi.Status

STATUS_RECEIVING, STATUS_SENDING

Fields inherited from interface org.jumpi.Status

STATUS_CANCELLED, STATUS_FAILURE, STATUS_SUCCESS, STATUS_TRANSIENT

Constructor Summary

OrderingController()

Method Summary

void	callbackRecv(Envelope env, boolean success, java.lang.String status) Notify of the recv operation completion by the Connector layer.
void	callbackSend(Envelope env, boolean success, java.lang.String status) Notify of the send operation completion by the Connector layer.
void	cancelRecv(Destination dest, Handle hdl) Cancel the Handle.
void	cancelSend(Destination dest, Handle hdl) Cancel the Handle.
protected boolean	checkJoinCondition() The clean shutdown condition is met when there are no Destination states present in the Destination cache.
void	configure(java.lang.String name, Properties props) Configure the OrderingController.
java.lang.String	getTaskId() A unique name for the task in the VM.
void	handleTimerInterrupt() Schedule the cache cleanup task.
boolean	isLongRunning() The cleanup task is not long running.
boolean	isSchedulable() The Destination cache cleanup task is schedulable when started and there is a pending Destination state cleanup.
void	manage(Component root, Component parent, java.lang.String operation, java.util.Hashtable parameters) Perform a recursive management operation through the Jumpi component tree, using the parameters provided.
Handle	recv(Destination dest, java.util.Hashtable clientProps) Receive a message from the Destination.
void	run() The cache cleanup task which removes any Destination state which has no pending or active receive or send operations.
Handle	send(Destination dest, java.lang.Object msg, java.util.Hashtable clientProps) Send a message to a Destination.
protected void	shutdownState() In order to shutdown cleanly, each Destination state is taken in turn and any pending send or receive operations therein are failed.

Methods inherited from class org.jumpi.impl.AbstractController

getDestination, getName, transferInternalProperties

Methods inherited from class java.lang.Object

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

Constructor Detail

OrderingController

public OrderingController()

Method Detail

configure

public void configure(java.lang.String name,
                      Properties props)
               throws java.lang.Exception

Configure the OrderingController.

The following configuration attributes are defined in addition to those of the superclass:

• maxrecvsperdestination - integer[1..MAX-INT] - optional[10].

  The maximum number of pending receives for a single Destination before blocking further receive operations for that Destination.

• maxsendsperdestination - integer[1..MAX-INT] - optional[10].

  The maximum number of pending sends for a single Destination before blocking further send operations for that Destination.

• cleanupinterval - integer[1000..MAX-INT] - optional[10000].

  The time interval in milliseconds between cache cleanups.

• statecachesize - integer[10..MAX-INT] - optional[50].

  The initial size of the state cache. The cache size is not fixed, so sizing this correctly will simply avoid unnecessary container size changes at runtime.

Specified by:

configure in interface Configurable

Overrides:

configure in class AbstractController

Parameters:

name - the fully qualified instance name.

props - the read-only properties to configure the instance with.

Throws:

java.lang.Exception - when configuration is unsuccessful.

manage

public void manage(Component root,
                   Component parent,
                   java.lang.String operation,
                   java.util.Hashtable parameters)
            throws java.lang.Exception

Perform a recursive management operation through the Jumpi component tree, using the parameters provided.

The following management actions are handled:

• STARTUP. Start a periodic timer for the cleanup task.
• SHUTDOWN. Stop the periodic timer for the cleanup task.

Specified by:

manage in interface Manageable

Overrides:

manage in class AbstractController

Parameters:

root - the Jumpi instance at the root of the component tree.

parent - the immediate parent of the component.

operation - the operation.

parameters - the operation parameters.

Throws:

java.lang.Exception - if any failure to perform management occurs.

getTaskId

public java.lang.String getTaskId()

A unique name for the task in the VM.

Specified by:

getTaskId in interface Task

Returns:

a unique name for the task in the VM.

isSchedulable

public boolean isSchedulable()

The Destination cache cleanup task is schedulable when started and there is a pending Destination state cleanup.

Specified by:

isSchedulable in interface Task

Returns:

true when started and Destination cache cleanup is scheduled.

isLongRunning

public boolean isLongRunning()

The cleanup task is not long running.

Specified by:

isLongRunning in interface Task

Returns:

false.

checkJoinCondition

protected boolean checkJoinCondition()

The clean shutdown condition is met when there are no Destination states present in the Destination cache.

Specified by:

checkJoinCondition in class AbstractController

Returns:

true when the Destination cache is empty.

shutdownState

protected void shutdownState()

In order to shutdown cleanly, each Destination state is taken in turn and any pending send or receive operations therein are failed.

Specified by:

shutdownState in class AbstractController

send

public Handle send(Destination dest,
                   java.lang.Object msg,
                   java.util.Hashtable clientProps)

Send a message to a Destination. Send operations are serialized for each Destination. Send operations for separate Destinations are handled concurrently. Send operations are delegated to a single Connector which created the Destination in getDestination(). There is a 1:1 relationship between calls to send and those passed down to the Connector layer.

The message format required on entry is not defined since messages are handled as opaque objects, subject to send transformation once before passing to the Connector layer.

Send operations to multiple Destinations are handled concurrently, and also multiple send operations to the same Destination.

Blocking or non blocking depending on the synchronization of the send method of the Connector and whether there are already pending sends waiting to be sent to the Connector. If there are already pending sends, the send operation is queued and the caller returns immediately unless the limit to the pending send queue has been reached.

If the Connector's send method is synchronous then the Controller's send is also synchronous since the calling thread is used to delegate to the Connector. In completing the send at the Connector, the callbackSend causes queued sends to be sent to the Connector in a recursive fashion. Therefore if send operations are continually placed in the send queue then there is the danger of Stack overflow due to too deep recursion with the synchronous sending Controller. Therefore it is advisable to use the OrderingController only with asynchronous Connectors.

If the Connector's send method is asynchronous, then the Controller's send method is also asynchronous. The calling thread drives the send operation to the Connector, but it is the Connector's asynchronous completion of the operation which drives further pending operations.

The Handles passed to the Connector are given the STATUS_SENDING transient state until either cancelSend is called by the client application or callbackSend is called by the Connector.

Specified by:

send in interface Controller

Overrides:

send in class AbstractController

Parameters:

dest - the Destination to which the message is sent.

msg - the message to be sent.

clientProps - not used but delegated in the Envelope to the Connector.

Returns:

a Handle associated to the send operation.

callbackSend

public void callbackSend(Envelope env,
                         boolean success,
                         java.lang.String status)

Notify of the send operation completion by the Connector layer. Since we only perform one send at a time, we expect to match up the last send Envelope with the returned value. This may not match up if a cancel took place, in which case a warning is logged. The next send operation which is queued pending the completion at the Connector layer is triggered to continue.

Specified by:

callbackSend in interface Controller

Overrides:

callbackSend in class AbstractController

Parameters:

env - the Envelope used in Connector.send

success - true if successfully sent or false if a failure ocured.

status - a message giving more information to the status.

recv

public Handle recv(Destination dest,
                   java.util.Hashtable clientProps)

Receive a message from the Destination. Receive operations are serialized for each Destination. Receive operations for seperate Destinations are handled concurrently. Receive operations are delegated to a single Connector which created the Destination in getDestination(). There is a 1:1 relationship between calls to receive and those passed down to the Connector layer.

The message format required on entry is not defined since messages are handled as opaque objects, subject to receive transformation once before passing to the client application in the Handle.

Receive operations to multiple Destinations are handled concurrently, and also multiple receive operations to the same Destination.

Blocking or non blocking depending on the synchronization of the receive method of the Connector and whether there are already pending receives waiting to be sent to the Connector. If there are already pending receives, the receive operation is queued and the caller returns immediately unless the limit to the pending receive queue has been reached.

If the Connector's recv method is synchronous then the Controller's recv is also synchronous since the calling thread is used to delegate to the Connector. In completing the receive at the Connector, the callbackRecv causes queued receives to be sent to the Connector in a recursive fashion. Therefore if receive operations are continually placed in the receive queue then there is the danger of Stack overflow due to too deep recursion with the synchronous receiving Controller. Therefore it is advisable to use the OrderingController only with asynchronous Connectors.

If the Connector's recv method is asynchronous, then the Controller's recv method is also asynchronous. The calling thread drives the receive operation to the Connector, but it is the Connector's asynchronous completion of the operation which drives further pending operations.

The Handles passed to the Connector are given the STATUS_RECEIVING transient state until either cancelRecv is called by the client application or callbackRecv is called by the Connector.

Specified by:

recv in interface Controller

Overrides:

recv in class AbstractController

Parameters:

dest - the Destination from which a message is to be received.

clientProps - not used but delegated in the Envelope to the Connector.

Returns:

a Handle associated to the recv operation.

callbackRecv

public void callbackRecv(Envelope env,
                         boolean success,
                         java.lang.String status)

Notify of the recv operation completion by the Connector layer. Since we only perform one recv at a time, we expect to match up the last recv Envelope with the returned value. This may not match up if a cancel took place, in which case a warning is logged. The next recv operation which is queued pending the completion at the Connector layer is triggered to continue.

Specified by:

callbackRecv in interface Controller

Overrides:

callbackRecv in class AbstractController

Parameters:

env - Envelope which has received a message or has failed at a Connector.

success - true if successful, else false.

status - description of the status.

cancelSend

public void cancelSend(Destination dest,
                       Handle hdl)

Cancel the Handle. Several circumstances are possible depending on timing.

Firsly if the Handle corresponds to the Handle currently at the Connector layer for the Destination, then we cancel the Handle at the Connector layer. Once this completes, we need to perform the next send as if callbackSend had been called. Note that the callbackSend may be in a race condition with the cancel at the Connector, so that the callbackSend may or may not be called. We must assure that we do not send two messages to the Connector if the callbackSend does come through although we do cancel.

Another possibility is that the cancel belongs to a Handle which is not the current Handle registered at the Connector layer. We do not remove the cancelled Handle from the pending send list, but wait for the normal processing at callbackSend time to skip sending the Handle to the Connector layer.

Unknown Handles are ignored.

Specified by:

cancelSend in interface Controller

Overrides:

cancelSend in class AbstractController

Parameters:

dest - the Destination of the send operation associated to hdl.

hdl - the Handle associated with the send operation to cancel.

cancelRecv

public void cancelRecv(Destination dest,
                       Handle hdl)

Cancel the Handle. Several circumstances are possible depending on timing.

Firsly if the Handle corresponds to the Handle currently at the Connector layer for the Destination, then we cancel the Handle at the Connector layer. Once this completes, we need to perform the next recv as if callbackRecv had been called. Note that the callbackRecv may be in a race condition with the cancel at the Connector, so that the callbackRecv may or may not be called. We must assure that we do not send two messages to the Connector if the callbackRecv does come through although we do cancel.

Another possibility is that the cancel belongs to a Handle which is not the current Handle registered at the Connector layer. We do not remove the cancelled Handle from the pending send list, but wait for the normal processing at callbackRecv time to skip sending the Handle to the Connector layer.

Unknown Handles are ignored.

Specified by:

cancelRecv in interface Controller

Overrides:

cancelRecv in class AbstractController

Parameters:

dest - the Destination of the send operation associated to hdl.

hdl - the Handle associated with the recv operation to cancel.

run

public void run()

The cache cleanup task which removes any Destination state which has no pending or active receive or send operations. Destination states are left in the cache since it is more efficient to reuse them than discard them immediately. Discarding takes place only when the number of Destination states in the cache has exceeded the maximum cache size.

Specified by:

run in interface java.lang.Runnable

handleTimerInterrupt

public void handleTimerInterrupt()

Schedule the cache cleanup task.

Specified by:

handleTimerInterrupt in interface TimerListener

See Also:

Timer.interruptAt(long, org.jumpi.spi.component.TimerListener), Timer.interruptAfterDelay(long, org.jumpi.spi.component.TimerListener), Timer.interruptPeriodically(long, org.jumpi.spi.component.TimerListener, boolean)