ReplyQueue.h

Go to the documentation of this file.

////////////////////////////////////////////////////////////////////////////////////
/*! \file ReplyQueue.h
*  \brief Declares and defines template classes for synchronization of network comm.
*  \author $Author: rsharo $
*  \version $Revision: 1.1 $
*  \date    $Date: 2004/04/07 13:45:04 $
*///////////////////////////////////////////////////////////////////////////////////
#ifndef REPLY_QUEUE_H
#define REPLY_QUEUE_H

#include "ace/Log_Msg.h"
#include "ace/Synch_T.h"

#include <list>

//
// Forward declaration of the PendingReply class. This forward declaration
// is needed for the declaration of ReplyQueue.
template <class T, ACE_SYNCH_DECL>
class PendingReply;

//
// This class encapsulates a synchronized queue through which solicited
// data (replies) are passed between threads.
//
// Command threads follow this sequence of execution:
//   (1) Acquire the queue's mutex exactly one time. Mutex acquisition must
//       NOT be recursive. (VERY IMPORTANT).
//   (2) Schedule some sort of input from another thread
//   (3) Call GetReply(), which will block until error or the result is ready
//   (4) Release the queue's mutex
//
// Reply threads follow this sequence of execution:
//   (1) Generate data or read data from the network
//   (2) The thread may or may not acquire the queue's mutex. In this thread
//       acquisition may be recursive if the operating system supports it.
//       In any event, SetReply() will acquire the mutex as part of its
//       operation.
//   (3) Call SetReply() to pass the data to the command thread(s).
//   (4) Release the queue's mutex if the thread had previously acquired it.
//
// Multiple threads can wait in GetReply() and will receive replies
// in FIFO order. This is possible because GetReply() temporarily
// relinquishes the mutex as part of its operation.
//
// If the waiter times out before data is received, it is removed from
// the queue.  Note that this could be problematic if a reply arrives
// after the deadline.  Waiters should set a deadline only if they are
// certain that missing a deadline means no reply will be coming.
//
// Any reply type for which the assignment operator is defined is allowed.
// Reply types that don't support the assignment operator may be used
// by specializing the Assign() method for your type.
//
// Examples of how to instantiate one of these queues:
//
//    A synchronized queue with string reply types:
//       ReplyQueue<std::string, ACE_MT_SYNCH> foo;
//
//    An unsynchronized queue with integer reply types:
//       ReplyQueue<int, ACE_NULL_SYNCH>       foo;
//
//    A queue with system-default synchronization and unspecified reply type:
//       ReplyQueue<void *, ACE_SYNCH>         foo;
template <class T, ACE_SYNCH_DECL>
class ReplyQueue
{
public:
        //
        // Constructs a reply queue with the specified mutex.
        ReplyQueue(ACE_SYNCH_MUTEX_T &myMutex);
        
        //
        // Destroys a reply queue. It calls Close() to ensure all waiters
        // are released.
        ~ReplyQueue();

        //
        // Closes the queue. The queue is cleared and all waiters are freed
        // with an error code. Any subsequent calls to GetReply() or SetReply()
        // will fail as well.
        //
        // If acquireMutex is false, then the method will not acquire its own
        // mutex. Do this only if the mutex is already externally acquired.
        //
        // Returns 0 on success or -1 if already closed.
        int Close(bool acquireMutex=true);

        //
        // Closes the queue. The queue is cleared and all waiters are freed
        // with an error code. Any subsequent calls to GetReply() or SetReply()
        // will fail as well.
        //
        // If acquireMutex is false, then the method will not acquire its own
        // mutex. Do this only if the mutex is already externally acquired.
        //
        // Returns 0 on success or -1 if already closed.
        inline int Close_i();

        //
        // Indicates whether there are any threads currently waiting for a reply.
        //
        // If acquireMutex is false, then the method will not acquire its own
        // mutex. Do this only if the mutex is already externally acquired.
        //
        // Returns true if there are waiters, false otherwise.
        bool IsEmpty(bool acquireMutex=false);

        //
        // Indicates whether there are any threads currently waiting for a reply.
        //
        // This method makes no attempt to acquire its own mutex. The mutex should
        // be externally acquired before calling this method.
        //
        // Returns true if there are waiters, false otherwise.
        inline bool IsEmpty_i();
        
