org.portico.core.fedex
Class Federation

java.lang.Object
  org.portico.core.fedex.Federation

public class Federation
extends Object

The main representation of a federation within Portico. This class groups together all the relevant components used to provide the federation features. These components are described below.

Messaging and Behaviour

The Federation uses a group of MessageSinks (see the littlebluefrog labs commons project). The Federation class contains all the messaging related components used to process requests from federates and do the general business of the RTI. Requests are processed in three separate "zones":

• The Request Sink: The request-sink processes direct RTIambassador requests from federates. The response message that is generated as part of this process becomes the return to the RTIambassador request on the other side of the connection. The federate that initiated the initial request blocks until the response to this request is received back from the RTI. As such, this allows return values/exceptions/etc... to flow back to the calling federate as a result of processing in this sink. However, any actions that need to occur asynchronously cannot be performed here (otherwise they'd hold up the federate). In these situations, the request-sink handler generates a new request and places it on the "action-queue" for later processing by the RTI. At this point, it can return a response and free up the federate to move on. Any number of threads can be processing requests through this sink at a time. The actual thread involved depends on the communications binding that is being used.

• The Action Sink: The action-sink is where all the asynchronous processing of requests take place. There is a single thread of execution that takes messages from the action-queue (which are placed there by the handlers in the request-sink) and passes them to the action-sink. Within the action-sink, any necessary processing for the request can take place. In the course of this processing, a number of callbacks may be generated (for example, when a time advance can be granted). As the sending of a callback may involve I/O, it is done by separate worker threads. When a callback is generated, it is placed on the "callback-queue" where those workers can access it. This frees up the action-thread to work on the next message sitting in the action-queue.

• The Callback Processor: The callback-processor extracts messages that were placed on the callback-queue and sends them to the appropriate federate. The processor enacpsulates a number of threads to perform this task.

Threading and Queues

Each federation requires a number of threads to function properly. There are three basic tasks that must be performed within the scope of the federation, and which thread executes them is shown below:

• Request Sink Messages: This is *binding specific*. The thread which calls the RTIConnection.receiveRequest(MessageContext) method is the thread that is used to execute request-sink processing. For example, in the JSOP bindings, the thread that is servicing a particular socket for a federate-RTI connection is the one that is used. In the JVM bindings, it is the same thread that invokes the RTIambassador. The source of the thread cannot be determined, but many can be running request-sink processing calls at the same time.

• Action Sink Requests: A dedicated thread extracts messages from the action-queue and passes them to the action-sink. All action-sink behaviour is executed from this thread. Even though there is only one thread, *locking of state is still required* as the request-sink handlers could be also using it.

• Federate Callbacks: The CallbackQueueProcessor manages this process. It encapsulates a number of threads that extract messages from the callback-queue and perform the actual callback (which may involve expensive I/O).

Internally, queues are used as handoff points between the various sinks and processors. The purpose of these queues is outlined below:

• The Action Queue: The action queue is used to store requests that involve asynchronous processing. These requests are placed on the queue by the handlers inside the request-sink. They are removed by the ActionQueueProcessor, which then passes them to the action-sink.

• The Callback Queue: Currently, whenever a callback message is generated (be it in either the request-sink or the action-sink) it is placed on the callback-queue. The threads inside the CallbackQueueProcessor then extract the messages from here and actually send the callback.

Federation State

All federation state is stored in the FederationState object. This includes all the various state-management/altering entities (such as the SyncPointManager and TimeManager). This data is shared between a large number of potential threads, and as such, locking is necessary. Many of the management entities already perform this locking where necessary on behalf of a user. See the javadoc for each component for specific information.

MOM Data

Two entities relating to the MOM are stored in the Federation. The first is an instance of the MomManager that handles all MOM-related requests. The second is a specially created instance of the Federate class that is to represent the actual Mom federate. This is used to interact with incoming HLA-requests at the actual HLA level, and as such, it behaves just like a regular federate with a minor exception. The MOM federate isn't stored in the same place as the other federates that have joined the federation. Rather, this store is kept to be only user created federates. The MOM-federate shouldn't have any bearing on things such as determining whether or not a federation can be destoyed or not (there should be no federates left), and as such, it is kept separate from the other federates.

See the javadoc for the MomManager for more information on MOM behaviour.

Nested Class Summary

private class

Federation.DefaultRequestHandler The default handler for messages sent to the request-sink.

Field Summary

private ActionQueueProcessor	actionProcessor
private BlockingQueue<com.lbf.commons.messaging.RequestMessage>	actionQueue
private com.lbf.commons.messaging.MessageSink	actionSink
private Thread	actionThread
private CallbackQueueProcessor	callbackProcessor
private BlockingQueue<FED_CallbackMessage>	callbackQueue
private HashMap<String,Federate>	federates
static String	KEY_FEDERATION
private org.apache.log4j.Logger	logger
private Federate	momFederate
private MomManager	momManager
private String	name
private com.lbf.commons.messaging.MessageSink	requestSink
private RTIExec	rti
private boolean	running
private FederationState	state

Constructor Summary

Federation(RTIExec rti, String name, ObjectModel model)
Create a new federation for the given object model (the model is then passed on to the FederationState where it is stored).

Method Summary

boolean	containsFederate(String name) Does this federation contain a federate by the given name?
int	federateCount() Get a count of the number of joined federates (excludes the mom federate)
BlockingQueue<com.lbf.commons.messaging.RequestMessage>	getActionQueue()
com.lbf.commons.messaging.MessageSink	getActionSink()
Set<Federate>	getAllFederates()
BlockingQueue<FED_CallbackMessage>	getCallbackQueue()
Federate	getFederate(int handle) Get the joined federate of the given handle.
Federate	getFederate(String name) Get the joined federate of the given name.
org.apache.log4j.Logger	getFederationLogger()
MomManager	getMomManager()
String	getName()
com.lbf.commons.messaging.MessageSink	getRequestSink()
RTIExec	getRTIExec()
FederationState	getState()
void	joinFederate(Federate federate) Try and add the given federate to this federation.
Federate	resignFederate(String name) Remove the joined federate from this federation.
void	shutdown() This method will stop the internal processing threads for this federation.
void	startup() This method will start the internal processing threads for this federation.

Methods inherited from class java.lang.Object

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

Field Detail

KEY_FEDERATION

public static final String KEY_FEDERATION

See Also:

Constant Field Values

rti

private RTIExec rti

name

private String name

federates

private HashMap<String,Federate> federates

running

private boolean running

state

private FederationState state

momManager

private MomManager momManager

momFederate

private Federate momFederate

requestSink

private com.lbf.commons.messaging.MessageSink requestSink

actionSink

private com.lbf.commons.messaging.MessageSink actionSink

actionQueue

private BlockingQueue<com.lbf.commons.messaging.RequestMessage> actionQueue

actionProcessor

private ActionQueueProcessor actionProcessor

actionThread

private Thread actionThread

callbackQueue

private BlockingQueue<FED_CallbackMessage> callbackQueue

callbackProcessor

private CallbackQueueProcessor callbackProcessor

logger

private org.apache.log4j.Logger logger

Constructor Detail

Federation

public Federation(RTIExec rti,
                  String name,
                  ObjectModel model)
           throws JRTIinternalError,
                  JErrorReadingFED

Create a new federation for the given object model (the model is then passed on to the FederationState where it is stored). After construction, the federation still needs to be configured and started (none of the internal threads are active following constuction).

Constructors for this class should never be invoked directly. The FedexFactory should be used to create federations as it will ensure that they are propertly setup and configured.

Throws:

JRTIinternalError

JErrorReadingFED

Method Detail

startup

public void startup()

This method will start the internal processing threads for this federation.

shutdown

public void shutdown()

This method will stop the internal processing threads for this federation. After this call no more callbacks will be sent and no more incoming action-sink messages will be processed.

federateCount

public int federateCount()

Get a count of the number of joined federates (excludes the mom federate)

joinFederate

public void joinFederate(Federate federate)
                  throws JFederateAlreadyExecutionMember

Try and add the given federate to this federation. If successful, it will set the federation property of the federate to this federation.

Parameters:

federate - The federation that wants to join this federation

Throws:

FederateAlreadyExecutionMember - If there is already a federate with the given name that has been joined to this federation

JFederateAlreadyExecutionMember

resignFederate

public Federate resignFederate(String name)

Remove the joined federate from this federation. If the remove is successful, the Federate that was removed will be returned, if not federate of the given name exists to be removed, null will be returned.

Parameters:

name - The name of the federate to remove

Returns:

The federate that was remove, or null

getFederate

public Federate getFederate(String name)

Get the joined federate of the given name.

Parameters:

name - The name of the federate to get

Returns:

The joined federate of the given name or null

getFederate

public Federate getFederate(int handle)

Get the joined federate of the given handle.

Parameters:

handle - The handle of the federate to get

Returns:

The joined federate of the given handle or null

containsFederate

public boolean containsFederate(String name)

Does this federation contain a federate by the given name?

getAllFederates

public Set<Federate> getAllFederates()

getRTIExec

public RTIExec getRTIExec()

getName

public String getName()

getRequestSink

public com.lbf.commons.messaging.MessageSink getRequestSink()

getActionSink

public com.lbf.commons.messaging.MessageSink getActionSink()

getActionQueue

public BlockingQueue<com.lbf.commons.messaging.RequestMessage> getActionQueue()

getCallbackQueue

public BlockingQueue<FED_CallbackMessage> getCallbackQueue()

getState

public FederationState getState()

getFederationLogger

public org.apache.log4j.Logger getFederationLogger()

getMomManager

public MomManager getMomManager()