broadcastQueue.h

Go to the documentation of this file.

#ifndef _DLR_THREAD_BROADCASTQUEUE_H_
#define _DLR_THREAD_BROADCASTQUEUE_H_

#include <map>
#include <boost/thread.hpp>
#include <dlrThread/clientID.h>
#include <dlrThread/exception.h>


namespace dlr {

  namespace thread {

    template <class Type>
    class BroadcastQueue
    {
    public:

      typedef ClientID<Type> ClientID;


      BroadcastQueue(size_t maximumLength, size_t numberOfClients=2);


      BroadcastQueue(const BroadcastQueue<Type>& other);


      virtual
      ~BroadcastQueue();


      BroadcastQueue<Type>&
      operator=(const BroadcastQueue<Type>& other);


      void
      catchUp(ClientID& clientID);
      

      void
      copyBack(const ClientID& clientID, Type& target);


      bool
      copyBack(const ClientID& clientID, Type& target, double timeout);


      size_t
      getMaximumLength() {return m_size;}
      

      void
      lockBack(const ClientID& clientID);


      bool
      lockBack(const ClientID& clientID, double timeout);
      

      inline void
      popBack(ClientID& clientID);
      

      void
      pushFront(const ClientID& clientID, const Type& element);


      void
      registerClient(ClientID& clientID,
                     const std::string& clientName="anonymous");


      void
      synchronize(ClientID& clientID);


      bool
      synchronize(ClientID& clientID, double timeout);


      void
      synchronizeQuorum(ClientID& clientID, size_t quorumSize);


      bool
      synchronizeQuorum(ClientID& clientID, size_t quorumSize, double timeout);
      
      
      void
      unlockBack(const ClientID& clientID);
      
    private:

      void
      copyOther(const BroadcastQueue<Type>& other);

      
      inline void
      declareReadLock(const ClientID& clientID,
                      bool& overflowFlag,
                      size_t& headElementNumber);
      

      void 
      handleSynchronize(ClientID& clientID,
                        size_t targetNumber,
                        boost::timed_mutex::scoped_lock& syncLock);
      

      bool
      handleSynchronize(ClientID& clientID,
                        size_t targetNumber,
                        boost::xtime xtimeout,
                        boost::timed_mutex::scoped_timed_lock& syncLock);


      void
      releaseResources();
      
      
      size_t* m_countPtr;
      size_t* m_headIndexPtr;
      size_t m_size;
      
      std::map<size_t, std::string>* m_clientNameMapPtr;
      Type* m_dataArray;
      size_t* m_maxNumberOfClientsPtr;
      size_t* m_numberOfClientsPtr;
      size_t* m_readLockCountArray;
      size_t** m_readLockingFlagsArray;
      size_t* m_syncClientCountPtr;
      size_t* m_syncElementCountPtr;
      size_t* m_syncIndexPtr;
      size_t* m_quorumSizePtr;

      boost::condition* m_headChangedConditionPtr;
      boost::timed_mutex* m_headIndexMutexPtr;
      // // This code left around in case we try to go back to
      // // individual mutexes for data elements.
      // boost::timed_mutex* m_mutexArray;
      boost::timed_mutex* m_numberOfClientsMutexPtr;

      boost::condition* m_syncConditionPtr;
      boost::timed_mutex* m_syncMutexPtr;

      // We don't use the pre-written ReferenceCount class becuase
      // it's not thread-safe.
      size_t* m_referenceCountPtr;
    };

  } // namespace thread

} // namespace dlr


/* =============       Implementation follows        ============= */

#include <algorithm>
#include <iostream>
#include <sstream>
#include <dlrThread/private.h>


namespace dlr {

  namespace thread {

    /* ===========    Implementation of BroadcastQueue      =========== */

