summaryrefslogtreecommitdiffstatshomepage
path: root/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp
diff options
context:
space:
mode:
Diffstat (limited to '3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp')
-rw-r--r--3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp288
1 files changed, 288 insertions, 0 deletions
diff --git a/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp b/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp
new file mode 100644
index 00000000000..f441e322688
--- /dev/null
+++ b/3rdparty/asio/src/examples/cpp11/executors/pipeline.cpp
@@ -0,0 +1,288 @@
+#include <asio/associated_executor.hpp>
+#include <asio/bind_executor.hpp>
+#include <asio/execution_context.hpp>
+#include <asio/post.hpp>
+#include <asio/system_executor.hpp>
+#include <asio/use_future.hpp>
+#include <condition_variable>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <vector>
+#include <cctype>
+
+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;
+namespace execution = asio::execution;
+
+// 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& query(execution::context_t) const
+ {
+ return asio::query(system_executor(), execution::context);
+ }
+
+ execution::blocking_t query(execution::blocking_t) const
+ {
+ return execution::blocking.never;
+ }
+
+ thread_executor require(execution::blocking_t::never_t) const
+ {
+ return *this;
+ }
+
+ template <class Func>
+ void execute(Func f) const
+ {
+ thread_bag& bag = use_service<thread_bag>(query(execution::context));
+ bag.add_thread(std::thread(std::move(f)));
+ }
+
+ 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]() 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]() 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]() 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(1);
+
+ auto f = pipeline(reader, filter, bind_executor(pool, upper), writer);
+ f.wait();
+}