        //
        // Blocks the calling thread until the deadline expires or
        // until a reply is sent received from the queue.
        // If a reply is enqueued before the deadline, its value
        // will be assigned to the "theReply" argument.
        // Note that the deadline argument is an absolute time -- not a relative
        // one. For example, if you want a 10 second timeout, you should compute
        // your deadline by calling ACE_OS::gettimeofday() and adding 10 seconds
        // to the result.
        // Passing a NULL pointer in the deadline argument means "wait forever".
        //
        // NOTE: You must acquire the queue's mutex exactly once when calling
        // this method. Acquiring the mutex more than once (i.e. recursively)
        // will result in a deadlock!
        //
        // Do not acquire the mutex externally before calling this method with
        // acquireMutex==true.
        //
        // If acquireMutex is false, then the method will not acquire its own
        // mutex. Do this only if the mutex is already externally acquired once.
        //
        // Returns 0 if the reply was successfully retrieved, -1 otherwise.
        int GetReply(T &theReply, const ACE_Time_Value *deadline, bool acquireMutex=false);
        
        //
        // Blocks the calling thread until the deadline expires or
        // until a reply is sent received from the queue.
        // If a reply is enqueued before the deadline, its value
        // will be assigned to the "theReply" argument.
        // Note that the deadline argument is an absolute time -- not a relative
        // one. For example, if you want a 10 second timeout, you should compute
        // your deadline by calling ACE_OS::gettimeofday() and adding 10 seconds
        // to the result.
        // Passing a NULL pointer in the deadline argument means "wait forever".
        //
        // NOTE: You must acquire the queue's mutex exactly once before calling
        // this method. Acquiring the mutex more than once (i.e. recursively)
        // will result in a deadlock! If you don't acquire the mutex before
        // calling, the method will fail.
        //
        // This method makes no attempt to acquire its own mutex. The mutex should
        // be externally acquired once before calling this method.
        //
        // Returns 0 if the reply was successfully retrieved, -1 otherwise.
        inline int GetReply_i(T &theReply, const ACE_Time_Value *deadline);

        //
        // Blocks the calling thread until a reply is sent received from the queue.
        // The reply is assigned to the "theReply" argument.
        //
        // NOTE: You must acquire the queue's mutex exactly once before calling
        // this method. Acquiring the mutex more than once (i.e. recursively)
        // will result in a deadlock! If you don't acquire the mutex before
        // calling, the method will fail.
        //
        // If acquireMutex is false, then the method will not acquire its own
        // mutex. Do this only if the mutex is already externally acquired.
        //
        // Returns 0 if the reply was successfully retrieved, -1 otherwise.
        int GetReply(T &theReply, bool acquireMutex=false);


        //
        // Blocks the calling thread until a reply is sent received from the queue.
        // The reply is assigned to the "theReply" argument.
        //
        // NOTE: You must acquire the queue's mutex exactly once before calling
        // this method. Acquiring the mutex more than once (i.e. recursively)
        // will result in a deadlock! If you don't acquire the mutex before
        // calling, the method will fail.
        //
        // This method makes no attempt to acquire its own mutex. The mutex should
        // be externally acquired before calling this method.
        //
        // Returns 0 if the reply was successfully retrieved, -1 otherwise.
        inline int GetReply_i(T &theReply);


        //
        // Sends a reply to the next waiter in GetReply().
        // This method should only be called when a GetReply() call is pending.
        // It will free the next GetReply() waiter and give them the "theReply"
        // argument.
        //
        // If acquireMutex is false, then the method will not acquire its own
        // mutex. Do this only if the mutex is already externally acquired.
        //
        // Returns 0 on success and -1 on failure.
        int SetReply(T &theReply, bool acquireMutex=true);

        //
        // Sends a reply to the next waiter in GetReply().
        // This method should only be called when a GetReply() call is pending.
        // It will free the next GetReply() waiter and give them the "theReply"
        // argument.
        //
        // This method makes no attempt to acquire its own mutex. The mutex should
        // be externally acquired before calling this method.
        //
        // Returns 0 on success and -1 on failure.
        inline int SetReply_i(T &theReply);

        //
        // Assigns the value of "from" to the object "to".
        // Returns a reference to "to".
        // This method exists solely to support partial template
        // specialization.
        inline T & Assign_i(T & from, T & to) { return to = from; }

