#include <MasterQueue.h>

Public Member Functions
int	disconnect ()
void	drain ()
int	id ()
	MasterQueue (unsigned int ind)
int	occupancy () const
pid_t	pidOfLastReceive () const
pid_t	pidOfLastSend () const
int	post (MsgBuf &ptr)
unsigned long	rcv (MsgBuf &ptr)
unsigned long	rcvNonBlocking (MsgBuf &ptr)
int	status ()
	~MasterQueue ()
Private Attributes
int	occup_
int	pidOfLastReceive_
int	pidOfLastSend_
int	queue_id_
int	status_

Detailed Description

Definition at line 22 of file MasterQueue.h.

Constructor & Destructor Documentation

evf::MasterQueue::MasterQueue ( unsigned int ind ) [inline]

Definition at line 26 of file MasterQueue.h.

References drain(), QUEUE_ID, and queue_id_.

                                  : status_(0)
      {
        
        /* create or attach a public message queue, with read/write access to every user. */
        queue_id_ = msgget(QUEUE_ID+ind, IPC_CREAT | 0666); 
        if (queue_id_ == -1) {
          std::ostringstream ost;
          ost << "failed to get message queue:" 
              << strerror(errno);
          XCEPT_RAISE(evf::Exception, ost.str());
        }
        // it may be necessary to drain the queue here if it already exists !!! 
        drain();
      }

evf::MasterQueue::~MasterQueue ( ) [inline]

Definition at line 40 of file MasterQueue.h.

References queue_id_, and status_.

      {
        if(status_>0) msgctl(queue_id_,IPC_RMID,0);
      }

Member Function Documentation

int evf::MasterQueue::disconnect ( void ) [inline]

Definition at line 81 of file MasterQueue.h.

References queue_id_, runTheMatrix::ret, and status_.

      {
        int ret = msgctl(queue_id_,IPC_RMID,0);
        status_ = -1000;
        return ret;
      }

void evf::MasterQueue::drain ( ) [inline]

Definition at line 107 of file MasterQueue.h.

References gather_cfg::cout, runTheMatrix::msg, evf::MsgBuf::msize(), occup_, evf::MsgBuf::ptr_, queue_id_, and status().

Referenced by MasterQueue().

                {
      status();
      if(occup_>0)
        std::cout << "message queue id " << queue_id_ << " contains " << occup_ << "leftover messages, going to drain " 
                  << std::endl;
      //drain the queue before using it
      MsgBuf msg;
      while(occup_>0)
        {
          msgrcv(queue_id_, msg.ptr_, msg.msize()+1, 0, 0);
          status();
          std::cout << "drained one message, occupancy now " << occup_ << std::endl;
        }
    }

int evf::MasterQueue::id ( void ) [inline]

Definition at line 87 of file MasterQueue.h.

References queue_id_.

{return queue_id_;}

int evf::MasterQueue::occupancy ( ) const [inline]

Definition at line 106 of file MasterQueue.h.

References occup_.

{return occup_;}

pid_t evf::MasterQueue::pidOfLastReceive ( ) const [inline]

Definition at line 122 of file MasterQueue.h.

References pidOfLastReceive_.

{return pidOfLastReceive_;}

pid_t evf::MasterQueue::pidOfLastSend ( ) const [inline]

Definition at line 121 of file MasterQueue.h.

References pidOfLastSend_.

{return pidOfLastSend_;}

int evf::MasterQueue::post ( MsgBuf & ptr ) [inline]

Definition at line 45 of file MasterQueue.h.

References gather_cfg::cout, evf::MsgBuf::msize(), evf::MsgBuf::ptr_, and queue_id_.

Referenced by evf::Vulture::start(), and evf::Vulture::stop().

      {
        int rc;                  /* error code returned by system calls. */
        rc = msgsnd(queue_id_, ptr.ptr_, ptr.msize()+1,0);
        if(rc==-1)
          std::cout << "snd::Master failed to post message - error:"
                    << strerror(errno) << std::endl;
        //      delete ptr;
        return rc;
      }

unsigned long evf::MasterQueue::rcv ( MsgBuf & ptr ) [inline]

Definition at line 55 of file MasterQueue.h.

References MSGQ_MESSAGE_TYPE_RANGE, evf::MsgBuf::msize(), evf::MsgBuf::ptr_, and queue_id_.

      {
        unsigned long msg_type = ptr->mtype;
        int rc = msgrcv(queue_id_, ptr.ptr_, ptr.msize()+1, ptr->mtype, 0);
        if (rc == -1 && errno != ENOMSG)  
          {
            std::string serr = "rcv::Master failed to get message from queue - error:";
            serr += strerror(errno);
            XCEPT_RAISE(evf::Exception, serr);
          }
        else if(rc == -1 && errno == ENOMSG) return MSGQ_MESSAGE_TYPE_RANGE;
        return msg_type;
      }

unsigned long evf::MasterQueue::rcvNonBlocking ( MsgBuf & ptr ) [inline]

Definition at line 68 of file MasterQueue.h.

References MSGQ_MESSAGE_TYPE_RANGE, evf::MsgBuf::msize(), evf::MsgBuf::ptr_, and queue_id_.

Referenced by evf::Vulture::hasStarted(), and evf::Vulture::hasStopped().

      {
        unsigned long msg_type = ptr->mtype;
        int rc = msgrcv(queue_id_, ptr.ptr_, ptr.msize()+1, msg_type, IPC_NOWAIT);
        if (rc == -1 && errno != ENOMSG)  
          {
            std::string serr = "rcvnb::Master failed to get message from queue - error:";
            serr += strerror(errno);
            XCEPT_RAISE(evf::Exception, serr);
          }
        else if(rc == -1 && errno == ENOMSG) return MSGQ_MESSAGE_TYPE_RANGE;
        return msg_type;
      }

int evf::MasterQueue::status ( void ) [inline]

Definition at line 88 of file MasterQueue.h.

References occup_, pidOfLastReceive_, pidOfLastSend_, queue_id_, runTheMatrix::ret, and status_.

Referenced by drain().

      {
        char cbuf[sizeof(struct msqid_ds)];
        struct msqid_ds *buf= (struct msqid_ds*)cbuf;
        int ret = msgctl(queue_id_,IPC_STAT,buf);       
        if(ret!=0) status_ = -1; 
        else
          {
            status_ = 1;
            occup_ = buf->msg_qnum;
            pidOfLastSend_ = buf->msg_lspid;
            pidOfLastReceive_ = buf->msg_lrpid;
            //      std::cout << "queue " << buf->msg_qnum << " " 
            //        << buf->msg_lspid << " " 
            //        << buf->msg_lrpid << std::endl;
          }
        return status_;
      }