diff options
Diffstat (limited to '3rdparty/asio/src/examples/cpp11/executors')
7 files changed, 90 insertions, 160 deletions
diff --git a/3rdparty/asio/src/examples/cpp11/executors/.gitignore b/3rdparty/asio/src/examples/cpp11/executors/.gitignore deleted file mode 100644 index 183cecf7165..00000000000 --- a/3rdparty/asio/src/examples/cpp11/executors/.gitignore +++ /dev/null @@ -1,5 +0,0 @@ -actor -bank_account_[0-9] -fork_join -pipeline -priority_scheduler diff --git a/3rdparty/asio/src/examples/cpp11/executors/actor.cpp b/3rdparty/asio/src/examples/cpp11/executors/actor.cpp index 22dbf206a17..1f0e2c11308 100644 --- a/3rdparty/asio/src/examples/cpp11/executors/actor.cpp +++ b/3rdparty/asio/src/examples/cpp11/executors/actor.cpp @@ -1,5 +1,5 @@ +#include <asio/any_io_executor.hpp> #include <asio/defer.hpp> -#include <asio/executor.hpp> #include <asio/post.hpp> #include <asio/strand.hpp> #include <asio/system_executor.hpp> @@ -10,8 +10,8 @@ #include <typeinfo> #include <vector> +using asio::any_io_executor; using asio::defer; -using asio::executor; using asio::post; using asio::strand; using asio::system_executor; @@ -106,7 +106,7 @@ public: protected: // Construct the actor to use the specified executor for all message handlers. - actor(executor e) + actor(any_io_executor e) : executor_(std::move(e)) { } @@ -124,7 +124,7 @@ protected: template <class Actor, class Message> void deregister_handler(void (Actor::* mf)(Message, actor_address)) { - const std::type_info& id = typeid(message_handler<Message>); + const std::type_info& id = typeid(Message); for (auto iter = handlers_.begin(); iter != handlers_.end(); ++iter) { if ((*iter)->message_id() == id) @@ -171,7 +171,7 @@ private: // All messages associated with a single actor object should be processed // non-concurrently. We use a strand to ensure non-concurrent execution even // if the underlying executor may use multiple threads. - strand<executor> executor_; + strand<any_io_executor> executor_; std::vector<std::shared_ptr<message_handler_base>> handlers_; }; @@ -221,7 +221,7 @@ using asio::thread_pool; class member : public actor { public: - explicit member(executor e) + explicit member(any_io_executor e) : actor(std::move(e)) { register_handler(&member::init_handler); diff --git a/3rdparty/asio/src/examples/cpp11/executors/bank_account_1.cpp b/3rdparty/asio/src/examples/cpp11/executors/bank_account_1.cpp index 85f7d9582af..a63d7fc059b 100644 --- a/3rdparty/asio/src/examples/cpp11/executors/bank_account_1.cpp +++ b/3rdparty/asio/src/examples/cpp11/executors/bank_account_1.cpp @@ -1,9 +1,9 @@ -#include <asio/post.hpp> -#include <asio/thread_pool.hpp> +#include <asio/execution.hpp> +#include <asio/static_thread_pool.hpp> #include <iostream> -using asio::post; -using asio::thread_pool; +using asio::static_thread_pool; +namespace execution = asio::execution; // Traditional active object pattern. // Member functions do not block. @@ -11,37 +11,40 @@ using asio::thread_pool; class bank_account { int balance_ = 0; - mutable thread_pool pool_{1}; + mutable static_thread_pool pool_{1}; public: void deposit(int amount) { - post(pool_, [=] - { - balance_ += amount; - }); + pool_.executor().execute( + [this, amount] + { + balance_ += amount; + }); } void withdraw(int amount) { - post(pool_, [=] - { - if (balance_ >= amount) - balance_ -= amount; - }); + pool_.executor().execute( + [this, amount] + { + if (balance_ >= amount) + balance_ -= amount; + }); } void print_balance() const { - post(pool_, [=] - { - std::cout << "balance = " << balance_ << "\n"; - }); + pool_.executor().execute( + [this] + { + std::cout << "balance = " << balance_ << "\n"; + }); } ~bank_account() { - pool_.join(); + pool_.wait(); } }; diff --git a/3rdparty/asio/src/examples/cpp11/executors/bank_account_2.cpp b/3rdparty/asio/src/examples/cpp11/executors/bank_account_2.cpp index b646cb29518..45a24abc352 100644 --- a/3rdparty/asio/src/examples/cpp11/executors/bank_account_2.cpp +++ b/3rdparty/asio/src/examples/cpp11/executors/bank_account_2.cpp @@ -1,11 +1,9 @@ -#include <asio/post.hpp> -#include <asio/thread_pool.hpp> -#include <asio/use_future.hpp> +#include <asio/execution.hpp> +#include <asio/static_thread_pool.hpp> #include <iostream> -using asio::post; -using asio::thread_pool; -using asio::use_future; +using asio::static_thread_pool; +namespace execution = asio::execution; // Traditional active object pattern. // Member functions block until operation is finished. @@ -13,35 +11,37 @@ using asio::use_future; class bank_account { int balance_ = 0; - mutable thread_pool pool_{1}; + mutable static_thread_pool pool_{1}; public: void deposit(int amount) { - post(pool_, - use_future([=] + asio::require(pool_.executor(), execution::blocking.always).execute( + [this, amount] { balance_ += amount; - })).get(); + }); } void withdraw(int amount) { - post(pool_, - use_future([=] + asio::require(pool_.executor(), execution::blocking.always).execute( + [this, amount] { if (balance_ >= amount) balance_ -= amount; - })).get(); + }); } int balance() const { - return post(pool_, - use_future([=] + int result = 0; + asio::require(pool_.executor(), execution::blocking.always).execute( + [this, &result] { - return balance_; - })).get(); + result = balance_; + }); + return result; } }; diff --git a/3rdparty/asio/src/examples/cpp11/executors/fork_join.cpp b/3rdparty/asio/src/examples/cpp11/executors/fork_join.cpp index 69716814823..7e8a827356d 100644 --- a/3rdparty/asio/src/examples/cpp11/executors/fork_join.cpp +++ b/3rdparty/asio/src/examples/cpp11/executors/fork_join.cpp @@ -1,20 +1,20 @@ -#include <asio/dispatch.hpp> -#include <asio/execution_context.hpp> -#include <asio/thread_pool.hpp> +#include <asio/execution.hpp> +#include <asio/static_thread_pool.hpp> +#include <algorithm> #include <condition_variable> #include <memory> #include <mutex> #include <queue> #include <thread> +#include <numeric> -using asio::dispatch; -using asio::execution_context; -using asio::thread_pool; +using asio::static_thread_pool; +namespace execution = asio::execution; // A fixed-size thread pool used to implement fork/join semantics. Functions // are scheduled using a simple FIFO queue. Implementing work stealing, or // using a queue based on atomic operations, are left as tasks for the reader. -class fork_join_pool : public execution_context +class fork_join_pool { public: // The constructor starts a thread pool with the specified number of threads. @@ -22,7 +22,7 @@ public: // Additional threads may temporarily be added to the pool if they join a // fork_executor. explicit fork_join_pool( - std::size_t thread_count = std::thread::hardware_concurrency() * 2) + std::size_t thread_count = std::max(std::thread::hardware_concurrency(), 1u) * 2) : use_count_(1), threads_(thread_count) { @@ -32,7 +32,8 @@ public: // it is time to shut down, i.e. the use count is zero. for (thread_count_ = 0; thread_count_ < thread_count; ++thread_count_) { - dispatch(threads_, [&] + threads_.executor().execute( + [this] { std::unique_lock<std::mutex> lock(mutex_); while (use_count_ > 0) @@ -44,7 +45,7 @@ public: catch (...) { stop_threads(); - threads_.join(); + threads_.wait(); throw; } } @@ -53,7 +54,7 @@ public: ~fork_join_pool() { stop_threads(); - threads_.join(); + threads_.wait(); } private: @@ -117,7 +118,7 @@ private: // Dispatch a function, executing it immediately if the queue is already // loaded. Otherwise adds the function to the queue and wakes a thread. - void do_dispatch(std::shared_ptr<function_base> p, + void do_execute(std::shared_ptr<function_base> p, const std::shared_ptr<std::size_t>& work_count) { std::unique_lock<std::mutex> lock(mutex_); @@ -135,16 +136,6 @@ private: } } - // Add a function to the queue and wake a thread. - void do_post(std::shared_ptr<function_base> p, - const std::shared_ptr<std::size_t>& work_count) - { - std::lock_guard<std::mutex> lock(mutex_); - queue_.push(p); - do_work_started(work_count); - condition_.notify_one(); - } - // Ask all threads to shut down. void stop_threads() { @@ -158,7 +149,7 @@ private: std::queue<std::shared_ptr<function_base>> queue_; std::size_t use_count_; std::size_t thread_count_; - thread_pool threads_; + static_thread_pool threads_; }; // A class that satisfies the Executor requirements. Every function or piece of @@ -172,45 +163,16 @@ public: { } - fork_join_pool& context() const noexcept + fork_join_pool& query(execution::context_t) const noexcept { return context_; } - void on_work_started() const noexcept - { - std::lock_guard<std::mutex> lock(context_.mutex_); - context_.do_work_started(work_count_); - } - - void on_work_finished() const noexcept - { - std::lock_guard<std::mutex> lock(context_.mutex_); - context_.do_work_finished(work_count_); - } - - template <class Func, class Alloc> - void dispatch(Func&& f, const Alloc& a) const - { - auto p(std::allocate_shared<function<Func>>( - typename std::allocator_traits<Alloc>::template rebind_alloc<char>(a), - std::move(f), work_count_)); - context_.do_dispatch(p, work_count_); - } - - template <class Func, class Alloc> - void post(Func f, const Alloc& a) const - { - auto p(std::allocate_shared<function<Func>>( - typename std::allocator_traits<Alloc>::template rebind_alloc<char>(a), - std::move(f), work_count_)); - context_.do_post(p, work_count_); - } - - template <class Func, class Alloc> - void defer(Func&& f, const Alloc& a) const + template <class Func> + void execute(Func f) const { - post(std::forward<Func>(f), a); + auto p(std::make_shared<function<Func>>(std::move(f), work_count_)); + context_.do_execute(p, work_count_); } friend bool operator==(const fork_executor& a, @@ -289,8 +251,8 @@ void fork_join_sort(Iterator begin, Iterator end) { fork_executor fork(pool); join_guard join(fork); - dispatch(fork, [=]{ fork_join_sort(begin, begin + n / 2); }); - dispatch(fork, [=]{ fork_join_sort(begin + n / 2, end); }); + fork.execute([=]{ fork_join_sort(begin, begin + n / 2); }); + fork.execute([=]{ fork_join_sort(begin + n / 2, end); }); } std::inplace_merge(begin, begin + n / 2, end); } diff --git a/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp b/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp index 86bf860cb9c..f441e322688 100644 --- a/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp +++ b/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp @@ -11,6 +11,7 @@ #include <queue> #include <thread> #include <vector> +#include <cctype> using asio::execution_context; using asio::executor_binder; @@ -19,9 +20,10 @@ using asio::post; using asio::system_executor; using asio::use_future; using asio::use_service; +namespace execution = asio::execution; // An executor that launches a new thread for each function submitted to it. -// This class satisfies the Executor requirements. +// This class satisfies the executor requirements. class thread_executor { private: @@ -54,40 +56,28 @@ private: }; public: - execution_context& context() const noexcept + execution_context& query(execution::context_t) const { - return system_executor().context(); + return asio::query(system_executor(), execution::context); } - void on_work_started() const noexcept + execution::blocking_t query(execution::blocking_t) const { - // This executor doesn't count work. + return execution::blocking.never; } - void on_work_finished() const noexcept + thread_executor require(execution::blocking_t::never_t) const { - // This executor doesn't count work. + return *this; } - template <class Func, class Alloc> - void dispatch(Func&& f, const Alloc& a) const + template <class Func> + void execute(Func f) const { - post(std::forward<Func>(f), a); - } - - template <class Func, class Alloc> - void post(Func f, const Alloc&) const - { - thread_bag& bag = use_service<thread_bag>(context()); + thread_bag& bag = use_service<thread_bag>(query(execution::context)); bag.add_thread(std::thread(std::move(f))); } - template <class Func, class Alloc> - void defer(Func&& f, const Alloc& a) const - { - post(std::forward<Func>(f), a); - } - friend bool operator==(const thread_executor&, const thread_executor&) noexcept { @@ -291,7 +281,7 @@ void writer(queue_back<std::string> in) int main() { - thread_pool pool; + thread_pool pool(1); auto f = pipeline(reader, filter, bind_executor(pool, upper), writer); f.wait(); diff --git a/3rdparty/asio/src/examples/cpp11/executors/priority_scheduler.cpp b/3rdparty/asio/src/examples/cpp11/executors/priority_scheduler.cpp index fb38fa2ce4b..d8ef79f3a51 100644 --- a/3rdparty/asio/src/examples/cpp11/executors/priority_scheduler.cpp +++ b/3rdparty/asio/src/examples/cpp11/executors/priority_scheduler.cpp @@ -8,6 +8,7 @@ using asio::dispatch; using asio::execution_context; +namespace execution = asio::execution; class priority_scheduler : public execution_context { @@ -21,47 +22,20 @@ public: { } - priority_scheduler& context() const noexcept + priority_scheduler& query(execution::context_t) const noexcept { return context_; } - void on_work_started() const noexcept + template <class Func> + void execute(Func f) const { - // This executor doesn't count work. Instead, the scheduler simply runs - // until explicitly stopped. - } - - void on_work_finished() const noexcept - { - // This executor doesn't count work. Instead, the scheduler simply runs - // until explicitly stopped. - } - - template <class Func, class Alloc> - void dispatch(Func&& f, const Alloc& a) const - { - post(std::forward<Func>(f), a); - } - - template <class Func, class Alloc> - void post(Func f, const Alloc& a) const - { - auto p(std::allocate_shared<item<Func>>( - typename std::allocator_traits< - Alloc>::template rebind_alloc<char>(a), - priority_, std::move(f))); + auto p(std::make_shared<item<Func>>(priority_, std::move(f))); std::lock_guard<std::mutex> lock(context_.mutex_); context_.queue_.push(p); context_.condition_.notify_one(); } - template <class Func, class Alloc> - void defer(Func&& f, const Alloc& a) const - { - post(std::forward<Func>(f), a); - } - friend bool operator==(const executor_type& a, const executor_type& b) noexcept { @@ -79,6 +53,12 @@ public: int priority_; }; + ~priority_scheduler() noexcept + { + shutdown(); + destroy(); + } + executor_type get_executor(int pri = 0) noexcept { return executor_type(*const_cast<priority_scheduler*>(this), pri); |