Implements a generic shared memory message queue for multiple subscribers. More...

Inheritance diagram for DtSmSubscribableMessageQueue__rti__:

Collaboration diagram for DtSmSubscribableMessageQueue__rti__:

Classes
struct	DtBucketHeader
	DtBucketHeader is a structure found at the front of each message to keep track of the number of bytes used, fragmented message indications, the subscriber that enqueued the message, and the sender's sequence number. More...

struct	DtQueueHeader
	DtQueueHeader exists at the front of the message queue and contains useful information about the state of the queue. More...

Public Member Functions
	DtSmSubscribableMessageQueue__rti__ (const MAKRti::DtString &name)
	Creates an instance of a Shared Memory Queue with the given name IMPORTANT: memory with the given name must have already been alocated somewhere, This class will only attach to it. More...

virtual	~DtSmSubscribableMessageQueue__rti__ ()

virtual int	subscribe (MAKRti::DtU32 subscriberId=DtSM_INVALID_SUBSCRIBER_ID)
	Subscribe to the queue and set up subscriber-specific fields in the queue header. More...

virtual void	unsubscribe (MAKRti::DtU32 id)
	Unsubscribes subscriber id from the queue. More...

virtual bool	isSubscribed (MAKRti::DtU32 subscriberId) const
	Test if subscriberId is subscribed. More...

virtual MAKRti::DtU32	subscriberId () const
	Return the assigned subscriber ID. More...

virtual int	subscriberCount () const
	Return the count of subscribers to the queue. More...

virtual bool	isEmpty ()
	Returns true if the subscriber's queue has no unread messages. More...

virtual bool	isEmpty (int *nextSize, MAKRti::DtU32 flagsMask=DtSMQ_FETCH_DEFAULT)

virtual void	setDefaultReadBehavior (MAKRti::DtU32 flagsMask)
	Set the default behavior for detecting messages in the queue to skip or return messages with certain characteristics. More...

virtual bool	sendMessage (const void *message, unsigned int size)
	Sends a packet at address message of size bytes to the queue. More...

virtual void *	message (unsigned int *outSize=0)
	Returns a pointer to the message copied from shared memory or null if there are no messages to read. More...

virtual void *	message (void recvBuffer, unsigned int outSize)

virtual void	resetQueue (MAKRti::DtU32 numberOfBuckets, MAKRti::DtU32 payloadSize)
	Emptys the queue. More...

virtual bool	sm_shutdown (bool ruthless=false)
	Shutdown the queue. Does not free shared memory or other resources. More...

virtual volatile MAKRti::DtU32	queueState () const
	returns current state indicating whether shutdown in progress More...

virtual void	initSyncVars ()
	Initializes Shared Memory locks. More...

virtual void	clearSyncVars ()
	This clears all synchronization variables and should only be called by the queue manager. More...

virtual std::ostream &	printDataToStream (std::ostream &str) const
	prints the state of the queue to the given stream More...

Public Member Functions inherited from DtMessageQueue__rti__
	DtMessageQueue__rti__ ()

virtual	~DtMessageQueue__rti__ ()

Static Public Member Functions
static unsigned int	poolSize (MAKRti::DtU32 numberOfBuckets, MAKRti::DtU32 payloadSize)
	Used to compute the size in bytes, required for a DtSmSubscribableMessageQueue instance using numberOfBuckets with a payload size of payloadSize. More...

Protected Member Functions
virtual int	assignSubscriberId ()
	Requests next available subscriber ID. More...

virtual unsigned int	maxSubscribers () const
	Report configured maximum subscriber count. More...

volatile DtQueueHeader *	queueHeader () const
	returns a pointer to the queue header. More...

virtual MAKRti::DtU32	queueHeaderSize () const
	report the size, in bytes, of the queue header More...

virtual MAKRti::DtU32	numberOfBuckets () const
	Returns the number of buckets in this queue. More...

virtual MAKRti::DtU32	payloadSize () const
	Returns the number of bytes available in an empty bucket payload. More...

virtual volatile MAKRti::DtU32	queueFront (unsigned int sub) const
	Returns the index of the specified subscribers queue front, which. More...

virtual volatile MAKRti::DtU32	queueBack () const
	Returns the back of this queue, which is the last bucket written to. More...

virtual volatile MAKRti::DtU32	queueBackstop () const
	Returns the backstop of this queue, which is the last bucket all subscribers have read. More...

