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
9using boost::asio::dispatch;
10using boost::asio::execution_context;
11namespace execution = boost::asio::execution;
12
13class priority_scheduler : public execution_context
14{
15public:
16 // A class that satisfies the Executor requirements.
17 class executor_type
18 {
19 public:
20 executor_type(priority_scheduler& ctx, int pri) noexcept
21 : context_(ctx), priority_(pri)
22 {
23 }
24
25 priority_scheduler& query(execution::context_t) const noexcept
26 {
27 return context_;
28 }
29
30 template <class Func>
31 void execute(Func f) const
32 {
33 auto p(std::make_shared<item<Func>>(priority_, std::move(f)));
34 std::lock_guard<std::mutex> lock(context_.mutex_);
35 context_.queue_.push(p);
36 context_.condition_.notify_one();
37 }
38
39 friend bool operator==(const executor_type& a,
40 const executor_type& b) noexcept
41 {
42 return &a.context_ == &b.context_;
43 }
44
45 friend bool operator!=(const executor_type& a,
46 const executor_type& b) noexcept
47 {
48 return &a.context_ != &b.context_;
49 }
50
51 private:
52 priority_scheduler& context_;
53 int priority_;
54 };
55
56 ~priority_scheduler() noexcept
57 {
58 shutdown();
59 destroy();
60 }
61
62 executor_type get_executor(int pri = 0) noexcept
63 {
64 return executor_type(*const_cast<priority_scheduler*>(this), pri);
65 }
66
67 void run()
68 {
69 std::unique_lock<std::mutex> lock(mutex_);
70 for (;;)
71 {
72 condition_.wait(lock&: lock, p: [&]{ return stopped_ || !queue_.empty(); });
73 if (stopped_)
74 return;
75 auto p(queue_.top());
76 queue_.pop();
77 lock.unlock();
78 p->execute_(p);
79 lock.lock();
80 }
81 }
82
83 void stop()
84 {
85 std::lock_guard<std::mutex> lock(mutex_);
86 stopped_ = true;
87 condition_.notify_all();
88 }
89
90private:
91 struct item_base
92 {
93 int priority_;
94 void (*execute_)(std::shared_ptr<item_base>&);
95 };
96
97 template <class Func>
98 struct item : item_base
99 {
100 item(int pri, Func f) : function_(std::move(f))
101 {
102 priority_ = pri;
103 execute_ = [](std::shared_ptr<item_base>& p)
104 {
105 Func tmp(std::move(static_cast<item*>(p.get())->function_));
106 p.reset();
107 tmp();
108 };
109 }
110
111 Func function_;
112 };
113
114 struct item_comp
115 {
116 bool operator()(
117 const std::shared_ptr<item_base>& a,
118 const std::shared_ptr<item_base>& b)
119 {
120 return a->priority_ < b->priority_;
121 }
122 };
123
124 std::mutex mutex_;
125 std::condition_variable condition_;
126 std::priority_queue<
127 std::shared_ptr<item_base>,
128 std::vector<std::shared_ptr<item_base>>,
129 item_comp> queue_;
130 bool stopped_ = false;
131};
132
133int main()
134{
135 priority_scheduler sched;
136 auto low = sched.get_executor(pri: 0);
137 auto med = sched.get_executor(pri: 1);
138 auto high = sched.get_executor(pri: 2);
139 dispatch(ex: low, token: []{ std::cout << "1\n"; });
140 dispatch(ex: low, token: []{ std::cout << "11\n"; });
141 dispatch(ex: med, token: []{ std::cout << "2\n"; });
142 dispatch(ex: med, token: []{ std::cout << "22\n"; });
143 dispatch(ex: high, token: []{ std::cout << "3\n"; });
144 dispatch(ex: high, token: []{ std::cout << "33\n"; });
145 dispatch(ex: high, token: []{ std::cout << "333\n"; });
146 dispatch(ex: sched.get_executor(pri: -1), token: [&]{ sched.stop(); });
147 sched.run();
148}
149

source code of boost/libs/asio/example/cpp11/executors/priority_scheduler.cpp