    template <class Type>
    BroadcastQueue<Type>::
    BroadcastQueue(size_t maximumLength, size_t numberOfClients)
      : m_countPtr(new size_t),
        m_headIndexPtr(new size_t),
        m_size(maximumLength),
        m_clientNameMapPtr(new std::map<size_t, std::string>),
        m_dataArray(new Type[maximumLength]),
        m_maxNumberOfClientsPtr(new size_t),
        m_numberOfClientsPtr(new size_t),
        m_readLockCountArray(new size_t[maximumLength]),
        m_readLockingFlagsArray(new size_t*[maximumLength]),
        m_syncClientCountPtr(new size_t),
        m_syncElementCountPtr(new size_t),
        m_syncIndexPtr(new size_t),
        m_quorumSizePtr(new size_t),
        m_headChangedConditionPtr(new boost::condition),
        m_headIndexMutexPtr(new boost::timed_mutex),
        // // This code left around in case we try to go back to
        // // individual mutexes for data elements.
        // m_mutexArray(new boost::timed_mutex[maximumLength]),
        m_numberOfClientsMutexPtr(new boost::timed_mutex),
        m_syncConditionPtr(new boost::condition),
        m_syncMutexPtr(new boost::timed_mutex),
        m_referenceCountPtr(new size_t)
    {
      if(numberOfClients == 0) {
        DLR_THROW(ValueException, "BroadcastQueue::BroadcastQueue()",
                  "Argument numberOfClients must be greater than zero.");
      }

      // No elements have been added yet.
      *m_countPtr = 0;
      *m_headIndexPtr = 0;

      // Set client capacity as instructed.
      *m_maxNumberOfClientsPtr = numberOfClients;

      // But indicate that no clients have registered yet.
      *m_numberOfClientsPtr = 0;

      // For each element in the queue.
      for(size_t index0 = 0; index0 < m_size; ++index0) {
        // Allocate storage for the expected number of clients.
        m_readLockingFlagsArray[index0] =
          new size_t[*m_maxNumberOfClientsPtr];
        size_t* beginIterator = m_readLockingFlagsArray[index0];
        size_t* endIterator = beginIterator + (*m_maxNumberOfClientsPtr);
        std::fill(beginIterator, endIterator, 0);

        // Indicate that no client is locking this element.
        m_readLockCountArray[index0] = 0;
      }

      // Initially, no clients are waiting for sync events.
      *m_syncClientCountPtr = 0;
      *m_syncElementCountPtr = 0;
      *m_syncIndexPtr = 0;
      *m_quorumSizePtr = 0;
      *m_referenceCountPtr = 1;
    }


    // The copy constructor does a shallow copy.
    template <class Type>
    BroadcastQueue<Type>::
    BroadcastQueue(const BroadcastQueue<Type>& other)
      : m_countPtr(0),
        m_headIndexPtr(0),
        m_size(0),
        m_clientNameMapPtr(0),
        m_dataArray(0),
        m_maxNumberOfClientsPtr(0),
        m_numberOfClientsPtr(0),
        m_readLockCountArray(0),
        m_readLockingFlagsArray(0),
        m_syncClientCountPtr(0),
        m_syncElementCountPtr(0),
        m_syncIndexPtr(0),
        m_quorumSizePtr(0),
        m_headChangedConditionPtr(0),
        m_headIndexMutexPtr(0),
        m_numberOfClientsMutexPtr(0),
        m_syncConditionPtr(0),
        m_syncMutexPtr(0),
        m_referenceCountPtr(0)
    {
      this->copyOther(other);
    }

    
    // Destroys the BroadcastQueue instance and releases any resources.
    template <class Type>
    BroadcastQueue<Type>::
    ~BroadcastQueue()
    {
      this->releaseResources();
    }


    // The assignment operator does a shallow copy.
    template <class Type>
    BroadcastQueue<Type>&
    BroadcastQueue<Type>::
    operator=(const BroadcastQueue<Type>& other)
    {
      if(&other != this) {
        this->releaseResources();
        this->copyOther(other);
      }
    }      


