1 2 // Copyright Oliver Kowalke 2017. 3 // Distributed under the Boost Software License, Version 1.0. 4 // (See accompanying file LICENSE_1_0.txt or copy at 5 // http://www.boost.org/LICENSE_1_0.txt) 6 // 7 8 #ifndef BOOST_FIBERS_NUMA_ALGO_WORK_STEALING_H 9 #define BOOST_FIBERS_NUMA_ALGO_WORK_STEALING_H 10 11 #include <condition_variable> 12 #include <chrono> 13 #include <cstddef> 14 #include <cstdint> 15 #include <mutex> 16 #include <vector> 17 18 #include <boost/config.hpp> 19 #include <boost/intrusive_ptr.hpp> 20 21 #include <boost/fiber/algo/algorithm.hpp> 22 #include <boost/fiber/context.hpp> 23 #include <boost/fiber/detail/config.hpp> 24 #include <boost/fiber/detail/context_spinlock_queue.hpp> 25 #include <boost/fiber/detail/context_spmc_queue.hpp> 26 #include <boost/fiber/numa/pin_thread.hpp> 27 #include <boost/fiber/numa/topology.hpp> 28 #include <boost/fiber/scheduler.hpp> 29 30 #ifdef BOOST_HAS_ABI_HEADERS 31 # include BOOST_ABI_PREFIX 32 #endif 33 34 namespace boost { 35 namespace fibers { 36 namespace numa { 37 namespace algo { 38 39 class BOOST_FIBERS_DECL work_stealing : public boost::fibers::algo::algorithm { 40 private: 41 static std::vector< intrusive_ptr< work_stealing > > schedulers_; 42 43 std::uint32_t cpu_id_; 44 std::vector< std::uint32_t > local_cpus_; 45 std::vector< std::uint32_t > remote_cpus_; 46 #ifdef BOOST_FIBERS_USE_SPMC_QUEUE 47 detail::context_spmc_queue rqueue_{}; 48 #else 49 detail::context_spinlock_queue rqueue_{}; 50 #endif 51 std::mutex mtx_{}; 52 std::condition_variable cnd_{}; 53 bool flag_{ false }; 54 bool suspend_; 55 56 static void init_( std::vector< boost::fibers::numa::node > const&, 57 std::vector< intrusive_ptr< work_stealing > > &); 58 59 public: 60 work_stealing( std::uint32_t, std::uint32_t, 61 std::vector< boost::fibers::numa::node > const&, 62 bool = false); 63 64 work_stealing( work_stealing const&) = delete; 65 work_stealing( work_stealing &&) = delete; 66 67 work_stealing & operator=( work_stealing const&) = delete; 68 work_stealing & operator=( work_stealing &&) = delete; 69 70 virtual void awakened( context *) noexcept; 71 72 virtual context * pick_next() noexcept; 73 steal()74 virtual context * steal() noexcept { 75 return rqueue_.steal(); 76 } 77 has_ready_fibers() const78 virtual bool has_ready_fibers() const noexcept { 79 return ! rqueue_.empty(); 80 } 81 82 virtual void suspend_until( std::chrono::steady_clock::time_point const&) noexcept; 83 84 virtual void notify() noexcept; 85 }; 86 87 }}}} 88 89 #ifdef BOOST_HAS_ABI_HEADERS 90 # include BOOST_ABI_SUFFIX 91 #endif 92 93 #endif // BOOST_FIBERS_NUMA_ALGO_WORK_STEALING_H 94