boost/lockfree/fifo.hpp

   1 //  lock-free fifo queue from
   2 //  Michael, M. M. and Scott, M. L.,
   3 //  "simple, fast and practical non-blocking and blocking concurrent queue algorithms"
   4 //
   5 //  implementation for c++
   6 //
   7 //  Copyright (C) 2008, 2009, 2010, 2011 Tim Blechmann
   8 //
   9 //  Distributed under the Boost Software License, Version 1.0. (See
  10 //  accompanying file LICENSE_1_0.txt or copy at
  11 //  http://www.boost.org/LICENSE_1_0.txt)
  12
  13 //  Disclaimer: Not a Boost library.
  14
  15 #ifndef BOOST_LOCKFREE_FIFO_HPP_INCLUDED
  16 #define BOOST_LOCKFREE_FIFO_HPP_INCLUDED
  17
  18 #include <memory>               /* std::auto_ptr */
  19
  20 #include <boost/noncopyable.hpp>
  21 #include <boost/scoped_ptr.hpp>
  22 #include <boost/shared_ptr.hpp>
  23 #include <boost/static_assert.hpp>
  24 #include <boost/type_traits/has_trivial_assign.hpp>
  25
  26 #include <boost/lockfree/detail/atomic.hpp>
  27 #include <boost/lockfree/detail/tagged_ptr.hpp>
  28 #include <boost/lockfree/detail/freelist.hpp>
  29
  30 namespace boost {
  31 namespace lockfree {
  32 namespace detail {
  33
  34 template <typename T, typename freelist_t, typename Alloc>
  35 class fifo:
  36     boost::noncopyable
  37 {
  38 private:
  39 #ifndef BOOST_DOXYGEN_INVOKED
  40     BOOST_STATIC_ASSERT(boost::is_pod<T>::value);
  41
  42     struct BOOST_LOCKFREE_CACHELINE_ALIGNMENT node
  43     {
  44         typedef tagged_ptr<node> tagged_node_ptr;
  45
  46         node(T const & v):
  47             data(v)
  48         {
  49             /* increment tag to avoid ABA problem */
  50             tagged_node_ptr old_next = next.load(memory_order_relaxed);
  51             tagged_node_ptr new_next (NULL, old_next.get_tag()+1);
  52             next.store(new_next, memory_order_release);
  53         }
  54
  55         node (void):
  56             next(tagged_node_ptr(NULL, 0))
  57         {}
  58
  59         atomic<tagged_node_ptr> next;
  60         T data;
  61     };
  62
  63     typedef tagged_ptr<node> tagged_node_ptr;
  64
  65     typedef typename Alloc::template rebind<node>::other node_allocator;
  66
  67     typedef typename boost::mpl::if_<boost::is_same<freelist_t, caching_freelist_t>,
  68                                      detail::freelist_stack<node, true, node_allocator>,
  69                                      detail::freelist_stack<node, false, node_allocator>
  70                                      >::type pool_t;
  71
  72     void initialize(void)
  73     {
  74         node * n = pool.construct();
  75         tagged_node_ptr dummy_node(n, 0);
  76         head_.store(dummy_node, memory_order_relaxed);
  77         tail_.store(dummy_node, memory_order_release);
  78     }
  79 #endif
  80
  81 public:
  82     /**
  83      * \return true, if implementation is lock-free.
  84      *
  85      * \warning \b Warning: It only checks, if the fifo head node is lockfree. On most platforms, the whole implementation is
  86      *                      lockfree, if this is true. Using c++0x-style atomics, there is no possibility to provide a completely
  87      *                      accurate implementation, because one would need to test every internal node, which is impossible
  88      *                      if further nodes will be allocated from the operating system.
  89      * */
  90     bool is_lock_free (void) const
  91     {
  92         return head_.is_lock_free() && pool.is_lock_free();
  93     }
  94
  95     //! Construct fifo.
  96     fifo(void)
  97     {
  98         pool.reserve(1);
  99         initialize();
 100     }
 101
 102     //! Construct fifo, allocate n nodes for the freelist.
 103     explicit fifo(std::size_t n)
 104     {
 105         pool.reserve(n+1);
 106         initialize();
 107     }
 108
 109     //! Allocate n nodes for freelist.
 110     void reserve(std::size_t n)
 111     {
 112         pool.reserve(n);
 113     }
 114
 115     /** Destroys fifo, free all nodes from freelist.
 116      * */
 117     ~fifo(void)
 118     {
 119         if (!empty()) {
 120             T dummy;
 121             while(dequeue(dummy))
 122                 ;
 123         }
 124         pool.destruct(head_.load(memory_order_relaxed).get_ptr());
 125     }
 126
 127     /** Check if the ringbuffer is empty
 128      *
 129      * \warning Not thread-safe, use for debugging purposes only
 130      * */
 131     bool empty(void)
 132     {
 133         return head_.load().get_ptr() == tail_.load().get_ptr();
 134     }
 135
 136     /** Enqueues object t to the fifo. Enqueueing may fail, if the freelist is not able to allocate a new fifo node.
 137      *
 138      * \returns true, if the enqueue operation is successful.
 139      *
 140      * \note Thread-safe and non-blocking
 141      * \warning \b Warning: May block if node needs to be allocated from the operating system
 142      * */
 143     bool enqueue(T const & t)
 144     {
 145         node * n = pool.construct(t);
 146
 147         if (n == NULL)
 148             return false;
 149
 150         for (;;) {
 151             tagged_node_ptr tail = tail_.load(memory_order_acquire);
 152             tagged_node_ptr next = tail->next.load(memory_order_acquire);
 153             node * next_ptr = next.get_ptr();
 154
 155             tagged_node_ptr tail2 = tail_.load(memory_order_acquire);
 156             if (likely(tail == tail2)) {
 157                 if (next_ptr == 0) {
 158                     if ( tail->next.compare_exchange_weak(next, tagged_node_ptr(n, next.get_tag() + 1)) ) {
 159                         tail_.compare_exchange_strong(tail, tagged_node_ptr(n, tail.get_tag() + 1));
 160                         return true;
 161                     }
 162                 }
 163                 else
 164                     tail_.compare_exchange_strong(tail, tagged_node_ptr(next_ptr, tail.get_tag() + 1));
 165             }
 166         }
 167     }
 168
 169     /** Dequeue object from fifo.
 170      *
 171      * if dequeue operation is successful, object is written to memory location denoted by ret.
 172      *
 173      * \returns true, if the dequeue operation is successful, false if fifo was empty.
 174      *
 175      * \note Thread-safe and non-blocking
 176      *
 177      * */
 178     bool dequeue (T & ret)
 179     {
 180         for (;;) {
 181             tagged_node_ptr head = head_.load(memory_order_acquire);
 182             tagged_node_ptr tail = tail_.load(memory_order_acquire);
 183             tagged_node_ptr next = head->next.load(memory_order_acquire);
 184             node * next_ptr = next.get_ptr();
 185
 186             tagged_node_ptr head2 = head_.load(memory_order_acquire);
 187             if (likely(head == head2)) {
 188                 if (head.get_ptr() == tail.get_ptr()) {
 189                     if (next_ptr == 0)
 190                         return false;
 191                     tail_.compare_exchange_strong(tail, tagged_node_ptr(next_ptr, tail.get_tag() + 1));
 192                 } else {
 193                     if (next_ptr == 0)
 194                         /* this check is not part of the original algorithm as published by michael and scott
 195                          *
 196                          * however we reuse the tagged_ptr part for the and clear the next part during node
 197                          * allocation. we can observe a null-pointer here.
 198                          * */
 199                         continue;
 200                     ret = next_ptr->data;
 201                     if (head_.compare_exchange_weak(head, tagged_node_ptr(next_ptr, head.get_tag() + 1))) {
 202                         pool.destruct(head.get_ptr());
 203                         return true;
 204                     }
 205                 }
 206             }
 207         }
 208     }
 209
 210 private:
 211 #ifndef BOOST_DOXYGEN_INVOKED
 212     atomic<tagged_node_ptr> head_;
 213     static const int padding_size = BOOST_LOCKFREE_CACHELINE_BYTES - sizeof(tagged_node_ptr);
 214     char padding1[padding_size];
 215     atomic<tagged_node_ptr> tail_;
 216     char padding2[padding_size];
 217
 218     pool_t pool;
 219 #endif
 220 };
 221
 222 } /* namespace detail */
 223
 224 /** The fifo class provides a multi-writer/multi-reader fifo, enqueueing and dequeueing is lockfree,
 225  *  construction/destruction has to be synchronized. It uses a freelist for memory management,
 226  *  freed nodes are pushed to the freelist and not returned to the os before the fifo is destroyed.
 227  *
 228  *  The memory management of the fifo can be controlled via its freelist_t template argument. Two different
 229  *  freelists can be used. struct caching_freelist_t selects a caching freelist, which can allocate more nodes
 230  *  from the operating system, and struct static_freelist_t uses a fixed-sized freelist. With a fixed-sized
 231  *  freelist, the enqueue operation may fail, while with a caching freelist, the enqueue operation may block.
 232  *
 233  *  \b Limitation: The class T is required to have a trivial assignment operator.
 234  *
 235  * */
 236 template <typename T,
 237           typename freelist_t = caching_freelist_t,
 238           typename Alloc = std::allocator<T>
 239          >
 240 class fifo:
 241     public detail::fifo<T, freelist_t, Alloc>
 242 {
 243     BOOST_STATIC_ASSERT(boost::has_trivial_assign<T>::value);
 244
 245 public:
 246     //! Construct fifo.
 247     fifo(void)
 248     {}
 249
 250     //! Construct fifo, allocate n nodes for the freelist.
 251     explicit fifo(std::size_t n):
 252         detail::fifo<T, freelist_t, Alloc>(n)
 253     {}
 254 };
 255
 256
 257 /** Template specialization of the fifo class for pointer arguments, that supports dequeue operations to
 258  *  stl/boost-style smart pointers
 259  *
 260  * */
 261 template <typename T,
 262           typename freelist_t,
 263           typename Alloc
 264          >
 265 class fifo<T*, freelist_t, Alloc>:
 266     public detail::fifo<T*, freelist_t, Alloc>
 267 {
 268 #ifndef BOOST_DOXYGEN_INVOKED
 269     typedef detail::fifo<T*, freelist_t, Alloc> fifo_t;
 270
 271     template <typename smart_ptr>
 272     bool dequeue_smart_ptr(smart_ptr & ptr)
 273     {
 274         T * result = 0;
 275         bool success = fifo_t::dequeue(result);
 276
 277         if (success)
 278             ptr.reset(result);
 279         return success;
 280     }
 281 #endif
 282
 283 public:
 284     //! Construct fifo.
 285     fifo(void)
 286     {}
 287
 288     //! Construct fifo, allocate n nodes for the freelist.
 289     explicit fifo(std::size_t n):
 290         fifo_t(n)
 291     {}
 292
 293     //! \copydoc detail::fifo::dequeue
 294     bool dequeue (T * & ret)
 295     {
 296         return fifo_t::dequeue(ret);
 297     }
 298
 299     /** Dequeue object from fifo to std::auto_ptr
 300      *
 301      * if dequeue operation is successful, object is written to memory location denoted by ret.
 302      *
 303      * \returns true, if the dequeue operation is successful, false if fifo was empty.
 304      *
 305      * \note Thread-safe and non-blocking
 306      *
 307      * */
 308     bool dequeue (std::auto_ptr<T> & ret)
 309     {
 310         return dequeue_smart_ptr(ret);
 311     }
 312
 313     /** Dequeue object from fifo to boost::scoped_ptr
 314      *
 315      * if dequeue operation is successful, object is written to memory location denoted by ret.
 316      *
 317      * \returns true, if the dequeue operation is successful, false if fifo was empty.
 318      *
 319      * \note Thread-safe and non-blocking
 320      *
 321      * */
 322     bool dequeue (boost::scoped_ptr<T> & ret)
 323     {
 324         BOOST_STATIC_ASSERT(sizeof(boost::scoped_ptr<T>) == sizeof(T*));
 325         return dequeue(reinterpret_cast<T*&>(ret));
 326     }
 327
 328     /** Dequeue object from fifo to boost::shared_ptr
 329      *
 330      * if dequeue operation is successful, object is written to memory location denoted by ret.
 331      *
 332      * \returns true, if the dequeue operation is successful, false if fifo was empty.
 333      *
 334      * \note Thread-safe and non-blocking
 335      *
 336      * */
 337     bool dequeue (boost::shared_ptr<T> & ret)
 338     {
 339         return dequeue_smart_ptr(ret);
 340     }
 341 };
 342
 343 } /* namespace lockfree */
 344 } /* namespace boost */
 345
 346 #endif /* BOOST_LOCKFREE_FIFO_HPP_INCLUDED */