distributionQueue.h

Go to the documentation of this file.

#ifndef _DLR_THREAD_DISTRIBUTIONQUEUE_H_
#define _DLR_THREAD_DISTRIBUTIONQUEUE_H_

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


namespace dlr {

  namespace thread {

    template <class Type>
    class DistributionQueue
      : public Monitor
    {
    public:

      typedef ClientID<Type> ClientID;


      DistributionQueue(size_t maximumLength, size_t numberOfClients=2);


      DistributionQueue(const DistributionQueue<Type>& source);


      virtual
      ~DistributionQueue();


      DistributionQueue<Type>&
      operator=(const DistributionQueue<Type>& source);


      void
      bufferBack(const ClientID& clientID);


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


      void
      clear();
      

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


      size_t
      getMaximumLength() {return m_size;}
      

      void
      lockBack(const ClientID& clientID);


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

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


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


      void
      unlockBack(const ClientID& clientID);
      
    private:

      inline void
      releaseLock();
      
      
      inline void
      waitForLock(Token& token);


      inline bool
      waitForLock(Token& token, double timeout);

      
        
      size_t* m_headIndexPtr;
      size_t m_size;
      size_t* m_tailIndexPtr;
      
      Type** m_bufferArrayPtr;
      std::map<size_t, std::string>* m_clientNameMapPtr;
      Type* m_dataArray;
      size_t* m_maxNumberOfClientsPtr;
      size_t* m_numberOfClientsPtr;

      Condition m_headChangedCondition;
      bool* m_isLockedPtr;
      Condition m_isLockedCondition;

      // We use a pointer to ReferenceCount, rather than a
      // ReferenceCount instance, so that we can manage construction,
      // destruction, etc. inside locked sections of code.
      ReferenceCount* m_referenceCountPtr;
    };

  } // namespace thread

} // namespace dlr


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

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


namespace dlr {

  namespace thread {

    /* ===========    Implementation of DistributionQueue      =========== */

    template <class Type>
    DistributionQueue<Type>::
    DistributionQueue(size_t maximumLength, size_t numberOfClients)
      : Monitor(),
        m_headIndexPtr(new size_t),
        m_size(maximumLength),
        m_tailIndexPtr(new size_t),
        m_bufferArrayPtr(new Type*),
        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_headChangedCondition(Monitor::createCondition()),
        m_isLockedPtr(new bool),
                m_isLockedCondition(Monitor::createCondition()),
        m_referenceCountPtr(new ReferenceCount)
    {
      if(numberOfClients == 0) {
        DLR_THROW(ValueException, "DistributionQueue::DistributionQueue()",
                  "Argument numberOfClients must be greater than zero.");
      }

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

      // But indicate that no clients have registered yet.
      *m_bufferArrayPtr = new Type[numberOfClients];
      *m_maxNumberOfClientsPtr = numberOfClients;
      *m_numberOfClientsPtr = 0;

      // And that nobody's called this->lockBack().
      *m_isLockedPtr = false;

      // Finally, allow copying.
      this->makeCopyable();
    }


    // The copy constructor does a shallow copy.
    template <class Type>
    DistributionQueue<Type>::
    DistributionQueue(const DistributionQueue<Type>& source)
      : Monitor(source),
        m_headIndexPtr(0),
        m_size(0),
        m_tailIndexPtr(0),
        m_bufferArrayPtr(0),
        m_clientNameMapPtr(0),
        m_dataArray(0),
        m_maxNumberOfClientsPtr(0),
        m_numberOfClientsPtr(0),
                m_headChangedCondition(Monitor::createCondition(false)),
        m_isLockedPtr(0),
                m_isLockedCondition(Monitor::createCondition(false)),
        m_referenceCountPtr(0)
    {
      Token token = this->getToken();
      // No need to wait for *m_isLockedPtr because we're not going to
      // read or change queue contents.
      
      m_headIndexPtr = source.m_headIndexPtr;
      m_size = source.m_size;
      m_tailIndexPtr = source.m_tailIndexPtr;
      m_bufferArrayPtr = source.m_bufferArrayPtr;
      m_clientNameMapPtr = source.m_clientNameMapPtr;
      m_dataArray = source.m_dataArray;
      m_maxNumberOfClientsPtr = source.m_maxNumberOfClientsPtr;
      m_numberOfClientsPtr = source.m_numberOfClientsPtr;
      m_headChangedCondition = source.m_headChangedCondition;
      m_isLockedPtr = source.m_isLockedPtr;
      m_isLockedCondition = source.m_isLockedCondition;
      m_referenceCountPtr = source.m_referenceCountPtr;
      ++(*m_referenceCountPtr);
    }

    
    // Destroys the DistributionQueue instance and releases any resources.
    template <class Type>
    DistributionQueue<Type>::
    ~DistributionQueue()
    {
      Token token = this->getToken();
      // No need to wait for *m_isLockedPtr because we're not going to
      // read or change queue contents unless we're the only instance
      // referencing this data.
      
      if(!(m_referenceCountPtr->isShared())) {
        delete m_headIndexPtr;
        delete m_tailIndexPtr;
        delete m_bufferArrayPtr;
        delete m_clientNameMapPtr;
        delete m_dataArray;
        delete m_maxNumberOfClientsPtr;
        delete m_numberOfClientsPtr;
        delete m_isLockedPtr;
        delete m_referenceCountPtr;
      } else {
        --(*m_referenceCountPtr);
      }
    }


