expand interface, internal deque, ultra sha1

Author: abeaucha

modules/c++/mt/include/mt/RequestQueue.h

@@ -23,7 +23,7 @@
 #ifndef __MT_REQUEST_QUEUE_H__
 #define __MT_REQUEST_QUEUE_H__
 
-#include <queue>
+#include <deque>
 #include "sys/Thread.h"
 #include "sys/ConditionVar.h"
 #include "sys/Mutex.h"
@@ -51,21 +51,38 @@ namespace mt
 template<typename T>
 struct RequestQueue
 {
+public:
     //! Default constructor
     RequestQueue() :
         mAvailableSpace(&mQueueLock),
         mAvailableItems(&mQueueLock)
     {
     }
 
-    // Put a (copy of, unless T is a pointer) request on the queue
+    //! Puts the request at the front of the queue
+    void priorityEnqueue(T request)
+    {
+#ifdef THREAD_DEBUG
+        dbg_printf("Locking (enqueue)\n");
+#endif
+        mQueueLock.lock();
+        mRequestQueue.push_front(request);
+#ifdef THREAD_DEBUG
+        dbg_printf("Unlocking (enqueue), new size [%d]\n", mRequestQueue.size());
+#endif
+        mQueueLock.unlock();
+
+        mAvailableItems.signal();
+    }
+
+    //! Put a (copy of, unless T is a pointer) request on the queue
     void enqueue(T request)
     {
 #ifdef THREAD_DEBUG
         dbg_printf("Locking (enqueue)\n");
 #endif
         mQueueLock.lock();
-        mRequestQueue.push(request);
+        mRequestQueue.push_back(request);
 #ifdef THREAD_DEBUG
         dbg_printf("Unlocking (enqueue), new size [%d]\n", mRequestQueue.size());
 #endif
@@ -74,7 +91,7 @@ struct RequestQueue
         mAvailableItems.signal();
     }
 
-    // Retrieve (by reference) T from the queue. blocks until ok
+    //! Retrieve (by reference) T from the queue. blocks until ok
     void dequeue(T& request)
     {
 #ifdef THREAD_DEBUG
@@ -87,22 +104,70 @@ struct RequestQueue
         }
 
         request = mRequestQueue.front();
-        mRequestQueue.pop();
+        mRequestQueue.pop_front();
 
 #ifdef THREAD_DEBUG
         dbg_printf("Unlocking (dequeue), new size [%d]\n", mRequestQueue.size());
 #endif
         mQueueLock.unlock();
         mAvailableSpace.signal();
     }
+    
+    //! Retrieves a copy of the n'th item from the front of the queue (0 = first item) without removing it
+    T peek(size_t n = 0)
+    {
+        T request;
+#ifdef THREAD_DEBUG
+        dbg_printf("Locking (peek)\n");
+#endif
+        mQueueLock.lock();
+        if (mRequestQueue.size() > n)
+        {
+            request = mRequestQueue[n];
+        }
+        else
+        {
+            mQueueLock.unlock();
+            throw except::Exception(Ctxt("Request queue cannot peek beyond end of queue"));
+        }
+        mQueueLock.unlock();
+#ifdef THREAD_DEBUG
+        dbg_printf("Unlocking (peek)\n");
+#endif
+
+        return request;
+    }
 
-    // Check to see if its empty
+    //! Lets the n'th request from the front cut in line and dequeue
+    void cutAndDequeue(size_t n, T& request)
+    {
+#ifdef THREAD_DEBUG
+        dbg_printf("Locking (peek)\n");
+#endif
+        mQueueLock.lock();
+        if (mRequestQueue.size() > n)
+        {
+            request = mRequestQueue[n];
+            mRequestQueue.erase(mRequestQueue.begin()+n);
+        }
+        else
+        {
+            mQueueLock.unlock();
+            throw except::Exception(Ctxt("Request queue cannot access beyond end of queue"));
+        }
+        mQueueLock.unlock();
+#ifdef THREAD_DEBUG
+        dbg_printf("Unlocking (peek)\n");
+#endif
+    }
+
+    //! Check to see if its empty
     bool isEmpty() const
     {
         return mRequestQueue.empty();
     }
 
-    // Check the length
+    //! Check the length
     int length() const
     {
         return mRequestQueue.size();
@@ -116,7 +181,7 @@ struct RequestQueue
         mQueueLock.lock();
         while (!isEmpty())
         {
-            mRequestQueue.pop();
+            mRequestQueue.pop_front();
         }
 
 #ifdef THREAD_DEBUG
@@ -126,12 +191,53 @@ struct RequestQueue
         mAvailableSpace.signal();
     }
 
+    //! Aggregates ProcFunctor of all of the elements of the queue
+    template <typename ProcFunctor, typename AggregateType>
+    AggregateType aggregate(const ProcFunctor& aggregate, const AggregateType& initial)
+    {
+        mQueueLock.lock();
+        AggregateType cumulative = initial;
+        for (typename std::deque<T>::iterator iter = mRequestQueue.begin();
+             iter != mRequestQueue.end();
+             ++iter)
+        {
+            cumulative = aggregate(*iter, cumulative);
+        }
+
+        mQueueLock.unlock();
+        return cumulative;
+    }
+
+    //! Remove the given request from the queue
+    // Does nothing if the given request is not in the queue
+    // \return true if an item was removed, false otherwise
+    template <typename CmpFunctor>
+    bool removeRequest(const CmpFunctor& compare)
+    {
+        mQueueLock.lock();
+        for (typename std::deque<T>::iterator iter = mRequestQueue.begin();
+             iter != mRequestQueue.end();
+             ++iter)
+        {
+            if (compare(*iter))
+            {
+                mRequestQueue.erase(iter);
+                mQueueLock.unlock();
+                mAvailableSpace.signal();
+                return true;
+            }
+        }
+        mQueueLock.unlock();
+        return false;
+    }
+
+private:
     RequestQueue(const RequestQueue&) = delete;
     RequestQueue& operator=(const RequestQueue&) = delete;
 
 private:
     //! The internal data structure
-    std::queue<T> mRequestQueue;
+    std::deque<T> mRequestQueue;
     //! The synchronizer
     sys::Mutex mQueueLock;
     //! This condition is "is there space?"