    // This member function moves the calling thread to the head of
    // the BroadcastQueue.
    template <class Type>
    void
    BroadcastQueue<Type>::
    catchUp(ClientID& clientID)
    {
      boost::timed_mutex::scoped_lock headLock(*m_headIndexMutexPtr);
      clientID.m_count = *m_countPtr;
      clientID.m_index = *m_headIndexPtr;
    }

    
    // This member function copies the tail element of the
    // BroadcastQueue.
    template <class Type>
    void
    BroadcastQueue<Type>::
    copyBack(const ClientID& clientID, Type& target)
    {
      this->lockBack(clientID);

      try {
        target = m_dataArray[clientID.m_index];
      } catch(...) {
        this->unlockBack(clientID);
        throw;
      }

      this->unlockBack(clientID);
    }


    // This member function copies the tail element of the
    // BroadcastQueue.
    template <class Type>
    bool
    BroadcastQueue<Type>::
    copyBack(const ClientID& clientID, Type& target, double timeout)
    {
      if(this->lockBack(clientID, timeout) == false) {
        return false;
      }

      try {
        target = m_dataArray[clientID.m_index];
      } catch(...) {
        this->unlockBack(clientID);
        throw;
      }

      this->unlockBack(clientID);
      return true;
    }


    // This member function locks the tail element of the BroadcastQueue
    // instance so that it cannot fall off the back of the queue.
    template <class Type>
    void
    BroadcastQueue<Type>::
    lockBack(const ClientID& clientID)
    {
      // These variables let us avoid time-consuming formatting of
      // error messages during the time that we have
      // *m_headIndexMutexPtr locked.
      bool overflowFlag = false;
      size_t headElementNumber = 0;
      
      // Get exclusive control of the broadcastQueue, check that we're not
      // too far ahead or behind, and mark our current element as
      // being read.  The scoped_lock instance will be released when
      // it goes out of scope.
      {
        boost::timed_mutex::scoped_lock headLock(*m_headIndexMutexPtr);

        // // This code left around in case we try to go back to
        // // individual mutexes for data elements.
        // boost::timed_mutex::scoped_lock elementLock(
        //   m_mutexArray[clientID.m_index]);
        
        // Make sure there's data available to copy.  That is, make
        // sure we're not ahead of the queue, and wait for a producer
        // to add an element if we are.
        while(clientID.m_count >= (*m_countPtr)) {
          m_headChangedConditionPtr->wait(headLock);
        }

        // Mark the appropriate element as being read by the calling
        // thread.
        this->declareReadLock(clientID, overflowFlag, headElementNumber);
        
      } // Release the lock.

      if(overflowFlag) {
        std::ostringstream message;
        message << "Thread "
                << (m_clientNameMapPtr->find(clientID.m_idNumber))->second
                << " "
                << "has fallen irretrievably far behind the front of the "
                << "queue.  Accessed element is number "
                << clientID.m_count << ", head element is number "
                << headElementNumber << ".";
        DLR_THROW(OverflowException, "BroadcastQueue::lockBack()",
                  message.str().c_str());
      }
    }


    // This member function locks the tail element of the BroadcastQueue
    // instance so that it cannot fall off the back of the queue.
    template <class Type>
    bool
    BroadcastQueue<Type>::
    lockBack(const ClientID& clientID, double timeout)
    {
      // Argument checking.
      if(timeout < 0.0) {
        timeout = 0.0;
      }
      boost::xtime xtimeout;
      if(getXTime(timeout, xtimeout) == false) {
        // Note(xxx): should we throw an exception here?
        return false;
      }
      
      // These variables let us avoid time-consuming formatting of
      // error messages during the time that we have
      // *m_headIndexMutexPtr locked.
      bool overflowFlag = false;
      size_t headElementNumber = 0;

      // Get exclusive control of the broadcastQueue, check that we're not
      // too far ahead or behind, and mark our current element as
      // being read.  The scoped_lock instance will be released when
      // it goes out of scope.
      try {
        // This call throws a boost::lock_error if the timeout expires.
        boost::timed_mutex::scoped_timed_lock headLock(
          *m_headIndexMutexPtr, xtimeout);

        // Make sure there's data available to copy.  That is, make
        // sure we're not ahead of the queue, and wait for a producer
        // to add an element if we are.
        while(clientID.m_count >= (*m_countPtr)) {
          if(m_headChangedConditionPtr->timed_wait(headLock, xtimeout)
             == false) {
            // No new data before timeout expires.
            return false;
          }
        }

        // Mark the appropriate element as being read by the calling
        // thread.
        this->declareReadLock(clientID, overflowFlag, headElementNumber);
        
      } catch(const boost::lock_error&) {
        // The timeout expired before we could get a lock.
        return false;
      }

      if(overflowFlag) {
        std::ostringstream message;
        message << "Thread "
                << (m_clientNameMapPtr->find(clientID.m_idNumber))->second
                << " "
                << "has fallen irretrievably far behind the front of the "
                << "queue.  Accessed element is number "
                << clientID.m_count << ", head element is number "
                << headElementNumber << ".";
        DLR_THROW(OverflowException, "BroadcastQueue::lockBack()",
                  message.str().c_str());
      }
      return true;
    }


