[DEV] add v1.76.0

libs/asio/example/cpp14/executors/Jamfile.v2

@@ -0,0 +1,38 @@
+#
+# Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+#
+# Distributed under the Boost Software License, Version 1.0. (See accompanying
+# file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+#
+
+lib socket ; # SOLARIS
+lib nsl ; # SOLARIS
+lib ws2_32 ; # NT
+lib mswsock ; # NT
+lib ipv6 ; # HPUX
+lib network ; # HAIKU
+
+project
+  : requirements
+    <library>/boost/system//boost_system
+    <define>BOOST_ALL_NO_LIB=1
+    <threading>multi
+    <target-os>solaris:<library>socket
+    <target-os>solaris:<library>nsl
+    <target-os>windows:<define>_WIN32_WINNT=0x0501
+    <target-os>windows,<toolset>gcc:<library>ws2_32
+    <target-os>windows,<toolset>gcc:<library>mswsock
+    <target-os>windows,<toolset>gcc-cygwin:<define>__USE_W32_SOCKETS
+    <target-os>hpux,<toolset>gcc:<define>_XOPEN_SOURCE_EXTENDED
+    <target-os>hpux:<library>ipv6
+    <target-os>haiku:<library>network
+  ;
+
+exe actor : actor.cpp ;
+exe async_1 : async_1.cpp ;
+exe async_2 : async_2.cpp ;
+exe bank_account_1 : bank_account_1.cpp ;
+exe bank_account_2 : bank_account_2.cpp ;
+exe fork_join : fork_join.cpp ;
+exe pipeline : pipeline.cpp ;
+exe priority_scheduler : priority_scheduler.cpp ;

libs/asio/example/cpp14/executors/actor.cpp