        //
        // Returns a reference to this queue's mutex. Use this method
        // if you wish to synchonize other objects with this queue.
        inline ACE_SYNCH_MUTEX_T & GetMutex() { return m_replyQueueMutex; }
        
protected:
        
        //
        // The PendingReply type corresponding to this template
        // instantiation.
        typedef PendingReply<T, ACE_SYNCH_USE>  PENDING_REPLY;
        //
        // The data type of the queue holding PendingReply's.
        typedef std::list<PENDING_REPLY *>              QUEUE_TYPE;
        
        //
        // The queue that holds PendingReply's.
        QUEUE_TYPE                                                              m_replyQueue;

        //
        // The mutex type that synchronizes this queue. If the synch
        // type is ACE_NULL_SYNCH, then this mutex is a functionless
        // dummy object.
        ACE_SYNCH_MUTEX_T &                                             m_replyQueueMutex;

        //
        // A flag that indicates whether the queue is closed. 
        bool                                                                    m_closed;
};


//
// This class encapsulates a reply that is waiting in a ReplyQueue.
template <class T, ACE_SYNCH_DECL>
class PendingReply
{
public:
        //
        // The ReplyQueue type that holds this PendingReply type.
        typedef ReplyQueue<T, ACE_SYNCH_USE> REPLY_QUEUE;

        //
        // The ReplyQueue, being a friend of this class, is allowed
        // to access this class's protected and private members.
        // We don't use the typedef in the friend declaration
        // because some (namely MSVC++) compilers can't handle it.
        friend class ReplyQueue<T, ACE_SYNCH_USE>;
        
        //
        // Constructs a PendingReply. The "theReply" argument is a reference to
        // the object that will be set by PendingQueue::SetReply().  The "myMutex"
        // argument is a reference to the PendingQueue's mutex.
        PendingReply(T & theReply, ACE_SYNCH_MUTEX_T &myMutex)
                : m_condition(myMutex), m_isEnqueued(false), m_theReply(theReply) {}

        //
        // Destroys a PendingReply.
        ~PendingReply() {}
        
protected:
        //
        // The condition variable used to signal PendingQueue::GetReply() 
        // on success or failure.
        ACE_SYNCH_CONDITION_T   m_condition;

        //
        // Indicates whether this object is currently enqueued in the PendingQueue.
        // This is important when determining the validity of PendingQueue's
        // internal iterators.
        bool                                    m_isEnqueued;

        //
        // A reference to the object that will be set by a successful call to
        // PendingQueue::SetReply().  The object is provided by the caller of
        // PendingQueue::GetReply().
        T &                                             m_theReply;
};





template <class T, ACE_SYNCH_DECL>
ReplyQueue<T, ACE_SYNCH_USE>::ReplyQueue(ACE_SYNCH_MUTEX_T &myMutex)
                : m_replyQueueMutex(myMutex),
                  m_closed(false)
{
}


template <class T, ACE_SYNCH_DECL>
ReplyQueue<T, ACE_SYNCH_USE>::~ReplyQueue()
{
        Close_i();
}



template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::Close(bool acquireMutex)
{
        if (acquireMutex)
        {
                ACE_Guard<ACE_SYNCH_MUTEX_T> g(m_replyQueueMutex);

                return Close_i();
        }

        return Close_i();
}


        
template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::Close_i()
{
        //
        // Double-check idiom. Ensure that the queue wasn't closed
        // while we were waiting to acquire the mutex.
        if (m_closed)
                return -1;
                
        //
        // Mark the queue as closed.
        m_closed = true;
                
        //
        // Iterate through the queue and release all the
        // waiters. They'll see the queue is closed and return an error.
        typename QUEUE_TYPE::iterator   i = m_replyQueue.begin(),
                end = m_replyQueue.end();
        
        //
        // A place to hold the results of system calls.
        int status = 0;

        //
        // Mark all the PendingReplies as no longer enqueued and signal
        // all the waiters.  The "status" variable collects any errors.
        for (; i != end; ++i)
        {
                (*i)->m_isEnqueued = false;
                status |= (*i)->m_condition.broadcast();
        }
                
        //
        // Clear out the queue. Its contents are now invalid.
        m_replyQueue.clear();

        //
        // Return 0 on success or -1 if signaling any waiters failed.
        return status;
}



