vlShmSubscribableMessageQueue__rti__.h

Go to the documentation of this file.

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


#include "rtiMsConfig.h"
#include "vlShmMessageQueue__rti__.h"
#include "vlShmPool__rti__.h"
#include <vlutil/vlTime.h>
#include "vlIpcRdWrLock__rti__.h"
#include "vlIpcMutex__rti__.h"
#include <vlutil/vlNetTypes.h>
#include <vector>
#include <limits.h>

#ifndef WIN32
#include <pthread.h>
#endif 

#ifdef DtUSE_UTILITIES_NAMESPACE
namespace DtUSE_UTILITIES_NAMESPACE
{
#endif

#define DtSm_QUEUE_STATE_UNINITIALIZED 0x0
#define DtSm_QUEUE_STATE_INITIALIZED   0xFAD2FADE
#define DtSm_QUEUE_STATE_SHUTDOWN      0xDEADBEEF
#define DtSM_INVALID_SUBSCRIBER_ID     UINT_MAX

#define DtSMQ_FETCH_DEFAULT   0x00  
#define DtSMQ_SKIP_MINE       0x40  
#define DtSMQ_FETCH_ALL       0x80  

#define DtSmSubMsgQ_theMaxSubscribers 255

//
class DT_DLL_RTIUTIL DtSmSubscribableMessageQueue__rti__: public DtMessageQueue__rti__
{

public:

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

private:

   DtSmSubscribableMessageQueue__rti__(const DtSmSubscribableMessageQueue__rti__ &other);

   DtSmSubscribableMessageQueue__rti__& operator=(const DtSmSubscribableMessageQueue__rti__ &other);

public:

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

   virtual void unsubscribe(MAKRti::DtU32 id);

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

   virtual MAKRti::DtU32 subscriberId() const;

   virtual int subscriberCount() const;

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

   //
   virtual void setDefaultReadBehavior(MAKRti::DtU32 flagsMask);

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

   virtual void* message(unsigned int *outSize = 0);
   virtual void* message(void *recvBuffer, unsigned int *outSize);

   //
   virtual void resetQueue(MAKRti::DtU32 numberOfBuckets, MAKRti::DtU32 payloadSize);

   virtual bool sm_shutdown( bool ruthless = false );

   virtual volatile MAKRti::DtU32 queueState() const;

   //
   virtual void initSyncVars();

   virtual void clearSyncVars();

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

public:

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

protected:

   struct DtQueueHeader;
   struct DtBucketHeader;

   virtual int assignSubscriberId();

   virtual unsigned int maxSubscribers() const;

   volatile DtQueueHeader *queueHeader() const;

   virtual MAKRti::DtU32 queueHeaderSize() const;

   virtual MAKRti::DtU32 numberOfBuckets() const;

   virtual MAKRti::DtU32 payloadSize() const;

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

   virtual volatile MAKRti::DtU32 queueBack() const;

   virtual volatile MAKRti::DtU32 queueBackstop() const;

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

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

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

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

   virtual int setSubscriberCount(MAKRti::DtU32 val);

   virtual int incrementSubscriberCount();

   virtual int decrementSubscriberCount();

   virtual void  setState(MAKRti::DtU32 val);

   virtual volatile char* payloadInBucket(unsigned int bucketNum) const;

   virtual volatile void * bucketHeader(unsigned int bucketNum) const;

   virtual MAKRti::DtU32 bucketHeaderSize() const;

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

   virtual void lockForWrite();

   virtual void unlockForWrite();

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

   virtual bool checkForStalls();

#ifndef WIN32
   virtual pthread_rwlock_t * rdWrLock();
   virtual pthread_mutex_t  * orcMutex();
#endif 

protected:

   DtSharedMemoryPoolClient__rti__ *myShmPool;
   MAKRti::DtString                  myShmName;
   MAKRti::DtU32                     mySubscriberId;
   MAKRti::DtU32                     myMessagesSentCnt;
   DtIpcRdWrLock__rti__   *  myRdWrLock;
   DtIpcMutex__rti__      *  myOrcMutex;
   std::vector<char>         myReadBuffer;
   MAKRti::DtU32                     myDefaultReadMask;

//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 = 255;
   unsigned int theMaxSubscribers;

protected:

   struct DtBucketHeader
   {
      MAKRti::DtU32 senderId;       
      MAKRti::DtU32 msgSn;            
      MAKRti::DtU32 outstandingReadCount;  
      MAKRti::DtU32 sizeInBytes : 24;  
      MAKRti::DtU32 reserved_1   : 6;  
      MAKRti::DtU32 last         : 1;  
      MAKRti::DtU32 first        : 1;  
   };

   struct DtQueueHeader
   {
      MAKRti::DtU32 state;         
      MAKRti::DtU32 payloadSize;
      MAKRti::DtU32 numberOfBuckets;
      MAKRti::DtU32 numberOfSubscribers;
      MAKRti::DtU32 front[DtSmSubMsgQ_theMaxSubscribers];
      MAKRti::DtU32 subscribed[DtSmSubMsgQ_theMaxSubscribers];
      MAKRti::DtU32 back;
      MAKRti::DtU32 backstop;
#ifndef WIN32
      pthread_rwlock_t  rwlock;
      pthread_mutex_t   ptOrcMutex;   
#endif 
   };
};

#ifdef DtUSE_UTILITIES_NAMESPACE
} 
#endif

#endif