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 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 sychronous IO or callbacks.
virtual bool	processConnections ()
	Check all the connections and process any incomming 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 queueing it if save pending.
virtual bool	flush ()
	Does nothing in Async.
virtual void	workerThread ()
	worker thread (called by run)
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.
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 *	run (void *usr)
	thread start calls worker thread
Protected Member Functions
	DtAsyncConnectionMgr (DtRIDParameters *params, bool isFederate=false, DtBoost::shared_ptr< DtRtiAssistantLrcClient > assistantClient=DtBoost::shared_ptr< DtRtiAssistantLrcClient >())
virtual bool	doWait (MAKRti::DtTime period)
	Perform actual wait, called by wait function.
virtual void	processCallbackMsg (const DtFedAmbCallbackMsg &callbackMsg)
	Message callback.
virtual DtQueuePacket *	getPacket ()
	Get a packet off of receive buffer Receive buffer is refreshed if empty.
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	waitOnIO (DtSelectParamWithWrite &params)
	Make processing wait on IO.
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.
Static Protected Member Functions
static void	processCallbackMsg (const DtFedAmbCallbackMsg &callbackMsg, void *usr)
	Message callback.
Protected Attributes
bool	myTimeToQuit
bool	myThreadActivated
bool	myAsynchronous
bool	myAsynchronousCallbacks
bool	myAsynchronousProcessMsg
bool	myReserved
bool	myMoreToFlush
DtMutex *	mySendLock
DtMutex *	myReceiveLock
pthread_t	myThreadHandle
int	mySendEvent [2]
int	myRecvEvent [2]
float	myIoPeriod
float	myTickWaitPeriod
unsigned int	myMaxQueue
unsigned int	myMaxCount
unsigned int	myLockQueue
long	myReceivedBufferCurrent
	Federate thread buffers.
long	myReceivedBufferCount
DtQueuePacket **	myReceivedBuffer
unsigned int	myCurrentSendPacket
	IO thread buffers.
unsigned int	myPacketsToSendCount
std::vector< DtQueuePacket * >	mySendPacketBuffer
std::vector< DtQueuePacket * >	myRecvPacketBuffer
unsigned int	myReadPacketCount
	Number of packets read until time to yield (== max count)
MAKRti::DtClock	myBestEffortRetryTimer
	Mechanism to retry best effort transmits which block.
unsigned int	myBestEffortRetryAttempts
MAKRti::DtTime	myBestEffortTimeToRetry
MAKRti::DtList *	mySendQueue
	Exchange queues between federate and IO threads these lists are used in asynchronous mode.
