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 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 <boost/atomic.hpp>
  19 #include <boost/lockfree/detail/tagged_ptr.hpp>
  20 #include <boost/lockfree/detail/freelist.hpp>
  21
  22 #include <boost/static_assert.hpp>
  23 #include <boost/type_traits/is_pod.hpp>
  24
  25 #include <memory>               /* std::auto_ptr */
  26 #include <boost/scoped_ptr.hpp>
  27 #include <boost/shared_ptr.hpp>
  28 #include <boost/noncopyable.hpp>
  29
  30 namespace boost
  31 {
  32 namespace lockfree
  33 {
  34
  35 namespace detail
  36 {
  37
  38 template <typename T, typename freelist_t, typename Alloc>
  39 class fifo:
  40     boost::noncopyable
  41 {
  42     BOOST_STATIC_ASSERT(boost::is_pod<T>::value);
  43
  44     struct BOOST_LOCKFREE_CACHELINE_ALIGNMENT node
  45     {
  46         typedef tagged_ptr<node> tagged_ptr_t;
  47
  48         node(T const & v):
  49             data(v)
  50         {
  51             /* increment tag to avoid ABA problem */
  52             tagged_ptr_t old_next = next.load(memory_order_acquire);
  53             tagged_ptr_t new_next (NULL, old_next.get_tag()+1);
  54             next.store(new_next, memory_order_release);
  55         }
  56
  57         node (void):
  58             next(tagged_ptr_t(NULL, 0))
  59         {}
  60
  61         atomic<tagged_ptr_t> next;
  62         T data;
  63     };
  64
  65     typedef tagged_ptr<node> atomic_node_ptr;
  66
  67     typedef typename Alloc::template rebind<node>::other node_allocator;
  68 /*     typedef typename select_freelist<node, node_allocator, freelist_t>::type pool_t; */
  69
  70     typedef typename boost::mpl::if_<boost::is_same<freelist_t, caching_freelist_t>,
  71                                      caching_freelist<node, node_allocator>,
  72                                      static_freelist<node, node_allocator>
  73                                      >::type pool_t;
  74
  75 public:
  76     static const bool is_lockfree = node::tagged_ptr_t::is_lockfree;
  77
  78     fifo(void):
  79         pool(128)
  80     {
  81         node * n = alloc_node();
  82         head_.set_ptr(n);
  83         tail_.set_ptr(n);
  84     }
  85
  86     explicit fifo(std::size_t initial_nodes):
  87         pool(initial_nodes)
  88     {
  89         node * n = alloc_node();
  90         head_.set_ptr(n);
  91         tail_.set_ptr(n);
  92     }
  93
  94     ~fifo(void)
  95     {
  96         if (!empty())
  97         {
  98             T dummy;
  99             for(;;)
 100             {
 101                 if (!dequeue(&dummy))
 102                     break;
 103             }
 104         }
 105         dealloc_node(head_.get_ptr());
 106     }
 107
 108     bool empty(void)
 109     {
 110         return head_.get_ptr() == tail_.get_ptr();
 111     }
 112
 113     bool enqueue(T const & t)
 114     {
 115         node * n = alloc_node(t);
 116
 117         if (n == NULL)
 118             return false;
 119
 120         for (;;)
 121         {
 122             atomic_node_ptr tail (tail_);
 123             read_memory_barrier();
 124             atomic_node_ptr next (tail->next);
 125
 126             if (likely(tail == tail_))
 127             {
 128                 if (next.get_ptr() == 0)
 129                 {
 130                     if ( tail->next.compare_exchange_strong(next, tagged_ptr<node>(n, next.get_tag() + 1)) )
 131                     {
 132                         tail_.cas(tail, n);
 133                         return true;
 134                     }
 135                 }
 136                 else
 137                     tail_.cas(tail, next.get_ptr());
 138             }
 139         }
 140     }
 141
 142     bool dequeue (T * ret)
 143     {
 144         for (;;)
 145         {
 146             atomic_node_ptr head (head_);
 147             read_memory_barrier();
 148
 149             atomic_node_ptr tail(tail_);
 150             node * next = head->next.load(memory_order_acquire).get_ptr();
 151             read_memory_barrier();
 152
 153             if (likely(head == head_))
 154             {
 155                 if (head.get_ptr() == tail.get_ptr())
 156                 {
 157                     if (next == 0)
 158                         return false;
 159
 160                     tail_.cas(tail, next);
 161                 }
 162                 else
 163                 {
 164                     *ret = next->data;
 165                     if (head_.cas(head, next))
 166                     {
 167                         dealloc_node(head.get_ptr());
 168
 169                         return true;
 170                     }
 171                 }
 172             }
 173         }
 174     }
 175
 176 private:
 177     node * alloc_node(void)
 178     {
 179         node * chunk = pool.allocate();
 180         new(chunk) node();
 181         return chunk;
 182     }
 183
 184     node * alloc_node(T const & t)
 185     {
 186         node * chunk = pool.allocate();
 187         new(chunk) node(t);
 188         return chunk;
 189     }
 190
 191     void dealloc_node(node * n)
 192     {
 193         n->~node();
 194         pool.deallocate(n);
 195     }
 196
 197     volatile atomic_node_ptr head_;
 198     static const int padding_size = 64 - sizeof(atomic_node_ptr); /* cache lines on current cpus seem to
 199                                                                    * be 64 byte */
 200     char padding1[padding_size];
 201     volatile atomic_node_ptr tail_;
 202     char padding2[padding_size];
 203
 204     pool_t pool;
 205 };
 206
 207 } /* namespace detail */
 208
 209 /** lockfree fifo
 210  *
 211  *  - wrapper for detail::fifo
 212  * */
 213 template <typename T,
 214           typename freelist_t = caching_freelist_t,
 215           typename Alloc = std::allocator<T>
 216           >
 217 class fifo:
 218     public detail::fifo<T, freelist_t, Alloc>
 219 {
 220 public:
 221     fifo(void)
 222     {}
 223
 224     explicit fifo(std::size_t initial_nodes):
 225         detail::fifo<T, freelist_t, Alloc>(initial_nodes)
 226     {}
 227 };
 228
 229
 230 /** lockfree fifo, template specialization for pointer-types
 231  *
 232  *  - wrapper for detail::fifo
 233  *  - overload dequeue to support smart pointers
 234  * */
 235 template <typename T, typename freelist_t, typename Alloc>
 236 class fifo<T*, freelist_t, Alloc>:
 237     public detail::fifo<T*, freelist_t, Alloc>
 238 {
 239     typedef detail::fifo<T*, freelist_t, Alloc> fifo_t;
 240
 241     template <typename smart_ptr>
 242     bool dequeue_smart_ptr(smart_ptr & ptr)
 243     {
 244         T * result = 0;
 245         bool success = fifo_t::dequeue(&result);
 246
 247         if (success)
 248             ptr.reset(result);
 249         return success;
 250     }
 251
 252 public:
 253     fifo(void)
 254     {}
 255
 256     explicit fifo(std::size_t initial_nodes):
 257         fifo_t(initial_nodes)
 258     {}
 259
 260     bool enqueue(T * t)
 261     {
 262         return fifo_t::enqueue(t);
 263     }
 264
 265     bool dequeue (T ** ret)
 266     {
 267         return fifo_t::dequeue(ret);
 268     }
 269
 270     bool dequeue (std::auto_ptr<T> & ret)
 271     {
 272         return dequeue_smart_ptr(ret);
 273     }
 274
 275     bool dequeue (boost::scoped_ptr<T> & ret)
 276     {
 277         BOOST_STATIC_ASSERT(sizeof(boost::scoped_ptr<T>) == sizeof(T*));
 278         return dequeue(reinterpret_cast<T**>((void*)&ret));
 279     }
 280
 281     bool dequeue (boost::shared_ptr<T> & ret)
 282     {
 283         return dequeue_smart_ptr(ret);
 284     }
 285 };
 286
 287 } /* namespace lockfree */
 288 } /* namespace boost */
 289
 290
 291 #endif /* BOOST_LOCKFREE_FIFO_HPP_INCLUDED */