@@ -0,0 +1,285 @@
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/defer.hpp>
+#include <boost/asio/post.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/asio/system_executor.hpp>
+#include <condition_variable>
+#include <deque>
+#include <memory>
+#include <mutex>
+#include <typeinfo>
+#include <vector>
+
+using boost::asio::any_io_executor;
+using boost::asio::defer;
+using boost::asio::post;
+using boost::asio::strand;
+using boost::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)]() mutable
+      {
+        to->call_handler(std::move(msg), from);
+      });
+  }
+
+protected:
+  // Construct the actor to use the specified executor for all message handlers.
+  actor(any_io_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);
+    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<any_io_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 <boost/asio/thread_pool.hpp>
+#include <iostream>
+
+using boost::asio::thread_pool;
+
+class member : public actor
+{
+public:
+  explicit member(any_io_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();
+}

libs/asio/example/cpp14/executors/async_1.cpp

@@ -0,0 +1,56 @@
+#include <boost/asio/associated_executor.hpp>
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/execution.hpp>
+#include <boost/asio/static_thread_pool.hpp>
+#include <iostream>
+#include <string>
+
+using boost::asio::bind_executor;
+using boost::asio::get_associated_executor;
+using boost::asio::static_thread_pool;
+namespace execution = boost::asio::execution;
+
+// A function to asynchronously read a single line from an input stream.
+template <class IoExecutor, class Handler>
+void async_getline(IoExecutor io_ex, std::istream& is, Handler handler)
+{
+  // Track work for the handler's associated executor.
+  auto work_ex = boost::asio::prefer(
+      get_associated_executor(handler, io_ex),
+      execution::outstanding_work.tracked);
+
+  // Post a function object to do the work asynchronously.
+  execution::execute(
+      boost::asio::require(io_ex, execution::blocking.never),
+      [&is, work_ex, handler=std::move(handler)]() mutable
+      {
+        std::string line;
+        std::getline(is, line);
+
+        // Pass the result to the handler, via the associated executor.
+        execution::execute(
+            boost::asio::prefer(work_ex, execution::blocking.possibly),
+            [line=std::move(line), handler=std::move(handler)]() mutable
+            {
+              handler(std::move(line));
+            });
+      });
+}
+
+int main()
+{
+  static_thread_pool io_pool(1);
+  static_thread_pool completion_pool(1);
+
+  std::cout << "Enter a line: ";
+
+  async_getline(io_pool.executor(), std::cin,
+      bind_executor(completion_pool.executor(),
+        [](std::string line)
+        {
+          std::cout << "Line: " << line << "\n";
+        }));
+
+  io_pool.wait();
+  completion_pool.wait();
+}

libs/asio/example/cpp14/executors/async_2.cpp

@@ -0,0 +1,81 @@
+#include <boost/asio/associated_executor.hpp>
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/execution.hpp>
+#include <boost/asio/static_thread_pool.hpp>
+#include <iostream>
+#include <string>
+
+using boost::asio::bind_executor;
+using boost::asio::get_associated_executor;
+using boost::asio::static_thread_pool;
+namespace execution = boost::asio::execution;
+
+// A function to asynchronously read a single line from an input stream.
+template <class IoExecutor, class Handler>
+void async_getline(IoExecutor io_ex, std::istream& is, Handler handler)
+{
+  // Track work for the handler's associated executor.
+  auto work_ex = boost::asio::prefer(
+      get_associated_executor(handler, io_ex),
+      execution::outstanding_work.tracked);
+
+  // Post a function object to do the work asynchronously.
+  execution::execute(
+      boost::asio::require(io_ex, execution::blocking.never),
+      [&is, work_ex, handler=std::move(handler)]() mutable
+      {
+        std::string line;
+        std::getline(is, line);
+
+        // Pass the result to the handler, via the associated executor.
+        execution::execute(
+            boost::asio::prefer(work_ex, execution::blocking.possibly),
+            [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 IoExecutor, class Handler>
+void async_getlines(IoExecutor io_ex, std::istream& is, std::string init, Handler handler)
+{
+  // Track work for the I/O executor.
+  auto io_work_ex = boost::asio::prefer(io_ex,
+      execution::outstanding_work.tracked);
+
+  // Track work for the handler's associated executor.
+  auto handler_work_ex = boost::asio::prefer(
+      get_associated_executor(handler, io_ex),
+      execution::outstanding_work.tracked);
+
+  // Use the associated executor for each operation in the composition.
+  async_getline(io_work_ex, is,
+      bind_executor(handler_work_ex,
+        [io_work_ex, &is, lines=std::move(init), handler=std::move(handler)]
+        (std::string line) mutable
+        {
+          if (line.empty())
+            handler(lines);
+          else
+            async_getlines(io_work_ex, is, lines + line + "\n", std::move(handler));
+        }));
+}
+
+int main()
+{
+  static_thread_pool io_pool(1);
+  static_thread_pool completion_pool(1);
+
+  std::cout << "Enter text, terminating with a blank line:\n";
+
+  async_getlines(io_pool.executor(), std::cin, "",
+      bind_executor(completion_pool.executor(), [](std::string lines)
+        {
+          std::cout << "Lines:\n" << lines << "\n";
+        }));
+
+  io_pool.wait();
+  completion_pool.wait();
+}

libs/asio/example/cpp14/executors/bank_account_1.cpp

@@ -0,0 +1,60 @@
+#include <boost/asio/execution.hpp>
+#include <boost/asio/static_thread_pool.hpp>
+#include <iostream>
+
+using boost::asio::static_thread_pool;
+namespace execution = boost::asio::execution;
+
+// Traditional active object pattern.
+// Member functions do not block.
+
+class bank_account
+{
+  int balance_ = 0;
+  mutable static_thread_pool pool_{1};
+
+public:
+  void deposit(int amount)
+  {
+    execution::execute(
+        pool_.executor(),
+        [this, amount]
+        {
+          balance_ += amount;
+        });
+  }
+
+  void withdraw(int amount)
+  {
+    execution::execute(
+        pool_.executor(),
+        [this, amount]
+        {
+          if (balance_ >= amount)
+            balance_ -= amount;
+        });
+  }
+
+  void print_balance() const
+  {
+    execution::execute(
+        pool_.executor(),
+        [this]
+        {
+          std::cout << "balance = " << balance_ << "\n";
+        });
+  }
+
+  ~bank_account()
+  {
+    pool_.wait();
+  }
+};
+
+int main()
+{
+  bank_account acct;
+  acct.deposit(20);
+  acct.withdraw(10);
+  acct.print_balance();
+}

libs/asio/example/cpp14/executors/bank_account_2.cpp

@@ -0,0 +1,60 @@
+#include <boost/asio/execution.hpp>
+#include <boost/asio/static_thread_pool.hpp>
+#include <iostream>
+
+using boost::asio::static_thread_pool;
+namespace execution = boost::asio::execution;
+
+// Traditional active object pattern.
+// Member functions block until operation is finished.
+
+class bank_account
+{
+  int balance_ = 0;
+  mutable static_thread_pool pool_{1};
+
+public:
+  void deposit(int amount)
+  {
+    execution::execute(
+        boost::asio::require(pool_.executor(),
+          execution::blocking.always),
+        [this, amount]
+        {
+          balance_ += amount;
+        });
+  }
+
+  void withdraw(int amount)
+  {
+    execution::execute(
+        boost::asio::require(pool_.executor(),
+          execution::blocking.always),
+        [this, amount]
+        {
+          if (balance_ >= amount)
+            balance_ -= amount;
+        });
+  }
+
+  int balance() const
+  {
+    int result = 0;
+    execution::execute(
+        boost::asio::require(pool_.executor(),
+          execution::blocking.always),
+        [this, &result]
+        {
+          result = balance_;
+        });
+    return result;
+  }
+};
+
+int main()
+{
+  bank_account acct;
+  acct.deposit(20);
+  acct.withdraw(10);
+  std::cout << "balance = " << acct.balance() << "\n";
+}

libs/asio/example/cpp14/executors/fork_join.cpp

@@ -0,0 +1,290 @@
+#include <boost/asio/execution.hpp>
+#include <boost/asio/static_thread_pool.hpp>
+#include <algorithm>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <numeric>
+
+using boost::asio::static_thread_pool;
+namespace execution = boost::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:
+  // 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::max(std::thread::hardware_concurrency(), 1u) * 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_)
+      {
+        execution::execute(
+            threads_.executor(),
+            [this]
+            {
+              std::unique_lock<std::mutex> lock(mutex_);
+              while (use_count_ > 0)
+                if (!execute_next(lock))
+                  condition_.wait(lock);
+            });
+      }
+    }
+    catch (...)
+    {
+      stop_threads();
+      threads_.wait();
+      throw;
+    }
+  }
+
+  // The destructor waits for the pool to finish executing functions.
+  ~fork_join_pool()
+  {
+    stop_threads();
+    threads_.wait();
+  }
+
+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_execute(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();
+    }
+  }
+
+  // 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_;
+  static_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& query(execution::context_t) const noexcept
+  {
+    return context_;
+  }
+
+  template <class Func>
+  void execute(Func f) const
+  {
+    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,
+      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);
+      execution::execute(fork, [=]{ fork_join_sort(begin, begin + n / 2); });
+      execution::execute(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;
+}

libs/asio/example/cpp14/executors/pipeline.cpp

@@ -0,0 +1,284 @@
+#include <boost/asio/associated_executor.hpp>
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/execution.hpp>
+#include <condition_variable>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <thread>
+#include <vector>
+#include <cctype>
+
+using boost::asio::executor_binder;
+using boost::asio::get_associated_executor;
+namespace execution = boost::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:
+  // Singleton execution context that manages threads launched by the new_thread_executor.
+  class thread_bag
+  {
+    friend class thread_executor;
+
+    void add_thread(std::thread&& t)
+    {
+      std::unique_lock<std::mutex> lock(mutex_);
+      threads_.push_back(std::move(t));
+    }
+
+    thread_bag() = default;
+
+    ~thread_bag()
+    {
+      for (auto& t : threads_)
+        t.join();
+    }
+
+    std::mutex mutex_;
+    std::vector<std::thread> threads_;
+  };
+
+public:
+  static thread_bag& query(execution::context_t)
+  {
+    static thread_bag threads;
+    return threads;
+  }
+
+  static constexpr auto query(execution::blocking_t)
+  {
+    return execution::blocking.never;
+  }
+
+  template <class Func>
+  void execute(Func f) const
+  {
+    thread_bag& 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.
+  std::packaged_task<void()> task(
+      [in, f = std::move(f)]() mutable
+      {
+        f(in);
+      });
+  std::future<void> fut = task.get_future();
+  execution::execute(
+      boost::asio::require(ex, execution::blocking.never),
+      std::move(task));
+  return fut;
+}
+
+// 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.
+  execution::execute(
+      boost::asio::require(ex, execution::blocking.never),
+      [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.
+  execution::execute(
+      boost::asio::require(ex, execution::blocking.never),
+      [out, f = std::move(f)]() mutable
+      {
+        f(out);
+        out.stop();
+      });
+
+  // Launch the rest of the pipeline.
+  return pipeline(next_in, std::move(t)...);
+}
+
+//------------------------------------------------------------------------------
+
+#include <boost/asio/static_thread_pool.hpp>
+#include <iostream>
+#include <string>
+
+using boost::asio::bind_executor;
+using boost::asio::static_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()
+{
+  static_thread_pool pool(1);
+
+  auto f = pipeline(reader, filter, bind_executor(pool.executor(), upper), writer);
+  f.wait();
+}

libs/asio/example/cpp14/executors/priority_scheduler.cpp

@@ -0,0 +1,181 @@
+#include <boost/asio/execution.hpp>
+#include <condition_variable>
+#include <iostream>
+#include <memory>
+#include <mutex>
+#include <queue>
+
+namespace execution = boost::asio::execution;
+
+namespace custom_props {
+
+  struct priority
+  {
+    template <typename T>
+    static constexpr bool is_applicable_property_v =
+      execution::is_executor<T>::value;
+
+    static constexpr bool is_requirable = true;
+    static constexpr bool is_preferable = true;
+
+    using polymorphic_query_result_type = int;
+
+    int value() const { return value_; }
+
+    int value_ = 1;
+  };
+
+  constexpr priority low_priority{0};
+  constexpr priority normal_priority{1};
+  constexpr priority high_priority{2};
+
+} // namespace custom_props
+
+class priority_scheduler
+{
+public:
+  // A class that satisfies the Executor requirements.
+  class executor_type
+  {
+  public:
+    executor_type(priority_scheduler& ctx) noexcept
+      : context_(ctx), priority_(custom_props::normal_priority.value())
+    {
+    }
+
+    priority_scheduler& query(execution::context_t) const noexcept
+    {
+      return context_;
+    }
+
+    int query(custom_props::priority) const noexcept
+    {
+      return priority_;
+    }
+
+    executor_type require(custom_props::priority pri) const
+    {
+      executor_type new_ex(*this);
+      new_ex.priority_ = pri.value();
+      return new_ex;
+    }
+
+    template <class Func>
+    void execute(Func f) const
+    {
+      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();
+    }
+
+    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 executor() noexcept
+  {
+    return executor_type(*const_cast<priority_scheduler*>(this));
+  }
+
+  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 ex = sched.executor();
+  auto prefer_low = boost::asio::prefer(ex, custom_props::low_priority);
+  auto low = boost::asio::require(ex, custom_props::low_priority);
+  auto med = boost::asio::require(ex, custom_props::normal_priority);
+  auto high = boost::asio::require(ex, custom_props::high_priority);
+  execution::any_executor<custom_props::priority> poly_high(high);
+  execution::execute(prefer_low, []{ std::cout << "1\n"; });
+  execution::execute(low, []{ std::cout << "11\n"; });
+  execution::execute(low, []{ std::cout << "111\n"; });
+  execution::execute(med, []{ std::cout << "2\n"; });
+  execution::execute(med, []{ std::cout << "22\n"; });
+  execution::execute(high, []{ std::cout << "3\n"; });
+  execution::execute(high, []{ std::cout << "33\n"; });
+  execution::execute(high, []{ std::cout << "333\n"; });
+  execution::execute(poly_high, []{ std::cout << "3333\n"; });
+  execution::execute(boost::asio::require(ex, custom_props::priority{-1}), [&]{ sched.stop(); });
+  sched.run();
+  std::cout << "polymorphic query result = " << boost::asio::query(poly_high, custom_props::priority{}) << "\n";
+}

libs/asio/example/cpp14/operations/Jamfile.v2

@@ -0,0 +1,39 @@
+#
+# Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+#
+# Distributed under the Boost Software License, Version 1.0. (See accompanying
+# file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+#
+
+lib socket ; # SOLARIS
+lib nsl ; # SOLARIS
+lib ws2_32 ; # NT
+lib mswsock ; # NT
+lib ipv6 ; # HPUX
+lib network ; # HAIKU
+
+project
+  : requirements
+    <library>/boost/system//boost_system
+    <library>/boost/chrono//boost_chrono
+    <define>BOOST_ALL_NO_LIB=1
+    <threading>multi
+    <target-os>solaris:<library>socket
+    <target-os>solaris:<library>nsl
+    <target-os>windows:<define>_WIN32_WINNT=0x0501
+    <target-os>windows,<toolset>gcc:<library>ws2_32
+    <target-os>windows,<toolset>gcc:<library>mswsock
+    <target-os>windows,<toolset>gcc-cygwin:<define>__USE_W32_SOCKETS
+    <target-os>hpux,<toolset>gcc:<define>_XOPEN_SOURCE_EXTENDED
+    <target-os>hpux:<library>ipv6
+    <target-os>haiku:<library>network
+  ;
+
+exe composed_1 : composed_1.cpp ;
+exe composed_2 : composed_2.cpp ;
+exe composed_3 : composed_3.cpp ;
+exe composed_4 : composed_4.cpp ;
+exe composed_5 : composed_5.cpp ;
+exe composed_6 : composed_6.cpp ;
+exe composed_7 : composed_7.cpp ;
+exe composed_8 : composed_8.cpp ;

libs/asio/example/cpp14/operations/composed_1.cpp

@@ -0,0 +1,113 @@
+//
+// composed_1.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <cstring>
+#include <iostream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+//------------------------------------------------------------------------------
+
+// This is the simplest example of a composed asynchronous operation, where we
+// simply repackage an existing operation. The asynchronous operation
+// requirements are met by delegating responsibility to the underlying
+// operation.
+
+template <typename CompletionToken>
+auto async_write_message(tcp::socket& socket,
+    const char* message, CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is void. However,
+  // when the completion token is boost::asio::yield_context (used for stackful
+  // coroutines) the return type would be std::size_t, and when the completion
+  // token is boost::asio::use_future it would be std::future<std::size_t>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of our underlying asynchronous operation
+{
+  // When delegating to the underlying operation we must take care to perfectly
+  // forward the completion token. This ensures that our operation works
+  // correctly with move-only function objects as callbacks, as well as other
+  // completion token types.
+  return boost::asio::async_write(socket,
+      boost::asio::buffer(message, std::strlen(message)),
+      std::forward<CompletionToken>(token));
+}
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_message(socket, "Testing callback\r\n",
+      [](const boost::system::error_code& error, std::size_t n)
+      {
+        if (!error)
+        {
+          std::cout << n << " bytes transferred\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<std::size_t> f = async_write_message(
+      socket, "Testing future\r\n", boost::asio::use_future);
+
+  io_context.run();
+
+  try
+  {
+    // Get the result of the operation.
+    std::size_t n = f.get();
+    std::cout << n << " bytes transferred\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Error: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}

libs/asio/example/cpp14/operations/composed_2.cpp

@@ -0,0 +1,131 @@
+//
+// composed_2.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <cstring>
+#include <iostream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+//------------------------------------------------------------------------------
+
+// This next simplest example of a composed asynchronous operation involves
+// repackaging multiple operations but choosing to invoke just one of them. All
+// of these underlying operations have the same completion signature. The
+// asynchronous operation requirements are met by delegating responsibility to
+// the underlying operations.
+
+template <typename CompletionToken>
+auto async_write_message(tcp::socket& socket,
+    const char* message, bool allow_partial_write,
+    CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is void. However,
+  // when the completion token is boost::asio::yield_context (used for stackful
+  // coroutines) the return type would be std::size_t, and when the completion
+  // token is boost::asio::use_future it would be std::future<std::size_t>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of our underlying asynchronous operation
+{
+  // As the return type of the initiating function is deduced solely from the
+  // CompletionToken and completion signature, we know that two different
+  // asynchronous operations having the same completion signature will produce
+  // the same return type, when passed the same CompletionToken. This allows us
+  // to trivially delegate to alternate implementations.
+  if (allow_partial_write)
+  {
+    // When delegating to an underlying operation we must take care to
+    // perfectly forward the completion token. This ensures that our operation
+    // works correctly with move-only function objects as callbacks, as well as
+    // other completion token types.
+    return socket.async_write_some(
+        boost::asio::buffer(message, std::strlen(message)),
+        std::forward<CompletionToken>(token));
+  }
+  else
+  {
+    // As above, we must perfectly forward the completion token when calling
+    // the alternate underlying operation.
+    return boost::asio::async_write(socket,
+        boost::asio::buffer(message, std::strlen(message)),
+        std::forward<CompletionToken>(token));
+  }
+}
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_message(socket, "Testing callback\r\n", false,
+      [](const boost::system::error_code& error, std::size_t n)
+      {
+        if (!error)
+        {
+          std::cout << n << " bytes transferred\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<std::size_t> f = async_write_message(
+      socket, "Testing future\r\n", false, boost::asio::use_future);
+
+  io_context.run();
+
+  try
+  {
+    // Get the result of the operation.
+    std::size_t n = f.get();
+    std::cout << n << " bytes transferred\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Error: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}

libs/asio/example/cpp14/operations/composed_3.cpp

@@ -0,0 +1,186 @@
+//
+// composed_3.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <cstring>
+#include <functional>
+#include <iostream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+// NOTE: This example requires the new boost::asio::async_initiate function. For
+// an example that works with the Networking TS style of completion tokens,
+// please see an older version of asio.
+
+//------------------------------------------------------------------------------
+
+// In this composed operation we repackage an existing operation, but with a
+// different completion handler signature. The asynchronous operation
+// requirements are met by delegating responsibility to the underlying
+// operation.
+
+template <typename CompletionToken>
+auto async_write_message(tcp::socket& socket,
+    const char* message, CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is always void.
+  // In this example, when the completion token is boost::asio::yield_context
+  // (used for stackful coroutines) the return type would be also be void, as
+  // there is no non-error argument to the completion handler. When the
+  // completion token is boost::asio::use_future it would be std::future<void>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of boost::asio::async_initiate.
+{
+  // In addition to determining the mechanism by which an asynchronous
+  // operation delivers its result, a completion token also determines the time
+  // when the operation commences. For example, when the completion token is a
+  // simple callback the operation commences before the initiating function
+  // returns. However, if the completion token's delivery mechanism uses a
+  // future, we might instead want to defer initiation of the operation until
+  // the returned future object is waited upon.
+  //
+  // To enable this, when implementing an asynchronous operation we must
+  // package the initiation step as a function object. The initiation function
+  // object's call operator is passed the concrete completion handler produced
+  // by the completion token. This completion handler matches the asynchronous
+  // operation's completion handler signature, which in this example is:
+  //
+  //   void(boost::system::error_code error)
+  //
+  // The initiation function object also receives any additional arguments
+  // required to start the operation. (Note: We could have instead passed these
+  // arguments in the lambda capture set. However, we should prefer to
+  // propagate them as function call arguments as this allows the completion
+  // token to optimise how they are passed. For example, a lazy future which
+  // defers initiation would need to make a decay-copy of the arguments, but
+  // when using a simple callback the arguments can be trivially forwarded
+  // straight through.)
+  auto initiation = [](auto&& completion_handler,
+      tcp::socket& socket, const char* message)
+  {
+    // The async_write operation has a completion handler signature of:
+    //
+    //   void(boost::system::error_code error, std::size n)
+    //
+    // This differs from our operation's signature in that it is also passed
+    // the number of bytes transferred as an argument of type std::size_t. We
+    // will adapt our completion handler to async_write's completion handler
+    // signature by using std::bind, which drops the additional argument.
+    //
+    // However, it is essential to the correctness of our composed operation
+    // that we preserve the executor of the user-supplied completion handler.
+    // The std::bind function will not do this for us, so we must do this by
+    // first obtaining the completion handler's associated executor (defaulting
+    // to the I/O executor - in this case the executor of the socket - if the
+    // completion handler does not have its own) ...
+    auto executor = boost::asio::get_associated_executor(
+        completion_handler, socket.get_executor());
+
+    // ... and then binding this executor to our adapted completion handler
+    // using the boost::asio::bind_executor function.
+    boost::asio::async_write(socket,
+        boost::asio::buffer(message, std::strlen(message)),
+        boost::asio::bind_executor(executor,
+          std::bind(std::forward<decltype(completion_handler)>(
+              completion_handler), std::placeholders::_1)));
+  };
+
+  // The boost::asio::async_initiate function takes:
+  //
+  // - our initiation function object,
+  // - the completion token,
+  // - the completion handler signature, and
+  // - any additional arguments we need to initiate the operation.
+  //
+  // It then asks the completion token to create a completion handler (i.e. a
+  // callback) with the specified signature, and invoke the initiation function
+  // object with this completion handler as well as the additional arguments.
+  // The return value of async_initiate is the result of our operation's
+  // initiating function.
+  //
+  // Note that we wrap non-const reference arguments in std::reference_wrapper
+  // to prevent incorrect decay-copies of these objects.
+  return boost::asio::async_initiate<
+    CompletionToken, void(boost::system::error_code)>(
+      initiation, token, std::ref(socket), message);
+}
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_message(socket, "Testing callback\r\n",
+      [](const boost::system::error_code& error)
+      {
+        if (!error)
+        {
+          std::cout << "Message sent\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<void> f = async_write_message(
+      socket, "Testing future\r\n", boost::asio::use_future);
+
+  io_context.run();
+
+  // Get the result of the operation.
+  try
+  {
+    // Get the result of the operation.
+    f.get();
+    std::cout << "Message sent\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Error: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}

libs/asio/example/cpp14/operations/composed_4.cpp

@@ -0,0 +1,201 @@
+//
+// composed_4.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/bind_executor.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <cstring>
+#include <functional>
+#include <iostream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+// NOTE: This example requires the new boost::asio::async_initiate function. For
+// an example that works with the Networking TS style of completion tokens,
+// please see an older version of asio.
+
+//------------------------------------------------------------------------------
+
+// In this composed operation we repackage an existing operation, but with a
+// different completion handler signature. We will also intercept an empty
+// message as an invalid argument, and propagate the corresponding error to the
+// user. The asynchronous operation requirements are met by delegating
+// responsibility to the underlying operation.
+
+template <typename CompletionToken>
+auto async_write_message(tcp::socket& socket,
+    const char* message, CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is always void.
+  // In this example, when the completion token is boost::asio::yield_context
+  // (used for stackful coroutines) the return type would be also be void, as
+  // there is no non-error argument to the completion handler. When the
+  // completion token is boost::asio::use_future it would be std::future<void>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of boost::asio::async_initiate.
+{
+  // In addition to determining the mechanism by which an asynchronous
+  // operation delivers its result, a completion token also determines the time
+  // when the operation commences. For example, when the completion token is a
+  // simple callback the operation commences before the initiating function
+  // returns. However, if the completion token's delivery mechanism uses a
+  // future, we might instead want to defer initiation of the operation until
+  // the returned future object is waited upon.
+  //
+  // To enable this, when implementing an asynchronous operation we must
+  // package the initiation step as a function object. The initiation function
+  // object's call operator is passed the concrete completion handler produced
+  // by the completion token. This completion handler matches the asynchronous
+  // operation's completion handler signature, which in this example is:
+  //
+  //   void(boost::system::error_code error)
+  //
+  // The initiation function object also receives any additional arguments
+  // required to start the operation. (Note: We could have instead passed these
+  // arguments in the lambda capture set. However, we should prefer to
+  // propagate them as function call arguments as this allows the completion
+  // token to optimise how they are passed. For example, a lazy future which
+  // defers initiation would need to make a decay-copy of the arguments, but
+  // when using a simple callback the arguments can be trivially forwarded
+  // straight through.)
+  auto initiation = [](auto&& completion_handler,
+      tcp::socket& socket, const char* message)
+  {
+    // The post operation has a completion handler signature of:
+    //
+    //   void()
+    //
+    // and the async_write operation has a completion handler signature of:
+    //
+    //   void(boost::system::error_code error, std::size n)
+    //
+    // Both of these operations' completion handler signatures differ from our
+    // operation's completion handler signature. We will adapt our completion
+    // handler to these signatures by using std::bind, which drops the
+    // additional arguments.
+    //
+    // However, it is essential to the correctness of our composed operation
+    // that we preserve the executor of the user-supplied completion handler.
+    // The std::bind function will not do this for us, so we must do this by
+    // first obtaining the completion handler's associated executor (defaulting
+    // to the I/O executor - in this case the executor of the socket - if the
+    // completion handler does not have its own) ...
+    auto executor = boost::asio::get_associated_executor(
+        completion_handler, socket.get_executor());
+
+    // ... and then binding this executor to our adapted completion handler
+    // using the boost::asio::bind_executor function.
+    std::size_t length = std::strlen(message);
+    if (length == 0)
+    {
+      boost::asio::post(
+          boost::asio::bind_executor(executor,
+            std::bind(std::forward<decltype(completion_handler)>(
+                completion_handler), boost::asio::error::invalid_argument)));
+    }
+    else
+    {
+      boost::asio::async_write(socket,
+          boost::asio::buffer(message, length),
+          boost::asio::bind_executor(executor,
+            std::bind(std::forward<decltype(completion_handler)>(
+                completion_handler), std::placeholders::_1)));
+    }
+  };
+
+  // The boost::asio::async_initiate function takes:
+  //
+  // - our initiation function object,
+  // - the completion token,
+  // - the completion handler signature, and
+  // - any additional arguments we need to initiate the operation.
+  //
+  // It then asks the completion token to create a completion handler (i.e. a
+  // callback) with the specified signature, and invoke the initiation function
+  // object with this completion handler as well as the additional arguments.
+  // The return value of async_initiate is the result of our operation's
+  // initiating function.
+  //
+  // Note that we wrap non-const reference arguments in std::reference_wrapper
+  // to prevent incorrect decay-copies of these objects.
+  return boost::asio::async_initiate<
+    CompletionToken, void(boost::system::error_code)>(
+      initiation, token, std::ref(socket), message);
+}
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_message(socket, "",
+      [](const boost::system::error_code& error)
+      {
+        if (!error)
+        {
+          std::cout << "Message sent\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<void> f = async_write_message(
+      socket, "", boost::asio::use_future);
+
+  io_context.run();
+
+  try
+  {
+    // Get the result of the operation.
+    f.get();
+    std::cout << "Message sent\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Exception: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}

libs/asio/example/cpp14/operations/composed_5.cpp

@@ -0,0 +1,238 @@
+//
+// composed_5.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <functional>
+#include <iostream>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+// NOTE: This example requires the new boost::asio::async_initiate function. For
+// an example that works with the Networking TS style of completion tokens,
+// please see an older version of asio.
+
+//------------------------------------------------------------------------------
+
+// This composed operation automatically serialises a message, using its I/O
+// streams insertion operator, before sending it on the socket. To do this, it
+// must allocate a buffer for the encoded message and ensure this buffer's
+// validity until the underlying async_write operation completes.
+
+template <typename T, typename CompletionToken>
+auto async_write_message(tcp::socket& socket,
+    const T& message, CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is always void.
+  // In this example, when the completion token is boost::asio::yield_context
+  // (used for stackful coroutines) the return type would be also be void, as
+  // there is no non-error argument to the completion handler. When the
+  // completion token is boost::asio::use_future it would be std::future<void>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of boost::asio::async_initiate.
+{
+  // In addition to determining the mechanism by which an asynchronous
+  // operation delivers its result, a completion token also determines the time
+  // when the operation commences. For example, when the completion token is a
+  // simple callback the operation commences before the initiating function
+  // returns. However, if the completion token's delivery mechanism uses a
+  // future, we might instead want to defer initiation of the operation until
+  // the returned future object is waited upon.
+  //
+  // To enable this, when implementing an asynchronous operation we must
+  // package the initiation step as a function object. The initiation function
+  // object's call operator is passed the concrete completion handler produced
+  // by the completion token. This completion handler matches the asynchronous
+  // operation's completion handler signature, which in this example is:
+  //
+  //   void(boost::system::error_code error)
+  //
+  // The initiation function object also receives any additional arguments
+  // required to start the operation. (Note: We could have instead passed these
+  // arguments in the lambda capture set. However, we should prefer to
+  // propagate them as function call arguments as this allows the completion
+  // token to optimise how they are passed. For example, a lazy future which
+  // defers initiation would need to make a decay-copy of the arguments, but
+  // when using a simple callback the arguments can be trivially forwarded
+  // straight through.)
+  auto initiation = [](auto&& completion_handler,
+      tcp::socket& socket, std::unique_ptr<std::string> encoded_message)
+  {
+    // In this example, the composed operation's intermediate completion
+    // handler is implemented as a hand-crafted function object, rather than
+    // using a lambda or std::bind.
+    struct intermediate_completion_handler
+    {
+      // The intermediate completion handler holds a reference to the socket so
+      // that it can obtain the I/O executor (see get_executor below).
+      tcp::socket& socket_;
+
+      // The allocated buffer for the encoded message. The std::unique_ptr
+      // smart pointer is move-only, and as a consequence our intermediate
+      // completion handler is also move-only.
+      std::unique_ptr<std::string> encoded_message_;
+
+      // The user-supplied completion handler.
+      typename std::decay<decltype(completion_handler)>::type handler_;
+
+      // The function call operator matches the completion signature of the
+      // async_write operation.
+      void operator()(const boost::system::error_code& error, std::size_t /*n*/)
+      {
+        // Deallocate the encoded message before calling the user-supplied
+        // completion handler.
+        encoded_message_.reset();
+
+        // Call the user-supplied handler with the result of the operation.
+        // The arguments must match the completion signature of our composed
+        // operation.
+        handler_(error);
+      }
+
+      // It is essential to the correctness of our composed operation that we
+      // preserve the executor of the user-supplied completion handler. With a
+      // hand-crafted function object we can do this by defining a nested type
+      // executor_type and member function get_executor. These obtain the
+      // completion handler's associated executor, and default to the I/O
+      // executor - in this case the executor of the socket - if the completion
+      // handler does not have its own.
+      using executor_type = boost::asio::associated_executor_t<
+          typename std::decay<decltype(completion_handler)>::type,
+          tcp::socket::executor_type>;
+
+      executor_type get_executor() const noexcept
+      {
+        return boost::asio::get_associated_executor(
+            handler_, socket_.get_executor());
+      }
+
+      // Although not necessary for correctness, we may also preserve the
+      // allocator of the user-supplied completion handler. This is achieved by
+      // defining a nested type allocator_type and member function
+      // get_allocator. These obtain the completion handler's associated
+      // allocator, and default to std::allocator<void> if the completion
+      // handler does not have its own.
+      using allocator_type = boost::asio::associated_allocator_t<
+          typename std::decay<decltype(completion_handler)>::type,
+          std::allocator<void>>;
+
+      allocator_type get_allocator() const noexcept
+      {
+        return boost::asio::get_associated_allocator(
+            handler_, std::allocator<void>{});
+      }
+    };
+
+    // Initiate the underlying async_write operation using our intermediate
+    // completion handler.
+    auto encoded_message_buffer = boost::asio::buffer(*encoded_message);
+    boost::asio::async_write(socket, encoded_message_buffer,
+        intermediate_completion_handler{socket, std::move(encoded_message),
+          std::forward<decltype(completion_handler)>(completion_handler)});
+  };
+
+  // Encode the message and copy it into an allocated buffer. The buffer will
+  // be maintained for the lifetime of the asynchronous operation.
+  std::ostringstream os;
+  os << message;
+  std::unique_ptr<std::string> encoded_message(new std::string(os.str()));
+
+  // The boost::asio::async_initiate function takes:
+  //
+  // - our initiation function object,
+  // - the completion token,
+  // - the completion handler signature, and
+  // - any additional arguments we need to initiate the operation.
+  //
+  // It then asks the completion token to create a completion handler (i.e. a
+  // callback) with the specified signature, and invoke the initiation function
+  // object with this completion handler as well as the additional arguments.
+  // The return value of async_initiate is the result of our operation's
+  // initiating function.
+  //
+  // Note that we wrap non-const reference arguments in std::reference_wrapper
+  // to prevent incorrect decay-copies of these objects.
+  return boost::asio::async_initiate<
+    CompletionToken, void(boost::system::error_code)>(
+      initiation, token, std::ref(socket),
+      std::move(encoded_message));
+}
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_message(socket, 123456,
+      [](const boost::system::error_code& error)
+      {
+        if (!error)
+        {
+          std::cout << "Message sent\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<void> f = async_write_message(
+      socket, 654.321, boost::asio::use_future);
+
+  io_context.run();
+
+  try
+  {
+    // Get the result of the operation.
+    f.get();
+    std::cout << "Message sent\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Exception: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}

libs/asio/example/cpp14/operations/composed_6.cpp

@@ -0,0 +1,298 @@
+//
+// composed_6.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <functional>
+#include <iostream>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+// NOTE: This example requires the new boost::asio::async_initiate function. For
+// an example that works with the Networking TS style of completion tokens,
+// please see an older version of asio.
+
+//------------------------------------------------------------------------------
+
+// This composed operation shows composition of multiple underlying operations.
+// It automatically serialises a message, using its I/O streams insertion
+// operator, before sending it N times on the socket. To do this, it must
+// allocate a buffer for the encoded message and ensure this buffer's validity
+// until all underlying async_write operation complete. A one second delay is
+// inserted prior to each write operation, using a steady_timer.
+
+template <typename T, typename CompletionToken>
+auto async_write_messages(tcp::socket& socket,
+    const T& message, std::size_t repeat_count,
+    CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is always void.
+  // In this example, when the completion token is boost::asio::yield_context
+  // (used for stackful coroutines) the return type would be also be void, as
+  // there is no non-error argument to the completion handler. When the
+  // completion token is boost::asio::use_future it would be std::future<void>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of boost::asio::async_initiate.
+{
+  // In addition to determining the mechanism by which an asynchronous
+  // operation delivers its result, a completion token also determines the time
+  // when the operation commences. For example, when the completion token is a
+  // simple callback the operation commences before the initiating function
+  // returns. However, if the completion token's delivery mechanism uses a
+  // future, we might instead want to defer initiation of the operation until
+  // the returned future object is waited upon.
+  //
+  // To enable this, when implementing an asynchronous operation we must
+  // package the initiation step as a function object. The initiation function
+  // object's call operator is passed the concrete completion handler produced
+  // by the completion token. This completion handler matches the asynchronous
+  // operation's completion handler signature, which in this example is:
+  //
+  //   void(boost::system::error_code error)
+  //
+  // The initiation function object also receives any additional arguments
+  // required to start the operation. (Note: We could have instead passed these
+  // arguments in the lambda capture set. However, we should prefer to
+  // propagate them as function call arguments as this allows the completion
+  // token to optimise how they are passed. For example, a lazy future which
+  // defers initiation would need to make a decay-copy of the arguments, but
+  // when using a simple callback the arguments can be trivially forwarded
+  // straight through.)
+  auto initiation = [](auto&& completion_handler, tcp::socket& socket,
+      std::unique_ptr<std::string> encoded_message, std::size_t repeat_count,
+      std::unique_ptr<boost::asio::steady_timer> delay_timer)
+  {
+    // In this example, the composed operation's intermediate completion
+    // handler is implemented as a hand-crafted function object.
+    struct intermediate_completion_handler
+    {
+      // The intermediate completion handler holds a reference to the socket as
+      // it is used for multiple async_write operations, as well as for
+      // obtaining the I/O executor (see get_executor below).
+      tcp::socket& socket_;
+
+      // The allocated buffer for the encoded message. The std::unique_ptr
+      // smart pointer is move-only, and as a consequence our intermediate
+      // completion handler is also move-only.
+      std::unique_ptr<std::string> encoded_message_;
+
+      // The repeat count remaining.
+      std::size_t repeat_count_;
+
+      // A steady timer used for introducing a delay.
+      std::unique_ptr<boost::asio::steady_timer> delay_timer_;
+
+      // To manage the cycle between the multiple underlying asychronous
+      // operations, our intermediate completion handler is implemented as a
+      // state machine.
+      enum { starting, waiting, writing } state_;
+
+      // As our composed operation performs multiple underlying I/O operations,
+      // we should maintain a work object against the I/O executor. This tells
+      // the I/O executor that there is still more work to come in the future.
+      typename std::decay<decltype(boost::asio::prefer(
+            std::declval<tcp::socket::executor_type>(),
+            boost::asio::execution::outstanding_work.tracked))>::type io_work_;
+
+      // The user-supplied completion handler, called once only on completion
+      // of the entire composed operation.
+      typename std::decay<decltype(completion_handler)>::type handler_;
+
+      // By having a default value for the second argument, this function call
+      // operator matches the completion signature of both the async_write and
+      // steady_timer::async_wait operations.
+      void operator()(const boost::system::error_code& error, std::size_t = 0)
+      {
+        if (!error)
+        {
+          switch (state_)
+          {
+          case starting:
+          case writing:
+            if (repeat_count_ > 0)
+            {
+              --repeat_count_;
+              state_ = waiting;
+              delay_timer_->expires_after(std::chrono::seconds(1));
+              delay_timer_->async_wait(std::move(*this));
+              return; // Composed operation not yet complete.
+            }
+            break; // Composed operation complete, continue below.
+          case waiting:
+            state_ = writing;
+            boost::asio::async_write(socket_,
+                boost::asio::buffer(*encoded_message_), std::move(*this));
+            return; // Composed operation not yet complete.
+          }
+        }
+
+        // This point is reached only on completion of the entire composed
+        // operation.
+
+        // Deallocate the encoded message before calling the user-supplied
+        // completion handler.
+        encoded_message_.reset();
+
+        // Call the user-supplied handler with the result of the operation.
+        handler_(error);
+      }
+
+      // It is essential to the correctness of our composed operation that we
+      // preserve the executor of the user-supplied completion handler. With a
+      // hand-crafted function object we can do this by defining a nested type
+      // executor_type and member function get_executor. These obtain the
+      // completion handler's associated executor, and default to the I/O
+      // executor - in this case the executor of the socket - if the completion
+      // handler does not have its own.
+      using executor_type = boost::asio::associated_executor_t<
+          typename std::decay<decltype(completion_handler)>::type,
+          tcp::socket::executor_type>;
+
+      executor_type get_executor() const noexcept
+      {
+        return boost::asio::get_associated_executor(
+            handler_, socket_.get_executor());
+      }
+
+      // Although not necessary for correctness, we may also preserve the
+      // allocator of the user-supplied completion handler. This is achieved by
+      // defining a nested type allocator_type and member function
+      // get_allocator. These obtain the completion handler's associated
+      // allocator, and default to std::allocator<void> if the completion
+      // handler does not have its own.
+      using allocator_type = boost::asio::associated_allocator_t<
+          typename std::decay<decltype(completion_handler)>::type,
+          std::allocator<void>>;
+
+      allocator_type get_allocator() const noexcept
+      {
+        return boost::asio::get_associated_allocator(
+            handler_, std::allocator<void>{});
+      }
+    };
+
+    // Initiate the underlying async_write operation using our intermediate
+    // completion handler.
+    auto encoded_message_buffer = boost::asio::buffer(*encoded_message);
+    boost::asio::async_write(socket, encoded_message_buffer,
+        intermediate_completion_handler{
+          socket, std::move(encoded_message),
+          repeat_count, std::move(delay_timer),
+          intermediate_completion_handler::starting,
+          boost::asio::prefer(socket.get_executor(),
+              boost::asio::execution::outstanding_work.tracked),
+          std::forward<decltype(completion_handler)>(completion_handler)});
+  };
+
+  // Encode the message and copy it into an allocated buffer. The buffer will
+  // be maintained for the lifetime of the composed asynchronous operation.
+  std::ostringstream os;
+  os << message;
+  std::unique_ptr<std::string> encoded_message(new std::string(os.str()));
+
+  // Create a steady_timer to be used for the delay between messages.
+  std::unique_ptr<boost::asio::steady_timer> delay_timer(
+      new boost::asio::steady_timer(socket.get_executor()));
+
+  // The boost::asio::async_initiate function takes:
+  //
+  // - our initiation function object,
+  // - the completion token,
+  // - the completion handler signature, and
+  // - any additional arguments we need to initiate the operation.
+  //
+  // It then asks the completion token to create a completion handler (i.e. a
+  // callback) with the specified signature, and invoke the initiation function
+  // object with this completion handler as well as the additional arguments.
+  // The return value of async_initiate is the result of our operation's
+  // initiating function.
+  //
+  // Note that we wrap non-const reference arguments in std::reference_wrapper
+  // to prevent incorrect decay-copies of these objects.
+  return boost::asio::async_initiate<
+    CompletionToken, void(boost::system::error_code)>(
+      initiation, token, std::ref(socket),
+      std::move(encoded_message), repeat_count,
+      std::move(delay_timer));
+}
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_messages(socket, "Testing callback\r\n", 5,
+      [](const boost::system::error_code& error)
+      {
+        if (!error)
+        {
+          std::cout << "Messages sent\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<void> f = async_write_messages(
+      socket, "Testing future\r\n", 5, boost::asio::use_future);
+
+  io_context.run();
+
+  try
+  {
+    // Get the result of the operation.
+    f.get();
+    std::cout << "Messages sent\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Error: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}

libs/asio/example/cpp14/operations/composed_7.cpp

@@ -0,0 +1,219 @@
+//
+// composed_7.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/compose.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <functional>
+#include <iostream>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+// NOTE: This example requires the new boost::asio::async_compose function. For
+// an example that works with the Networking TS style of completion tokens,
+// please see an older version of asio.
+
+//------------------------------------------------------------------------------
+
+// This composed operation shows composition of multiple underlying operations.
+// It automatically serialises a message, using its I/O streams insertion
+// operator, before sending it N times on the socket. To do this, it must
+// allocate a buffer for the encoded message and ensure this buffer's validity
+// until all underlying async_write operation complete. A one second delay is
+// inserted prior to each write operation, using a steady_timer.
+
+template <typename T, typename CompletionToken>
+auto async_write_messages(tcp::socket& socket,
+    const T& message, std::size_t repeat_count,
+    CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is always void.
+  // In this example, when the completion token is boost::asio::yield_context
+  // (used for stackful coroutines) the return type would be also be void, as
+  // there is no non-error argument to the completion handler. When the
+  // completion token is boost::asio::use_future it would be std::future<void>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of boost::asio::async_initiate.
+{
+  // Encode the message and copy it into an allocated buffer. The buffer will
+  // be maintained for the lifetime of the composed asynchronous operation.
+  std::ostringstream os;
+  os << message;
+  std::unique_ptr<std::string> encoded_message(new std::string(os.str()));
+
+  // Create a steady_timer to be used for the delay between messages.
+  std::unique_ptr<boost::asio::steady_timer> delay_timer(
+      new boost::asio::steady_timer(socket.get_executor()));
+
+  // To manage the cycle between the multiple underlying asychronous
+  // operations, our implementation is a state machine.
+  enum { starting, waiting, writing };
+
+  // The boost::asio::async_compose function takes:
+  //
+  // - our asynchronous operation implementation,
+  // - the completion token,
+  // - the completion handler signature, and
+  // - any I/O objects (or executors) used by the operation
+  //
+  // It then wraps our implementation, which is implemented here as a state
+  // machine in a lambda, in an intermediate completion handler that meets the
+  // requirements of a conforming asynchronous operation. This includes
+  // tracking outstanding work against the I/O executors associated with the
+  // operation (in this example, this is the socket's executor).
+  //
+  // The first argument to our lambda is a reference to the enclosing
+  // intermediate completion handler. This intermediate completion handler is
+  // provided for us by the boost::asio::async_compose function, and takes care
+  // of all the details required to implement a conforming asynchronous
+  // operation. When calling an underlying asynchronous operation, we pass it
+  // this enclosing intermediate completion handler as the completion token.
+  //
+  // All arguments to our lambda after the first must be defaulted to allow the
+  // state machine to be started, as well as to allow the completion handler to
+  // match the completion signature of both the async_write and
+  // steady_timer::async_wait operations.
+  return boost::asio::async_compose<
+    CompletionToken, void(boost::system::error_code)>(
+      [
+        // The implementation holds a reference to the socket as it is used for
+        // multiple async_write operations.
+        &socket,
+
+        // The allocated buffer for the encoded message. The std::unique_ptr
+        // smart pointer is move-only, and as a consequence our lambda
+        // implementation is also move-only.
+        encoded_message = std::move(encoded_message),
+
+        // The repeat count remaining.
+        repeat_count,
+
+        // A steady timer used for introducing a delay.
+        delay_timer = std::move(delay_timer),
+
+        // To manage the cycle between the multiple underlying asychronous
+        // operations, our implementation is a state machine.
+        state = starting
+      ]
+      (
+        auto& self,
+        const boost::system::error_code& error = {},
+        std::size_t = 0
+      ) mutable
+      {
+        if (!error)
+        {
+          switch (state)
+          {
+          case starting:
+          case writing:
+            if (repeat_count > 0)
+            {
+              --repeat_count;
+              state = waiting;
+              delay_timer->expires_after(std::chrono::seconds(1));
+              delay_timer->async_wait(std::move(self));
+              return; // Composed operation not yet complete.
+            }
+            break; // Composed operation complete, continue below.
+          case waiting:
+            state = writing;
+            boost::asio::async_write(socket,
+                boost::asio::buffer(*encoded_message), std::move(self));
+            return; // Composed operation not yet complete.
+          }
+        }
+
+        // This point is reached only on completion of the entire composed
+        // operation.
+
+        // Deallocate the encoded message and delay timer before calling the
+        // user-supplied completion handler.
+        encoded_message.reset();
+        delay_timer.reset();
+
+        // Call the user-supplied handler with the result of the operation.
+        self.complete(error);
+      },
+      token, socket);
+}
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_messages(socket, "Testing callback\r\n", 5,
+      [](const boost::system::error_code& error)
+      {
+        if (!error)
+        {
+          std::cout << "Messages sent\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<void> f = async_write_messages(
+      socket, "Testing future\r\n", 5, boost::asio::use_future);
+
+  io_context.run();
+
+  try
+  {
+    // Get the result of the operation.
+    f.get();
+    std::cout << "Messages sent\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Error: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}

libs/asio/example/cpp14/operations/composed_8.cpp

@@ -0,0 +1,212 @@
+//
+// composed_8.cpp
+// ~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2021 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#include <boost/asio/compose.hpp>
+#include <boost/asio/coroutine.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/asio/use_future.hpp>
+#include <boost/asio/write.hpp>
+#include <functional>
+#include <iostream>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+using boost::asio::ip::tcp;
+
+// NOTE: This example requires the new boost::asio::async_compose function. For
+// an example that works with the Networking TS style of completion tokens,
+// please see an older version of asio.
+
+//------------------------------------------------------------------------------
+
+// This composed operation shows composition of multiple underlying operations,
+// using asio's stackless coroutines support to express the flow of control. It
+// automatically serialises a message, using its I/O streams insertion
+// operator, before sending it N times on the socket. To do this, it must
+// allocate a buffer for the encoded message and ensure this buffer's validity
+// until all underlying async_write operation complete. A one second delay is
+// inserted prior to each write operation, using a steady_timer.
+
+#include <boost/asio/yield.hpp>
+
+template <typename T, typename CompletionToken>
+auto async_write_messages(tcp::socket& socket,
+    const T& message, std::size_t repeat_count,
+    CompletionToken&& token)
+  // The return type of the initiating function is deduced from the combination
+  // of CompletionToken type and the completion handler's signature. When the
+  // completion token is a simple callback, the return type is always void.
+  // In this example, when the completion token is boost::asio::yield_context
+  // (used for stackful coroutines) the return type would be also be void, as
+  // there is no non-error argument to the completion handler. When the
+  // completion token is boost::asio::use_future it would be std::future<void>.
+  //
+  // In C++14 we can omit the return type as it is automatically deduced from
+  // the return type of boost::asio::async_initiate.
+{
+  // Encode the message and copy it into an allocated buffer. The buffer will
+  // be maintained for the lifetime of the composed asynchronous operation.
+  std::ostringstream os;
+  os << message;
+  std::unique_ptr<std::string> encoded_message(new std::string(os.str()));
+
+  // Create a steady_timer to be used for the delay between messages.
+  std::unique_ptr<boost::asio::steady_timer> delay_timer(
+      new boost::asio::steady_timer(socket.get_executor()));
+
+  // The boost::asio::async_compose function takes:
+  //
+  // - our asynchronous operation implementation,
+  // - the completion token,
+  // - the completion handler signature, and
+  // - any I/O objects (or executors) used by the operation
+  //
+  // It then wraps our implementation, which is implemented here as a stackless
+  // coroutine in a lambda, in an intermediate completion handler that meets the
+  // requirements of a conforming asynchronous operation. This includes
+  // tracking outstanding work against the I/O executors associated with the
+  // operation (in this example, this is the socket's executor).
+  //
+  // The first argument to our lambda is a reference to the enclosing
+  // intermediate completion handler. This intermediate completion handler is
+  // provided for us by the boost::asio::async_compose function, and takes care
+  // of all the details required to implement a conforming asynchronous
+  // operation. When calling an underlying asynchronous operation, we pass it
+  // this enclosing intermediate completion handler as the completion token.
+  //
+  // All arguments to our lambda after the first must be defaulted to allow the
+  // state machine to be started, as well as to allow the completion handler to
+  // match the completion signature of both the async_write and
+  // steady_timer::async_wait operations.
+  return boost::asio::async_compose<
+    CompletionToken, void(boost::system::error_code)>(
+      [
+        // The implementation holds a reference to the socket as it is used for
+        // multiple async_write operations.
+        &socket,
+
+        // The allocated buffer for the encoded message. The std::unique_ptr
+        // smart pointer is move-only, and as a consequence our lambda
+        // implementation is also move-only.
+        encoded_message = std::move(encoded_message),
+
+        // The repeat count remaining.
+        repeat_count,
+
+        // A steady timer used for introducing a delay.
+        delay_timer = std::move(delay_timer),
+
+        // The coroutine state.
+        coro = boost::asio::coroutine()
+      ]
+      (
+        auto& self,
+        const boost::system::error_code& error = {},
+        std::size_t = 0
+      ) mutable
+      {
+        reenter (coro)
+        {
+          while (repeat_count > 0)
+          {
+            --repeat_count;
+
+            delay_timer->expires_after(std::chrono::seconds(1));
+            yield delay_timer->async_wait(std::move(self));
+            if (error)
+              break;
+
+            yield boost::asio::async_write(socket,
+                boost::asio::buffer(*encoded_message), std::move(self));
+            if (error)
+              break;
+          }
+
+          // Deallocate the encoded message and delay timer before calling the
+          // user-supplied completion handler.
+          encoded_message.reset();
+          delay_timer.reset();
+
+          // Call the user-supplied handler with the result of the operation.
+          self.complete(error);
+        }
+      },
+      token, socket);
+}
+
+#include <boost/asio/unyield.hpp>
+
+//------------------------------------------------------------------------------
+
+void test_callback()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using a lambda as a callback.
+  async_write_messages(socket, "Testing callback\r\n", 5,
+      [](const boost::system::error_code& error)
+      {
+        if (!error)
+        {
+          std::cout << "Messages sent\n";
+        }
+        else
+        {
+          std::cout << "Error: " << error.message() << "\n";
+        }
+      });
+
+  io_context.run();
+}
+
+//------------------------------------------------------------------------------
+
+void test_future()
+{
+  boost::asio::io_context io_context;
+
+  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
+  tcp::socket socket = acceptor.accept();
+
+  // Test our asynchronous operation using the use_future completion token.
+  // This token causes the operation's initiating function to return a future,
+  // which may be used to synchronously wait for the result of the operation.
+  std::future<void> f = async_write_messages(
+      socket, "Testing future\r\n", 5, boost::asio::use_future);
+
+  io_context.run();
+
+  try
+  {
+    // Get the result of the operation.
+    f.get();
+    std::cout << "Messages sent\n";
+  }
+  catch (const std::exception& e)
+  {
+    std::cout << "Error: " << e.what() << "\n";
+  }
+}
+
+//------------------------------------------------------------------------------
+
+int main()
+{
+  test_callback();
+  test_future();
+}