MAKRti::DtList *	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.

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]`

References DtRIDParameters::forwarderGroupSize(), myReceiveLock, myRecvEvent, mySendEvent, and mySendLock.

Referenced by create().

DtAsyncConnectionMgr::~DtAsyncConnectionMgr ( ) [virtual]

References destroyPackets(), DtWaitEvent_AncillaryReliableConnStart, DtWaitEvent_BestEffortConn, DtWaitEvent_PrimaryReliableConn, DtWaitEvent_Send, isAsynchronous(), DtConnectionMgr::myCallbackMutex, myCallbackQueue, myReceivedBuffer, myReceiveLock, myReceiveQueue, mySendLock, mySendQueue, processCallbackMsg(), DtFedAmbCallbackMsg::removeCallback(), and DtConnectionMgr::removeFedAmbWrapper().

Member Function Documentation

void DtAsyncConnectionMgr::clearReceiveEvent ( ) [virtual]

Clear receive event Used by federate thread to clear signal.

References myRecvEvent.

Referenced by doWait(), and getPacket().

void DtAsyncConnectionMgr::clearSendEvent ( ) [virtual]

Clear send event Used by IO thread to clear signal.

References mySendEvent.

Referenced by workerThread().

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.

References DtWaitEvent_AncillaryReliableConnStart, DtWaitEvent_BestEffortConn, DtWaitEvent_PrimaryReliableConn, DtWaitEvent_Send, myAsynchronous, myMoreToFlush, myReceivedBuffer, myThreadActivated, myThreadHandle, myTimeToQuit, NULL, DtConnectionMgr::reliableConn(), and run().

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

Creator function.

Reimplemented from DtConnectionMgr.

References DtAsyncConnectionMgr().

Referenced by DtRtiAmbassadorBaseImplementor::setupConnection().

void DtAsyncConnectionMgr::destroyAncillaryRelConn ( DtRtiConnection * conn ) [protected, virtual]

Destroy the given ancillary reliable connection.

Reimplemented from DtConnectionMgr.

References myNextReliableConnToRead.

void DtAsyncConnectionMgr::destroyAncillaryRelConns ( ) [protected, virtual]

Destroy ancillary reliable connections.

Reimplemented from DtConnectionMgr.

References myNextReliableConnToRead.

void DtAsyncConnectionMgr::destroyPackets ( ) [protected, virtual]

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

The buffers and queues themselves are not destroyed

References isAsynchronous(), myCurrentSendPacket, myLockQueue, myPacketsToSendCount, myReceivedBuffer, myReceivedBufferCount, myReceivedBufferCurrent, myReceiveQueue, myRecvPacketBuffer, mySendPacketBuffer, and mySendQueue.

Referenced by disconnectFederation(), workerThread(), and ~DtAsyncConnectionMgr().

void DtAsyncConnectionMgr::destroyRelConn ( ) [protected, virtual]

Destroy reliable connection.

Reimplemented from DtConnectionMgr.

References myNextReliableConnToRead.

Referenced by receivePackets(), and sendPackets().

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.

References destroyPackets(), DtConnectionMgr::myCallbackMutex, myCurrentSendPacket, DtConnectionMgr::myFedMgr, DtConnectionMgr::myInService, DtConnectionMgr::myLockService, myMoreToFlush, myPacketsToSendCount, mySendQueue, myThreadActivated, myThreadHandle, myTimeToQuit, DtFederateMgr::posttick(), and sendPackets().

bool DtAsyncConnectionMgr::doWait ( MAKRti::DtTime period ) [protected, virtual]

Perform actual wait, called by wait function.

Reimplemented from DtConnectionMgr.

References clearReceiveEvent(), myRecvEvent, myTickWaitPeriod, DtConnectionMgr::params(), period, and signalSendEvent().

Referenced by wait().

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.

References myThreadActivated.

DtRtiMsg * DtAsyncConnectionMgr::getMessage ( ) [virtual]

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

Reimplemented from DtConnectionMgr.

References DtRtiMsg::cloneWithTransport(), getPacket(), DtQueuePacket::msg, and myThreadActivated.

DtQueuePacket * DtAsyncConnectionMgr::getPacket ( ) [protected, virtual]

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

References clearReceiveEvent(), myLockQueue, myReceivedBuffer, myReceivedBufferCount, myReceivedBufferCurrent, myReceiveLock, myReceiveQueue, and myThreadActivated.

Referenced by getMessage(), and processConnections().

DtNetReadStatus DtAsyncConnectionMgr::handleMsg ( DtRtiMsg & msg ) [virtual]

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

Reimplemented from DtConnectionMgr.

References DtNET_READ_SUCCESS, DtConnectionMgr::lockServiceCall(), myAsynchronousCallbacks, DtConnectionMgr::myFedMgr, DtFederateMgr::posttick(), and DtConnectionMgr::unlockServiceCall().

Referenced by DtRtiShmQNetMgr::processConnection(), processConnections(), processMsgQueueCallbacks(), and receivePacket().

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.

References DtFedAmbCallbackMsg::addCallback(), DtRIDParameters::DtAsynchronousCallbacks, DtRIDParameters::DtSynchronousMode, DtWaitEvent_AncillaryReliableConnStart, DtWaitEvent_BestEffortConn, DtWaitEvent_PrimaryReliableConn, DtWaitEvent_Send, DtRIDParameters::IOLockQueue, DtRIDParameters::maxIOQueue, myAsynchronous, myAsynchronousCallbacks, myBestEffortRetryTimer, DtConnectionMgr::myCallbackMutex, myCurrentSendPacket, myIoPeriod, myLockQueue, myMaxCount, myMaxQueue, myPacketsToSendCount, myReadPacketCount, myReceivedBuffer, myReceivedBufferCount, myReceivedBufferCurrent, myReceiveQueue, myRecvEvent, myRecvPacketBuffer, mySendEvent, mySendPacketBuffer, mySendQueue, myTickWaitPeriod, processCallbackMsg(), and DtRIDParameters::processingModel().

bool DtAsyncConnectionMgr::isAsynchronous ( ) [virtual]

return if in asynchronous mode

References myAsynchronous.

Referenced by destroyPackets(), and ~DtAsyncConnectionMgr().

void DtAsyncConnectionMgr::processCallbackMsg ( const DtFedAmbCallbackMsg & callbackMsg ) [protected, virtual]

Message callback.

References DtToWString(), DtFedAmbCallbackMsg::evokeCallback(), and DtConnectionMgr::fedAmb().

Referenced by init(), and ~DtAsyncConnectionMgr().

void DtAsyncConnectionMgr::processCallbackMsg	(	const DtFedAmbCallbackMsg &	callbackMsg,
		void *	usr
	)		`[static, protected]`

Message callback.

bool DtAsyncConnectionMgr::processConnections ( ) [virtual]

Check all the connections and process any incomming 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.

References getPacket(), handleMsg(), DtQueuePacket::msg, myAsynchronousConnectionIsOk, and myThreadActivated.

DtNetReadStatus DtAsyncConnectionMgr::processMsgQueueCallbacks	(	DtRtiMsg &	msg,
		unsigned int &	receivedPacketCount
	)		`[protected, virtual]`

Process message but queue any resulting federate ambassador callbacks.

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

References DtTestManager::addAsyncQueueMsgOnRecieve(), DtNET_READ_NO_PACKETS, DtNET_READ_SUCCESS, handleMsg(), DtConnectionMgr::lockServiceCall(), myCallbackQueue, DtConnectionMgr::myFedMgr, myLockQueue, myRecvPacketBuffer, DtFederateMgr::posttick(), DtFederateMgr::testMgr(), and DtConnectionMgr::unlockServiceCall().

Referenced by receivePacket().

void DtAsyncConnectionMgr::processPendingMsgMakeCallbacks ( ) [protected, virtual]

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

References DtConnectionMgr::lockServiceCall(), DtConnectionMgr::myFedMgr, myTimeToQuit, DtFederateMgr::posttick(), DtConnectionMgr::processPendingMsg(), DtConnectionMgr::setEvokingCallbacks(), and DtConnectionMgr::unlockServiceCall().

Referenced by workerThread().

void DtAsyncConnectionMgr::processPendingMsgQueueCallbacks ( ) [protected, virtual]

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

Any resulting callback messages are put on receive queue.

References DtConnectionMgr::lockServiceCall(), myCallbackQueue, DtConnectionMgr::myFedMgr, myReceiveLock, myReceiveQueue, myTimeToQuit, DtFederateMgr::posttick(), DtConnectionMgr::processPendingMsg(), signalReceiveEvent(), and DtConnectionMgr::unlockServiceCall().

Referenced by workerThread().

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

References DtConnectionMgr::myFedMgr, myIoPeriod, myMaxQueue, mySendLock, mySendQueue, myThreadActivated, DtTestManager::queueMsgForSend(), signalSendEvent(), and DtFederateMgr::testMgr().

Referenced by send(), and sendAndDelete().

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.

References DtConnectionMgr::readFileDescriptor().

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

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

References DtTestManager::addReceiveTimeStampIfNecessary(), DtRtiMsg::clone(), DtNET_READ_NO_PACKETS, handleMsg(), myAsynchronousCallbacks, myAsynchronousProcessMsg, myCallbackQueue, DtConnectionMgr::myFedMgr, myLockQueue, myReadPacketCount, myRecvPacketBuffer, DtRtiConnection::netRead(), processMsgQueueCallbacks(), DtConnectionMgr::processReadResult(), DtRtiMsg::setTransportType(), and DtFederateMgr::testMgr().

Referenced by receivePackets().

bool DtAsyncConnectionMgr::receivePackets ( ) [protected, virtual]

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?

References DtTestManager::addAsyncQueueMsgOnRecieve(), DtConnectionMgr::bestEffortConn(), DtConnectionMgr::checkRelConn(), destroyRelConn(), DtNET_READ_NO_PACKETS, DtRtiConnection::isOk(), DtQueuePacket::msg, myAsynchronousConnectionIsOk, DtConnectionMgr::myFedMgr, myLockQueue, myMoreToFlush, myNextReliableConnToRead, myReceiveLock, myReceiveQueue, myRecvPacketBuffer, DtConnectionMgr::myReliableConns, myTimeToQuit, receivePacket(), signalReceiveEvent(), and DtFederateMgr::testMgr().

Referenced by workerThread().

void * DtAsyncConnectionMgr::run ( void * usr ) [static]

thread start calls worker thread

References workerThread().

Referenced by connectFederation().

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

Send a message.

Reimplemented from DtConnectionMgr.

References DtRtiMsg::clone(), DtRtiMsg::fedExHandle(), DtFederateMgr::isTicking(), DtConnectionMgr::logicalTimeFactory(), DtConnectionMgr::myClock, DtConnectionMgr::myFedMgr, DtConnectionMgr::mySentMsgCount, myThreadActivated, DtConnectionMgr::params(), DtRtiMsg::print(), queueMsg(), setMessageTracking(), and DtFederateMgr::tickCount().

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.

References DtRIDParameters::destAddr(), DtRtiMsg::fedExHandle(), DtFederateMgr::isTicking(), DtConnectionMgr::logicalTimeFactory(), DtConnectionMgr::myClock, DtConnectionMgr::myFedMgr, DtConnectionMgr::mySentMsgCount, myThreadActivated, DtConnectionMgr::params(), DtRtiMsg::print(), queueMsg(), setMessageTracking(), and DtFederateMgr::tickCount().

bool DtAsyncConnectionMgr::sendPackets ( ) [protected, virtual]

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)?

References DtQueuePacket::addr, DtTestManager::addSendTimeStampMetaDataIfNecessary(), DtConnectionMgr::bestEffortConn(), DtRIDParameters::bestEffortSendRetryWaitUsec, DtConnectionMgr::checkRelConn(), destroyRelConn(), DtMetadataTypeMcastAddr, DtMetadataTypeTransport, DtRIDParameters::enableBestEffortSendRetry, DtRtiConnection::flush(), DtConnectionMgr::isMetadataEnabled(), DtRtiConnection::isOk(), DtQueuePacket::msg, myAsynchronousConnectionIsOk, myBestEffortRetryAttempts, myBestEffortRetryTimer, myBestEffortTimeToRetry, myCurrentSendPacket, DtConnectionMgr::myFedMgr, myLockQueue, myMoreToFlush, myPacketsToSendCount, mySendLock, mySendPacketBuffer, mySendQueue, DtConnectionMgr::params(), DtConnectionMgr::reliableConn(), DtRtiConnection::send(), DtRtiMtdTransportType::set(), DtRtiMtdMcastAddr::set(), DtFederateMgr::testMgr(), and DtRtiMsg::transportType().