    template <class Type>
    inline void
    BroadcastQueue<Type>::
    popBack(ClientID& clientID)
    {
      ++clientID.m_count;
      ++clientID.m_index;
      if(clientID.m_index >= m_size) {
        clientID.m_index = 0;
      }
      if(clientID.m_count == std::numeric_limits<size_t>::max()) {
        std::ostringstream message;
        message << "Thread "
                << (m_clientNameMapPtr->find(clientID.m_idNumber))->second
                << " "
                << "has exceeded the maximum allowable queue position of "
                << std::numeric_limits<size_t>::max() - 1 << ".";
        DLR_THROW(LogicException, "BroadcastQueue::popBack()",
                  message.str().c_str());
      }
    }


    template <class Type>
    void
    BroadcastQueue<Type>::
    pushFront(const ClientID& clientID, const Type& element)
    {
      // These variables let us avoid time-consuming formatting of
      // error messages during the time that we have
      // *m_headIndexMutexPtr locked.
      bool readLockedFlag = false;
      bool countOverflowFlag = false;
      size_t threadIDNumber = 0;

      // Get exclusive control of the BroadcastQueue, check that noone's
      // read-locked an element that's so far behind the head that the
      // head has wrapped around and is about to stomp on the locked
      // data, copy the new element into it, and advance the head.
      {
        // This lock will be automatically released when it goes out
        // of scope.
        boost::timed_mutex::scoped_lock headLock(*m_headIndexMutexPtr);

        if(m_readLockCountArray[*m_headIndexPtr] != 0) {
          // Ack! This element is read-locked.  Figure out which
          // thread is the culprit.
          readLockedFlag = true;
          size_t* flagsPtr = m_readLockingFlagsArray[*m_headIndexPtr];
          while(threadIDNumber < (*m_numberOfClientsPtr)) {
            if(flagsPtr[threadIDNumber] != 0) {
              break;
            }
            ++threadIDNumber;
          }
        } else {
          // Not read-locked, proceed with the copying, etc.
          m_dataArray[*m_headIndexPtr] = element;
          ++(*m_headIndexPtr);
          ++(*m_countPtr);
          if((*m_headIndexPtr) >= m_size) {
            (*m_headIndexPtr) = 0;
          }
          if(*m_countPtr == std::numeric_limits<size_t>::max()) {
            countOverflowFlag = true;
          }

          // Wake up any threads that are waiting for data.
          m_headChangedConditionPtr->notify_all();
        }
      }

      // Report any errors.
      if(readLockedFlag) {
        // Sanity check.
        if(threadIDNumber == *m_numberOfClientsPtr) {
          DLR_THROW(LogicException, "BroadcastQueue::pushFront()",
                    "Read lock accounting is in an inconsistent state.");
        }
        std::ostringstream message;
        message << "Thread "
                << (m_clientNameMapPtr->find(clientID.m_idNumber))->second
                << " "
                << "can't advance the queue because thread "
                << (m_clientNameMapPtr->find(threadIDNumber))->second
                << " "
                << "is still read-locking the tail element.";
        DLR_THROW(OverflowException, "BroadcastQueue::pushFront()",
                  message.str().c_str());
      }
      if(countOverflowFlag) {
        std::ostringstream message;
        message << "Queue size exceeds the allowable maximum of "
                << std::numeric_limits<size_t>::max() - 1 << " "
                << "on a function call from thread " 
                << (m_clientNameMapPtr->find(clientID.m_idNumber))->second
                << ".";
        DLR_THROW(OverflowException, "BroadcastQueue::pushFront()",
                  message.str().c_str());
      }
    }