    // The assignment operator does a shallow copy.
    template <class Type>
    DistributionQueue<Type>&
    DistributionQueue<Type>::
    operator=(const DistributionQueue<Type>& source)
    {
      if(&source != this) {
        Monitor::operator=(source);
        Token token = this->getToken();
        // No need to wait for *m_isLockedPtr because we're not going
        // to read or change queue contents unless we're the only
        // instance referencing the data.

        this->release();
        this->copyOther(source);
      }
    }      


    // This member function copies the tail element of the
    // DistributionQueue.
    template <class Type>
    void
    DistributionQueue<Type>::
    bufferBack(const ClientID& clientID)
    {
      // All locking, synchronizing, etc. handled inside lockBack().
      this->lockBack(clientID);
      
      try {
        (*m_bufferArrayPtr)[clientID.m_idNumber]
          = m_dataArray[*m_tailIndexPtr];
      } catch(...) {
        this->unlockBack(clientID);
        throw;
      }

      ++(*m_tailIndexPtr);
      if(*m_tailIndexPtr == m_size) {
        *m_tailIndexPtr = 0;
      }

      this->unlockBack(clientID);
    }


    // This member function copies the tail element of the
    // DistributionQueue.
    template <class Type>
    bool
    DistributionQueue<Type>::
    bufferBack(const ClientID& clientID, double timeout)
    {
      // All locking, synchronizing, etc. handled inside lockBack().
      if(this->lockBack(clientID, timeout) == false) {
        return false;
      }

      try {
        (*m_bufferArrayPtr)[clientID.m_idNumber]
          = m_dataArray[*m_tailIndexPtr];
      } catch(...) {
        this->unlockBack(clientID);
        throw;
      }

      ++(*m_tailIndexPtr);
      if(*m_tailIndexPtr == m_size) {
        *m_tailIndexPtr = 0;
      }

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


    // This member function empties the queue.
    template <class Type>
    void
    DistributionQueue<Type>::
    clear()
    {
      Token token = this->getToken();
      this->waitForLock();
      *m_tailIndexPtr = *m_headIndexPtr;
      this->releaseLock();
    }

    
    // This member function copies the buffered element (see member
    // function bufferBack() from the internal buffer into user
    // code.
    template <class Type>
    void
    DistributionQueue<Type>::
    copyBuffer(const ClientID& clientID, Type& target)
    {
      target = (*m_bufferArrayPtr)[clientID.m_idNumber];
    }


    // This member function locks the tail element of the DistributionQueue
    // instance so that it cannot fall off the back of the queue.
    template <class Type>
    void
    DistributionQueue<Type>::
    lockBack(const ClientID& clientID)
    {
      Token token = this->getToken();
      this->waitForLock(token);

      // 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(*m_tailIndexPtr == *m_headIndexPtr) {
        this->releaseLock();
        this->wait(m_headChangedCondition, token);
        this->waitForLock(token);
      }

      // Note that we still have the lock!
      return;
    }


    // This member function locks the tail element of the DistributionQueue
    // instance so that it cannot fall off the back of the queue.
    template <class Type>
    bool
    DistributionQueue<Type>::
    lockBack(const ClientID& clientID, double timeout)
    {
      // Get exclusive control of the distributionQueue.
      double deadline = getCurrentTime() + timeout;
      Token token = this->getToken(timeout);
      if(!token) {
        return false;
      }
      if(!this->waitForLock(token, deadline - getCurrentTime())) {
        return false;
      }

      // 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(*m_tailIndexPtr == *m_headIndexPtr) {
        this->releaseLock();
        if(!this->wait(m_headChangedCondition, token,
                       deadline - getCurrentTime())) {
          return false;
        }
        if(!this->waitForLock(token, deadline - getCurrentTime())) {
          return false;
        }
      }
        
