# common/include/boost/lockfree/fifo.hpp

    00001 //  lock-free fifo queue from
    00002 //  Michael, M. M. and Scott, M. L.,
    00003 //  "simple, fast and practical non-blocking and blocking concurrent queue algorithms"
    00004 //
    00005 //  implementation for c++
    00006 //
    00007 //  Copyright (C) 2008 Tim Blechmann
    00008 //
    00009 //  Distributed under the Boost Software License, Version 1.0. (See
    00010 //  accompanying file LICENSE_1_0.txt or copy at
    00011 //  http://www.boost.org/LICENSE_1_0.txt)
    00012
    00013 //  Disclaimer: Not a Boost library.
    00014
    00015 #ifndef BOOST_LOCKFREE_FIFO_HPP_INCLUDED
    00016 #define BOOST_LOCKFREE_FIFO_HPP_INCLUDED
    00017
    00018 #include <boost/lockfree/atomic_int.hpp>
    00019 #include <boost/lockfree/detail/tagged_ptr.hpp>
    00020 #include <boost/lockfree/detail/freelist.hpp>
    00021
    00022 #include <boost/static_assert.hpp>
    00023 #include <boost/type_traits/is_pod.hpp>
    00024
    00025 #include <memory>               /* std::auto_ptr */
    00026 #include <boost/scoped_ptr.hpp>
    00027 #include <boost/shared_ptr.hpp>
    00028 #include <boost/noncopyable.hpp>
    00029
    00030 namespace boost
    00031 {
    00032 namespace lockfree
    00033 {
    00034
    00035 namespace detail
    00036 {
    00037
    00038 template <typename T, typename freelist_t, typename Alloc>
    00039 class fifo:
    00040     boost::noncopyable
    00041 {
    00042     BOOST_STATIC_ASSERT(boost::is_pod<T>::value);
    00043
    00044     struct BOOST_LOCKFREE_CACHELINE_ALIGNMENT node
    00045     {
    00046         typedef tagged_ptr<node> tagged_ptr_t;
    00047
    00048         node(T const & v):
    00049             data(v)
    00050         {
    00051             next.set(NULL, next.get_tag()+1); /* increment tag to avoid ABA problem */
    00052         }
    00053
    00054         node (void):
    00055             next(NULL)
    00056         {}
    00057
    00058         tagged_ptr_t next;
    00059         T data;
    00060     };
    00061
    00062     typedef tagged_ptr<node> atomic_node_ptr;
    00063
    00064     typedef typename Alloc::template rebind<node>::other node_allocator;
    00065 /*     typedef typename select_freelist<node, node_allocator, freelist_t>::type pool_t; */
    00066
    00067     typedef typename boost::mpl::if_<boost::is_same<freelist_t, caching_freelist_t>,
    00068                                      caching_freelist<node, node_allocator>,
    00069                                      static_freelist<node, node_allocator>
    00070                                      >::type pool_t;
    00071
    00072 public:
    00073     static const bool is_lockfree = node::tagged_ptr_t::is_lockfree;
    00074
    00075     fifo(void):
    00076         pool(128)
    00077     {
    00078         node * n = alloc_node();
    00079         head_.set_ptr(n);
    00080         tail_.set_ptr(n);
    00081     }
    00082
    00083     explicit fifo(std::size_t initial_nodes):
    00084         pool(initial_nodes)
    00085     {
    00086         node * n = alloc_node();
    00087         head_.set_ptr(n);
    00088         tail_.set_ptr(n);
    00089     }
    00090
    00091     ~fifo(void)
    00092     {
    00093         assert(empty());
    00094         dealloc_node(head_.get_ptr());
    00095     }
    00096
    00097     bool empty(void)
    00098     {
    00099         return head_.get_ptr() == tail_.get_ptr();
    00100     }
    00101
    00102     bool enqueue(T const & t)
    00103     {
    00104         node * n = alloc_node(t);
    00105
    00106         if (n == NULL)
    00107             return false;
    00108
    00109         for (;;)
    00110         {
    00111             atomic_node_ptr tail (tail_);
    00112             read_memory_barrier();
    00113             atomic_node_ptr next (tail->next);
    00114
    00115             if (likely(tail == tail_))
    00116             {
    00117                 if (next.get_ptr() == 0)
    00118                 {
    00119                     if ( tail->next.cas(next, n) )
    00120                     {
    00121                         tail_.cas(tail, n);
    00122                         return true;
    00123                     }
    00124                 }
    00125                 else
    00126                     tail_.cas(tail, next.get_ptr());
    00127             }
    00128         }
    00129     }
    00130
    00131     bool dequeue (T * ret)
    00132     {
    00133         for (;;)
    00134         {
    00135             atomic_node_ptr head (head_);
    00136             read_memory_barrier();
    00137
    00138             atomic_node_ptr tail(tail_);
    00139             node * next = head->next.get_ptr();
    00140
    00141             if (likely(head == head_))
    00142             {
    00143                 if (head.get_ptr() == tail.get_ptr())
    00144                 {
    00145                     if (next == 0)
    00146                         return false;
    00147
    00148                     tail_.cas(tail, next);
    00149                 }
    00150                 else
    00151                 {
    00152                     *ret = next->data;
    00153                     if (head_.cas(head, next))
    00154                     {
    00155                         dealloc_node(head.get_ptr());
    00156
    00157                         return true;
    00158                     }
    00159                 }
    00160             }
    00161         }
    00162     }
    00163
    00164 private:
    00165     node * alloc_node(void)
    00166     {
    00167         node * chunk = pool.allocate();
    00168         new(chunk) node();
    00169         return chunk;
    00170     }
    00171
    00172     node * alloc_node(T const & t)
    00173     {
    00174         node * chunk = pool.allocate();
    00175         new(chunk) node(t);
    00176         return chunk;
    00177     }
    00178
    00179     void dealloc_node(node * n)
    00180     {
    00181         n->~node();
    00182         pool.deallocate(n);
    00183     }
    00184
    00185     atomic_node_ptr head_;
    00186     static const int padding_size = 64 - sizeof(atomic_node_ptr); /* cache lines on current cpus seem to
    00187                                                                    * be 64 byte */
    00188     char padding1[padding_size];
    00189     atomic_node_ptr tail_;
    00190     char padding2[padding_size];
    00191
    00192     pool_t pool;
    00193 };
    00194
    00195 } /* namespace detail */
    00196
    00201 template <typename T,
    00202           typename freelist_t = caching_freelist_t,
    00203           typename Alloc = std::allocator<T>
    00204           >
    00205 class fifo:
    00206     public detail::fifo<T, freelist_t, Alloc>
    00207 {
    00208 public:
    00209     fifo(void)
    00210     {}
    00211
    00212     explicit fifo(std::size_t initial_nodes):
    00213         detail::fifo<T, freelist_t, Alloc>(initial_nodes)
    00214     {}
    00215 };
    00216
    00217
    00223 template <typename T, typename freelist_t, typename Alloc>
    00224 class fifo<T*, freelist_t, Alloc>:
    00225     public detail::fifo<T*, freelist_t, Alloc>
    00226 {
    00227     typedef detail::fifo<T*, freelist_t, Alloc> fifo_t;
    00228
    00229     template <typename smart_ptr>
    00230     bool dequeue_smart_ptr(smart_ptr & ptr)
    00231     {
    00232         T * result = 0;
    00233         bool success = fifo_t::dequeue(&result);
    00234
    00235         if (success)
    00236             ptr.reset(result);
    00237         return success;
    00238     }
    00239
    00240 public:
    00241     fifo(void)
    00242     {}
    00243
    00244     explicit fifo(std::size_t initial_nodes):
    00245         fifo_t(initial_nodes)
    00246     {}
    00247
    00248     bool enqueue(T * t)
    00249     {
    00250         return fifo_t::enqueue(t);
    00251     }
    00252
    00253     bool dequeue (T ** ret)
    00254     {
    00255         return fifo_t::dequeue(ret);
    00256     }
    00257
    00258     bool dequeue (std::auto_ptr<T> & ret)
    00259     {
    00260         return dequeue_smart_ptr(ret);
    00261     }
    00262
    00263     bool dequeue (boost::scoped_ptr<T> & ret)
    00264     {
    00265         BOOST_STATIC_ASSERT(sizeof(boost::scoped_ptr<T>) == sizeof(T*));
    00266         return dequeue(reinterpret_cast<T**>(&ret));
    00267     }
    00268
    00269     bool dequeue (boost::shared_ptr<T> & ret)
    00270     {
    00271         return dequeue_smart_ptr(ret);
    00272     }
    00273 };
    00274
    00275 } /* namespace lockfree */
    00276 } /* namespace boost */
    00277
    00278
    00279 #endif /* BOOST_LOCKFREE_FIFO_HPP_INCLUDED */