    // This member function returns a ClientID instance that should
    // be used by the calling thread in all subsequent interactions
    // with the BroadcastQueue instance.
    template <class Type>
    void 
    BroadcastQueue<Type>::
    registerClient(ClientID& clientID, const std::string& clientName)
    {
      // Registration must be strictly serialized.  Only one thread
      // can register at once.  Also, producer threads must be locked
      // out so they don't throw an exception when they can't get
      // locks on m_mutexArray elements.
      boost::timed_mutex::scoped_lock headLock(*m_headIndexMutexPtr);

      // If adding another client will overflow our bookkeeping space,
      // then allocate a bigger bookkeeping space and copy the old
      // data into it.
      if(*m_numberOfClientsPtr >= (*m_maxNumberOfClientsPtr)) {
        size_t newMaxNumberOfClients = (*m_maxNumberOfClientsPtr) << 1;
        for(size_t index0 = 0; index0 < m_size; ++index0) {
          {
            // // This code left around in case we try to go back to
            // // individual mutexes for data elements.
            // boost::timed_mutex::scoped_lock elementLock(
            //   m_mutexArray[index0]);
          
            size_t* newFlagsArray = new size_t[newMaxNumberOfClients];
            size_t* beginIterator = m_readLockingFlagsArray[index0];
            size_t* endIterator = beginIterator + (*m_maxNumberOfClientsPtr);
            std::copy(beginIterator, endIterator, newFlagsArray);
            
            beginIterator = newFlagsArray;
            endIterator = beginIterator + newMaxNumberOfClients;
            beginIterator += (*m_maxNumberOfClientsPtr);
            std::fill(beginIterator, endIterator, 0);
            
            delete[] m_readLockingFlagsArray[index0];
            m_readLockingFlagsArray[index0] = newFlagsArray;
          }
        }
        *m_maxNumberOfClientsPtr = newMaxNumberOfClients;
      }
      m_clientNameMapPtr->insert(
        std::make_pair(*m_numberOfClientsPtr, clientName));
      clientID.m_count = 0;
      clientID.m_idNumber = (*m_numberOfClientsPtr)++;
      clientID.m_index = 0;
    }
    

    // This member function blocks until all clients of the
    // BroadcastQueue have called it, then moves the client to the head
    // of the queue.
    template <class Type>
    void 
    BroadcastQueue<Type>::
    synchronize(ClientID& clientID)
    {
      boost::timed_mutex::scoped_lock syncLock(*m_syncMutexPtr);
      
      if((*m_syncClientCountPtr) != 0) {
        if((*m_quorumSizePtr) != 0) {
          std::ostringstream message;
          message << "Can't synchronize with all registered clients "
                  << "while another thread is waiting for a quorum of size "
                  << (*m_quorumSizePtr) << "."; 
          DLR_THROW(StateException, "BroadcastQueue::synchronize()",
                    message.str().c_str());
        }
      }
      (*m_quorumSizePtr) = 0;

      this->handleSynchronize(clientID, (*m_numberOfClientsPtr), syncLock);
    }


