smSubscribableMessageQueue.h

Go to the documentation of this file.

/*********************************************************************
 ** Copyright (c) 2005 MaK Technologies, Inc.
 ** All rights reserved.
 *********************************************************************/
/*********************************************************************
 ** $RCSfile: smSubscribableMessageQueue.h,v $ $Revision: 1.14 $ $State: Exp $
 *********************************************************************/
#ifndef smSubscribableMessageQueue_H_
#define smSubscribableMessageQueue_H_


#include "rtiMsConfig.h"
#include "vlShmSubscribableMessageQueue__rti__.h"
#include "vlIpcCondVar__rti__.h"
#include "vlIpcMutex__rti__.h"
#include <vlutil/vlTime.h>
#include <vlutil/vlUtil.h>
#include <vlutil/vlInetAddr.h>
#include <list>

using namespace MAKRti;

#define DtSMQ_PRIO_0          0x00  
#define DtSMQ_PRIO_1          0x01  
#define DtSMQ_PRIO_2          0x02  
#define DtSMQ_PRIO_3          0x03  
#define DtSMQ_PRIO_4          0x04  
#define DtSMQ_PRIO_5          0x05  
#define DtSMQ_PRIO_6          0x06  
#define DtSMQ_PRIO_7          0x07  
#define DtSMQ_PRIO_MASK       0x07  
#define DtSMQ_MSG_RELIABLE    0x10  
#define DtSMQ_MSG_BESTEFFORT  0x20  
//#define DtSMQ_MSG_TBD         0x40   
#define DtSMQ_SKIP_RELIABLE   0x10  
#define DtSMQ_SKIP_BESTEFFORT 0x20  

enum DtSmqEnqueueMode
{
   DtSmqEnqueueBestEffortMode = 0,
   DtSmqEnqueueReliableMode   = 1,

   DtSmqEnqueueInvalidMode        
};

#define DtRtiSubMsgQ_theMaxSubscribers 16384

class DT_DLL_LRC DtRtiSubscribableMessageQueue: public MAKRti::DtSmSubscribableMessageQueue__rti__
{

public:

   DtRtiSubscribableMessageQueue(const MAKRti::DtString &name);
   virtual ~DtRtiSubscribableMessageQueue();

private:

   DtRtiSubscribableMessageQueue(const DtRtiSubscribableMessageQueue &other);

   DtRtiSubscribableMessageQueue& operator=(const DtRtiSubscribableMessageQueue &other);

public:

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

   virtual void unsubscribeSelf();

   virtual void unsubscribe(DtU32 id);

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

   virtual bool queueIsReliable() const;

   virtual bool isSubscribed(DtU32 subscriberId) const;

   virtual int subscriberCount() const;

   virtual bool sendMessage(const void *message, unsigned int size);
   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);

   virtual void getRcvMsgSenderInfo(DtU32* senderId, DtU32* msgSn = NULL);

   virtual int waitForMessages(MAKRti::DtTime waitTime = 0);

   virtual int waitUntilFull(MAKRti::DtTime waitTime = 0);

   virtual void resetQueue(DtU32 numberOfBuckets, DtU32 payloadSize);

   virtual void setEnqueueMode(DtSmqEnqueueMode mode);

   virtual void initSyncVars();

   virtual void clearSyncVars();

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

public:

   static unsigned int poolSize(DtU32 numberOfBuckets, DtU32 payloadSize);

protected:
   struct DtRtiQueueHeader;
   struct DtRtiBucketHeader;
   class DtStalledSubscriber;
   typedef std::list< DtStalledSubscriber * > DtStalledSubscriberList;

   virtual unsigned int maxSubscribers() const;

   volatile DtRtiQueueHeader *queueHeader() const;

   DtU32 queueHeaderSize() const;

   virtual DtU32 numberOfBuckets() const;

   virtual DtU32 payloadSize() const;

   virtual volatile DtU32 queueState() const;

   virtual volatile DtU32 queueFront(unsigned int sub) const;

   virtual volatile DtU32 queueBack() const;

   virtual volatile DtU32 queueBackstop() const;

   virtual DtU32 updateFront(unsigned int sub, unsigned int val);

   virtual DtU32 updateBack(unsigned int val);

   virtual DtU32 updateBackstop(unsigned int val);

   virtual void setSubscribed(unsigned int sub, unsigned int val);

   virtual int setSubscriberCount(DtU32 val);

   virtual int incrementSubscriberCount();

   virtual int decrementSubscriberCount();

   virtual void setState(DtU32 val);

   virtual void lockForRead();
   virtual void unlockForRead();

   virtual bool stalledSubscribers(unsigned int bucketNum, unsigned int numStalled);

   virtual bool handleStall(unsigned int bucketNum, unsigned int sub);

   virtual DtU32 bucketHeaderSize() const;

   virtual DtU32 bucketSenderId(unsigned int bucketNum) const;
   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);
   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);

#ifndef WIN32
   virtual pthread_rwlock_t * rdWrLock();
   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();
#endif 

protected:

   DtStalledSubscriberList   myStalledSubs;
   MAKRti::DtClock                   myTimestampClock;
   DtIpcCondVar__rti__           *  myNewMessageCond;
   DtIpcMutex__rti__             *  myNewMessageMutex;
   DtIpcCondVar__rti__           *  myQueueFullCond;
   DtIpcMutex__rti__             *  myQueueFullMutex;

   unsigned int theMaxSubscribers;

   unsigned int theMaxStallTime;

protected:

   struct DtRtiBucketHeader
   {
      DtU32 senderId;         
      DtU32 msgSn;            
      DtU32 outstandingReadCount;  
      DtU32 sizeInBytes : 24; 
      DtU32 msgPriority : 3;  
      DtU32 reliable    : 1;  
      DtU32 besteffort  : 1;  
      DtU32 ipv6        : 1;  
      DtU32 last        : 1;  
      DtU32 first       : 1;  
      DtU128 destAddr;        
   };

   struct DtRtiQueueHeader
   {
      DtU32 state;             
      DtU32 payloadSize : 24;  
      DtU32 reliable    :  1;
      DtU32 reserved_1  :  7;
      DtU32 numberOfBuckets;
      DtU32 numberOfSubscribers;
      DtU32 front[DtRtiSubMsgQ_theMaxSubscribers];
      DtU32 subscribed[DtRtiSubMsgQ_theMaxSubscribers];
      DtU32 back;
      DtU32 backstop;
#ifndef WIN32
      pthread_rwlock_t  rwlock;
      pthread_cond_t    newMessage;   
      pthread_mutex_t   newMessageMutex;
      pthread_cond_t    queueFull;
      pthread_mutex_t   queueFullMutex;
      pthread_mutex_t   ptOrcMutex;   
#endif 
   };

   class DtStalledSubscriber
   {
   public:
      DtStalledSubscriber(DtU32 bucket, DtU32 id, DtU32 sender, DtU32 seq, MAKRti::DtTime tstamp) :
         bucketIndex(bucket), stalledId(id), senderId(sender), senderSn(seq), posted(tstamp)
         {};
      ~DtStalledSubscriber(){};
      DtU32    bucketIndex; 
      DtU32    stalledId;  
      DtU32    senderId;   
      DtU32    senderSn;   
      MAKRti::DtTime   posted;     
   };

};

#endif