virtual MAKRti::DtU32	updateFront (unsigned int sub, unsigned int val)
	Sets a new value for the specified subscriber's queue front. More...

virtual MAKRti::DtU32	updateBack (unsigned int val)
	Sets a new value for the queue's back. More...

virtual MAKRti::DtU32	updateBackstop (unsigned int val)
	Sets a new value for the queue's backstop. More...

virtual void	setSubscribed (unsigned int sub, unsigned int val)
	Set or clear the subscriber's subscribed bit in the queue header. More...

virtual int	setSubscriberCount (MAKRti::DtU32 val)
	Set the number of current subscribers. More...

virtual int	incrementSubscriberCount ()
	Increment the number of subscribers by one. More...

virtual int	decrementSubscriberCount ()
	Decrement the number of subscribers by one. More...

virtual void	setState (MAKRti::DtU32 val)
	Set the queue state. More...

virtual volatile char *	payloadInBucket (unsigned int bucketNum) const
	returns a pointer to the payload in the bucket. More...

virtual volatile void *	bucketHeader (unsigned int bucketNum) const
	returns a pointer to the message header at index More...

virtual MAKRti::DtU32	bucketHeaderSize () const
	report the size, in bytes, of the bucket header More...

virtual MAKRti::DtU32	bucketSenderId (unsigned int bucketNum) const
	return the bucket header field values More...

virtual MAKRti::DtU32	bucketMsgSn (unsigned int bucketNum) const

virtual MAKRti::DtU32	bucketOutstandingReadCount (unsigned int bucketNum) const

virtual MAKRti::DtU32	bucketPayloadLen (unsigned int bucketNum) const

virtual bool	bucketIsFirst (unsigned int bucketNum) const

virtual bool	bucketIsLast (unsigned int bucketNum) const

virtual void	setSenderId (unsigned int bucketNum, MAKRti::DtU32 sid)
	change the value of bucket header fields More...

virtual void	setMsgSn (unsigned int bucketNum, MAKRti::DtU32 seqnum)

virtual void	setOutstandingReadCount (unsigned int bucketNum, MAKRti::DtU32 val)

virtual void	decrementOutstandingReadCount (unsigned int bucketNum)

virtual void	setPayloadLen (unsigned int bucketNum, MAKRti::DtU32 len)

virtual void	setFirst (unsigned int bucketNum, MAKRti::DtU32 val)

virtual void	setLast (unsigned int bucketNum, MAKRti::DtU32 val)

virtual void	lockForRead ()
	Locks the queue for a read. More...

virtual void	unlockForRead ()
	Unlocks the queue for a read. More...

virtual void	lockForWrite ()
	Locks the queue for a write. More...

virtual void	unlockForWrite ()
	Unlocks the queue for writes. More...

virtual bool	stalledSubscribers (unsigned int bucketNum, unsigned int numStalled)
	Record information about stalled subscriber. More...

virtual bool	checkForStalls ()
	Determine if there are any subscribers stalled on the bucket immediately following the 'backstop' bucket. More...

virtual pthread_rwlock_t *	rdWrLock ()
	For non-Win32 implementations the rwlock structure in the queue header must be accessed via inheritable methods in order for derived classes to be able to define a different queue header structure. More...

virtual pthread_mutex_t *	orcMutex ()

Protected Attributes
DtSharedMemoryPoolClient__rti__ *	myShmPool
	!WIN32 More...

MAKRti::DtString	myShmName

MAKRti::DtU32	mySubscriberId

MAKRti::DtU32	myMessagesSentCnt

DtIpcRdWrLock__rti__ *	myRdWrLock

DtIpcMutex__rti__ *	myOrcMutex

std::vector< char >	myReadBuffer

MAKRti::DtU32	myDefaultReadMask

unsigned int	theMaxSubscribers
	The maximum number of subscribers allowed in this queue. More...

Private Member Functions
	DtSmSubscribableMessageQueue__rti__ (const DtSmSubscribableMessageQueue__rti__ &other)
	Copy Ctor Not Implemented – Class Not Duplicable. More...

DtSmSubscribableMessageQueue__rti__ &	operator= (const DtSmSubscribableMessageQueue__rti__ &other)
	Assignment Operator Not Implimented – Class Not Duplicable. More...

Detailed Description

Implements a generic shared memory message queue for multiple subscribers.