    // This member function blocks until all clients of the
    // BroadcastQueue have called it, then moves the client to the head
    // of the queue.
    template <class Type>
    bool
    BroadcastQueue<Type>::
    synchronize(ClientID& clientID, double timeout)
    {
      // Argument checking.
      if(timeout < 0.0) {
        timeout = 0.0;
      }
      boost::xtime xtimeout;
      if(getXTime(timeout, xtimeout) == false) {
        // Note(xxx): should we throw an exception here?
        return false;
      }

      bool returnValue = false;
      try {
        // This call throws a boost::lock_error if the timeout expires.
        boost::timed_mutex::scoped_timed_lock syncLock(
          *m_syncMutexPtr, xtimeout);

        // State checking.
        if((*m_syncClientCountPtr) != 0) {
          if((*m_quorumSizePtr) != 0) {
            std::ostringstream message;
            message << "Can't synchronize with all registered clients "
                    << "while another thread is waiting for a quorum of size "
                    << (*m_quorumSizePtr) << "."; 
            DLR_THROW(StateException, "BroadcastQueue::synchronize()",
                      message.str().c_str());
          }
        }
        (*m_quorumSizePtr) = 0;
        
        returnValue =
          this->handleSynchronize(
            clientID, (*m_numberOfClientsPtr), xtimeout, syncLock);
      } catch(const boost::lock_error&) {
        // The timeout expired before we could get a lock.
      }
      return returnValue;
    }


    // This member function blocks until the specified number of
    // BroadcastQueue clients have called it, then moves the client to
    // the head of the queue.
    template <class Type>
    void
    BroadcastQueue<Type>::
    synchronizeQuorum(ClientID& clientID, size_t quorumSize)
    {
      boost::timed_mutex::scoped_lock syncLock(*m_syncMutexPtr);

      if(quorumSize == 0) {
        quorumSize = (*m_numberOfClientsPtr);
      }
      
      if((*m_syncClientCountPtr) != 0) {
        if((*m_quorumSizePtr) == 0) {
          DLR_THROW(StateException, "BroadcastQueue::synchronizeQuorum()",
                    "Can't start a quorum while another thread is blocking "
                    "in member function synchronize()");         
        } else if((*m_quorumSizePtr) != quorumSize) {
          std::ostringstream message;
          message << "Can't start a quorum of size " << quorumSize << " "
                  << "while another thread is waiting for a quorum of size "
                  << (*m_quorumSizePtr) << ".";
          DLR_THROW(StateException, "BroadcastQueue::synchronizeQuorum()",
                    message.str().c_str());
        }
      }
      (*m_quorumSizePtr) = quorumSize;

      this->handleSynchronize(clientID, (*m_quorumSizePtr), syncLock);
    }


    // This member function blocks until the specified number of
    // BroadcastQueue clients have called it, then moves the client to
    // the head of the queue.
    template <class Type>
    bool
    BroadcastQueue<Type>::
    synchronizeQuorum(ClientID& clientID, size_t quorumSize, double timeout)
    {
      // Argument checking.
      if(timeout < 0.0) {
        timeout = 0.0;
      }
      boost::xtime xtimeout;
      if(getXTime(timeout, xtimeout) == false) {
        // Note(xxx): should we throw an exception here?
        return false;
      }
      
      bool returnValue = false;
      try {
        // This call throws a boost::lock_error if the timeout expires.
        boost::timed_mutex::scoped_timed_lock syncLock(
          *m_syncMutexPtr, xtimeout);

        // State checking.
        if(quorumSize == 0) {
          quorumSize = (*m_numberOfClientsPtr);
        }
      
        if((*m_syncClientCountPtr) != 0) {
          if((*m_quorumSizePtr) == 0) {
            DLR_THROW(StateException, "BroadcastQueue::synchronizeQuorum()",
                      "Can't start a quorum while another thread is blocking "
                      "in member function synchronize()");         
          } else if((*m_quorumSizePtr) != quorumSize) {
            std::ostringstream message;
            message << "Can't start a quorum of size " << quorumSize << " "
                    << "while another thread is waiting for a quorum of size "
                    << (*m_quorumSizePtr) << ".";
            DLR_THROW(StateException, "BroadcastQueue::synchronizeQuorum()",
                      message.str().c_str());
          }
        }
        (*m_quorumSizePtr) = quorumSize;
        
        returnValue =
          this->handleSynchronize(
            clientID, (*m_quorumSizePtr), xtimeout, syncLock);
      } catch(const boost::lock_error&) {
        // The timeout expired before we could get a lock.
      }
      return returnValue;
    }

    
    // This member function releases a read lock acquired by
    // lockBack().
    template <class Type>
    void
    BroadcastQueue<Type>::
    unlockBack(const ClientID& clientID)
    {
      // Get exclusive control of the broadcastQueue, and undo the read
      // lock.  The scoped_lock instance will be released when it goes
      // out of scope.
      {
        boost::timed_mutex::scoped_lock headLock(*m_headIndexMutexPtr);

        // m_readLockingFlagsArray keeps track of which threads are
        // read-locking which elements.  Note that lockedFlag is a
        // reference.
        size_t& lockedFlag =
          m_readLockingFlagsArray[clientID.m_index][clientID.m_idNumber];

        // We only need to release the lock if this client is actually
        // locking this element.
        if(lockedFlag != 0) {
          --lockedFlag;
          --(m_readLockCountArray[clientID.m_index]);
        }
      } // Release the lock.
    }
    
    
    /* =========== Private member functions of BroadcastQueue =========== */

