boost/fiber/detail/context_mpsc_queue.hpp

   1
   2 //          Copyright Dmitry Vyukov 2010-2011.
   3 //          Copyright Oliver Kowalke 2016.
   4 // Distributed under the Boost Software License, Version 1.0.
   5 //    (See accompanying file LICENSE_1_0.txt or copy at
   6 //          http://www.boost.org/LICENSE_1_0.txt)
   7 //
   8 // based on Dmitry Vyukov's intrusive MPSC queue
   9 // http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue
  10 // https://groups.google.com/forum/#!topic/lock-free/aFHvZhu1G-0
  11
  12 #ifndef BOOST_FIBERS_DETAIL_CONTEXT_MPSC_QUEUE_H
  13 #define BOOST_FIBERS_DETAIL_CONTEXT_MPSC_QUEUE_H
  14
  15 #include <atomic>
  16 #include <memory>
  17 #include <type_traits>
  18
  19 #include <boost/assert.hpp>
  20 #include <boost/config.hpp>
  21
  22 #include <boost/fiber/context.hpp>
  23 #include <boost/fiber/detail/config.hpp>
  24
  25 #ifdef BOOST_HAS_ABI_HEADERS
  26 #  include BOOST_ABI_PREFIX
  27 #endif
  28
  29 namespace boost {
  30 namespace fibers {
  31 namespace detail {
  32
  33 // a MPSC queue
  34 // multiple threads push ready fibers (belonging to local scheduler)
  35 // (thread) local scheduler pops fibers
  36 class context_mpsc_queue {
  37 private:
  38     // not default constructor for context - use aligned_storage instead
  39     alignas(cache_alignment) std::aligned_storage< sizeof( context), alignof( context) >::type  storage_{};
  40     context                                         *   dummy_;
  41     alignas(cache_alignment) std::atomic< context * >   head_;
  42     alignas(cache_alignment) context                *   tail_;
  43     char                                                pad_[cacheline_length];
  44
  45 public:
  46     context_mpsc_queue() :
  47         dummy_{ reinterpret_cast< context * >( std::addressof( storage_) ) },
  48         head_{ dummy_ },
  49         tail_{ dummy_ } {
  50         dummy_->remote_nxt_.store( nullptr, std::memory_order_release);
  51     }
  52
  53     context_mpsc_queue( context_mpsc_queue const&) = delete;
  54     context_mpsc_queue & operator=( context_mpsc_queue const&) = delete;
  55
  56     void push( context * ctx) noexcept {
  57         BOOST_ASSERT( nullptr != ctx);
  58         ctx->remote_nxt_.store( nullptr, std::memory_order_release);
  59         context * prev = head_.exchange( ctx, std::memory_order_acq_rel);
  60         prev->remote_nxt_.store( ctx, std::memory_order_release);
  61     }
  62
  63     context * pop() noexcept {
  64         context * tail = tail_;
  65         context * next = tail->remote_nxt_.load( std::memory_order_acquire);
  66         if ( dummy_ == tail) {
  67             if ( nullptr == next) {
  68                 return nullptr;
  69             }
  70             tail_ = next;
  71             tail = next;
  72             next = next->remote_nxt_.load( std::memory_order_acquire);;
  73         }
  74         if ( nullptr != next) {
  75             tail_ = next;
  76             return tail;
  77         }
  78         context * head = head_.load( std::memory_order_acquire);
  79         if ( tail != head) {
  80             return nullptr;
  81         }
  82         push( dummy_);
  83         next = tail->remote_nxt_.load( std::memory_order_acquire);
  84         if ( nullptr != next) {
  85             tail_= next;
  86             return tail;
  87         }
  88         return nullptr;
  89     }
  90 };
  91
  92 }}}
  93
  94 #ifdef BOOST_HAS_ABI_HEADERS
  95 #  include BOOST_ABI_SUFFIX
  96 #endif
  97
  98 #endif // BOOST_FIBERS_DETAIL_CONTEXT_MPSC_QUEUE_H