This is a fixed length queue comprised of fixed length message containers. It is implemented as a circular list that will not fill up. Any subscriber which has a queue full of unread messages will be unsubscribed. Subscribers are given a 32 bit ID and register their interest in messages merely by creating an instance of this class. A subscriber can unsubscribe themselves at any time. Before an instance of this class can be created there must already be a shared memory pool of the correct size open. The correct size is computed using DtSubscribableMessageQueue::poolSize(). Before a queue can be used one subscriber (and only one subscriber) must call queue->reset(...); queue->initSyncVars();

Constructor & Destructor Documentation

DtSmSubscribableMessageQueue__rti__::DtSmSubscribableMessageQueue__rti__ ( const MAKRti::DtString & name )

Creates an instance of a Shared Memory Queue with the given name IMPORTANT: memory with the given name must have already been alocated somewhere, This class will only attach to it.

The memory must be at least as big as ::poolSize(), this queue can't check.

virtual DtSmSubscribableMessageQueue__rti__::~DtSmSubscribableMessageQueue__rti__ ( )

virtual

DtSmSubscribableMessageQueue__rti__::DtSmSubscribableMessageQueue__rti__ ( const DtSmSubscribableMessageQueue__rti__ & other )

private

Copy Ctor Not Implemented – Class Not Duplicable.

Member Function Documentation

virtual int DtSmSubscribableMessageQueue__rti__::assignSubscriberId ( )

protectedvirtual

Requests next available subscriber ID.

virtual volatile void* DtSmSubscribableMessageQueue__rti__::bucketHeader ( unsigned int bucketNum ) const

protectedvirtual

returns a pointer to the message header at index

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::bucketHeaderSize ( ) const

protectedvirtual

report the size, in bytes, of the bucket header

virtual bool DtSmSubscribableMessageQueue__rti__::bucketIsFirst ( unsigned int bucketNum ) const

protectedvirtual

virtual bool DtSmSubscribableMessageQueue__rti__::bucketIsLast ( unsigned int bucketNum ) const

protectedvirtual

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::bucketMsgSn ( unsigned int bucketNum ) const

protectedvirtual

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::bucketOutstandingReadCount ( unsigned int bucketNum ) const

protectedvirtual

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::bucketPayloadLen ( unsigned int bucketNum ) const

protectedvirtual

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::bucketSenderId ( unsigned int bucketNum ) const

protectedvirtual

return the bucket header field values

virtual bool DtSmSubscribableMessageQueue__rti__::checkForStalls ( )

protectedvirtual

Determine if there are any subscribers stalled on the bucket immediately following the 'backstop' bucket.

virtual void DtSmSubscribableMessageQueue__rti__::clearSyncVars ( )

virtual

This clears all synchronization variables and should only be called by the queue manager.

virtual void DtSmSubscribableMessageQueue__rti__::decrementOutstandingReadCount ( unsigned int bucketNum )

protectedvirtual

virtual int DtSmSubscribableMessageQueue__rti__::decrementSubscriberCount ( )

protectedvirtual

Decrement the number of subscribers by one.

virtual int DtSmSubscribableMessageQueue__rti__::incrementSubscriberCount ( )

protectedvirtual

Increment the number of subscribers by one.

virtual void DtSmSubscribableMessageQueue__rti__::initSyncVars ( )

virtual

Initializes Shared Memory locks.

This should only be called after the queue is created, and only by one party, the queue manager.

virtual bool DtSmSubscribableMessageQueue__rti__::isEmpty ( )

virtual

Returns true if the subscriber's queue has no unread messages.

Returns true if the queue has messages no messages in it, false if there is a message in the queue. If nextSize is not NULL the size of the next message is stored in that location. The flagsMask argument can control which messages the subscriber will detect in the queue (overrides default read behavior)

Implements DtMessageQueue__rti__.

virtual bool DtSmSubscribableMessageQueue__rti__::isEmpty	(	int *	nextSize,
		MAKRti::DtU32	flagsMask = `DtSMQ_FETCH_DEFAULT`
	)

virtual

virtual bool DtSmSubscribableMessageQueue__rti__::isSubscribed ( MAKRti::DtU32 subscriberId ) const

virtual

Test if subscriberId is subscribed.

virtual void DtSmSubscribableMessageQueue__rti__::lockForRead ( )

protectedvirtual

Locks the queue for a read.

virtual void DtSmSubscribableMessageQueue__rti__::lockForWrite ( )

protectedvirtual

Locks the queue for a write.

virtual unsigned int DtSmSubscribableMessageQueue__rti__::maxSubscribers ( ) const