    // This private member function is called by the copy
    // constructor and assignment operator to copy another
    // BroadcastQueue instance in a thread-safe way.
    template <class Type>
    void
    BroadcastQueue<Type>::
    copyOther(const BroadcastQueue<Type>& other)
    {
      {
        boost::timed_mutex::scoped_lock
          headLock(*(other.m_headIndexMutexPtr));

        m_countPtr = other.m_countPtr;
        m_headIndexPtr = other.m_headIndexPtr;
        m_size = other.m_size;
        m_clientNameMapPtr = other.m_clientNameMapPtr;
        m_dataArray = other.m_dataArray;
        m_maxNumberOfClientsPtr = other.m_maxNumberOfClientsPtr;
        m_numberOfClientsPtr = other.m_numberOfClientsPtr;
        m_readLockCountArray = other.m_readLockCountArray;
        m_readLockingFlagsArray = other.m_readLockingFlagsArray;
        m_syncClientCountPtr = other.m_syncClientCountPtr;
        m_syncElementCountPtr = other.m_syncElementCountPtr;
        m_syncIndexPtr = other.m_syncIndexPtr;
        m_quorumSizePtr = other.m_quorumSizePtr;
        m_headIndexMutexPtr = other.m_headIndexMutexPtr;
        m_headChangedConditionPtr = other.m_headChangedConditionPtr;
        // // This code left around in case we try to go back to
        // // individual mutexes for data elements.
        // m_mutexArray = other.m_mutexArray;
        m_numberOfClientsMutexPtr = other.m_numberOfClientsMutexPtr;
        m_syncConditionPtr = other.m_syncConditionPtr;
        m_syncMutexPtr = other.m_syncMutexPtr;
        m_referenceCountPtr = other.m_referenceCountPtr;
        ++(*m_referenceCountPtr);
      }
    }

    
    // This private member function should only be called when
    // m_headIndexMutexPtr is locked.
    template <class Type>
    inline void
    BroadcastQueue<Type>::
    declareReadLock(const ClientID& clientID,
                    bool& overflowFlag,
                    size_t& headElementNumber)
    {
      // Check that we're not too behind the head of the queue, and
      // mark our current element as being read.
      if((clientID.m_count + m_size) < (*m_countPtr)) {
        overflowFlag = true;
        headElementNumber = *m_countPtr;
      } else {
        // No overflow.
        ++(m_readLockingFlagsArray[clientID.m_index][clientID.m_idNumber]);
        ++(m_readLockCountArray[clientID.m_index]);
      }
    }
    