template <class T, ACE_SYNCH_DECL>
bool ReplyQueue<T, ACE_SYNCH_USE>::IsEmpty(bool acquireMutex)
{
        if (acquireMutex)
        {
                //
                // Don't bother acquiring the mutex if the queue is already closed. Just 
                // report that it's empty (which it must be).
                if (m_closed)
                        return true;

                ACE_Guard<ACE_SYNCH_MUTEX_T> g(m_replyQueueMutex);

                return IsEmpty_i();
        }

        return this->IsEmpty_i();
        return false;
}



template <class T, ACE_SYNCH_DECL>
bool ReplyQueue<T, ACE_SYNCH_USE>::IsEmpty_i()
{
        return m_replyQueue.empty();
}



template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::GetReply(T &theReply, const ACE_Time_Value *deadline, bool acquireMutex)
{
        if (acquireMutex)
        {
                //
                // Don't bother acquiring the mutex if the queue is already closed. Just 
                // report failure.
                if (m_closed)
                        return -1;

                ACE_Guard<ACE_SYNCH_MUTEX_T> g(m_replyQueueMutex);

                return GetReply_i(theReply, deadline);
        }

        return GetReply_i(theReply, deadline);
}



template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::GetReply_i(T &theReply, const ACE_Time_Value *deadline)
{
        //
        // Check to make sure the queue isn't closed already.
        if (m_closed)
        {
                ACE_ERROR_RETURN((LM_ERROR,
                        "(%N:%l) Attempt to get a reply from a closed queue.\n"),
                        -1);
        }

        //
        // Construct a PendingReply so we can put it in the queue.
        // Do this before acquiring the mutex so we don't make other threads
        // wait through construction.
        PENDING_REPLY replyObj(theReply, m_replyQueueMutex);
        
        //
        // Mark the PendingReply as residing on the queue, and then add it to
        // the queue.
        replyObj.m_isEnqueued = true;
        typename QUEUE_TYPE::iterator it = m_replyQueue.insert(m_replyQueue.end(), &replyObj);
        
        //
        // Wait until signaled by a SetReply() call or until the deadline passes.
        // Note that m_replyQueueMutex is relinquished during the time we are 
        // waiting.  Its re-acquired on return from wait().
        int status = replyObj.m_condition.wait(deadline);
        
        //
        // If m_isEnqueued is still set, then remove the object from the queue
        // now. Since the queue is an STL list, the previously-recorded iterator
        // is still valid.
        if (replyObj.m_isEnqueued)
        {
                m_replyQueue.erase(it);
        }
        
        //
        // If wait() failed, then this whole call failed.
        // Check with the ACE logger to see what the last error was.
        // It the failure was something other than a timeout, then log the
        // error. Timeout is a failure, but it isn't really an "error".
        if (status == -1 && ACE_Log_Msg::last_error_adapter() != ETIME)
        {
                ACE_ERROR((LM_ERROR, "(%N:%l) %p\n", "acquire"));
        }
        //
        // If the queue is closed at this point then we didn't get a value. Log a
        // warning to aid in debugging and return -1.
        else if (m_closed)
        {
                status = -1;
                ACE_ERROR((LM_WARNING,
                        "(%N:%l) GetReply() failed because reply queue was closed before a response was set.\n"));
        }

        return status;
}



template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::GetReply(T &theReply, bool acquireMutex)
{
        if (acquireMutex)
        {
                //
                // Don't bother acquiring the mutex if the queue is already closed. Just 
                // report failure.
                if (m_closed)
                        return -1;

                ACE_Guard<ACE_SYNCH_MUTEX_T> g(m_replyQueueMutex);

                return GetReply_i(theReply);
        }

        return GetReply_i(theReply);
}



