Merge pull request ceph#58348 from cbodley/wip-async-co-algorithms

common/async: add primitives for structured concurrency with c++20 coroutines

Reviewed-by: Adam Emerson <[email protected]>

Author: cbodley

src/common/async/co_spawn_group.h

@@ -0,0 +1,101 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#pragma once
+
+#include <boost/asio/awaitable.hpp>
+#include <boost/asio/execution/executor.hpp>
+#include "cancel_on_error.h"
+#include "detail/co_spawn_group.h"
+
+namespace ceph::async {
+
+/// \brief Tracks a group of coroutines to await all of their completions.
+///
+/// The wait() function can be used to await the completion of all children.
+/// If any child coroutines exit with an exception, the first such exception
+/// is rethrown by wait(). The cancel_on_error option controls whether these
+/// exceptions trigger the cancellation of other children.
+///
+/// All child coroutines are canceled by cancel() or co_spawn_group destruction.
+/// This allows the parent coroutine to share memory with its child coroutines
+/// without fear of dangling references.
+///
+/// This class is not thread-safe, so a strand executor should be used in
+/// multi-threaded contexts.
+///
+/// Example:
+/// \code
+/// awaitable<void> child(task& t);
+///
+/// awaitable<void> parent(std::span<task> tasks)
+/// {
+///   // process all tasks in parallel
+///   auto ex = co_await boost::asio::this_coro::executor;
+///   auto group = co_spawn_group{ex, tasks.size()};
+///
+///   for (auto& t : tasks) {
+///     group.spawn(child(t));
+///   }
+///   co_await group.wait();
+/// }
+/// \endcode
+template <boost::asio::execution::executor Executor>
+class co_spawn_group {
+  using impl_type = detail::co_spawn_group_impl<Executor>;
+  boost::intrusive_ptr<impl_type> impl;
+
+ public:
+  co_spawn_group(Executor ex, size_t limit,
+                 cancel_on_error on_error = cancel_on_error::none)
+    : impl(new impl_type(ex, limit, on_error))
+  {
+  }
+
+  ~co_spawn_group()
+  {
+    impl->cancel();
+  }
+
+  using executor_type = Executor;
+  executor_type get_executor() const
+  {
+    return impl->get_executor();
+  }
+
+  /// Spawn the given coroutine \ref cr on the group's executor. Throws a
+  /// std::length_error exception if the number of outstanding coroutines
+  /// would exceed the group's limit.
+  void spawn(boost::asio::awaitable<void, executor_type> cr)
+  {
+    impl->spawn(std::move(cr));
+  }
+
+  /// Wait for all outstanding coroutines before returning. If any of the
+  /// spawned coroutines exit with an exception, the first exception is
+  /// rethrown.
+  ///
+  /// After wait() completes, whether by exception or co_return, the spawn
+  /// group can be reused to spawn and await additional coroutines.
+  boost::asio::awaitable<void, executor_type> wait()
+  {
+    return impl->wait();
+  }
+
+  /// Cancel all outstanding coroutines.
+  void cancel()
+  {
+    impl->cancel();
+  }
+};
+
+} // namespace ceph::async

src/common/async/co_throttle.h

