diff options
Diffstat (limited to '3rdparty/asio/src/examples/cpp14/executors')
9 files changed, 1295 insertions, 0 deletions
diff --git a/3rdparty/asio/src/examples/cpp14/executors/.gitignore b/3rdparty/asio/src/examples/cpp14/executors/.gitignore new file mode 100644 index 00000000000..d51d7f00295 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/.gitignore @@ -0,0 +1,6 @@ +actor +async_[0-9] +bank_account_[0-9] +fork_join +pipeline +priority_scheduler diff --git a/3rdparty/asio/src/examples/cpp14/executors/actor.cpp b/3rdparty/asio/src/examples/cpp14/executors/actor.cpp new file mode 100644 index 00000000000..26d2fb4bf11 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/actor.cpp @@ -0,0 +1,281 @@ +#include <asio/ts/executor.hpp> +#include <condition_variable> +#include <deque> +#include <memory> +#include <mutex> +#include <typeinfo> +#include <vector> + +using asio::defer; +using asio::executor; +using asio::post; +using asio::strand; +using asio::system_executor; + +//------------------------------------------------------------------------------ +// A tiny actor framework +// ~~~~~~~~~~~~~~~~~~~~~~ + +class actor; + +// Used to identify the sender and recipient of messages. +typedef actor* actor_address; + +// Base class for all registered message handlers. +class message_handler_base +{ +public: + virtual ~message_handler_base() {} + + // Used to determine which message handlers receive an incoming message. + virtual const std::type_info& message_id() const = 0; +}; + +// Base class for a handler for a specific message type. +template <class Message> +class message_handler : public message_handler_base +{ +public: + // Handle an incoming message. + virtual void handle_message(Message msg, actor_address from) = 0; +}; + +// Concrete message handler for a specific message type. +template <class Actor, class Message> +class mf_message_handler : public message_handler<Message> +{ +public: + // Construct a message handler to invoke the specified member function. + mf_message_handler(void (Actor::* mf)(Message, actor_address), Actor* a) + : function_(mf), actor_(a) + { + } + + // Used to determine which message handlers receive an incoming message. + virtual const std::type_info& message_id() const + { + return typeid(Message); + } + + // Handle an incoming message. + virtual void handle_message(Message msg, actor_address from) + { + (actor_->*function_)(std::move(msg), from); + } + + // Determine whether the message handler represents the specified function. + bool is_function(void (Actor::* mf)(Message, actor_address)) const + { + return mf == function_; + } + +private: + void (Actor::* function_)(Message, actor_address); + Actor* actor_; +}; + +// Base class for all actors. +class actor +{ +public: + virtual ~actor() + { + } + + // Obtain the actor's address for use as a message sender or recipient. + actor_address address() + { + return this; + } + + // Send a message from one actor to another. + template <class Message> + friend void send(Message msg, actor_address from, actor_address to) + { + // Execute the message handler in the context of the target's executor. + post(to->executor_, + [=, msg=std::move(msg)] + { + to->call_handler(std::move(msg), from); + }); + } + +protected: + // Construct the actor to use the specified executor for all message handlers. + actor(executor e) + : executor_(std::move(e)) + { + } + + // Register a handler for a specific message type. Duplicates are permitted. + template <class Actor, class Message> + void register_handler(void (Actor::* mf)(Message, actor_address)) + { + handlers_.push_back( + std::make_shared<mf_message_handler<Actor, Message>>( + mf, static_cast<Actor*>(this))); + } + + // Deregister a handler. Removes only the first matching handler. + template <class Actor, class Message> + void deregister_handler(void (Actor::* mf)(Message, actor_address)) + { + const std::type_info& id = typeid(message_handler<Message>); + for (auto iter = handlers_.begin(); iter != handlers_.end(); ++iter) + { + if ((*iter)->message_id() == id) + { + auto mh = static_cast<mf_message_handler<Actor, Message>*>(iter->get()); + if (mh->is_function(mf)) + { + handlers_.erase(iter); + return; + } + } + } + } + + // Send a message from within a message handler. + template <class Message> + void tail_send(Message msg, actor_address to) + { + // Execute the message handler in the context of the target's executor. + defer(to->executor_, + [=, msg=std::move(msg), from=this] + { + to->call_handler(std::move(msg), from); + }); + } + +private: + // Find the matching message handlers, if any, and call them. + template <class Message> + void call_handler(Message msg, actor_address from) + { + const std::type_info& message_id = typeid(Message); + for (auto& h: handlers_) + { + if (h->message_id() == message_id) + { + auto mh = static_cast<message_handler<Message>*>(h.get()); + mh->handle_message(msg, from); + } + } + } + + // 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_; + + std::vector<std::shared_ptr<message_handler_base>> handlers_; +}; + +// A concrete actor that allows synchronous message retrieval. +template <class Message> +class receiver : public actor +{ +public: + receiver() + : actor(system_executor()) + { + register_handler(&receiver::message_handler); + } + + // Block until a message has been received. + Message wait() + { + std::unique_lock<std::mutex> lock(mutex_); + condition_.wait(lock, [this]{ return !message_queue_.empty(); }); + Message msg(std::move(message_queue_.front())); + message_queue_.pop_front(); + return msg; + } + +private: + // Handle a new message by adding it to the queue and waking a waiter. + void message_handler(Message msg, actor_address /* from */) + { + std::lock_guard<std::mutex> lock(mutex_); + message_queue_.push_back(std::move(msg)); + condition_.notify_one(); + } + + std::mutex mutex_; + std::condition_variable condition_; + std::deque<Message> message_queue_; +}; + +//------------------------------------------------------------------------------ + +#include <asio/thread_pool.hpp> +#include <iostream> + +using asio::thread_pool; + +class member : public actor +{ +public: + explicit member(executor e) + : actor(std::move(e)) + { + register_handler(&member::init_handler); + } + +private: + void init_handler(actor_address next, actor_address from) + { + next_ = next; + caller_ = from; + + register_handler(&member::token_handler); + deregister_handler(&member::init_handler); + } + + void token_handler(int token, actor_address /*from*/) + { + int msg(token); + actor_address to(caller_); + + if (token > 0) + { + msg = token - 1; + to = next_; + } + + tail_send(msg, to); + } + + actor_address next_; + actor_address caller_; +}; + +int main() +{ + const std::size_t num_threads = 16; + const int num_hops = 50000000; + const std::size_t num_actors = 503; + const int token_value = (num_hops + num_actors - 1) / num_actors; + const std::size_t actors_per_thread = num_actors / num_threads; + + struct single_thread_pool : thread_pool { single_thread_pool() : thread_pool(1) {} }; + single_thread_pool pools[num_threads]; + std::vector<std::shared_ptr<member>> members(num_actors); + receiver<int> rcvr; + + // Create the member actors. + for (std::size_t i = 0; i < num_actors; ++i) + members[i] = std::make_shared<member>(pools[(i / actors_per_thread) % num_threads].get_executor()); + + // Initialise the actors by passing each one the address of the next actor in the ring. + for (std::size_t i = num_actors, next_i = 0; i > 0; next_i = --i) + send(members[next_i]->address(), rcvr.address(), members[i - 1]->address()); + + // Send exactly one token to each actor, all with the same initial value, rounding up if required. + for (std::size_t i = 0; i < num_actors; ++i) + send(token_value, rcvr.address(), members[i]->address()); + + // Wait for all signal messages, indicating the tokens have all reached zero. + for (std::size_t i = 0; i < num_actors; ++i) + rcvr.wait(); +} diff --git a/3rdparty/asio/src/examples/cpp14/executors/async_1.cpp b/3rdparty/asio/src/examples/cpp14/executors/async_1.cpp new file mode 100644 index 00000000000..d282b74f206 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/async_1.cpp @@ -0,0 +1,47 @@ +#include <asio/ts/executor.hpp> +#include <asio/thread_pool.hpp> +#include <iostream> +#include <string> + +using asio::bind_executor; +using asio::dispatch; +using asio::make_work_guard; +using asio::post; +using asio::thread_pool; + +// A function to asynchronously read a single line from an input stream. +template <class Handler> +void async_getline(std::istream& is, Handler handler) +{ + // Create executor_work for the handler's associated executor. + auto work = make_work_guard(handler); + + // Post a function object to do the work asynchronously. + post([&is, work, handler=std::move(handler)]() mutable + { + std::string line; + std::getline(is, line); + + // Pass the result to the handler, via the associated executor. + dispatch(work.get_executor(), + [line=std::move(line), handler=std::move(handler)]() mutable + { + handler(std::move(line)); + }); + }); +} + +int main() +{ + thread_pool pool; + + std::cout << "Enter a line: "; + + async_getline(std::cin, + bind_executor(pool, [](std::string line) + { + std::cout << "Line: " << line << "\n"; + })); + + pool.join(); +} diff --git a/3rdparty/asio/src/examples/cpp14/executors/async_2.cpp b/3rdparty/asio/src/examples/cpp14/executors/async_2.cpp new file mode 100644 index 00000000000..b39156dcb23 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/async_2.cpp @@ -0,0 +1,68 @@ +#include <asio/ts/executor.hpp> +#include <asio/thread_pool.hpp> +#include <iostream> +#include <string> + +using asio::bind_executor; +using asio::dispatch; +using asio::get_associated_executor; +using asio::make_work_guard; +using asio::post; +using asio::thread_pool; + +// A function to asynchronously read a single line from an input stream. +template <class Handler> +void async_getline(std::istream& is, Handler handler) +{ + // Create executor_work for the handler's associated executor. + auto work = make_work_guard(handler); + + // Post a function object to do the work asynchronously. + post([&is, work, handler=std::move(handler)]() mutable + { + std::string line; + std::getline(is, line); + + // Pass the result to the handler, via the associated executor. + dispatch(work.get_executor(), + [line=std::move(line), handler=std::move(handler)]() mutable + { + handler(std::move(line)); + }); + }); +} + +// A function to asynchronously read multiple lines from an input stream. +template <class Handler> +void async_getlines(std::istream& is, std::string init, Handler handler) +{ + // Get the final handler's associated executor. + auto ex = get_associated_executor(handler); + + // Use the associated executor for each operation in the composition. + async_getline(is, + bind_executor(ex, + [&is, lines=std::move(init), handler=std::move(handler)] + (std::string line) mutable + { + if (line.empty()) + handler(lines); + else + async_getlines(is, lines + line + "\n", std::move(handler)); + })); +} + +int main() +{ + thread_pool pool; + + std::cout << "Enter text, terminating with a blank line:\n"; + + async_getlines(std::cin, "", + bind_executor(pool, [](std::string lines) + { + std::cout << "Lines:\n" << lines << "\n"; + })); + + pool.join(); +} diff --git a/3rdparty/asio/src/examples/cpp14/executors/bank_account_1.cpp b/3rdparty/asio/src/examples/cpp14/executors/bank_account_1.cpp new file mode 100644 index 00000000000..f85a1852b56 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/bank_account_1.cpp @@ -0,0 +1,54 @@ +#include <asio/ts/executor.hpp> +#include <asio/thread_pool.hpp> +#include <iostream> + +using asio::post; +using asio::thread_pool; + +// Traditional active object pattern. +// Member functions do not block. + +class bank_account +{ + int balance_ = 0; + mutable thread_pool pool_{1}; + +public: + void deposit(int amount) + { + post(pool_, [=] + { + balance_ += amount; + }); + } + + void withdraw(int amount) + { + post(pool_, [=] + { + if (balance_ >= amount) + balance_ -= amount; + }); + } + + void print_balance() const + { + post(pool_, [=] + { + std::cout << "balance = " << balance_ << "\n"; + }); + } + + ~bank_account() + { + pool_.join(); + } +}; + +int main() +{ + bank_account acct; + acct.deposit(20); + acct.withdraw(10); + acct.print_balance(); +} diff --git a/3rdparty/asio/src/examples/cpp14/executors/bank_account_2.cpp b/3rdparty/asio/src/examples/cpp14/executors/bank_account_2.cpp new file mode 100644 index 00000000000..5233e77dec3 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/bank_account_2.cpp @@ -0,0 +1,53 @@ +#include <asio/ts/executor.hpp> +#include <asio/thread_pool.hpp> +#include <iostream> + +using asio::post; +using asio::thread_pool; +using asio::use_future; + +// Traditional active object pattern. +// Member functions block until operation is finished. + +class bank_account +{ + int balance_ = 0; + mutable thread_pool pool_{1}; + +public: + void deposit(int amount) + { + post(pool_, + use_future([=] + { + balance_ += amount; + })).get(); + } + + void withdraw(int amount) + { + post(pool_, + use_future([=] + { + if (balance_ >= amount) + balance_ -= amount; + })).get(); + } + + int balance() const + { + return post(pool_, + use_future([=] + { + return balance_; + })).get(); + } +}; + +int main() +{ + bank_account acct; + acct.deposit(20); + acct.withdraw(10); + std::cout << "balance = " << acct.balance() << "\n"; +} diff --git a/3rdparty/asio/src/examples/cpp14/executors/fork_join.cpp b/3rdparty/asio/src/examples/cpp14/executors/fork_join.cpp new file mode 100644 index 00000000000..fe6746ad7e0 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/fork_join.cpp @@ -0,0 +1,326 @@ +#include <asio/ts/executor.hpp> +#include <asio/thread_pool.hpp> +#include <condition_variable> +#include <memory> +#include <mutex> +#include <queue> +#include <thread> + +using asio::dispatch; +using asio::execution_context; +using asio::thread_pool; + +// 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 +{ +public: + // The constructor starts a thread pool with the specified number of threads. + // Note that the thread_count is not a fixed limit on the pool's concurrency. + // 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) + : use_count_(1), + threads_(thread_count) + { + try + { + // Ask each thread in the pool to dequeue and execute functions until + // 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_, [&] + { + std::unique_lock<std::mutex> lock(mutex_); + while (use_count_ > 0) + if (!execute_next(lock)) + condition_.wait(lock); + }); + } + } + catch (...) + { + stop_threads(); + threads_.join(); + throw; + } + } + + // The destructor waits for the pool to finish executing functions. + ~fork_join_pool() + { + stop_threads(); + threads_.join(); + } + +private: + friend class fork_executor; + + // The base for all functions that are queued in the pool. + struct function_base + { + std::shared_ptr<std::size_t> work_count_; + void (*execute_)(std::shared_ptr<function_base>& p); + }; + + // Execute the next function from the queue, if any. Returns true if a + // function was executed, and false if the queue was empty. + bool execute_next(std::unique_lock<std::mutex>& lock) + { + if (queue_.empty()) + return false; + auto p(queue_.front()); + queue_.pop(); + lock.unlock(); + execute(lock, p); + return true; + } + + // Execute a function and decrement the outstanding work. + void execute(std::unique_lock<std::mutex>& lock, + std::shared_ptr<function_base>& p) + { + std::shared_ptr<std::size_t> work_count(std::move(p->work_count_)); + try + { + p->execute_(p); + lock.lock(); + do_work_finished(work_count); + } + catch (...) + { + lock.lock(); + do_work_finished(work_count); + throw; + } + } + + // Increment outstanding work. + void do_work_started(const std::shared_ptr<std::size_t>& work_count) noexcept + { + if (++(*work_count) == 1) + ++use_count_; + } + + // Decrement outstanding work. Notify waiting threads if we run out. + void do_work_finished(const std::shared_ptr<std::size_t>& work_count) noexcept + { + if (--(*work_count) == 0) + { + --use_count_; + condition_.notify_all(); + } + } + + // 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, + const std::shared_ptr<std::size_t>& work_count) + { + std::unique_lock<std::mutex> lock(mutex_); + if (queue_.size() > thread_count_ * 16) + { + do_work_started(work_count); + lock.unlock(); + execute(lock, p); + } + else + { + queue_.push(p); + do_work_started(work_count); + condition_.notify_one(); + } + } + + // 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() + { + std::lock_guard<std::mutex> lock(mutex_); + --use_count_; + condition_.notify_all(); + } + + std::mutex mutex_; + std::condition_variable condition_; + std::queue<std::shared_ptr<function_base>> queue_; + std::size_t use_count_; + std::size_t thread_count_; + thread_pool threads_; +}; + +// A class that satisfies the Executor requirements. Every function or piece of +// work associated with a fork_executor is part of a single, joinable group. +class fork_executor +{ +public: + fork_executor(fork_join_pool& ctx) + : context_(ctx), + work_count_(std::make_shared<std::size_t>(0)) + { + } + + fork_join_pool& context() 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 + { + post(std::forward<Func>(f), a); + } + + friend bool operator==(const fork_executor& a, + const fork_executor& b) noexcept + { + return a.work_count_ == b.work_count_; + } + + friend bool operator!=(const fork_executor& a, + const fork_executor& b) noexcept + { + return a.work_count_ != b.work_count_; + } + + // Block until all work associated with the executor is complete. While it is + // waiting, the thread may be borrowed to execute functions from the queue. + void join() const + { + std::unique_lock<std::mutex> lock(context_.mutex_); + while (*work_count_ > 0) + if (!context_.execute_next(lock)) + context_.condition_.wait(lock); + } + +private: + template <class Func> + struct function : fork_join_pool::function_base + { + explicit function(Func f, const std::shared_ptr<std::size_t>& w) + : function_(std::move(f)) + { + work_count_ = w; + execute_ = [](std::shared_ptr<fork_join_pool::function_base>& p) + { + Func tmp(std::move(static_cast<function*>(p.get())->function_)); + p.reset(); + tmp(); + }; + } + + Func function_; + }; + + fork_join_pool& context_; + std::shared_ptr<std::size_t> work_count_; +}; + +// Helper class to automatically join a fork_executor when exiting a scope. +class join_guard +{ +public: + explicit join_guard(const fork_executor& ex) : ex_(ex) {} + join_guard(const join_guard&) = delete; + join_guard(join_guard&&) = delete; + ~join_guard() { ex_.join(); } + +private: + fork_executor ex_; +}; + +//------------------------------------------------------------------------------ + +#include <algorithm> +#include <iostream> +#include <random> +#include <vector> + +fork_join_pool pool; + +template <class Iterator> +void fork_join_sort(Iterator begin, Iterator end) +{ + std::size_t n = end - begin; + if (n > 32768) + { + { + 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); }); + } + std::inplace_merge(begin, begin + n / 2, end); + } + else + { + std::sort(begin, end); + } +} + +int main(int argc, char* argv[]) +{ + if (argc != 2) + { + std::cerr << "Usage: fork_join <size>\n"; + return 1; + } + + std::vector<double> vec(std::atoll(argv[1])); + std::iota(vec.begin(), vec.end(), 0); + + std::random_device rd; + std::mt19937 g(rd()); + std::shuffle(vec.begin(), vec.end(), g); + + std::chrono::steady_clock::time_point start = std::chrono::steady_clock::now(); + + fork_join_sort(vec.begin(), vec.end()); + + std::chrono::steady_clock::duration elapsed = std::chrono::steady_clock::now() - start; + + std::cout << "sort took "; + std::cout << std::chrono::duration_cast<std::chrono::microseconds>(elapsed).count(); + std::cout << " microseconds" << std::endl; +} diff --git a/3rdparty/asio/src/examples/cpp14/executors/pipeline.cpp b/3rdparty/asio/src/examples/cpp14/executors/pipeline.cpp new file mode 100644 index 00000000000..c66e283fcbc --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/pipeline.cpp @@ -0,0 +1,293 @@ +#include <asio/ts/executor.hpp> +#include <condition_variable> +#include <future> +#include <memory> +#include <mutex> +#include <queue> +#include <thread> +#include <vector> + +using asio::execution_context; +using asio::executor_binder; +using asio::get_associated_executor; +using asio::post; +using asio::system_executor; +using asio::use_future; +using asio::use_service; + +// An executor that launches a new thread for each function submitted to it. +// This class satisfies the Executor requirements. +class thread_executor +{ +private: + // Service to track all threads started through a thread_executor. + class thread_bag : public execution_context::service + { + public: + typedef thread_bag key_type; + + explicit thread_bag(execution_context& ctx) + : execution_context::service(ctx) + { + } + + void add_thread(std::thread&& t) + { + std::unique_lock<std::mutex> lock(mutex_); + threads_.push_back(std::move(t)); + } + + private: + virtual void shutdown() + { + for (auto& t : threads_) + t.join(); + } + + std::mutex mutex_; + std::vector<std::thread> threads_; + }; + +public: + execution_context& context() const noexcept + { + return system_executor().context(); + } + + void on_work_started() const noexcept + { + // This executor doesn't count work. + } + + void on_work_finished() const noexcept + { + // This executor doesn't count work. + } + + 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&) const + { + thread_bag& bag = use_service<thread_bag>(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 + { + return true; + } + + friend bool operator!=(const thread_executor&, + const thread_executor&) noexcept + { + return false; + } +}; + +// Base class for all thread-safe queue implementations. +class queue_impl_base +{ + template <class> friend class queue_front; + template <class> friend class queue_back; + std::mutex mutex_; + std::condition_variable condition_; + bool stop_ = false; +}; + +// Underlying implementation of a thread-safe queue, shared between the +// queue_front and queue_back classes. +template <class T> +class queue_impl : public queue_impl_base +{ + template <class> friend class queue_front; + template <class> friend class queue_back; + std::queue<T> queue_; +}; + +// The front end of a queue between consecutive pipeline stages. +template <class T> +class queue_front +{ +public: + typedef T value_type; + + explicit queue_front(std::shared_ptr<queue_impl<T>> impl) + : impl_(impl) + { + } + + void push(T t) + { + std::unique_lock<std::mutex> lock(impl_->mutex_); + impl_->queue_.push(std::move(t)); + impl_->condition_.notify_one(); + } + + void stop() + { + std::unique_lock<std::mutex> lock(impl_->mutex_); + impl_->stop_ = true; + impl_->condition_.notify_one(); + } + +private: + std::shared_ptr<queue_impl<T>> impl_; +}; + +// The back end of a queue between consecutive pipeline stages. +template <class T> +class queue_back +{ +public: + typedef T value_type; + + explicit queue_back(std::shared_ptr<queue_impl<T>> impl) + : impl_(impl) + { + } + + bool pop(T& t) + { + std::unique_lock<std::mutex> lock(impl_->mutex_); + while (impl_->queue_.empty() && !impl_->stop_) + impl_->condition_.wait(lock); + if (!impl_->queue_.empty()) + { + t = impl_->queue_.front(); + impl_->queue_.pop(); + return true; + } + return false; + } + +private: + std::shared_ptr<queue_impl<T>> impl_; +}; + +// Launch the last stage in a pipeline. +template <class T, class F> +std::future<void> pipeline(queue_back<T> in, F f) +{ + // Get the function's associated executor, defaulting to thread_executor. + auto ex = get_associated_executor(f, thread_executor()); + + // Run the function, and as we're the last stage return a future so that the + // caller can wait for the pipeline to finish. + return post(ex, use_future([in, f = std::move(f)]() mutable { f(in); })); +} + +// Launch an intermediate stage in a pipeline. +template <class T, class F, class... Tail> +std::future<void> pipeline(queue_back<T> in, F f, Tail... t) +{ + // Determine the output queue type. + typedef typename executor_binder<F, thread_executor>::second_argument_type::value_type output_value_type; + + // Create the output queue and its implementation. + auto out_impl = std::make_shared<queue_impl<output_value_type>>(); + queue_front<output_value_type> out(out_impl); + queue_back<output_value_type> next_in(out_impl); + + // Get the function's associated executor, defaulting to thread_executor. + auto ex = get_associated_executor(f, thread_executor()); + + // Run the function. + post(ex, [in, out, f = std::move(f)]() mutable + { + f(in, out); + out.stop(); + }); + + // Launch the rest of the pipeline. + return pipeline(next_in, std::move(t)...); +} + +// Launch the first stage in a pipeline. +template <class F, class... Tail> +std::future<void> pipeline(F f, Tail... t) +{ + // Determine the output queue type. + typedef typename executor_binder<F, thread_executor>::argument_type::value_type output_value_type; + + // Create the output queue and its implementation. + auto out_impl = std::make_shared<queue_impl<output_value_type>>(); + queue_front<output_value_type> out(out_impl); + queue_back<output_value_type> next_in(out_impl); + + // Get the function's associated executor, defaulting to thread_executor. + auto ex = get_associated_executor(f, thread_executor()); + + // Run the function. + post(ex, [out, f = std::move(f)]() mutable + { + f(out); + out.stop(); + }); + + // Launch the rest of the pipeline. + return pipeline(next_in, std::move(t)...); +} + +//------------------------------------------------------------------------------ + +#include <asio/thread_pool.hpp> +#include <iostream> +#include <string> + +using asio::bind_executor; +using asio::thread_pool; + +void reader(queue_front<std::string> out) +{ + std::string line; + while (std::getline(std::cin, line)) + out.push(line); +} + +void filter(queue_back<std::string> in, queue_front<std::string> out) +{ + std::string line; + while (in.pop(line)) + if (line.length() > 5) + out.push(line); +} + +void upper(queue_back<std::string> in, queue_front<std::string> out) +{ + std::string line; + while (in.pop(line)) + { + std::string new_line; + for (char c : line) + new_line.push_back(std::toupper(c)); + out.push(new_line); + } +} + +void writer(queue_back<std::string> in) +{ + std::size_t count = 0; + std::string line; + while (in.pop(line)) + std::cout << count++ << ": " << line << std::endl; +} + +int main() +{ + thread_pool pool; + + auto f = pipeline(reader, filter, bind_executor(pool, upper), writer); + f.wait(); +} diff --git a/3rdparty/asio/src/examples/cpp14/executors/priority_scheduler.cpp b/3rdparty/asio/src/examples/cpp14/executors/priority_scheduler.cpp new file mode 100644 index 00000000000..3b28af1eec0 --- /dev/null +++ b/3rdparty/asio/src/examples/cpp14/executors/priority_scheduler.cpp @@ -0,0 +1,167 @@ +#include <asio/ts/executor.hpp> +#include <condition_variable> +#include <iostream> +#include <memory> +#include <mutex> +#include <queue> + +using asio::dispatch; +using asio::execution_context; + +class priority_scheduler : public execution_context +{ +public: + // A class that satisfies the Executor requirements. + class executor_type + { + public: + executor_type(priority_scheduler& ctx, int pri) noexcept + : context_(ctx), priority_(pri) + { + } + + priority_scheduler& context() const noexcept + { + return context_; + } + + void on_work_started() const noexcept + { + // 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))); + 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 + { + return &a.context_ == &b.context_; + } + + friend bool operator!=(const executor_type& a, + const executor_type& b) noexcept + { + return &a.context_ != &b.context_; + } + + private: + priority_scheduler& context_; + int priority_; + }; + + executor_type get_executor(int pri = 0) noexcept + { + return executor_type(*const_cast<priority_scheduler*>(this), pri); + } + + void run() + { + std::unique_lock<std::mutex> lock(mutex_); + for (;;) + { + condition_.wait(lock, [&]{ return stopped_ || !queue_.empty(); }); + if (stopped_) + return; + auto p(queue_.top()); + queue_.pop(); + lock.unlock(); + p->execute_(p); + lock.lock(); + } + } + + void stop() + { + std::lock_guard<std::mutex> lock(mutex_); + stopped_ = true; + condition_.notify_all(); + } + +private: + struct item_base + { + int priority_; + void (*execute_)(std::shared_ptr<item_base>&); + }; + + template <class Func> + struct item : item_base + { + item(int pri, Func f) : function_(std::move(f)) + { + priority_ = pri; + execute_ = [](std::shared_ptr<item_base>& p) + { + Func tmp(std::move(static_cast<item*>(p.get())->function_)); + p.reset(); + tmp(); + }; + } + + Func function_; + }; + + struct item_comp + { + bool operator()( + const std::shared_ptr<item_base>& a, + const std::shared_ptr<item_base>& b) + { + return a->priority_ < b->priority_; + } + }; + + std::mutex mutex_; + std::condition_variable condition_; + std::priority_queue< + std::shared_ptr<item_base>, + std::vector<std::shared_ptr<item_base>>, + item_comp> queue_; + bool stopped_ = false; +}; + +int main() +{ + priority_scheduler sched; + auto low = sched.get_executor(0); + auto med = sched.get_executor(1); + auto high = sched.get_executor(2); + dispatch(low, []{ std::cout << "1\n"; }); + dispatch(low, []{ std::cout << "11\n"; }); + dispatch(med, []{ std::cout << "2\n"; }); + dispatch(med, []{ std::cout << "22\n"; }); + dispatch(high, []{ std::cout << "3\n"; }); + dispatch(high, []{ std::cout << "33\n"; }); + dispatch(high, []{ std::cout << "333\n"; }); + dispatch(sched.get_executor(-1), [&]{ sched.stop(); }); + sched.run(); +} |