common/include/pion/PionLockedQueue.hpp

00001 // -----------------------------------------------------------------------
00002 // pion-common: a collection of common libraries used by the Pion Platform
00003 // -----------------------------------------------------------------------
00004 // Copyright (C) 2007-2008 Atomic Labs, Inc.  (http://www.atomiclabs.com)
00005 //
00006 // Distributed under the Boost Software License, Version 1.0.
00007 // See http://www.boost.org/LICENSE_1_0.txt
00008 //
00009 
00010 #ifndef __PION_PIONLOCKEDQUEUE_HEADER__
00011 #define __PION_PIONLOCKEDQUEUE_HEADER__
00012 
00013 #include <new>
00014 #include <boost/cstdint.hpp>
00015 #include <boost/noncopyable.hpp>
00016 #include <boost/thread/thread.hpp>
00017 #include <boost/thread/mutex.hpp>
00018 #include <boost/thread/condition.hpp>
00019 #include <boost/detail/atomic_count.hpp>
00020 #include <pion/PionConfig.hpp>
00021 #include <pion/PionException.hpp>
00022 #if defined(PION_HAVE_LOCKFREE) && !defined(_MSC_VER)
00023     #include <boost/lockfree/detail/freelist.hpp>
00024 #endif
00025 
00026 
00027 // NOTE: the data structures contained in this file are based upon algorithms
00028 // published in the paper "Simple, Fast, and Practical Non-Blocking and Blocking
00029 // Concurrent Queue Algorithms" (1996, Maged M. Michael and Michael L. Scott,
00030 // Department of Computer Science, University of Rochester).
00031 // See http://www.cs.rochester.edu/u/scott/papers/1996_PODC_queues.pdf
00032 
00033 
00034 namespace pion {    // begin namespace pion
00035 
00036 
00040 template <typename T,
00041     boost::uint32_t MaxSize = 250000,
00042     boost::uint32_t SleepMilliSec = 10 >
00043 class PionLockedQueue :
00044     private boost::noncopyable
00045 {
00046 protected:
00047 
00049     struct QueueNode {
00050         T       data;       //< data wrapped by the node item
00051         QueueNode * next;       //< points to the next node in the queue
00052         boost::uint32_t version;    //< the node item's version number
00053     };
00054 
00056     inline QueueNode *createNode(void) {
00057 #if defined(PION_HAVE_LOCKFREE) && !defined(_MSC_VER)
00058         return new (m_free_list.allocate()) QueueNode();
00059 #else
00060         return new QueueNode();
00061 #endif
00062     }
00063 
00065     inline void destroyNode(QueueNode *node_ptr) {
00066 #if defined(PION_HAVE_LOCKFREE) && !defined(_MSC_VER)
00067         node_ptr->~QueueNode();
00068         m_free_list.deallocate(node_ptr);
00069 #else
00070         delete node_ptr;
00071 #endif
00072     }
00073     
00082     inline bool dequeue(T& t, boost::uint32_t& version) {
00083         // just return if the list is empty
00084         boost::mutex::scoped_lock head_lock(m_head_mutex);
00085         QueueNode *new_head_ptr = m_head_ptr->next;
00086         if (! new_head_ptr) {
00087             version = m_head_ptr->version;
00088             return false;
00089         }
00090 
00091         // get a copy of the item at the head of the list
00092         version = new_head_ptr->version;
00093         t = new_head_ptr->data;
00094 
00095         // update the pointer to the head of the list
00096         QueueNode *old_head_ptr = m_head_ptr;
00097         m_head_ptr = new_head_ptr;
00098         head_lock.unlock();
00099 
00100         // free the QueueNode for the old head of the list
00101         destroyNode(old_head_ptr);
00102 
00103         // decrement size
00104         --m_size;
00105 
00106         // item successfully dequeued
00107         return true;
00108     }
00109 
00110 
00111 public:
00112 
00114     class ConsumerThread {
00115     public:
00116 
00121         ConsumerThread(void) : m_is_running(true), m_next_ptr(NULL),
00122             m_wakeup_time(boost::posix_time::not_a_date_time) {}
00123 
00130         template <typename DurationType>
00131         ConsumerThread(const DurationType& d)
00132             : m_is_running(true), m_next_ptr(NULL), m_wakeup_time(d)
00133         {}
00134 
00136         inline bool isRunning(void) const { return m_is_running; }
00137 
00139         inline void stop(void) { m_is_running = false; m_wakeup_event.notify_one(); }
00140 
00142         inline void reset(void) { m_is_running = true; m_next_ptr = NULL; }
00143         
00145         inline bool hasWakeupTimer(void) const { return !m_wakeup_time.is_not_a_date_time(); }
00146         
00148         inline const boost::posix_time::time_duration& getWakeupTimer(void) const {
00149             return m_wakeup_time;
00150         }
00151 
00152     private:
00153 
00155         friend class PionLockedQueue;
00156 
00157         volatile bool       m_is_running;       //< true while the thread is running/active
00158         ConsumerThread *    m_next_ptr;     //< pointer to the next idle thread
00159         boost::condition    m_wakeup_event; //< triggered when a new item is available
00160         boost::posix_time::time_duration    m_wakeup_time;  //< inactivity wakeup timer duration
00161     };
00162 
00163 
00165     PionLockedQueue(void)
00166         : m_head_ptr(NULL), m_tail_ptr(NULL), m_idle_ptr(NULL),
00167         m_next_version(1), m_size(0)
00168     {
00169         // initialize with a dummy node since m_head_ptr is always 
00170         // pointing to the item before the head of the list
00171         m_head_ptr = m_tail_ptr = createNode();
00172         m_head_ptr->next = NULL;
00173         m_head_ptr->version = 0;
00174     }
00175     
00177     virtual ~PionLockedQueue() {
00178         clear();
00179         destroyNode(m_tail_ptr);
00180     }
00181     
00183     inline bool empty(void) const { return (m_head_ptr->next == NULL); }
00184     
00186     std::size_t size(void) const {
00187         return m_size;
00188     }
00189     
00191     void clear(void) {
00192         boost::mutex::scoped_lock tail_lock(m_tail_mutex);
00193         boost::mutex::scoped_lock head_lock(m_head_mutex);
00194         while (m_head_ptr->next) {
00195             m_tail_ptr = m_head_ptr;
00196             m_head_ptr = m_head_ptr->next;
00197             destroyNode(m_tail_ptr);
00198             --m_size;
00199         }
00200         m_tail_ptr = m_head_ptr;
00201     }
00202 
00208     void push(const T& t) {
00209         // sleep while MaxSize is exceeded
00210         if (MaxSize > 0) {
00211             boost::system_time wakeup_time;
00212             while (size() >= MaxSize) {
00213                 wakeup_time = boost::get_system_time()
00214                     + boost::posix_time::millisec(SleepMilliSec);
00215                 boost::thread::sleep(wakeup_time);
00216             }
00217         }
00218 
00219         // create a new list node for the queue item
00220         QueueNode *node_ptr = createNode();
00221         node_ptr->data = t;
00222         node_ptr->next = NULL;
00223         node_ptr->version = 0;
00224 
00225         // append node to the end of the list
00226         boost::mutex::scoped_lock tail_lock(m_tail_mutex);
00227         node_ptr->version = (m_next_version += 2);
00228         m_tail_ptr->next = node_ptr;
00229 
00230         // update the tail pointer for the new node
00231         m_tail_ptr = node_ptr;
00232 
00233         // increment size
00234         ++m_size;
00235 
00236         // wake up an idle thread (if any)
00237         if (m_idle_ptr) {
00238             ConsumerThread *idle_ptr = m_idle_ptr;
00239             m_idle_ptr = m_idle_ptr->m_next_ptr;
00240             idle_ptr->m_wakeup_event.notify_one();
00241         }
00242     }
00243     
00254     bool pop(T& t, ConsumerThread& thread_info) {
00255         boost::uint32_t last_known_version;
00256         while (thread_info.isRunning()) {
00257             // try to get the next value
00258             if ( dequeue(t, last_known_version) )
00259                 return true;    // got an item
00260 
00261             // queue is empty
00262             boost::mutex::scoped_lock tail_lock(m_tail_mutex);
00263             if (m_tail_ptr->version == last_known_version) {
00264                 // still empty after acquiring lock
00265                 thread_info.m_next_ptr = m_idle_ptr;
00266                 m_idle_ptr = & thread_info;
00267                 // get wakeup time (if any)
00268                 if (thread_info.hasWakeupTimer()) {
00269                     // wait for an item to become available
00270                     const boost::posix_time::ptime wakeup_time(boost::get_system_time() + thread_info.getWakeupTimer());
00271                     if (!thread_info.m_wakeup_event.timed_wait(tail_lock, wakeup_time))
00272                         return false;   // timer expired if timed_wait() returns false
00273                 } else {
00274                     // wait for an item to become available
00275                     thread_info.m_wakeup_event.wait(tail_lock);
00276                 }
00277             }
00278         }
00279         return false;
00280     }
00281 
00289     inline bool pop(T& t) { boost::uint32_t version; return dequeue(t, version); }
00290     
00291 
00292 private:
00293 
00294 #if defined(PION_HAVE_LOCKFREE) && !defined(_MSC_VER)
00296     boost::lockfree::caching_freelist<QueueNode>    m_free_list;
00297 #endif
00298     
00300     boost::mutex                    m_head_mutex;
00301 
00303     boost::mutex                    m_tail_mutex;
00304 
00306     QueueNode *                     m_head_ptr;
00307 
00309     QueueNode *                     m_tail_ptr;
00310 
00312     ConsumerThread *                m_idle_ptr;
00313 
00315     boost::uint32_t                 m_next_version;
00316     
00318     boost::detail::atomic_count     m_size;
00319 };
00320 
00321     
00322 }   // end namespace pion
00323 
00324 #endif

Generated on Tue Mar 8 13:42:14 2011 for pion-net by  doxygen 1.4.7