protectedvirtual

Report configured maximum subscriber count.

virtual void* DtSmSubscribableMessageQueue__rti__::message ( unsigned int * outSize = 0 )

virtual

Returns a pointer to the message copied from shared memory or null if there are no messages to read.

If recvBuffer pointer is supplied the message will be copied to that location, rather than the intermediate local buffer. If outsize is passed then the size of the message is returned.

Implements DtMessageQueue__rti__.

virtual void* DtSmSubscribableMessageQueue__rti__::message	(	void *	recvBuffer,
		unsigned int *	outSize
	)

virtual

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::numberOfBuckets ( ) const

protectedvirtual

Returns the number of buckets in this queue.

DtSmSubscribableMessageQueue__rti__& DtSmSubscribableMessageQueue__rti__::operator= ( const DtSmSubscribableMessageQueue__rti__ & other )

private

Assignment Operator Not Implimented – Class Not Duplicable.

virtual pthread_mutex_t* DtSmSubscribableMessageQueue__rti__::orcMutex ( )

protectedvirtual

virtual volatile char* DtSmSubscribableMessageQueue__rti__::payloadInBucket ( unsigned int bucketNum ) const

protectedvirtual

returns a pointer to the payload in the bucket.

Where a bucket is composed of the bucketHeader and the bucketPayload.

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::payloadSize ( ) const

protectedvirtual

Returns the number of bytes available in an empty bucket payload.

static unsigned int DtSmSubscribableMessageQueue__rti__::poolSize	(	MAKRti::DtU32	numberOfBuckets,
		MAKRti::DtU32	payloadSize
	)

static

Used to compute the size in bytes, required for a DtSmSubscribableMessageQueue instance using numberOfBuckets with a payload size of payloadSize.

Each bucket is the size of the payload + the size of a bucketHeader

virtual std::ostream& DtSmSubscribableMessageQueue__rti__::printDataToStream ( std::ostream & str ) const

virtual

prints the state of the queue to the given stream

Implements DtMessageQueue__rti__.

virtual volatile MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::queueBack ( ) const

protectedvirtual

Returns the back of this queue, which is the last bucket written to.

virtual volatile MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::queueBackstop ( ) const

protectedvirtual

Returns the backstop of this queue, which is the last bucket all subscribers have read.

virtual volatile MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::queueFront ( unsigned int sub ) const

protectedvirtual

Returns the index of the specified subscribers queue front, which.

is the last bucket the subscriber read from.

volatile DtQueueHeader* DtSmSubscribableMessageQueue__rti__::queueHeader ( ) const

protected

returns a pointer to the queue header.

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::queueHeaderSize ( ) const

protectedvirtual

report the size, in bytes, of the queue header

virtual volatile MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::queueState ( ) const

virtual

returns current state indicating whether shutdown in progress

virtual pthread_rwlock_t* DtSmSubscribableMessageQueue__rti__::rdWrLock ( )

protectedvirtual

For non-Win32 implementations the rwlock structure in the queue header must be accessed via inheritable methods in order for derived classes to be able to define a different queue header structure.

virtual void DtSmSubscribableMessageQueue__rti__::resetQueue	(	MAKRti::DtU32	numberOfBuckets,
		MAKRti::DtU32	payloadSize
	)

virtual

Emptys the queue.

This should only be called by the queue manager, as all unread messages are lost and all subscribers are unsubscribed.

virtual bool DtSmSubscribableMessageQueue__rti__::sendMessage	(	const void *	message,
		unsigned int	size
	)

virtual

Sends a packet at address message of size bytes to the queue.

If the queue is full this message will knock the oldest message off the queue.

Implements DtMessageQueue__rti__.

virtual void DtSmSubscribableMessageQueue__rti__::setDefaultReadBehavior ( MAKRti::DtU32 flagsMask )

virtual

Set the default behavior for detecting messages in the queue to skip or return messages with certain characteristics.

virtual void DtSmSubscribableMessageQueue__rti__::setFirst	(	unsigned int	bucketNum,
		MAKRti::DtU32	val
	)

protectedvirtual

virtual void DtSmSubscribableMessageQueue__rti__::setLast	(	unsigned int	bucketNum,
		MAKRti::DtU32	val
	)

protectedvirtual

virtual void DtSmSubscribableMessageQueue__rti__::setMsgSn	(	unsigned int	bucketNum,
		MAKRti::DtU32	seqnum
	)

protectedvirtual

