 |
MAK RTIspy API Documentation for HLA 1516
|
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Properties Friends Macros Groups Pages
|
MAK RTIspy API Documentation for HLA 1516
|
Implements a message queue for multiple subscribers. More...
Inheritance diagram for DtRtiSubscribableMessageQueue: Collaboration diagram for DtRtiSubscribableMessageQueue:Classes | |
| struct | DtRtiBucketHeader |
| 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 | DtRtiQueueHeader |
| DtQueueHeader exists at the front of the message queue and contains useful information about the state of the queue. More... | |
| class | DtStalledSubscriber |
| DtStalledSubscriber keeps track of which queue subscribers have not read a bucket as well as information about the message in the unread bucket for later comparison. More... | |
Public Member Functions | |
| DtRtiSubscribableMessageQueue (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. | |
| virtual | ~DtRtiSubscribableMessageQueue () |
| virtual int | subscribe (MAKRti::DtU32 subscriberId=DtSM_INVALID_SUBSCRIBER_ID) |
| Set up and subscribe to the queue. | |
| virtual void | unsubscribeSelf () |
| Unsubscribes self from the queue. | |
| virtual void | unsubscribe (DtU32 id) |
| Unsubscribes subscriber id from the queue. | |
| virtual bool | isEmpty (int *nextSize, DtU32 flagsMask=DtSMQ_FETCH_DEFAULT) |
| Returns true if the queue has messages no messages in it, false if there is a message in the queue. | |
| virtual bool | queueIsReliable () const |
| Test if queue operates in reliable or best effort delivery mode. | |
| virtual bool | isSubscribed (DtU32 subscriberId) const |
| Test if subscribed to queue. | |
| virtual int | subscriberCount () const |
| Return the count of subscribers to the queue. | |
| virtual bool | sendMessage (const void *message, unsigned int size) |
| Sends a packet at address message of size bytes to the queue. | |
| virtual bool | sendMessage (const void *message, unsigned int size, unsigned int flags, const MAKRti::DtInetAddr &destAddr=MAKRti::DtInetAddr::inaddrAny()) |
| virtual void | getRcvMsgTransportInfo (MAKRti::DtInetAddr *destAddr, DtU32 *flags) |
| Returns the destination address and flags field from the next available bucket header. | |
| virtual void | getRcvMsgSenderInfo (DtU32 *senderId, DtU32 *msgSn=NULL) |
| Returns the contents of the sender ID and sender seq. | |
| virtual int | waitForMessages (MAKRti::DtTime waitTime=0) |
| Wait (block) for the specified time until a message can be read from the queue. | |
| virtual int | waitUntilFull (MAKRti::DtTime waitTime=0) |
| Wait (block) for the specified time until the queue is full. | |
| virtual void | resetQueue (DtU32 numberOfBuckets, DtU32 payloadSize) |
| Emptys the queue. | |
| virtual void | setEnqueueMode (DtSmqEnqueueMode mode) |
| Set the queue to operate in reliable or best effort delivery mode. | |
| virtual void | initSyncVars () |
| Initialize condition variables and mutexes used to synchronize subscribers. | |
| virtual void | clearSyncVars () |
| This clears all synchronization variables and should only be called by the queue manager. | |
| virtual std::ostream & | printDataToStream (std::ostream &str) const |
| prints the state of the queue to the given stream | |
Static Public Member Functions | |
| static unsigned int | poolSize (DtU32 numberOfBuckets, DtU32 payloadSize) |
| Used to compute the size in bytes, required for a DtSmSubscribableMessageQueue instance using numberOfBuckets with a payload size of payloadSize. | |
Protected Types | |
| typedef std::list < DtStalledSubscriber * > | DtStalledSubscriberList |
Protected Member Functions | |
| virtual unsigned int | maxSubscribers () const |
| Report configured maximum subscriber count. | |
| volatile DtRtiQueueHeader * | queueHeader () const |
| returns a pointer to the queue header, called from CTOR | |
| DtU32 | queueHeaderSize () const |
| report the size, in bytes, of the queue header | |
| virtual DtU32 | numberOfBuckets () const |
| Returns the number of buckets in this queue. | |
| virtual DtU32 | payloadSize () const |
| Returns the number of bytes available in an empty bucket payload. | |
| virtual volatile DtU32 | queueState () const |
| returns current state indicating whether shutdown in progress | |
| virtual volatile DtU32 | queueFront (unsigned int sub) const |
| Returns the index of the specified subscribers queue front, which is the last bucket the subscriber read from. | |
| virtual volatile DtU32 | queueBack () const |
| Returns the back of this queue, which is the last bucket written to. | |
| virtual volatile DtU32 | queueBackstop () const |
| Returns the backstop of this queue, which is the last bucket all subscribers have read. | |
| virtual DtU32 | updateFront (unsigned int sub, unsigned int val) |
| Sets a new value for the specified subscriber's queue front. | |
| virtual DtU32 | updateBack (unsigned int val) |
| Sets a new value for the queue's back. | |
| virtual DtU32 | updateBackstop (unsigned int val) |
| Sets a new value for the queue's backstop. | |
| virtual void | setSubscribed (unsigned int sub, unsigned int val) |
| Set or clear the subscriber's subscribed bit in the queue header. | |
| virtual int | setSubscriberCount (DtU32 val) |
| Set the number of subscribers. | |
| virtual int | incrementSubscriberCount () |
| Increment the number of subscribers by one. | |
| virtual int | decrementSubscriberCount () |
| Decrement the number of subscribers by one. | |
| virtual void | setState (DtU32 val) |
| set the queue state | |
| virtual void | lockForRead () |
| Locks the queue for a read. | |
| virtual void | unlockForRead () |
| Unlocks the queue for a read. | |
| virtual bool | stalledSubscribers (unsigned int bucketNum, unsigned int numStalled) |
| Determine which subscribers are stalled and handle them. | |
| virtual bool | handleStall (unsigned int bucketNum, unsigned int sub) |
| Record information about stalled subscriber. | |
| virtual DtU32 | bucketHeaderSize () const |
| report the size, in bytes, of the bucket header | |
| virtual DtU32 | bucketSenderId (unsigned int bucketNum) const |
| return the bucket header field values | |
| virtual DtU32 | bucketMsgSn (unsigned int bucketNum) const |
| virtual DtU32 | bucketOutstandingReadCount (unsigned int bucketNum) const |
| virtual DtInetAddr | bucketDestinationAddr (unsigned int bucketNum) const |
| virtual DtU32 | bucketPayloadLen (unsigned int bucketNum) const |
| virtual DtU32 | bucketPriority (unsigned int bucketNum) const |
| virtual bool | bucketIsReliable (unsigned int bucketNum) const |
| virtual bool | bucketIsBestEffort (unsigned int bucketNum) const |
| virtual bool | bucketIsFirst (unsigned int bucketNum) const |
| virtual bool | bucketIsLast (unsigned int bucketNum) const |
| virtual void | setSenderId (unsigned int bucketNum, DtU32 sid) |
| change the value of bucket header fields | |
| virtual void | setMsgSn (unsigned int bucketNum, DtU32 seqnum) |
| virtual void | setOutstandingReadCount (unsigned int bucketNum, DtU32 val) |
| virtual void | decrementOutstandingReadCount (unsigned int bucketNum) |
| virtual void | setDestinationAddr (unsigned int bucketNum, const DtInetAddr &dest) |
| virtual void | setPayloadLen (unsigned int bucketNum, DtU32 len) |
| virtual void | setPriority (unsigned int bucketNum, DtU32 priority) |
| virtual void | setReliable (unsigned int bucketNum, DtU32 val) |
| virtual void | setBestEffort (unsigned int bucketNum, DtU32 val) |
| virtual void | setFirst (unsigned int bucketNum, DtU32 val) |
| virtual void | setLast (unsigned int bucketNum, DtU32 val) |
| virtual pthread_rwlock_t * | rdWrLock () |
| For non-Win32 implementations the pthread structures 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 pthread_mutex_t * | orcMutex () |
| virtual pthread_cond_t * | newMessageCond () |
| virtual pthread_mutex_t * | newMessageCondMutex () |
| virtual pthread_cond_t * | queueFullCond () |
| virtual pthread_mutex_t * | queueFullCondMutex () |
Protected Attributes | |
| DtStalledSubscriberList | myStalledSubs |
| !WIN32 | |
| MAKRti::DtClock | myTimestampClock |
| DtIpcCondVar__rti__ * | myNewMessageCond |
| DtIpcMutex__rti__ * | myNewMessageMutex |
| DtIpcCondVar__rti__ * | myQueueFullCond |
| DtIpcMutex__rti__ * | myQueueFullMutex |
| unsigned int | theMaxSubscribers |
| The maximum number of subscribers allowed in this queue. | |
| unsigned int | theMaxStallTime |
| The maximum time to permit a subscriber to be stalled, in seconds. | |
Private Member Functions | |
| DtRtiSubscribableMessageQueue (const DtRtiSubscribableMessageQueue &other) | |
| Copy Ctor Not Implimented – Class Not Duplicable. | |
| DtRtiSubscribableMessageQueue & | operator= (const DtRtiSubscribableMessageQueue &other) |
| Assignment Operator Not Implimented – Class Not Duplicable. | |
Implements a 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 dealt with in one of two ways, depending upon whether per-subscriber asynchronous read threads are employed. If async read threads are used the queue will never fill up because the async read threads will drain the queue on behalf of the subscriber process that forked the async read thread. Otherwise the list will be temporarily closed to new messages and subscribers will create private pending queues. The queue manager will monitor the stalled subscribers and unsubscribe them after a configurable period of inactivity. Subscribers register their interest in messages merely by creating an instance of this class. Upon subscribing they are assigned a unique 32 bit ID. The exception is the queue manager, the process that creates and initializes the queue, which assigns itself ID 0. 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 DtRtiSubscribableMessageQueue::poolSize(). Before a queue can be used the queue manager (ID 0) must call queue->reset(...); queue->initSyncVars();
|
protected |
| DtRtiSubscribableMessageQueue::DtRtiSubscribableMessageQueue | ( | 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 |
|
private |
Copy Ctor Not Implimented – Class Not Duplicable.
|
protectedvirtual |
|
protectedvirtual |
report the size, in bytes, of the bucket header
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
return the bucket header field values
|
virtual |
This clears all synchronization variables and should only be called by the queue manager.
|
protectedvirtual |
|
protectedvirtual |
Decrement the number of subscribers by one.
|
virtual |
Returns the contents of the sender ID and sender seq.
num fields from the next available bucket header.
|
virtual |
Returns the destination address and flags field from the next available bucket header.
|
protectedvirtual |
Record information about stalled subscriber.
Unsubscribe stalled subscriber if necessary
|
protectedvirtual |
Increment the number of subscribers by one.
|
virtual |
Initialize condition variables and mutexes used to synchronize subscribers.
This should only be called after the queue is created by the queue manager.
|
virtual |
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. messages not matching the priority specifications of flagsMask are skipped.
|
virtual |
Test if subscribed to queue.
|
protectedvirtual |
Locks the queue for a read.
|
protectedvirtual |
Report configured maximum subscriber count.
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
Returns the number of buckets in this queue.
|
private |
Assignment Operator Not Implimented – Class Not Duplicable.
|
protectedvirtual |
|
protectedvirtual |
Returns the number of bytes available in an empty bucket payload.
|
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 |
prints the state of the queue to the given stream
|
protectedvirtual |
Returns the back of this queue, which is the last bucket written to.
|
protectedvirtual |
Returns the backstop of this queue, which is the last bucket all subscribers have read.
|
protectedvirtual |
Returns the index of the specified subscribers queue front, which is the last bucket the subscriber read from.
|
protectedvirtual |
|
protectedvirtual |
|
protected |
returns a pointer to the queue header, called from CTOR
|
protected |
report the size, in bytes, of the queue header
|
virtual |
Test if queue operates in reliable or best effort delivery mode.
|
protectedvirtual |
returns current state indicating whether shutdown in progress
|
protectedvirtual |
For non-Win32 implementations the pthread structures 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 |
Emptys the queue.
This should only be called by the queue manager, as all unread messages are lost and all subscribers are unsubscribed.
|
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. Return boolean: message was successfully enqueued?
|
virtual |
|
protectedvirtual |
|
protectedvirtual |
|
virtual |
Set the queue to operate in reliable or best effort delivery mode.
This should only be called after the queue is created by the queue manager.
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
|
protectedvirtual |
change the value of bucket header fields
|
protectedvirtual |
set the queue state
|
protectedvirtual |
Set or clear the subscriber's subscribed bit in the queue header.
|
protectedvirtual |
Set the number of subscribers.
|
protectedvirtual |
Determine which subscribers are stalled and handle them.
|
virtual |
Set up and subscribe to the queue.
Subscribe to the queue and set up subscriber-specific fields in the queue header. Return 0 on failure, 1 on success
|
virtual |
Return the count of subscribers to the queue.
|
protectedvirtual |
Unlocks the queue for a read.
|
virtual |
Unsubscribes subscriber id from the queue.
This is called by the DTOR or to clean up a stalled subscriber. It should only be called by users of this class with caution. 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 |
Unsubscribes self 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. 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.
|
protectedvirtual |
Sets a new value for the queue's back.
Returns the next index, wrapping if necessary.
|
protectedvirtual |
Sets a new value for the queue's backstop.
Returns the next index, wrapping if necessary.
|
protectedvirtual |
Sets a new value for the specified subscriber's queue front.
Returns the next index, wrapping if necessary.
|
virtual |
Wait (block) for the specified time until a message can be read from the queue.
Return status: 0 if messages are ready, > 0 if wait expired, or < 0 if error
|
virtual |
Wait (block) for the specified time until the queue is full.
Return status: 0 if queue is full, > 0 if wait expired, or < 0 if error
|
protected |
|
protected |
|
protected |
|
protected |
|
protected |
!WIN32
|
protected |
|
protected |
The maximum time to permit a subscriber to be stalled, in seconds.
static const unsigned int theMaxStallTime = 5;
|
protected |
The maximum number of subscribers allowed in this queue.
TO DO – DAA workaround for MSVC6 bug ID #241569 A bug in MSVC++6 prevents initializing a static const member within a class declaration. It must be initialized outside the class. The bug is fixed in MSVC7, so when support for MSVC6 is dropped we can do away with this kludgy workaround static const unsigned int theMaxSubscribers = 16384;