template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::GetReply_i(T &theReply)
{
        //
        // Check to make sure the queue isn't closed already.
        if (m_closed)
        {
                ACE_ERROR_RETURN((LM_ERROR,
                        "(%N:%l) Attempt to get a reply from a closed queue.\n"),
                        -1);
        }

        //
        // Construct a PendingReply so we can put it in the queue.
        // Do this before acquiring the mutex so we don't make other threads
        // wait through construction.
        PENDING_REPLY replyObj(theReply, m_replyQueueMutex);
                        
        //
        // Mark the PendingReply as residing on the queue, and then add it to
        // the queue.
        replyObj.m_isEnqueued = true;
        typename QUEUE_TYPE::iterator it = m_replyQueue.insert(m_replyQueue.end(), &replyObj);
        
        //
        // Wait until signaled by a SetReply() call or until an error occurs.
        // Note that m_replyQueueMutex is relinquished during the time we are 
        // waiting.  Its re-acquired on return from wait().
        int status = replyObj.m_condition.wait();
        
        //
        // If m_isEnqueued is still set, then remove the object from the queue
        // now. Since the queue is an STL list, the previously-recorded iterator
        // is still valid.
        if (replyObj.m_isEnqueued)
        {
                m_replyQueue.erase(it);
        }
        
        //
        // If wait() failed, then this whole call failed. Log an error message.
        if (status == -1)
        {
                ACE_ERROR((LM_ERROR, "(%N:%l) %p\n", "LOCK::acquire"));
        }
        //
        // If the queue is closed at this point then we didn't get a value. Log a
        // warning to aid in debugging and return -1.
        else if (m_closed)
        {
                status = -1;
                ACE_ERROR((LM_WARNING,
                        "(%N:%l) GetReply() failed because reply queue was closed before a response was set.\n"));
        }
        
        return status;
}



template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::SetReply(T &theReply, bool acquireMutex)
{
        if (acquireMutex)
        {
                //
                // Don't bother acquiring the mutex if the queue is already closed. Just 
                // report failure.
                if (m_closed)
                        return -1;

                ACE_Guard<ACE_SYNCH_MUTEX_T> g(m_replyQueueMutex);

                return SetReply_i(theReply);
        }

        return SetReply_i(theReply);
}



template <class T, ACE_SYNCH_DECL>
int ReplyQueue<T, ACE_SYNCH_USE>::SetReply_i(T &theReply)
{
        //
        // A place to hold the status of system calls.
        int status = -1;

        //
        // If the queue is already closed the log an error and return -1.
        if (m_closed)
        {
                ACE_ERROR_RETURN((LM_ERROR,
                        "(%N:%l) Attempt to get a reply from a closed queue.\n"),
                        -1);
        }
        
        //
        // If the queue is empty, then someone called this method in error.
        // Log an error and return -1.
        if (m_replyQueue.empty())
        {
                ACE_ERROR_RETURN((LM_ERROR,
                        "(%N:%l) Attempt to set a reply when none are pending.\n"), -1);
        }
        else
        {
                //
                // Get the next PendingReply and mark it as no longer enqueued.
                PENDING_REPLY * replyObj = m_replyQueue.front();
                replyObj->m_isEnqueued = false;
                m_replyQueue.pop_front();

                //
                // Assigns the value of theReply to replyObj->m_theReply
                Assign_i(theReply, replyObj->m_theReply);
                
                //
                // Notify the waiter that their reply is ready.
                status = replyObj->m_condition.signal();
                
                //
                // If notification failed, then log a message.
                if (status == -1)
                {
                        ACE_ERROR((LM_ERROR,
                                "(%N:%l) %p\n", "ACE_Condition<>::signal"));
                }
        }
        
        return status;
}

#if 0
//
// The following class is used to test compilation. It is not intended to be used.
namespace
{
        class ReplyQueueTester
        {
                void TestCompilation()
                {
                        ReplyQueue<int, ACE_MT_SYNCH>   foo1;
                        ReplyQueue<int, ACE_NULL_SYNCH> foo2;
                        ReplyQueue<int, ACE_SYNCH>              foo3;
                        
                        
                        ACE_Thread_Mutex        &m1 = foo1.GetMutex();
                        ACE_Null_Mutex          &m2 = foo2.GetMutex();
                        ACE_SYNCH_MUTEX         &m3 = foo3.GetMutex();
                        
                        int                                     status,
                                                                arg;
                        
                        ACE_Time_Value          tv = ACE_OS::gettimeofday()
                                                                                + ACE_Time_Value(10,0);
                        
                        status = foo1.GetReply(arg, &tv);
                        status = foo2.GetReply(arg, &tv);
                        status = foo3.GetReply(arg, &tv);
                        ;
                        status = foo1.GetReply(arg);
                        status = foo2.GetReply(arg);
                        status = foo3.GetReply(arg);
                        
                        status = foo1.SetReply(status);
                        status = foo2.SetReply(status);
                        status = foo3.SetReply(status);
                }
        };
}
#endif // 0

#endif // REPLY_QUEUE_H

---