DtAsyncConnectionMgr Class Reference

the Asynchronous Connection Mgr adds asynchronous I/O to the connectionMgr class. More...

Inheritance diagram for DtAsyncConnectionMgr:

Collaboration diagram for DtAsyncConnectionMgr:

Public Member Functions
virtual	~DtAsyncConnectionMgr ()
virtual int	init (DtRIDParameters *params)
	Initialize the RTI's exercise connections Users who wish to override the MAK RTI's network connections should override this function After the network connections are set it up creates a thread to monitor the sockets.
virtual void	setFederateMgr (DtFederateMgr *fedMgrPtr)
	Set the federate manager.
virtual void	shutdown ()
	Request shutdown of communication activities.
virtual bool	isAsynchronous ()
	return if in asynchronous mode
virtual void	disconnectFederation ()
	The LRC disassociates its connections from the federation.
virtual void	connectFederation (DtLogicalTimeFactory *logicalTimeFactoryPtr, MAKRti::DtString fedExName, DtFederateHandle fedHandle, int fedExHandle)
	The LRC associates its federation information with its network connections.
virtual void	queueMsg (DtRtiMsg *msg, MAKRti::DtTransportType transport, const MAKRti::DtInetAddr &addr=MAKRti::DtInetAddr::inaddrAny())
	method uses a mutex to lock the send queue while adding a a packet to the queue
virtual int	send (const DtRtiMsg &msg, MAKRti::DtTransportType transport, const MAKRti::DtInetAddr &addr=MAKRti::DtInetAddr::inaddrAny())
	Send a message.
virtual int	sendAndDelete (DtRtiMsg *msg, MAKRti::DtTransportType transport, const MAKRti::DtInetAddr &addr=MAKRti::DtInetAddr::inaddrAny())
	Send a message and delete the message after it has been sent Retraction record of timestamped messages must made here.
virtual bool	tick ()
	Process the connections for synchronous IO or callbacks.