    // This private member function does the actual mechanics of
    // synchronizing threads.
    template <class Type>
    void 
    BroadcastQueue<Type>::
    handleSynchronize(ClientID& clientID, size_t targetNumber,
                      boost::timed_mutex::scoped_lock& syncLock)
    {
      // We know (because one of our arguments is a lock object) that
      // it's safe to mess with the m_sync* data members.
      ++(*m_syncClientCountPtr);
      if((*m_syncClientCountPtr) >= targetNumber) {
        {
          // We need to lock headlock before accessing m_countPtr and
          // m_headIndexPtr so they don't change out from under us.
          boost::timed_mutex::scoped_lock headLock(*m_headIndexMutexPtr);
          *m_syncElementCountPtr = *m_countPtr;
          *m_syncIndexPtr = *m_headIndexPtr;
        }
        // Wake up any threads that are waiting for sync.
        (*m_syncClientCountPtr) = 0;
        m_syncConditionPtr->notify_all();
      } else {
        m_syncConditionPtr->wait(syncLock);
      }
      clientID.m_count = *m_syncElementCountPtr;
      clientID.m_index = *m_syncIndexPtr;
    }
    

    // This private member function does the actual mechanics of
    // synchronizing threads.
    template <class Type>
    bool
    BroadcastQueue<Type>::
    handleSynchronize(ClientID& clientID,
                      size_t targetNumber,
                      boost::xtime xtimeout,
                      boost::timed_mutex::scoped_timed_lock& syncLock)
    {
      // We know (because one of our arguments is a lock object) that
      // it's safe to mess with the m_sync* data members.
      ++(*m_syncClientCountPtr);
      if((*m_syncClientCountPtr) >= targetNumber) {
        try {
          // We need to lock headlock before accessing m_countPtr and
          // m_headIndexPtr so they don't change out from under us.
          boost::timed_mutex::scoped_timed_lock
            headLock(*m_headIndexMutexPtr, xtimeout);
          *m_syncElementCountPtr = *m_countPtr;
          *m_syncIndexPtr = *m_headIndexPtr;
        } catch(const boost::lock_error&) {
          --(*m_syncClientCountPtr);
          return false;
        }
        // Wake up any threads that are waiting for sync.
        (*m_syncClientCountPtr) = 0;
        m_syncConditionPtr->notify_all();
      } else {
        if(m_syncConditionPtr->timed_wait(syncLock, xtimeout) == false) {
          // Sync event didn't happen.  Bail out.
          --(*m_syncClientCountPtr);
          return false;
        }
      }
      clientID.m_count = *m_syncElementCountPtr;
      clientID.m_index = *m_syncIndexPtr;
      return true;
    }


    // This protected member function is called by the destructor
    // and copy constructor to decrement reference counts, etc.
    template <class Type>
    void
    BroadcastQueue<Type>::    
    releaseResources()
    {
      bool isShared = true;
      {
        boost::timed_mutex::scoped_lock headLock(*m_headIndexMutexPtr);
        if(--(*m_referenceCountPtr) == 0) {
          isShared = false;
        }
      }
        
      if(!isShared) {
        delete m_countPtr;
        delete m_headIndexPtr;
        delete m_clientNameMapPtr;
        delete[] m_dataArray;
        delete m_maxNumberOfClientsPtr;
        delete m_numberOfClientsPtr;
        delete[] m_readLockCountArray;
        for(size_t index0 = 0; index0 < m_size; ++index0) {
          delete[] m_readLockingFlagsArray[index0];
        }
        delete[] m_readLockingFlagsArray;
        delete m_syncClientCountPtr;
        delete m_syncElementCountPtr;
        delete m_syncIndexPtr;
        delete m_quorumSizePtr;
        delete m_headChangedConditionPtr;
        delete m_headIndexMutexPtr;
        // // This code left around in case we try to go back to
        // // individual mutexes for data elements.
        // delete[] m_mutexArray;
        delete m_numberOfClientsMutexPtr;
        delete m_syncConditionPtr;
        delete m_syncMutexPtr;
        delete m_referenceCountPtr;
      }
    }
    
  } // namespace thread

} // namespace dlr



#endif /* #ifndef _DLR_THREAD_BROADCASTQUEUE_H_ */

---