virtual void DtSmSubscribableMessageQueue__rti__::setOutstandingReadCount	(	unsigned int	bucketNum,
		MAKRti::DtU32	val
	)

protectedvirtual

virtual void DtSmSubscribableMessageQueue__rti__::setPayloadLen	(	unsigned int	bucketNum,
		MAKRti::DtU32	len
	)

protectedvirtual

virtual void DtSmSubscribableMessageQueue__rti__::setSenderId	(	unsigned int	bucketNum,
		MAKRti::DtU32	sid
	)

protectedvirtual

change the value of bucket header fields

virtual void DtSmSubscribableMessageQueue__rti__::setState ( MAKRti::DtU32 val )

protectedvirtual

Set the queue state.

virtual void DtSmSubscribableMessageQueue__rti__::setSubscribed	(	unsigned int	sub,
		unsigned int	val
	)

protectedvirtual

Set or clear the subscriber's subscribed bit in the queue header.

virtual int DtSmSubscribableMessageQueue__rti__::setSubscriberCount ( MAKRti::DtU32 val )

protectedvirtual

Set the number of current subscribers.

virtual bool DtSmSubscribableMessageQueue__rti__::sm_shutdown ( bool ruthless = false )

virtual

Shutdown the queue. Does not free shared memory or other resources.

virtual bool DtSmSubscribableMessageQueue__rti__::stalledSubscribers	(	unsigned int	bucketNum,
		unsigned int	numStalled
	)

protectedvirtual

Record information about stalled subscriber.

Unsubscribe stalled subscribers if necessary

virtual int DtSmSubscribableMessageQueue__rti__::subscribe ( MAKRti::DtU32 subscriberId = DtSM_INVALID_SUBSCRIBER_ID )

virtual

Subscribe to the queue and set up subscriber-specific fields in the queue header.

Return 0 on failure, 1 on success

virtual int DtSmSubscribableMessageQueue__rti__::subscriberCount ( ) const

virtual

Return the count of subscribers to the queue.

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::subscriberId ( ) const

virtual

Return the assigned subscriber ID.

virtual void DtSmSubscribableMessageQueue__rti__::unlockForRead ( )

protectedvirtual

Unlocks the queue for a read.

virtual void DtSmSubscribableMessageQueue__rti__::unlockForWrite ( )

protectedvirtual

Unlocks the queue for writes.

virtual void DtSmSubscribableMessageQueue__rti__::unsubscribe ( MAKRti::DtU32 id )

virtual

Unsubscribes subscriber id from the queue.

This is called by the DTOR and should only be called by users of this class with caution. Unsubscribing will clear subscriber ID. (except the queue manager's ID, which is always reserved as 0) The queue manager may unsubscribe other subscribers without clearing that subscriber objects subscriber ID, so should be done only in extreme cases. Once unsubscribed, use subscribe() to get a new subscriber ID.

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::updateBack ( unsigned int val )

protectedvirtual

Sets a new value for the queue's back.

Returns the next index, wrapping if necessary.

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::updateBackstop ( unsigned int val )

protectedvirtual

Sets a new value for the queue's backstop.

Returns the next index, wrapping if necessary.

virtual MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::updateFront	(	unsigned int	sub,
		unsigned int	val
	)

protectedvirtual

Sets a new value for the specified subscriber's queue front.

Returns the next index, wrapping if necessary.

Member Data Documentation

MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::myDefaultReadMask

protected

MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::myMessagesSentCnt

protected

DtIpcMutex__rti__* DtSmSubscribableMessageQueue__rti__::myOrcMutex

protected

DtIpcRdWrLock__rti__* DtSmSubscribableMessageQueue__rti__::myRdWrLock

protected

std::vector<char> DtSmSubscribableMessageQueue__rti__::myReadBuffer

protected

MAKRti::DtString DtSmSubscribableMessageQueue__rti__::myShmName

protected

DtSharedMemoryPoolClient__rti__* DtSmSubscribableMessageQueue__rti__::myShmPool

protected

!WIN32

MAKRti::DtU32 DtSmSubscribableMessageQueue__rti__::mySubscriberId

protected

unsigned int DtSmSubscribableMessageQueue__rti__::theMaxSubscribers

protected

The maximum number of subscribers allowed in this queue.

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

vlShmSubscribableMessageQueue__rti__.h

Classes

Public Member Functions

Static Public Member Functions

Protected Member Functions

Protected Attributes

Private Member Functions

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation