 |
MAK RTIspy API Documentation for HLA 1.3
|
|
MAK RTIspy API Documentation for HLA 1.3
|
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();
typedef std::list< DtStalledSubscriber * > DtRtiSubscribableMessageQueue::DtStalledSubscriberList [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.
References DtRtiSubMsgQ_theMaxSubscribers, myTimestampClock, theMaxStallTime, and theMaxSubscribers.
| DtRtiSubscribableMessageQueue::DtRtiSubscribableMessageQueue | ( | const DtRtiSubscribableMessageQueue & | other | ) | [private] |
Copy Ctor Not Implimented -- Class Not Duplicable.
| DtInetAddr DtRtiSubscribableMessageQueue::bucketDestinationAddr | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
Referenced by getRcvMsgTransportInfo().
| DtU32 DtRtiSubscribableMessageQueue::bucketHeaderSize | ( | ) | const [protected, virtual] |
report the size, in bytes, of the bucket header
| bool DtRtiSubscribableMessageQueue::bucketIsBestEffort | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
Referenced by getRcvMsgTransportInfo(), and isEmpty().
| bool DtRtiSubscribableMessageQueue::bucketIsFirst | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
Referenced by getRcvMsgSenderInfo(), and getRcvMsgTransportInfo().
| bool DtRtiSubscribableMessageQueue::bucketIsLast | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
| bool DtRtiSubscribableMessageQueue::bucketIsReliable | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
Referenced by getRcvMsgTransportInfo(), and isEmpty().
| DtU32 DtRtiSubscribableMessageQueue::bucketMsgSn | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
Referenced by getRcvMsgSenderInfo(), and handleStall().
| DtU32 DtRtiSubscribableMessageQueue::bucketOutstandingReadCount | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
| DtU32 DtRtiSubscribableMessageQueue::bucketPayloadLen | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
Referenced by isEmpty().
| DtU32 DtRtiSubscribableMessageQueue::bucketPriority | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
Referenced by getRcvMsgTransportInfo(), and isEmpty().
| DtU32 DtRtiSubscribableMessageQueue::bucketSenderId | ( | unsigned int | bucketNum | ) | const [protected, virtual] |
return the bucket header field values
Referenced by getRcvMsgSenderInfo(), handleStall(), and isEmpty().
| void DtRtiSubscribableMessageQueue::clearSyncVars | ( | ) | [virtual] |
This clears all synchronization variables and should only be called by the queue manager.
References DtCondVar__rti__::destroy(), myNewMessageCond, myNewMessageMutex, myQueueFullCond, and myQueueFullMutex.
| void DtRtiSubscribableMessageQueue::decrementOutstandingReadCount | ( | unsigned int | bucketNum | ) | [protected, virtual] |
References updateBackstop().
Referenced by handleStall(), isEmpty(), and unsubscribe().
| int DtRtiSubscribableMessageQueue::decrementSubscriberCount | ( | ) | [protected, virtual] |
Decrement the number of subscribers by one.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::numberOfSubscribers, queueHeader(), and subscriberCount().
Referenced by unsubscribe().
| void DtRtiSubscribableMessageQueue::getRcvMsgSenderInfo | ( | DtU32 * | senderId, |
| DtU32 * | msgSn = NULL |
||
| ) | [virtual] |
Returns the contents of the sender ID and sender seq.
num fields from the next available bucket header.
References bucketIsFirst(), bucketMsgSn(), bucketSenderId(), isSubscribed(), lockForRead(), numberOfBuckets(), queueFront(), and unlockForRead().
Referenced by DtRtiSmSocket::recv().
| void DtRtiSubscribableMessageQueue::getRcvMsgTransportInfo | ( | MAKRti::DtInetAddr * | destAddr, |
| DtU32 * | flags | ||
| ) | [virtual] |
Returns the destination address and flags field from the next available bucket header.
References bucketDestinationAddr(), bucketIsBestEffort(), bucketIsFirst(), bucketIsReliable(), bucketPriority(), DtSMQ_MSG_BESTEFFORT, DtSMQ_MSG_RELIABLE, isSubscribed(), lockForRead(), numberOfBuckets(), queueFront(), and unlockForRead().
Referenced by DtRtiSmSocket::getMsgTransportInfo().
| bool DtRtiSubscribableMessageQueue::handleStall | ( | unsigned int | bucketNum, |
| unsigned int | sub | ||
| ) | [protected, virtual] |
Record information about stalled subscriber.
Unsubscribe stalled subscriber if necessary
References DtCondVar__rti__::broadcast(), bucketMsgSn(), bucketSenderId(), decrementOutstandingReadCount(), DtRtiSubscribableMessageQueue::DtRtiQueueHeader::front, myQueueFullCond, myStalledSubs, myTimestampClock, numberOfBuckets(), queueHeader(), queueIsReliable(), theMaxStallTime, and unsubscribe().
Referenced by stalledSubscribers().
| int DtRtiSubscribableMessageQueue::incrementSubscriberCount | ( | ) | [protected, virtual] |
Increment the number of subscribers by one.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::numberOfSubscribers, queueHeader(), subscriberCount(), and theMaxSubscribers.
Referenced by resetQueue().
| void DtRtiSubscribableMessageQueue::initSyncVars | ( | ) | [virtual] |
Initialize condition variables and mutexes used to synchronize subscribers.
This should only be called after the queue is created by the queue manager.
References DtIpcCondVar__rti__::init(), DtIpcMutex__rti__::init(), myNewMessageCond, myNewMessageMutex, myQueueFullCond, and myQueueFullMutex.
Referenced by DtRtiSmSocket::initializeSmQueue().
| bool DtRtiSubscribableMessageQueue::isEmpty | ( | int * | nextSize, |
| DtU32 | flagsMask = DtSMQ_FETCH_DEFAULT |
||
| ) | [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.
References bucketIsBestEffort(), bucketIsReliable(), bucketPayloadLen(), bucketPriority(), bucketSenderId(), decrementOutstandingReadCount(), DtSMQ_FETCH_ALL, DtSMQ_FETCH_DEFAULT, DtSMQ_PRIO_MASK, DtSMQ_SKIP_BESTEFFORT, DtSMQ_SKIP_MINE, DtSMQ_SKIP_RELIABLE, numberOfBuckets(), queueBack(), queueFront(), and updateFront().
Referenced by DtRtiSmSocket::getMsgTransportInfo(), and unsubscribe().
| bool DtRtiSubscribableMessageQueue::isSubscribed | ( | DtU32 | subscriberId | ) | const [virtual] |
Test if subscribed to queue.
References queueHeader(), and DtRtiSubscribableMessageQueue::DtRtiQueueHeader::subscribed.
Referenced by getRcvMsgSenderInfo(), getRcvMsgTransportInfo(), and printDataToStream().
| void DtRtiSubscribableMessageQueue::lockForRead | ( | ) | [protected, virtual] |
Locks the queue for a read.
Referenced by getRcvMsgSenderInfo(), and getRcvMsgTransportInfo().
| unsigned int DtRtiSubscribableMessageQueue::maxSubscribers | ( | ) | const [protected, virtual] |
Report configured maximum subscriber count.
References theMaxSubscribers.
Referenced by printDataToStream(), resetQueue(), and stalledSubscribers().
| pthread_cond_t * DtRtiSubscribableMessageQueue::newMessageCond | ( | ) | [protected, virtual] |
References queueHeader().
Referenced by subscribe().
| pthread_mutex_t * DtRtiSubscribableMessageQueue::newMessageCondMutex | ( | ) | [protected, virtual] |
References queueHeader().
Referenced by subscribe().
| DtU32 DtRtiSubscribableMessageQueue::numberOfBuckets | ( | ) | const [protected, virtual] |
Returns the number of buckets in this queue.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::numberOfBuckets, and queueHeader().
Referenced by getRcvMsgSenderInfo(), getRcvMsgTransportInfo(), handleStall(), isEmpty(), printDataToStream(), resetQueue(), sendMessage(), unsubscribe(), updateBack(), updateBackstop(), updateFront(), and waitUntilFull().
| DtRtiSubscribableMessageQueue& DtRtiSubscribableMessageQueue::operator= | ( | const DtRtiSubscribableMessageQueue & | other | ) | [private] |
Assignment Operator Not Implimented -- Class Not Duplicable.
| pthread_mutex_t * DtRtiSubscribableMessageQueue::orcMutex | ( | ) | [protected, virtual] |
References queueHeader().
| DtU32 DtRtiSubscribableMessageQueue::payloadSize | ( | ) | const [protected, virtual] |
Returns the number of bytes available in an empty bucket payload.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::payloadSize, and queueHeader().
Referenced by resetQueue(), and sendMessage().
| unsigned int DtRtiSubscribableMessageQueue::poolSize | ( | DtU32 | numberOfBuckets, |
| 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
| std::ostream & DtRtiSubscribableMessageQueue::printDataToStream | ( | std::ostream & | str | ) | const [virtual] |
prints the state of the queue to the given stream
References DtRtiSubscribableMessageQueue::DtRtiBucketHeader::besteffort, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::first, isSubscribed(), DtRtiSubscribableMessageQueue::DtRtiBucketHeader::last, maxSubscribers(), DtRtiSubscribableMessageQueue::DtRtiBucketHeader::msgPriority, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::msgSn, numberOfBuckets(), queueBack(), queueFront(), DtRtiSubscribableMessageQueue::DtRtiBucketHeader::reliable, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::senderId, and DtRtiSubscribableMessageQueue::DtRtiBucketHeader::sizeInBytes.
| volatile DtU32 DtRtiSubscribableMessageQueue::queueBack | ( | ) | const [protected, virtual] |
Returns the back of this queue, which is the last bucket written to.
References queueHeader().
Referenced by isEmpty(), printDataToStream(), sendMessage(), and unsubscribe().
| volatile DtU32 DtRtiSubscribableMessageQueue::queueBackstop | ( | ) | const [protected, virtual] |
Returns the backstop of this queue, which is the last bucket all subscribers have read.
References queueHeader().
Referenced by sendMessage().
| volatile DtU32 DtRtiSubscribableMessageQueue::queueFront | ( | unsigned int | sub | ) | const [protected, virtual] |
Returns the index of the specified subscribers queue front, which is the last bucket the subscriber read from.
References queueHeader().
Referenced by getRcvMsgSenderInfo(), getRcvMsgTransportInfo(), isEmpty(), printDataToStream(), and unsubscribe().
| pthread_cond_t * DtRtiSubscribableMessageQueue::queueFullCond | ( | ) | [protected, virtual] |
References queueHeader().
Referenced by subscribe().
| pthread_mutex_t * DtRtiSubscribableMessageQueue::queueFullCondMutex | ( | ) | [protected, virtual] |
References queueHeader().
Referenced by subscribe().
| volatile DtRtiSubscribableMessageQueue::DtRtiQueueHeader * DtRtiSubscribableMessageQueue::queueHeader | ( | ) | const [protected] |
returns a pointer to the queue header, called from CTOR
Referenced by decrementSubscriberCount(), handleStall(), incrementSubscriberCount(), isSubscribed(), newMessageCond(), newMessageCondMutex(), numberOfBuckets(), orcMutex(), payloadSize(), queueBack(), queueBackstop(), queueFront(), queueFullCond(), queueFullCondMutex(), queueIsReliable(), queueState(), rdWrLock(), resetQueue(), sendMessage(), setEnqueueMode(), setState(), setSubscribed(), setSubscriberCount(), stalledSubscribers(), subscriberCount(), updateBack(), updateBackstop(), updateFront(), waitForMessages(), and waitUntilFull().
| DtU32 DtRtiSubscribableMessageQueue::queueHeaderSize | ( | ) | const [protected] |
report the size, in bytes, of the queue header
| bool DtRtiSubscribableMessageQueue::queueIsReliable | ( | ) | const [virtual] |
Test if queue operates in reliable or best effort delivery mode.
References queueHeader(), and DtRtiSubscribableMessageQueue::DtRtiQueueHeader::reliable.
Referenced by handleStall(), and DtRtiSmSocket::queueIsReliable().
| volatile DtU32 DtRtiSubscribableMessageQueue::queueState | ( | ) | const [protected, virtual] |
returns current state indicating whether shutdown in progress
References queueHeader(), and DtRtiSubscribableMessageQueue::DtRtiQueueHeader::state.
| pthread_rwlock_t * DtRtiSubscribableMessageQueue::rdWrLock | ( | ) | [protected, virtual] |
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.
References queueHeader().
| void DtRtiSubscribableMessageQueue::resetQueue | ( | DtU32 | numberOfBuckets, |
| 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.
References DtRtiSubscribableMessageQueue::DtRtiBucketHeader::besteffort, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::first, incrementSubscriberCount(), DtRtiSubscribableMessageQueue::DtRtiBucketHeader::ipv6, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::last, maxSubscribers(), DtRtiSubscribableMessageQueue::DtRtiBucketHeader::msgPriority, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::msgSn, numberOfBuckets(), DtRtiSubscribableMessageQueue::DtRtiQueueHeader::numberOfBuckets, DtRtiSubscribableMessageQueue::DtRtiQueueHeader::numberOfSubscribers, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::outstandingReadCount, payloadSize(), DtRtiSubscribableMessageQueue::DtRtiQueueHeader::payloadSize, queueHeader(), DtRtiSubscribableMessageQueue::DtRtiBucketHeader::reliable, DtRtiSubscribableMessageQueue::DtRtiQueueHeader::reliable, DtRtiSubscribableMessageQueue::DtRtiQueueHeader::reserved_1, DtRtiSubscribableMessageQueue::DtRtiBucketHeader::senderId, setState(), setSubscribed(), DtRtiSubscribableMessageQueue::DtRtiBucketHeader::sizeInBytes, DtRtiSubscribableMessageQueue::DtRtiQueueHeader::subscribed, updateBack(), updateBackstop(), and updateFront().
Referenced by DtRtiSmSocket::initializeSmQueue().
| bool DtRtiSubscribableMessageQueue::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. Return boolean: message was successfully enqueued?
Referenced by DtRtiSmSocket::sendMessage().
| bool DtRtiSubscribableMessageQueue::sendMessage | ( | const void * | message, |
| unsigned int | size, | ||
| unsigned int | flags, | ||
| const MAKRti::DtInetAddr & | destAddr = MAKRti::DtInetAddr::inaddrAny() |
||
| ) | [virtual] |
References DtCondVar__rti__::broadcast(), DtSMQ_MSG_BESTEFFORT, DtSMQ_MSG_RELIABLE, DtSMQ_PRIO_MASK, DtIpcMutex__rti__::lock(), myNewMessageCond, myNewMessageMutex, NULL, numberOfBuckets(), payloadSize(), DtRtiSubscribableMessageQueue::DtRtiQueueHeader::payloadSize, queueBack(), queueBackstop(), queueHeader(), setBestEffort(), setDestinationAddr(), setFirst(), setLast(), setMsgSn(), setOutstandingReadCount(), setPayloadLen(), setPriority(), setReliable(), setSenderId(), subscriberCount(), DtIpcMutex__rti__::unlock(), and updateBack().
| void DtRtiSubscribableMessageQueue::setBestEffort | ( | unsigned int | bucketNum, |
| DtU32 | val | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setDestinationAddr | ( | unsigned int | bucketNum, |
| const DtInetAddr & | dest | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setEnqueueMode | ( | DtSmqEnqueueMode | mode | ) | [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.
References DtSmqEnqueueReliableMode, queueHeader(), and DtRtiSubscribableMessageQueue::DtRtiQueueHeader::reliable.
Referenced by DtRtiSmSocket::initializeSmQueue().
| void DtRtiSubscribableMessageQueue::setFirst | ( | unsigned int | bucketNum, |
| DtU32 | val | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setLast | ( | unsigned int | bucketNum, |
| DtU32 | val | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setMsgSn | ( | unsigned int | bucketNum, |
| DtU32 | seqnum | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setOutstandingReadCount | ( | unsigned int | bucketNum, |
| DtU32 | val | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setPayloadLen | ( | unsigned int | bucketNum, |
| DtU32 | len | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setPriority | ( | unsigned int | bucketNum, |
| DtU32 | priority | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setReliable | ( | unsigned int | bucketNum, |
| DtU32 | val | ||
| ) | [protected, virtual] |
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setSenderId | ( | unsigned int | bucketNum, |
| DtU32 | sid | ||
| ) | [protected, virtual] |
change the value of bucket header fields
Referenced by sendMessage().
| void DtRtiSubscribableMessageQueue::setState | ( | DtU32 | val | ) | [protected, virtual] |
set the queue state
References queueHeader(), and DtRtiSubscribableMessageQueue::DtRtiQueueHeader::state.
Referenced by resetQueue().
| void DtRtiSubscribableMessageQueue::setSubscribed | ( | unsigned int | sub, |
| unsigned int | val | ||
| ) | [protected, virtual] |
Set or clear the subscriber's subscribed bit in the queue header.
References queueHeader(), and DtRtiSubscribableMessageQueue::DtRtiQueueHeader::subscribed.
Referenced by resetQueue(), and unsubscribe().
| int DtRtiSubscribableMessageQueue::setSubscriberCount | ( | DtU32 | val | ) | [protected, virtual] |
Set the number of subscribers.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::numberOfSubscribers, queueHeader(), subscriberCount(), and theMaxSubscribers.
| bool DtRtiSubscribableMessageQueue::stalledSubscribers | ( | unsigned int | bucketNum, |
| unsigned int | numStalled | ||
| ) | [protected, virtual] |
Determine which subscribers are stalled and handle them.
References handleStall(), maxSubscribers(), and queueHeader().
| int DtRtiSubscribableMessageQueue::subscribe | ( | MAKRti::DtU32 | subscriberId = DtSM_INVALID_SUBSCRIBER_ID | ) | [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
References myNewMessageCond, myNewMessageMutex, myQueueFullCond, myQueueFullMutex, newMessageCond(), newMessageCondMutex(), queueFullCond(), and queueFullCondMutex().
Referenced by DtRtiSmSocket::initializeSmQueue().
| int DtRtiSubscribableMessageQueue::subscriberCount | ( | ) | const [virtual] |
Return the count of subscribers to the queue.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::numberOfSubscribers, and queueHeader().
Referenced by decrementSubscriberCount(), incrementSubscriberCount(), sendMessage(), and setSubscriberCount().
| void DtRtiSubscribableMessageQueue::unlockForRead | ( | ) | [protected, virtual] |
Unlocks the queue for a read.
Referenced by getRcvMsgSenderInfo(), and getRcvMsgTransportInfo().
| void DtRtiSubscribableMessageQueue::unsubscribe | ( | DtU32 | id | ) | [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.
References decrementOutstandingReadCount(), decrementSubscriberCount(), DtSMQ_FETCH_ALL, isEmpty(), NULL, numberOfBuckets(), queueBack(), queueFront(), and setSubscribed().
Referenced by handleStall(), and unsubscribeSelf().
| void DtRtiSubscribableMessageQueue::unsubscribeSelf | ( | ) | [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.
References DtSM_INVALID_SUBSCRIBER_ID, and unsubscribe().
Referenced by ~DtRtiSubscribableMessageQueue().
| DtU32 DtRtiSubscribableMessageQueue::updateBack | ( | unsigned int | val | ) | [protected, virtual] |
Sets a new value for the queue's back.
Returns the next index, wrapping if necessary.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::back, numberOfBuckets(), and queueHeader().
Referenced by resetQueue(), and sendMessage().
| DtU32 DtRtiSubscribableMessageQueue::updateBackstop | ( | unsigned int | val | ) | [protected, virtual] |
Sets a new value for the queue's backstop.
Returns the next index, wrapping if necessary.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::backstop, numberOfBuckets(), and queueHeader().
Referenced by decrementOutstandingReadCount(), and resetQueue().
| DtU32 DtRtiSubscribableMessageQueue::updateFront | ( | unsigned int | sub, |
| unsigned int | val | ||
| ) | [protected, virtual] |
Sets a new value for the specified subscriber's queue front.
Returns the next index, wrapping if necessary.
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::front, numberOfBuckets(), and queueHeader().
Referenced by isEmpty(), and resetQueue().
| int DtRtiSubscribableMessageQueue::waitForMessages | ( | MAKRti::DtTime | waitTime = 0 | ) | [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
References DtIpcMutex__rti__::lock(), myNewMessageCond, myNewMessageMutex, queueHeader(), DtIpcMutex__rti__::unlock(), and DtCondVar__rti__::wait().
Referenced by DtRtiSmSocket::waitForMessages().
| int DtRtiSubscribableMessageQueue::waitUntilFull | ( | MAKRti::DtTime | waitTime = 0 | ) | [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
References DtRtiSubscribableMessageQueue::DtRtiQueueHeader::back, DtIpcMutex__rti__::lock(), myQueueFullCond, myQueueFullMutex, numberOfBuckets(), queueHeader(), DtIpcMutex__rti__::unlock(), and DtCondVar__rti__::wait().
Referenced by DtRtiSmSocket::waitUntilFull().
Referenced by clearSyncVars(), initSyncVars(), sendMessage(), subscribe(), waitForMessages(), and ~DtRtiSubscribableMessageQueue().
Referenced by clearSyncVars(), initSyncVars(), sendMessage(), subscribe(), waitForMessages(), and ~DtRtiSubscribableMessageQueue().
Referenced by clearSyncVars(), handleStall(), initSyncVars(), subscribe(), waitUntilFull(), and ~DtRtiSubscribableMessageQueue().
Referenced by clearSyncVars(), initSyncVars(), subscribe(), waitUntilFull(), and ~DtRtiSubscribableMessageQueue().
!WIN32
Referenced by handleStall(), and ~DtRtiSubscribableMessageQueue().
MAKRti::DtClock DtRtiSubscribableMessageQueue::myTimestampClock [protected] |
Referenced by DtRtiSubscribableMessageQueue(), and handleStall().
unsigned int DtRtiSubscribableMessageQueue::theMaxStallTime [protected] |
The maximum time to permit a subscriber to be stalled, in seconds.
static const unsigned int theMaxStallTime = 5;
Referenced by DtRtiSubscribableMessageQueue(), and handleStall().
unsigned int DtRtiSubscribableMessageQueue::theMaxSubscribers [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;
Referenced by DtRtiSubscribableMessageQueue(), incrementSubscriberCount(), maxSubscribers(), and setSubscriberCount().