virtual bool	processConnections ()
	Check all the connections and process any incoming messages Return boolean: message was received? (i.e.
virtual DtRtiMsg *	getMessage ()
	Get a message if one exists Message must be destroyed by caller.
virtual DtNetReadStatus	handleMsg (DtRtiMsg &msg)
	Handle this message by either processing it or queuing it if save pending.
virtual bool	flush ()
	Does nothing in Async.
virtual void	receiveThread ()
	worker thread (called by run)
virtual void	sendThread ()
virtual bool	wait (MAKRti::DtTime periodInSeconds, bool unlockWhileWaiting=false)
	Wait for data to be ready or the given period of time to expire Return boolean: data is ready for reading?
virtual int	readFileDescriptor (RTI::TransportType transport)
	Returns a file descriptor on which data will be present for the given transport type whenever there is incoming data waiting to be read.
virtual int	waitEvent ()
	return the waiting handle
virtual void	signalReceiveEvent ()
	Signal the receive event Used by IO thread to signal federate thread.
virtual void	clearReceiveEvent ()
	Clear receive event Used by federate thread to clear signal.
virtual void	signalSendEvent ()
	Signal the send event Used by federate thread to signal IO thread.
virtual void	clearSendEvent ()
	Clear send event Used by IO thread to clear signal.
virtual void	yield ()
	Yield processing to allow more data to arrive.
virtual bool	processConnectionsInCallback ()
	Process the connections while in a callback to get a message (e.g., processing MOM resign) Return OK status of connections.
virtual void	exitServiceCall ()
	Called by RTI Ambassador services to release mutually exclusive access to RTI managers.
Public Member Functions inherited from DtConnectionMgr
virtual	~DtConnectionMgr ()
	Destructor.
virtual void	initParams (DtRIDParameters *params)
	Initialize the parameters used by the connection manager.
virtual void	enableSmartForwarding (bool enableSmartForwarding, bool enableInternalMessageSmartForwarding, int maxNumFederates)
	Enables or disables smart forwarding.
virtual DtRIDParameters *	params ()
	Get a pointer to the RID parameters.
virtual bool	isShuttingDown ()
	Shutdown acknowledged.
virtual int	sendStamped (const DtRtiMsg &msg, MAKRti::DtTransportType transport, const MAKRti::DtInetAddr &=MAKRti::DtInetAddr::inaddrAny())
	Send a message stamped with fedEx name and federate ID Retraction record of timestamped messages made here.
virtual int	sendStampedAnonymous (const DtRtiMsg &msg, MAKRti::DtTransportType transport, const MAKRti::DtInetAddr &addr=MAKRti::DtInetAddr::inaddrAny())
	Send a message stamped with fedEx name, but not federate ID Does not make retraction record of sent messages.
virtual int	sendDirectedAndDelete (DtRtiMsg *msg, MAKRti::DtTransportType transport, DtFederateHandle fedHandle, const MAKRti::DtInetAddr &addr=MAKRti::DtInetAddr::inaddrAny())
	Send and delete a message stamped with federate ID These messages are sent to a specific entity by using internal smart forwarding if enabled.
virtual int	sendStampedAndDelete (DtRtiMsg *msg, MAKRti::DtTransportType transport, const MAKRti::DtInetAddr &=MAKRti::DtInetAddr::inaddrAny())
	Send and delete a message stamped with fedEx name and federate ID Retraction record of timestamped messages made here.
virtual void	sendLoopback (DtRtiMsg *msg)
	Send a message to be processed by this federate.
virtual bool	drainBacklog (MAKRti::DtTime timeout=0.0)
	Attempts to flush backlogged messages repeatedly until either all messages are sent or the timeout expires.
virtual void	activateTransmitDelay ()
	Activate a delay timer (based on RID) used by managers to control transmission of internal state Expired delays disable activation for current tick.
virtual bool	delayTransmit () const
	Return true if retransmission of internal state should be delayed.
virtual void	testTransmitDelay ()
	Test if retransmit delay time has expired.
virtual void	activateAutoProvideDelay ()
	Activate a delay timer (based on RID) used by managers to control timing for autoprovide callbacks Expired delays disable activation for current tick.
virtual bool	delayAutoProvide () const
	Return true if autoprovides should be delayed.
bool	allProvideUpdateRequestsDelayed () const
	Return true if all provide update calls should be delayed (not just auto provide).
virtual void	testAutoProvideDelay ()
	Test if autoprovide delay time has expired.
virtual void	processThrottledMessages ()
	Process messages in throttle queue that are being sent at specified rate.
virtual void	flushThrottledMessages ()
	Flush messages in throttle queue.
virtual unsigned long	incrementServiceCallCount ()
	Increment a counter to identify the service call being processed.
virtual unsigned long	serviceCallCount () const
	Return service call counter.
virtual bool	processConnection (DtRtiConnection *conn, MAKRti::DtTransportType trans)
	Read from the given connection and process the message (if received) Return boolean: message was received? (i.e.
virtual void	updateBufferSizes ()
	Update Buffer sizes from Rid params object.
virtual DtNetReadStatus	processMsg (const DtRtiMsg &msg)
	this function applies the callbacks to the message
virtual DtNetReadStatus	momProcessMsg (const DtRtiMsg &msg)
	apply MOM callbacks to message
DtFederateMgr *	fedMgr ()
	The federate manager.
DtPartnerMgr *	partnerMgr ()
	Partner manager.
virtual MAKRti::DtTransportType	internalTransportType () const
	Returns the TransportType associated for the internal connection.
virtual bool	setupSelectParams (DtSelectParam *params)
	adds Read File descriptors to a DtSelectParams Return boolean: file descriptor was added? In case no descriptors have been added, should sleep instead of select as there will be no input.
virtual bool	hasTransport (MAKRti::DtTransportType trans)
	returns true if the specified transport type is available
virtual DtRtiMsg *	create (DtRtiMsgKind kind)
	Request the RTI message factory for a msg.
virtual void	setMsgFactory (DtRtiMsgFactory *fact)
	Get/Set the message factory.
virtual DtRtiMsgFactory *	msgFactory () const
virtual void	forceDropMulticastGroup (const MAKRti::DtInetAddr &addr, const MAKRti::DtInetAddr &netInterface=MAKRti::DtInetAddr::inaddrAny())
	Override listen count and force drop multicast address.
virtual bool	repairFederationConnection ()
	Attempt to re-establish a broken connection to the federation Returns true if successful.
virtual RTI::FederateAmbassador *	fedAmb () const
	Return pointer to Federate Ambassador.
virtual void	addFedAmbWrapper (MAKRti::DtString const &name, DtFedAmbWrapper *fedAmb)
	Register a Federate Wrapper with the RTI A federate wrapper allows your plugin to monitor any RTI invoked services.
virtual bool	addFedAmbWrapperAtFront (MAKRti::DtString const &name, DtFedAmbWrapper *fedAmb)
	When the fedAmb Wrapper is installed make certain that this particular wrapper is the first to process the msg.
virtual DtFedAmbWrapper *	removeFedAmbWrapper (MAKRti::DtString const &name)
	Remove federate wrapper.
virtual unsigned long	connectionId () const
	Return connection ID.
virtual DtFederateHandle	fedHandle () const
	Return federate handle.
virtual unsigned long	fedExHandle () const
	Return federation execution handle.
virtual const MAKRti::DtString &	fedExName () const
	Return federation execution name.
virtual void	setFederateHandle (DtFederateHandle handle)
	set federate handle ( this is used by restore and should not be called otherwise )
virtual DtLogicalTimeFactory *	logicalTimeFactory (DtFederationHandle federationHandle=0) const
	Get the factory for the given federation that can create logical time/interval instances.
virtual void	addLogicalTimeFactory (DtFederationHandle federationHandle, DtLogicalTimeFactory *timeFactory)
	Add the factory for the given federation that can create logical time/interval instances.
virtual void	removeLogicalTimeFactory (DtFederationHandle federationHandle)
	Remove the factory for the given federation that can create logical time/interval instances.
virtual DtRtiMsgQueue *	msgQueue ()
	Set/Get the time managed message queue.
virtual void	setMsgQueue (DtRtiMsgQueue *msgQueuePtr)
virtual DtRetractInfoList *	retractList ()
	Set/Get the time managed retract list.
virtual void	setRetractList (DtRetractInfoList *listPtr)
virtual bool	saveInProgress () const
	Set/Get the flag indicating whether a save is in progress Unless a message is marked safe to process during a save a received message is queued for later processing.
virtual void	setSaveInProgress (bool saveIsInProgress)
virtual bool	restoreInProgress () const
	Set/Get the flag indicating whether a restore is in progress Unless a message is marked safe to process during a save a received message is queued for later processing.
virtual void	setRestoreInProgress (bool saveIsInProgress)
virtual bool	saveRestoreInProgress () const
	Return bool: Is save or restore in progress.
virtual void	setMetadataControl (DtMetadataType type, bool onOff)
	Set/Get metadata control indicators.
virtual bool	isMetadataEnabled (DtMetadataType type)
virtual void	sendShutdownForwarderMsg ()
	Called by the rtiExec when it is exiting, only if it started this forwarder.
virtual unsigned long	sequenceNumber () const
	Get the next sequence number.
virtual void	incrementSequenceNumber ()
	Increment the sequence number.
virtual int	sendDirectedStampedAndDelete (DtRtiMsg *msg, MAKRti::DtTransportType transport, DtFederateHandle fedHandle, const MAKRti::DtInetAddr &=MAKRti::DtInetAddr::inaddrAny())
virtual int	sendDirectedStampedAndDelete (DtRtiMsg *msg, MAKRti::DtTransportType transport, DtBitMask *sendMask, const MAKRti::DtInetAddr &=MAKRti::DtInetAddr::inaddrAny())
virtual int	sendStampedAnonymousAndDelete (DtRtiMsg *msg, MAKRti::DtTransportType transport, const MAKRti::DtInetAddr &addr=MAKRti::DtInetAddr::inaddrAny())
virtual int	sendDirectedStampedAnonymousAndDelete (DtRtiMsg *msg, MAKRti::DtTransportType transport, DtFederateHandle fedHandle, const MAKRti::DtInetAddr &addr=MAKRti::DtInetAddr::inaddrAny())
virtual void	setProcessingTick (bool processingTick)
	Set/Get tick processing state.
virtual bool	processingTick () const
	Set/Get tick processing state.
virtual void	setEvokingCallbacks (bool evokingCbs)
	Set/Get the evoke callbacks state.
virtual bool	evokingCallbacks () const
	Set/Get the evoke callbacks state.
virtual void	addPendingMsg (DtRtiMsg *msg)
	Add/get/process a pending message Messages will be processed during the next tick.
virtual bool	processPendingMsg ()
	Add/get/process a pending message Messages will be processed during the next tick.
virtual DtRtiMsg *	nextPendingMsg ()
	Add/get/process a pending message Messages will be processed during the next tick.
virtual void	emptyPendingMsgList ()
	Add/get/process a pending message Messages will be processed during the next tick.
bool	isPendingMsgQueueEmpty ()
	Add/get/process a pending message Messages will be processed during the next tick.
virtual void	addMsgCallback (DtRtiMsgKind msgKind, DtRtiMsgCbMgr::DtCallbackFcn fcn, void *usr, bool isMomCallback=false)
	Message Callbacks These callback functions operate on all of the connections.
virtual void	removeMsgCallback (DtRtiMsgKind msgKind, DtRtiMsgCbMgr::DtCallbackFcn fcn, void *usr, bool isMomCallback=false)
	Message Callbacks These callback functions operate on all of the connections.
virtual void	addWildcardMsgCallback (DtRtiMsgCbMgr::DtCallbackFcn fcn, void *usr, bool isMomCallback=false)
	Message Callbacks These callback functions operate on all of the connections.
virtual void	removeWildcardMsgCallback (DtRtiMsgCbMgr::DtCallbackFcn fcn, void *usr, bool isMomCallback=false)
	Message Callbacks These callback functions operate on all of the connections.
virtual DtRtiConnection *	reliableConn () const
	Returns the connection created by createTcpConn().
virtual DtRtiConnection *	bestEffortConn () const
	Returns the connection created by createBestEffortConn().
virtual void	listenToMulticastGroup (const MAKRti::DtInetAddr &addr, const MAKRti::DtInetAddr &netInterface=MAKRti::DtInetAddr::inaddrAny())
	Add/Drop multicast groups from the best effort connection Listens are accumulated and drops decrement.
virtual void	dropMulticastGroup (const MAKRti::DtInetAddr &addr, const MAKRti::DtInetAddr &netInterface=MAKRti::DtInetAddr::inaddrAny())
	Add/Drop multicast groups from the best effort connection Listens are accumulated and drops decrement.
virtual bool	lockServiceCall ()
	Called by asynchronous processing thread to gain mutually exclusive access to RTI managers.
virtual void	unlockServiceCall ()
	Called by asychronous processing thread to release mutually exclusive access to RTI managers.
virtual bool	serviceCallLocked ()
	Called by RTI Ambassador services query if mutex was set using lockServiceCall()
virtual bool	enterServiceCall ()
	Called by RTI Ambassador services to gain mutually exclusive access to RTI managers.
virtual bool	inServiceCall ()
	Called by RTI Ambassador services query if mutex was set using enterServiceCall()

Static Public Member Functions
static DtConnectionMgr *	create (DtRIDParameters *p, bool isFederate=false, DtBoost::shared_ptr< DtRtiAssistantLrcClient > assistantClient=DtBoost::shared_ptr< DtRtiAssistantLrcClient >())
	Creator function.
static void *	runReceive (void *usr)
	thread start calls worker thread
static void *	runSend (void *usr)
Static Public Member Functions inherited from DtConnectionMgr
static void	setCreatorFunction (DtConnectionMgrCreator fn)
	Register a creator function.
static void	setInternalCreatorFunction (DtConnectionMgrCreator fn)
	Register alternate internal creators.

Protected Member Functions
	DtAsyncConnectionMgr (DtRIDParameters *params, bool isFederate=false, DtBoost::shared_ptr< DtRtiAssistantLrcClient > assistantClient=DtBoost::shared_ptr< DtRtiAssistantLrcClient >())
virtual bool	doReceiveProcessing (unsigned long timeout)
	Do the receive processing of the connections.
virtual void	doDisconnectFederation ()
	Perform actual disconnect.
virtual bool	doWait (MAKRti::DtTime period)
	Perform actual wait, called by wait function.
virtual void	processCallbackMsg (const DtFedAmbCallbackMsg &callbackMsg)
	Message callback.
virtual bool	sendPackets ()
	Send buffered packets.
virtual DtNetReadStatus	receivePacket (DtRtiConnection *conn, MAKRti::DtTransportType transportType, unsigned int &receivedPacketCount)
	Process a packet from the given connection.
virtual bool	receivePackets ()
	Process connections and either place packets in receive queue or process messages directly if asynchronous callbacks are enabled.
virtual DtNetReadStatus	processMsgQueueCallbacks (DtRtiMsg &msg, unsigned int &receivedPacketCount)
	Process message but queue any resulting federate ambassador callbacks.
virtual void	processPendingMsgQueueCallbacks ()
	Process a pending message but queue any resulting federate ambassador callbacks.
virtual void	processPendingMsgMakeCallbacks ()
	Process a pending message and make any resulting federate ambassador callbacks.
virtual void	setMessageTracking (DtRtiMsg &msg)
	Set the appropriate tracking values in the message.
virtual void	destroyPackets ()
	Destroy any packets in the send and receive buffers and queues.
virtual void	destroyRelConn ()
	Destroy reliable connection.
virtual void	destroyAncillaryRelConn (DtRtiConnection *conn)
	Destroy the given ancillary reliable connection.
virtual void	destroyAncillaryRelConns ()
	Destroy ancillary reliable connections.
virtual void	startAsynchronousReceiveProcessing ()
	Start asynchronous processing.
virtual void	startAsynchronousSendProcessing ()
virtual void	stopAsynchronousReceiveProcessing ()
	Stop asynchronous processing.
virtual void	stopAsynchronousSendProcessing ()
DtQueuePacket *	getPacket ()
	Get a packet off of receive buffer Receive buffer is refreshed if empty.
virtual bool	createBestEffortConn ()
	Create best effort connection; Returns true if successful Could be that it already exist in which case it is still used.
virtual bool	queueReceivePacket (DtRtiMsg *rtiMessage, MAKRti::DtTransportType transportType, bool &overflowWarning)
	Put message on receive queue.
virtual void	testIoIndicator (const char *label, unsigned int queueLevel, unsigned int &lowWaterLine, unsigned int &highWaterLine, bool &overflowWarning)
	Test if the number of messages in an IO queue has exceeded the diagnostic indicator level.
Protected Member Functions inherited from DtConnectionMgr
	DtConnectionMgr (DtRIDParameters *params, bool isFederate=false, DtBoost::shared_ptr< DtRtiAssistantLrcClient > assistantClient=DtBoost::shared_ptr< DtRtiAssistantLrcClient >())
	Constructor is performed through create.
virtual int	initConnections ()
	Initialize the RTI's exercise connections Users who wish to override the MAK RTI's network connections should override this function.
virtual void	destroyBestEffortConn ()
	Destroy best effort connection.
virtual bool	createRelConn ()
	Create reliable connection; Returns true if successful.
virtual unsigned int	createAncillaryRelConns ()
	Create ancillary reliable connections; Returns count of successful connections.
virtual unsigned int	ancillaryReliableConnsCount () const
	Report number of ancillary reliable connections successfully established.
virtual bool	checkRelConn ()
	Check the reliable connection status.
virtual void	processReadResult (int readError)
	Process read result to print error diagnostics for example.
virtual int	sendStampedUtilityFunction (DtRtiMsg *msg, MAKRti::DtTransportType transport, bool performDelete, bool isAnonymous, const MAKRti::DtInetAddr &addr)
	Called by sendStamped, sendStampedAndDelete, sendAnonymousStamped, etc.

Static Protected Member Functions
static void	processCallbackMsg (const DtFedAmbCallbackMsg &callbackMsg, void *usr)
	Message callback.

Protected Attributes
bool	myTimeToQuitSending
bool	myTimeToQuitReceiving
bool	mySendThreadActivated
bool	myReceiveThreadActivated
bool	myAsynchronous
bool	myAsynchronousCallbacks
bool	myAsynchronousProcessMsg
bool	myReserved
bool	myMoreToFlush
bool	myDoDisconnect
bool	myNonReflectedList [DtInvalidMsgKind+1]
DtSelectParamWithWrite	myWaitEventsParams
pthread_t	myReceiveThreadHandle
pthread_t	mySendThreadHandle
int	mySendEvent [2]
int	myRecvEvent [2]
float	myIoPeriod
float	myTickWaitPeriod
unsigned int	myMaxQueue
unsigned int	myMaxCount
unsigned int	mySendPacketCount
	Number of packets until time to yield (== max count)
unsigned int	myReadPacketCount
DtAsycQueueType *	mySendQueue
	Exchange queues between federate and IO threads these lists are used in asynchronous mode.
DtAsycQueueType *	myReceiveQueue
DtCallbackQueue *	myCallbackQueue
	A federate ambassador wrapper that queues up federate ambassador calls.
DtThreadSafe< bool >	myAsynchronousConnectionIsOk
unsigned int	myNextReliableConnToRead
	Mechanism to track which reliable connection to fetch packets from when multiple ancillary reliable connections exist.
unsigned int	myIoQueueIndicator
	These are pointers to the locations in the WaitEvent list they need to be dynamic in case one of them doesn't exist.
unsigned int	myReceiveQueueLowWaterLine
unsigned int	myReceiveQueueHighWaterLine
bool	myReceiveOverflowWarning
unsigned int	mySendQueueLowWaterLine
unsigned int	mySendQueueHighWaterLine
bool	mySendOverflowWarning
bool	myShouldWaitOnCurrentBundle
	This determines whether the send thread should sleep after it attempts to send all packets, but there is still more to send due to bundling or socket send failures.
Protected Attributes inherited from DtConnectionMgr
DtRtiMsgFactory *	myMsgFactory
	The message factory.
DtRtiConnection *	myBestEffortConn
	The best effort connection.
DtRtiConnection *	myPrimaryReliableConn
	The primary reliable connection.
DtRtiConnectionList	myReliableConns
DtRIDParameters *	myParams
	The RTI initialization parameters.
DtFederateMgr *	myFedMgr
	The federate manager.
std::map< MAKRti::DtInetAddr, DtMulticastEntry * >	myMulticastGroups
	LRC listening to these multicast groups.
DtFedAmbWrapperChain	myFedAmbChain
	Chain of Federate Ambassador Wrappers.
DtDfltMsgFactory	myDfltMsgFactory
	Default message factory.
DtRtiMsgCbMgr	myMsgCbMgr
	Message callback manager.
DtRtiMsgCbMgr	myMomMsgCbMgr
	MOM message callback manager.
MAKRti::DtTransportType	myInternalTransportType
	Transport type used for internal messages.
bool	myAddTransportTypeMetadata
	Metadata control indicators.
bool	myAddMcastMetadata
std::map< DtFederationHandle, DtLogicalTimeFactory * >	myLogicalTimeFactories
	Factories for creating logical time instances.
DtRtiMsgQueue *	myMsgQueue
	Message queue.
DtRetractInfoList *	myRetractList
	Message retractions.
DtMessageQueueType	myPendingMessages
	Pending messages.
bool	mySaveInProgress
	Save status.
bool	myRestoreInProgress
	Restore status.
unsigned long	myConnectionId
	"unique" identifier for this connMgr (if there is a good source of unique ids)
MAKRti::DtString	myFedExName
	Federation execution name.
DtFederateHandle	myFedExHandle
	Federation execution handle.
DtFederateHandle	myFederateHandle
	Federate handle.
unsigned long	myReceivedMsgCount
	Number of received messages.
unsigned long	mySentMsgCount
	Number of sent messages.
MAKRti::DtList	myWorkingMsgId
	Diagnostic message ID stack.
DtMessageThrottle *	myMessageThrottle
	Message output throttle.
bool	myAllProvideUpdateRequestsDelayed
	Extend auto provide delay to all attribute update requests.
DtMutex *	myCallbackMutex
	Mutex to synchronize access to RTI and Federate Ambassadors.
unsigned int	myLockService
	RTI is processing a message.
unsigned int	myInService
	RTI is processing a service call.
MAKRti::DtClock	myClock
	Clock for timing.
DtBoost::shared_ptr < DtRtiAssistantLrcClient >	myAssistantClient
	The assistant client.
bool	myIsFederate
	Whether this manager is for a federate.
unsigned long	myVersionConflictCount
	The count of message version conflicts.
MAKRti::DtTime	myVersionConflictNoticationTimer
unsigned long	mySizeMismatchCount
	The count of message size mismatches.
MAKRti::DtTime	mySizeMismatchNoticationTimer
DtPartnerMgr *	myPartnerMgr
unsigned long	myServiceCallCount
	Service call counter - increment before processing each service call.
bool	myProcessingTick
	Tick processing state.
bool	myEvokingCallbacks
	Indicates if callbacks are being evoked.
bool	myProcessingMessage
	Indicates if message is being processed.
bool	myShutdown
	Shutdown requested.
bool	mySendShuttingDown
	Shutdown acknowledged.
bool	myReceiveShuttingDown
	Shutdown acknowledged.
unsigned long	myTickCount

Additional Inherited Members
Public Types inherited from DtConnectionMgr
typedef DtConnectionMgr *(*	DtConnectionMgrCreator )(DtRIDParameters *p, bool isFederate, DtBoost::shared_ptr< DtRtiAssistantLrcClient > assistantClient)
	Connection Manager Creator.
Protected Types inherited from DtConnectionMgr
typedef std::list < DtRtiConnection * >	DtRtiConnectionList
	List of all reliable connections (for forwarder load balancing)
Static Protected Attributes inherited from DtConnectionMgr
static DtConnectionMgrCreator	theCreator
	The plugin creator.
static DtConnectionMgrCreator	theInternalCreator
	The internal creator.

Detailed Description

the Asynchronous Connection Mgr adds asynchronous I/O to the connectionMgr class.

A thread is created to manage the send and receive queues. It can operate in a synchronous mode if the myAsynchronous flag is set to false.

Constructor & Destructor Documentation

DtAsyncConnectionMgr::DtAsyncConnectionMgr ( DtRIDParameters *  params, bool  isFederate = false, DtBoost::shared_ptr< DtRtiAssistantLrcClient >  assistantClient = DtBoost::shared_ptr< DtRtiAssistantLrcClient >()  )

protected

virtual DtAsyncConnectionMgr::~DtAsyncConnectionMgr ( )

virtual

Member Function Documentation

virtual void DtAsyncConnectionMgr::clearReceiveEvent ( )

virtual

Clear receive event Used by federate thread to clear signal.

virtual void DtAsyncConnectionMgr::clearSendEvent ( )

virtual

Clear send event Used by IO thread to clear signal.

virtual void DtAsyncConnectionMgr::connectFederation ( DtLogicalTimeFactory *  logicalTimeFactoryPtr, MAKRti::DtString  fedExName, DtFederateHandle  federateHandle, int  fedExHandle  )

virtual

The LRC associates its federation information with its network connections.

If the LRC is communicating with the rtiexec through a reliable connection, then the best effort connection is created here.

Reimplemented from DtConnectionMgr.

static DtConnectionMgr* DtAsyncConnectionMgr::create ( DtRIDParameters *  p, bool  isFederate = false, DtBoost::shared_ptr< DtRtiAssistantLrcClient >  assistantClient = DtBoost::shared_ptr< DtRtiAssistantLrcClient >()  )

static

Creator function.

Reimplemented from DtConnectionMgr.

Reimplemented in DtNetStatsAsyncConnMgr.

virtual bool DtAsyncConnectionMgr::createBestEffortConn ( )

protectedvirtual

Create best effort connection; Returns true if successful Could be that it already exist in which case it is still used.

Reimplemented from DtConnectionMgr.

virtual void DtAsyncConnectionMgr::destroyAncillaryRelConn ( DtRtiConnection *  conn )

protectedvirtual

Destroy the given ancillary reliable connection.

Reimplemented from DtConnectionMgr.

virtual void DtAsyncConnectionMgr::destroyAncillaryRelConns ( )

protectedvirtual

Destroy ancillary reliable connections.

Reimplemented from DtConnectionMgr.

virtual void DtAsyncConnectionMgr::destroyPackets ( )

protectedvirtual

Destroy any packets in the send and receive buffers and queues.

The buffers and queues themselves are not destroyed

virtual void DtAsyncConnectionMgr::destroyRelConn ( )

protectedvirtual

Destroy reliable connection.

Reimplemented from DtConnectionMgr.

virtual void DtAsyncConnectionMgr::disconnectFederation ( )

virtual

The LRC disassociates its connections from the federation.

If the LRC is communicating with the rtiexec through a reliable connection, then the best effort connection is destroyed here.

Reimplemented from DtConnectionMgr.

virtual void DtAsyncConnectionMgr::doDisconnectFederation ( )

protectedvirtual

Perform actual disconnect.

virtual bool DtAsyncConnectionMgr::doReceiveProcessing ( unsigned long  timeout )

protectedvirtual

Do the receive processing of the connections.

Takes timeout in milliseconds. Return OK status of connections.

virtual bool DtAsyncConnectionMgr::doWait ( MAKRti::DtTime  period )

protectedvirtual

Perform actual wait, called by wait function.

Reimplemented from DtConnectionMgr.

virtual void DtAsyncConnectionMgr::exitServiceCall ( )

virtual

Called by RTI Ambassador services to release mutually exclusive access to RTI managers.

Reimplemented from DtConnectionMgr.

virtual bool DtAsyncConnectionMgr::flush ( )

virtual

Does nothing in Async.

I/O mode In sync. I/O mode, Flush the send queues Send buffered messages. Returns indication of whether flush is complete or incomplete.

Reimplemented from DtConnectionMgr.

virtual DtRtiMsg* DtAsyncConnectionMgr::getMessage ( )

virtual

Get a message if one exists Message must be destroyed by caller.

Reimplemented from DtConnectionMgr.

Reimplemented in DtNetStatsAsyncConnMgr.

DtQueuePacket* DtAsyncConnectionMgr::getPacket ( )

inlineprotected

Get a packet off of receive buffer Receive buffer is refreshed if empty.

virtual DtNetReadStatus DtAsyncConnectionMgr::handleMsg ( DtRtiMsg &  msg )

virtual

Handle this message by either processing it or queuing it if save pending.

Reimplemented from DtConnectionMgr.

virtual int DtAsyncConnectionMgr::init ( DtRIDParameters *  params )

virtual

Initialize the RTI's exercise connections Users who wish to override the MAK RTI's network connections should override this function After the network connections are set it up creates a thread to monitor the sockets.

Reimplemented from DtConnectionMgr.

Reimplemented in DtRtiShmQNetMgr.

virtual bool DtAsyncConnectionMgr::isAsynchronous ( )

virtual

return if in asynchronous mode

virtual void DtAsyncConnectionMgr::processCallbackMsg ( const DtFedAmbCallbackMsg &  callbackMsg )

protectedvirtual

Message callback.

static void DtAsyncConnectionMgr::processCallbackMsg ( const DtFedAmbCallbackMsg &  callbackMsg, void *  usr  )

staticprotected

Message callback.

virtual bool DtAsyncConnectionMgr::processConnections ( )

virtual

Check all the connections and process any incoming messages Return boolean: message was received? (i.e.

could be more input). method uses a mutex to receive the send queue while adding a packet to the queue

Reimplemented from DtConnectionMgr.

virtual bool DtAsyncConnectionMgr::processConnectionsInCallback ( )

virtual

Process the connections while in a callback to get a message (e.g., processing MOM resign) Return OK status of connections.

Reimplemented from DtConnectionMgr.

virtual DtNetReadStatus DtAsyncConnectionMgr::processMsgQueueCallbacks ( DtRtiMsg &  msg, unsigned int &  receivedPacketCount  )

protectedvirtual

Process message but queue any resulting federate ambassador callbacks.

Packet count is incremented if messages placed on receive buffer. Return status of message processing.

virtual void DtAsyncConnectionMgr::processPendingMsgMakeCallbacks ( )

protectedvirtual

Process a pending message and make any resulting federate ambassador callbacks.

virtual void DtAsyncConnectionMgr::processPendingMsgQueueCallbacks ( )

protectedvirtual

Process a pending message but queue any resulting federate ambassador callbacks.

Any resulting callback messages are put on receive queue.

virtual void DtAsyncConnectionMgr::queueMsg ( DtRtiMsg *  msg, MAKRti::DtTransportType  transport, const MAKRti::DtInetAddr &  addr = MAKRti::DtInetAddr::inaddrAny()  )

virtual

method uses a mutex to lock the send queue while adding a a packet to the queue

virtual bool DtAsyncConnectionMgr::queueReceivePacket ( DtRtiMsg *  rtiMessage, MAKRti::DtTransportType  transportType, bool &  overflowWarning  )

protectedvirtual

Put message on receive queue.

If queue is full, message will be deleted. Warning message will be produced for first deletion (repeated appropriately if IO queue diagnostics are enabled). Return true if message was placed on queue.

virtual int DtAsyncConnectionMgr::readFileDescriptor ( RTI::TransportType  transport )

virtual

Returns a file descriptor on which data will be present for the given transport type whenever there is incoming data waiting to be read.

Returns -1 if no such file descriptor.

On windows, always returns -1, use readFileEvent instead. On unix returns the file descriptor of the pipe used for inter-thread communication instead of the socket file descriptor. Transport type is ignored.

Reimplemented from DtConnectionMgr.

virtual DtNetReadStatus DtAsyncConnectionMgr::receivePacket ( DtRtiConnection *  conn, MAKRti::DtTransportType  transportType, unsigned int &  receivedPacketCount  )

protectedvirtual

Process a packet from the given connection.

Depending on processing mode and the packet contents, the packet may result in messages placed on receive buffer. Packet count is incremented if messages are placed on receive buffer. Returns status of read on connection

virtual bool DtAsyncConnectionMgr::receivePackets ( )

protectedvirtual

Process connections and either place packets in receive queue or process messages directly if asynchronous callbacks are enabled.

Number processed packets is governed by lock queue ceiling Returns bool: more packets to read?

virtual void DtAsyncConnectionMgr::receiveThread ( )

virtual

worker thread (called by run)

static void* DtAsyncConnectionMgr::runReceive ( void *  usr )

static

thread start calls worker thread

static void* DtAsyncConnectionMgr::runSend ( void *  usr )

static

virtual int DtAsyncConnectionMgr::send ( const DtRtiMsg &  msg, MAKRti::DtTransportType  transport, const MAKRti::DtInetAddr &  addr = MAKRti::DtInetAddr::inaddrAny()  )

virtual

Send a message.

Reimplemented from DtConnectionMgr.

virtual int DtAsyncConnectionMgr::sendAndDelete ( DtRtiMsg *  msg, MAKRti::DtTransportType  transport, const MAKRti::DtInetAddr &  addr = MAKRti::DtInetAddr::inaddrAny()  )

virtual

Send a message and delete the message after it has been sent Retraction record of timestamped messages must made here.

Reimplemented from DtConnectionMgr.

virtual bool DtAsyncConnectionMgr::sendPackets ( )

protectedvirtual

Send buffered packets.

Moves a block of packets from send queue to send packet buffer and sends them on network. If blocked sending, then holds packets in buffer for next attempt. Flushes connections on exit. Returns bool: packets sent without being blocked (including no sends)?

virtual void DtAsyncConnectionMgr::sendThread ( )

virtual

virtual void DtAsyncConnectionMgr::setFederateMgr ( DtFederateMgr *  fedMgrPtr )

virtual

Set the federate manager.

Reimplemented from DtConnectionMgr.

virtual void DtAsyncConnectionMgr::setMessageTracking ( DtRtiMsg &  msg )

protectedvirtual

Set the appropriate tracking values in the message.

virtual void DtAsyncConnectionMgr::shutdown ( )

virtual

Request shutdown of communication activities.

Reimplemented from DtConnectionMgr.

Reimplemented in DtNetStatsAsyncConnMgr.

virtual void DtAsyncConnectionMgr::signalReceiveEvent ( )

virtual

Signal the receive event Used by IO thread to signal federate thread.

virtual void DtAsyncConnectionMgr::signalSendEvent ( )

virtual

Signal the send event Used by federate thread to signal IO thread.

virtual void DtAsyncConnectionMgr::startAsynchronousReceiveProcessing ( )

protectedvirtual

Start asynchronous processing.

virtual void DtAsyncConnectionMgr::startAsynchronousSendProcessing ( )

protectedvirtual

virtual void DtAsyncConnectionMgr::stopAsynchronousReceiveProcessing ( )

protectedvirtual

Stop asynchronous processing.

virtual void DtAsyncConnectionMgr::stopAsynchronousSendProcessing ( )

protectedvirtual

virtual void DtAsyncConnectionMgr::testIoIndicator ( const char *  label, unsigned int  queueLevel, unsigned int &  lowWaterLine, unsigned int &  highWaterLine, bool &  overflowWarning  )

protectedvirtual

Test if the number of messages in an IO queue has exceeded the diagnostic indicator level.

Update indicator and output diagnostic if level has been reached.

virtual bool DtAsyncConnectionMgr::tick ( )

virtual

Process the connections for synchronous IO or callbacks.

Reimplemented from DtConnectionMgr.

virtual bool DtAsyncConnectionMgr::wait ( MAKRti::DtTime  periodInSeconds, bool  unlockWhileWaiting = false  )

virtual

Wait for data to be ready or the given period of time to expire Return boolean: data is ready for reading?

Reimplemented from DtConnectionMgr.

virtual int DtAsyncConnectionMgr::waitEvent ( )

virtual

return the waiting handle

virtual void DtAsyncConnectionMgr::yield ( )

virtual

Yield processing to allow more data to arrive.

Reimplemented from DtConnectionMgr.

Member Data Documentation

bool DtAsyncConnectionMgr::myAsynchronous

protected

bool DtAsyncConnectionMgr::myAsynchronousCallbacks

protected

DtThreadSafe<bool> DtAsyncConnectionMgr::myAsynchronousConnectionIsOk

protected

bool DtAsyncConnectionMgr::myAsynchronousProcessMsg

protected

DtCallbackQueue* DtAsyncConnectionMgr::myCallbackQueue

protected

A federate ambassador wrapper that queues up federate ambassador calls.

bool DtAsyncConnectionMgr::myDoDisconnect

protected

float DtAsyncConnectionMgr::myIoPeriod

protected

unsigned int DtAsyncConnectionMgr::myIoQueueIndicator

protected

These are pointers to the locations in the WaitEvent list they need to be dynamic in case one of them doesn't exist.

unsigned int DtAsyncConnectionMgr::myMaxCount

protected

unsigned int DtAsyncConnectionMgr::myMaxQueue

protected

bool DtAsyncConnectionMgr::myMoreToFlush

protected

unsigned int DtAsyncConnectionMgr::myNextReliableConnToRead

protected

Mechanism to track which reliable connection to fetch packets from when multiple ancillary reliable connections exist.

Must be set to zero whenever reliable connection is deleted.

bool DtAsyncConnectionMgr::myNonReflectedList[DtInvalidMsgKind+1]

protected

unsigned int DtAsyncConnectionMgr::myReadPacketCount

protected

bool DtAsyncConnectionMgr::myReceiveOverflowWarning

protected

DtAsycQueueType* DtAsyncConnectionMgr::myReceiveQueue

protected

unsigned int DtAsyncConnectionMgr::myReceiveQueueHighWaterLine

protected

unsigned int DtAsyncConnectionMgr::myReceiveQueueLowWaterLine

protected

bool DtAsyncConnectionMgr::myReceiveThreadActivated

protected

pthread_t DtAsyncConnectionMgr::myReceiveThreadHandle

protected

int DtAsyncConnectionMgr::myRecvEvent[2]

protected

bool DtAsyncConnectionMgr::myReserved

protected

int DtAsyncConnectionMgr::mySendEvent[2]

protected

bool DtAsyncConnectionMgr::mySendOverflowWarning

protected

unsigned int DtAsyncConnectionMgr::mySendPacketCount

protected

Number of packets until time to yield (== max count)

DtAsycQueueType* DtAsyncConnectionMgr::mySendQueue

protected

Exchange queues between federate and IO threads these lists are used in asynchronous mode.

They contain pointers to DtQueuePacket's. They are not used when not operating asynchronously

unsigned int DtAsyncConnectionMgr::mySendQueueHighWaterLine

protected

unsigned int DtAsyncConnectionMgr::mySendQueueLowWaterLine

protected

bool DtAsyncConnectionMgr::mySendThreadActivated

protected

pthread_t DtAsyncConnectionMgr::mySendThreadHandle

protected

bool DtAsyncConnectionMgr::myShouldWaitOnCurrentBundle

protected

This determines whether the send thread should sleep after it attempts to send all packets, but there is still more to send due to bundling or socket send failures.

float DtAsyncConnectionMgr::myTickWaitPeriod

protected

bool DtAsyncConnectionMgr::myTimeToQuitReceiving

protected

bool DtAsyncConnectionMgr::myTimeToQuitSending

protected

DtSelectParamWithWrite DtAsyncConnectionMgr::myWaitEventsParams

protected

---

The documentation for this class was generated from the following file:

• asyConMgr.h