@@ -0,0 +1,113 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2023 Red Hat <[email protected]>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation.  See file COPYING.
+ *
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <limits>
+#include <boost/intrusive_ptr.hpp>
+#include "common/async/cancel_on_error.h"
+#include "common/async/detail/co_throttle_impl.h"
+
+namespace ceph::async {
+
+/// A coroutine throttle that allows a parent coroutine to spawn and manage
+/// multiple child coroutines, while enforcing an upper bound on concurrency.
+///
+/// Child coroutines must be of type awaitable<void>. Exceptions thrown by
+/// children are rethrown to the parent on its next call to spawn() or wait().
+/// The cancel_on_error option controls whether these exceptions errors trigger
+/// the cancellation of other children.
+///
+/// All child coroutines are canceled by cancel() or co_throttle destruction.
+/// This allows the parent coroutine to share memory with its child coroutines
+/// without fear of dangling references.
+///
+/// This class is not thread-safe, so a strand executor should be used in
+/// multi-threaded contexts.
+///
+/// Example:
+/// \code
+/// awaitable<void> child(task& t);
+///
+/// awaitable<void> parent(std::span<task> tasks)
+/// {
+///   // process all tasks, up to 10 at a time
+///   auto ex = co_await boost::asio::this_coro::executor;
+///   auto throttle = co_throttle{ex, 10};
+///
+///   for (auto& t : tasks) {
+///     co_await throttle.spawn(child(t));
+///   }
+///   co_await throttle.wait();
+/// }
+/// \endcode
+template <boost::asio::execution::executor Executor>
+class co_throttle {
+  using impl_type = detail::co_throttle_impl<Executor>;
+  boost::intrusive_ptr<impl_type> impl;
+
+ public:
+  using executor_type = Executor;
+  executor_type get_executor() const noexcept { return impl->get_executor(); }
+
+  static constexpr size_t max_limit = std::numeric_limits<size_t>::max();
+
+  co_throttle(const executor_type& ex, size_t limit,
+              cancel_on_error on_error = cancel_on_error::none)
+    : impl(new impl_type(ex, limit, on_error))
+  {
+  }
+
+  ~co_throttle()
+  {
+    cancel();
+  }
+
+  co_throttle(const co_throttle&) = delete;
+  co_throttle& operator=(const co_throttle&) = delete;
+
+  /// Try to spawn the given coroutine \ref cr. If this would exceed the
+  /// concurrency limit, wait for another coroutine to complete first. This
+  /// default limit can be overridden with the optional \ref smaller_limit
+  /// argument.
+  ///
+  /// If any spawned coroutines exit with an exception, the first exception is
+  /// rethrown by the next call to spawn() or wait(). If spawn() has an
+  /// exception to rethrow, it will spawn \cr first only in the case of
+  /// cancel_on_error::none. New coroutines can be spawned by later calls to
+  /// spawn() regardless of cancel_on_error.
+  auto spawn(boost::asio::awaitable<void, executor_type> cr,
+             size_t smaller_limit = max_limit)
+      -> boost::asio::awaitable<void, executor_type>
+  {
+    return impl->spawn(std::move(cr), smaller_limit);
+  }
+
+  /// Wait for all associated coroutines to complete. If any of these coroutines
+  /// exit with an exception, the first of those exceptions is rethrown.
+  auto wait()
+      -> boost::asio::awaitable<void, executor_type>
+  {
+    return impl->wait();
+  }
+
+  /// Cancel all associated coroutines.
+  void cancel()
+  {
+    impl->cancel();
+  }
+};
+
+} // namespace ceph::async

src/common/async/co_waiter.h

