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   executor_type get_executor(int pri = 0) noexcept
  83   {
  84     return executor_type(*const_cast<priority_scheduler*>(this), pri);
  85   }
  86
  87   void run()
  88   {
  89     std::unique_lock<std::mutex> lock(mutex_);
  90     for (;;)
  91     {
  92       condition_.wait(lock, [&]{ return stopped_ || !queue_.empty(); });
  93       if (stopped_)
  94         return;
  95       auto p(queue_.top());
  96       queue_.pop();
  97       lock.unlock();
  98       p->execute_(p);
  99       lock.lock();
 100     }
 101   }
 102
 103   void stop()
 104   {
 105     std::lock_guard<std::mutex> lock(mutex_);
 106     stopped_ = true;
 107     condition_.notify_all();
 108   }
 109
 110 private:
 111   struct item_base
 112   {
 113     int priority_;
 114     void (*execute_)(std::shared_ptr<item_base>&);
 115   };
 116
 117   template <class Func>
 118   struct item : item_base
 119   {
 120     item(int pri, Func f) : function_(std::move(f))
 121     {
 122       priority_ = pri;
 123       execute_ = [](std::shared_ptr<item_base>& p)
 124       {
 125         Func tmp(std::move(static_cast<item*>(p.get())->function_));
 126         p.reset();
 127         tmp();
 128       };
 129     }
 130
 131     Func function_;
 132   };
 133
 134   struct item_comp
 135   {
 136     bool operator()(
 137         const std::shared_ptr<item_base>& a,
 138         const std::shared_ptr<item_base>& b)
 139     {
 140       return a->priority_ < b->priority_;
 141     }
 142   };
 143
 144   std::mutex mutex_;
 145   std::condition_variable condition_;
 146   std::priority_queue<
 147     std::shared_ptr<item_base>,
 148     std::vector<std::shared_ptr<item_base>>,
 149     item_comp> queue_;
 150   bool stopped_ = false;
 151 };
 152
 153 int main()
 154 {
 155   priority_scheduler sched;
 156   auto low = sched.get_executor(0);
 157   auto med = sched.get_executor(1);
 158   auto high = sched.get_executor(2);
 159   dispatch(low, []{ std::cout << "1\n"; });
 160   dispatch(low, []{ std::cout << "11\n"; });
 161   dispatch(med, []{ std::cout << "2\n"; });
 162   dispatch(med, []{ std::cout << "22\n"; });
 163   dispatch(high, []{ std::cout << "3\n"; });
 164   dispatch(high, []{ std::cout << "33\n"; });
 165   dispatch(high, []{ std::cout << "333\n"; });
 166   dispatch(sched.get_executor(-1), [&]{ sched.stop(); });
 167   sched.run();
 168 }