ceph/src/boost/libs/asio/example/cpp11/executors/priority_scheduler.cpp

   1 #include <boost/asio/dispatch.hpp>
   2 #include <boost/asio/execution_context.hpp>
   3 #include <condition_variable>
   4 #include <iostream>
   5 #include <memory>
   6 #include <mutex>
   7 #include <queue>
   8
   9 using boost::asio::dispatch;
  10 using boost::asio::execution_context;
  11
  12 class priority_scheduler : public execution_context
  13 {
  14 public:
  15   // A class that satisfies the Executor requirements.
  16   class executor_type
  17   {
  18   public:
  19     executor_type(priority_scheduler& ctx, int pri) noexcept
  20       : context_(ctx), priority_(pri)
  21     {
  22     }
  23
  24     priority_scheduler& context() const noexcept
  25     {
  26       return context_;
  27     }
  28
  29     void on_work_started() const noexcept
  30     {
  31       // This executor doesn't count work. Instead, the scheduler simply runs
  32       // until explicitly stopped.
  33     }
  34
  35     void on_work_finished() const noexcept
  36     {
  37       // This executor doesn't count work. Instead, the scheduler simply runs
  38       // until explicitly stopped.
  39     }
  40
  41     template <class Func, class Alloc>
  42     void dispatch(Func&& f, const Alloc& a) const
  43     {
  44       post(std::forward<Func>(f), a);
  45     }
  46
  47     template <class Func, class Alloc>
  48     void post(Func f, const Alloc& a) const
  49     {
  50       auto p(std::allocate_shared<item<Func>>(
  51             typename std::allocator_traits<
  52               Alloc>::template rebind_alloc<char>(a),
  53             priority_, std::move(f)));
  54       std::lock_guard<std::mutex> lock(context_.mutex_);
  55       context_.queue_.push(p);
  56       context_.condition_.notify_one();
  57     }
  58
  59     template <class Func, class Alloc>
  60     void defer(Func&& f, const Alloc& a) const
  61     {
  62       post(std::forward<Func>(f), a);
  63     }
  64
  65     friend bool operator==(const executor_type& a,
  66         const executor_type& b) noexcept
  67     {
  68       return &a.context_ == &b.context_;
  69     }
  70
  71     friend bool operator!=(const executor_type& a,
  72         const executor_type& b) noexcept
  73     {
  74       return &a.context_ != &b.context_;
  75     }
  76
  77   private:
  78     priority_scheduler& context_;
  79     int priority_;
  80   };
  81
  82   ~priority_scheduler() noexcept
  83   {
  84     shutdown();
  85     destroy();
  86   }
  87
  88   executor_type get_executor(int pri = 0) noexcept
  89   {
  90     return executor_type(*const_cast<priority_scheduler*>(this), pri);
  91   }
  92
  93   void run()
  94   {
  95     std::unique_lock<std::mutex> lock(mutex_);
  96     for (;;)
  97     {
  98       condition_.wait(lock, [&]{ return stopped_ || !queue_.empty(); });
  99       if (stopped_)
 100         return;
 101       auto p(queue_.top());
 102       queue_.pop();
 103       lock.unlock();
 104       p->execute_(p);
 105       lock.lock();
 106     }
 107   }
 108
 109   void stop()
 110   {
 111     std::lock_guard<std::mutex> lock(mutex_);
 112     stopped_ = true;
 113     condition_.notify_all();
 114   }
 115
 116 private:
 117   struct item_base
 118   {
 119     int priority_;
 120     void (*execute_)(std::shared_ptr<item_base>&);
 121   };
 122
 123   template <class Func>
 124   struct item : item_base
 125   {
 126     item(int pri, Func f) : function_(std::move(f))
 127     {
 128       priority_ = pri;
 129       execute_ = [](std::shared_ptr<item_base>& p)
 130       {
 131         Func tmp(std::move(static_cast<item*>(p.get())->function_));
 132         p.reset();
 133         tmp();
 134       };
 135     }
 136
 137     Func function_;
 138   };
 139
 140   struct item_comp
 141   {
 142     bool operator()(
 143         const std::shared_ptr<item_base>& a,
 144         const std::shared_ptr<item_base>& b)
 145     {
 146       return a->priority_ < b->priority_;
 147     }
 148   };
 149
 150   std::mutex mutex_;
 151   std::condition_variable condition_;
 152   std::priority_queue<
 153     std::shared_ptr<item_base>,
 154     std::vector<std::shared_ptr<item_base>>,
 155     item_comp> queue_;
 156   bool stopped_ = false;
 157 };
 158
 159 int main()
 160 {
 161   priority_scheduler sched;
 162   auto low = sched.get_executor(0);
 163   auto med = sched.get_executor(1);
 164   auto high = sched.get_executor(2);
 165   dispatch(low, []{ std::cout << "1\n"; });
 166   dispatch(low, []{ std::cout << "11\n"; });
 167   dispatch(med, []{ std::cout << "2\n"; });
 168   dispatch(med, []{ std::cout << "22\n"; });
 169   dispatch(high, []{ std::cout << "3\n"; });
 170   dispatch(high, []{ std::cout << "33\n"; });
 171   dispatch(high, []{ std::cout << "333\n"; });
 172   dispatch(sched.get_executor(-1), [&]{ sched.stop(); });
 173   sched.run();
 174 }