      // Note that we still have the lock!
      return true;
    }


    template <class Type>
    void
    DistributionQueue<Type>::
    pushFront(const ClientID& clientID, const Type& element)
    {
      // This variable lets us avoid time-consuming formatting of
      // error messages during the time that we have *m_accessMutexPtr
      // locked.
      bool overflowFlag = false;
      {
        // Get exclusive control of the DistributionQueue.
        Token token = this->getToken();
        this->waitForLock(token);
        try {
          // Increment head, and wrap around if necessary.
          size_t newHeadIndex = *m_headIndexPtr + 1;
          if(newHeadIndex == m_size) {
            newHeadIndex = 0;
          }
          
          // Check for overflow.
          if(newHeadIndex == *m_tailIndexPtr) {
            overflowFlag = true;
          } else {
            // Proceed with the copying, etc.
            m_dataArray[*m_headIndexPtr] = element;
          }
          *m_headIndexPtr = newHeadIndex;
          
          // Wake up any threads that are waiting for data.
          this->signalAll(m_headChangedCondition);
        } catch(...) {
          this->releaseLock();
          throw;
        }
        this->releaseLock();
      }

      // Report any errors.
      if(overflowFlag) {
        std::ostringstream message;
        message << "Queue size exceeds the allowable maximum of "
                << m_size << " " << "on a function call from thread " 
                << (m_clientNameMapPtr->find(clientID.m_idNumber))->second
                << ".";
        DLR_THROW(OverflowException, "DistributionQueue::pushFront()",
                  message.str().c_str());
      }
    }


    // This member function introduces a thread to a
    // DistributionQueue instance.
    template <class Type>
    void 
    DistributionQueue<Type>::
    registerClient(ClientID& clientID, const std::string& clientName)
    {
      // Registration must be strictly serialized.  Only one thread
      // can register at once.
      Token token = this->getToken();
      this->waitForLock(token);

      try {
        m_clientNameMapPtr->insert(
          std::make_pair(*m_numberOfClientsPtr, clientName));
        clientID.m_count = 0;
        clientID.m_idNumber = (*m_numberOfClientsPtr)++;
        clientID.m_index = 0;

        // If we now have more than the expected number of clients, we
        // have to resize our internal storage.
        if(*m_numberOfClientsPtr > *m_maxNumberOfClientsPtr) {
          size_t oldMaxNumberOfClients = *m_maxNumberOfClientsPtr;
          Type* oldBufferArray = *m_bufferArrayPtr;
        
          *m_maxNumberOfClientsPtr *= 2;
          *m_bufferArrayPtr = new Type[*m_maxNumberOfClientsPtr];

          std::copy(oldBufferArray, oldBufferArray + oldMaxNumberOfClients,
                    *m_bufferArrayPtr);
          delete[] oldBufferArray;        
        }
      } catch(...) {
        this->releaseLock();
        throw;
      }
      this->releaseLock();
    }
    

    // This member function releases a lock acquired by
    // lockBack().
    template <class Type>
    void
    DistributionQueue<Type>::
    unlockBack(const ClientID& clientID)
    {
      Token token = this->getToken();
      this->releaseLock();
    }

    
    template <class Type>
    void
    DistributionQueue<Type>::
    releaseLock()
    {
      *m_isLockedPtr = false;
      this->signalOne(m_isLockedCondition);
    }

    
    template <class Type>
    void
    DistributionQueue<Type>::
    waitForLock(Token& token)
    {
      while(*m_isLockedPtr) {
        this->wait(this->m_isLockedCondition, token);
      }
      *m_isLockedPtr = true;
    }


    template <class Type>
    bool
    DistributionQueue<Type>::
    waitForLock(Token& token, double timeout)
    {
      double deadline = getCurrentTime() + timeout;
      while(*m_isLockedPtr) {
        if(this->wait(this->m_isLockedCondition, token,
                      deadline - getCurrentTime()) == false) {
          return false;
        }
      }
      *m_isLockedPtr = true;
      return true;
    }
    
    
  } // namespace thread

} // namespace dlr

#endif /* #ifndef _DLR_THREAD_DISTRIBUTIONQUEUE_H_ */

---