@@ -0,0 +1,166 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#pragma once
+
+#include <exception>
+#include <optional>
+#include <boost/asio/append.hpp>
+#include <boost/asio/async_result.hpp>
+#include <boost/asio/dispatch.hpp>
+#include <boost/asio/execution/executor.hpp>
+#include <boost/asio/use_awaitable.hpp>
+#include "include/ceph_assert.h"
+
+namespace ceph::async {
+
+/// Captures an awaitable handler for deferred completion or cancellation.
+template <typename Ret, boost::asio::execution::executor Executor>
+class co_waiter {
+  using signature = void(std::exception_ptr, Ret);
+  using token_type = boost::asio::use_awaitable_t<Executor>;
+  using handler_type = typename boost::asio::async_result<
+      token_type, signature>::handler_type;
+  std::optional<handler_type> handler;
+
+  struct op_cancellation {
+    co_waiter* self;
+    op_cancellation(co_waiter* self) : self(self) {}
+    void operator()(boost::asio::cancellation_type_t type) {
+      if (type != boost::asio::cancellation_type::none) {
+        self->cancel();
+      }
+    }
+  };
+ public:
+  co_waiter() = default;
+
+  // copy and move are disabled because the cancellation handler captures 'this'
+  co_waiter(const co_waiter&) = delete;
+  co_waiter& operator=(const co_waiter&) = delete;
+
+  /// Returns true if there's a handler awaiting completion.
+  bool waiting() const { return handler.has_value(); }
+
+  /// Returns an awaitable that blocks until complete() or cancel().
+  boost::asio::awaitable<Ret, Executor> get()
+  {
+    ceph_assert(!handler);
+    token_type token;
+    return boost::asio::async_initiate<token_type, signature>(
+        [this] (handler_type h) {
+          auto slot = boost::asio::get_associated_cancellation_slot(h);
+          if (slot.is_connected()) {
+            slot.template emplace<op_cancellation>(this);
+          }
+          handler.emplace(std::move(h));
+        }, token);
+  }
+
+  /// Schedule the completion handler with the given arguments.
+  void complete(std::exception_ptr eptr, Ret value)
+  {
+    ceph_assert(handler);
+    auto h = boost::asio::append(std::move(*handler), eptr, std::move(value));
+    handler.reset();
+    boost::asio::dispatch(std::move(h));
+  }
+
+  /// Cancel the coroutine with an operation_aborted exception.
+  void cancel()
+  {
+    if (handler) {
+      auto eptr = std::make_exception_ptr(
+          boost::system::system_error(
+              boost::asio::error::operation_aborted));
+      complete(eptr, Ret{});
+    }
+  }
+
+  /// Destroy the completion handler.
+  void shutdown()
+  {
+    handler.reset();
+  }
+};
+
+// specialization for Ret=void
+template <boost::asio::execution::executor Executor>
+class co_waiter<void, Executor> {
+  using signature = void(std::exception_ptr);
+  using token_type = boost::asio::use_awaitable_t<Executor>;
+  using handler_type = typename boost::asio::async_result<
+      token_type, signature>::handler_type;
+  std::optional<handler_type> handler;
+
+  struct op_cancellation {
+    co_waiter* self;
+    op_cancellation(co_waiter* self) : self(self) {}
+    void operator()(boost::asio::cancellation_type_t type) {
+      if (type != boost::asio::cancellation_type::none) {
+        self->cancel();
+      }
+    }
+  };
+ public:
+  co_waiter() = default;
+
+  // copy and move are disabled because the cancellation handler captures 'this'
+  co_waiter(const co_waiter&) = delete;
+  co_waiter& operator=(const co_waiter&) = delete;
+
+  /// Returns true if there's a handler awaiting completion.
+  bool waiting() const { return handler.has_value(); }
+
+  /// Returns an awaitable that blocks until complete() or cancel().
+  boost::asio::awaitable<void, Executor> get()
+  {
+    ceph_assert(!handler);
+    token_type token;
+    return boost::asio::async_initiate<token_type, signature>(
+        [this] (handler_type h) {
+          auto slot = boost::asio::get_associated_cancellation_slot(h);
+          if (slot.is_connected()) {
+            slot.template emplace<op_cancellation>(this);
+          }
+          handler.emplace(std::move(h));
+        }, token);
+  }
+
+  /// Schedule the completion handler with the given arguments.
+  void complete(std::exception_ptr eptr)
+  {
+    ceph_assert(handler);
+    auto h = boost::asio::append(std::move(*handler), eptr);
+    handler.reset();
+    boost::asio::dispatch(std::move(h));
+  }
+
+  /// Cancel the coroutine with an operation_aborted exception.
+  void cancel()
+  {
+    if (handler) {
+      auto eptr = std::make_exception_ptr(
+          boost::system::system_error(
+              boost::asio::error::operation_aborted));
+      complete(eptr);
+    }
+  }
+
+  /// Destroy the completion handler.
+  void shutdown()
+  {
+    handler.reset();
+  }
+};
+
+} // namespace ceph::async