feat: Move/copy support in async framework (#1609)

Fixes #1608

benchmarks/CMakeLists.txt

@@ -12,5 +12,5 @@ target_sources(
 include(deps/gbench)
 
 target_include_directories(clio_benchmark PRIVATE .)
-target_link_libraries(clio_benchmark PUBLIC clio benchmark::benchmark_main)
+target_link_libraries(clio_benchmark PUBLIC clio_etl benchmark::benchmark_main)
 set_target_properties(clio_benchmark PROPERTIES RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})

benchmarks/util/async/ExecutionContextBenchmarks.cpp

@@ -31,7 +31,6 @@
 #include <cstdint>
 #include <latch>
 #include <optional>
-#include <stdexcept>
 #include <thread>
 #include <vector>
 

src/util/Assert.hpp

@@ -74,4 +74,5 @@ assertImpl(
 
 }  // namespace util
 
-#define ASSERT(condition, ...) util::assertImpl(CURRENT_SRC_LOCATION, #condition, (condition), __VA_ARGS__)
+#define ASSERT(condition, ...) \
+    util::assertImpl(CURRENT_SRC_LOCATION, #condition, static_cast<bool>(condition), __VA_ARGS__)

src/util/async/AnyExecutionContext.hpp

@@ -43,16 +43,37 @@ class AnyExecutionContext {
     /**
      * @brief Construct a new type-erased Execution Context object
      *
+     * @note Stores the Execution Context by reference.
+     *
      * @tparam CtxType The type of the execution context to wrap
      * @param ctx The execution context to wrap
      */
-    template <typename CtxType>
-        requires(not std::is_same_v<std::decay_t<CtxType>, AnyExecutionContext>)
+    template <NotSameAs<AnyExecutionContext> CtxType>
+    /* implicit */
+    AnyExecutionContext(CtxType& ctx) : pimpl_{std::make_shared<Model<CtxType&>>(ctx)}
+    {
+    }
+
+    /**
+     * @brief Construct a new type-erased Execution Context object
+     *
+     * @note Stores the Execution Context by moving it into the AnyExecutionContext.
+     *
+     * @tparam CtxType The type of the execution context to wrap
+     * @param ctx The execution context to wrap
+     */
+    template <RValueNotSameAs<AnyExecutionContext> CtxType>
     /* implicit */
-    AnyExecutionContext(CtxType&& ctx) : pimpl_{std::make_unique<Model<CtxType>>(std::forward<CtxType>(ctx))}
+    AnyExecutionContext(CtxType&& ctx) : pimpl_{std::make_shared<Model<CtxType>>(std::forward<CtxType>(ctx))}
     {
     }
 
+    AnyExecutionContext(AnyExecutionContext const&) = default;
+    AnyExecutionContext(AnyExecutionContext&&) = default;
+    AnyExecutionContext&
+    operator=(AnyExecutionContext const&) = default;
+    AnyExecutionContext&
+    operator=(AnyExecutionContext&&) = default;
     ~AnyExecutionContext() = default;
 
     /**
@@ -206,7 +227,7 @@ class AnyExecutionContext {
      * @brief Stop the execution context
      */
     void
-    stop()
+    stop() const
     {
         pimpl_->stop();
     }
@@ -215,7 +236,7 @@ class AnyExecutionContext {
      * @brief Join the execution context
      */
     void
-    join()
+    join() const
     {
         pimpl_->join();
     }
@@ -237,64 +258,65 @@ class AnyExecutionContext {
         virtual AnyStrand
         makeStrand() = 0;
         virtual void
-        stop() = 0;
+        stop() const = 0;
         virtual void
-        join() = 0;
+        join() const = 0;
     };
 
     template <typename CtxType>
     struct Model : Concept {
-        std::reference_wrapper<std::decay_t<CtxType>> ctx;
+        CtxType ctx;
 
-        Model(CtxType& ctx) : ctx{std::ref(ctx)}
+        template <typename Type>
+        Model(Type&& ctx) : ctx(std::forward<Type>(ctx))
         {
         }
 
         impl::ErasedOperation
         execute(std::function<std::any(AnyStopToken)> fn, std::optional<std::chrono::milliseconds> timeout) override
         {
-            return ctx.get().execute(std::move(fn), timeout);
+            return ctx.execute(std::move(fn), timeout);
         }
 
         impl::ErasedOperation
         execute(std::function<std::any()> fn) override
         {
-            return ctx.get().execute(std::move(fn));
+            return ctx.execute(std::move(fn));
         }
 
         impl::ErasedOperation
         scheduleAfter(std::chrono::milliseconds delay, std::function<std::any(AnyStopToken)> fn) override
         {
-            return ctx.get().scheduleAfter(delay, std::move(fn));
+            return ctx.scheduleAfter(delay, std::move(fn));
         }
 
         impl::ErasedOperation
         scheduleAfter(std::chrono::milliseconds delay, std::function<std::any(AnyStopToken, bool)> fn) override
         {
-            return ctx.get().scheduleAfter(delay, std::move(fn));
+            return ctx.scheduleAfter(delay, std::move(fn));
         }
 
         AnyStrand
         makeStrand() override
         {
-            return ctx.get().makeStrand();
+            return ctx.makeStrand();
         }
 
         void
-        stop() override
+        stop() const override
         {
-            ctx.get().stop();
+            ctx.stop();
         }
 
         void
-        join() override
+        join() const override
         {
-            ctx.get().join();
+            ctx.join();
         }
     };
 
 private:
-    std::unique_ptr<Concept> pimpl_;
+    std::shared_ptr<Concept> pimpl_;
 };
 
 }  // namespace util::async

src/util/async/AnyOperation.hpp

@@ -47,12 +47,9 @@ class AnyOperation {
     /**
      * @brief Construct a new type-erased Operation object
      *
-     * @tparam OpType The type of the operation to wrap
      * @param operation The operation to wrap
      */
-    template <SomeOperation OpType>
-        requires std::is_same_v<std::decay_t<OpType>, impl::ErasedOperation>
-    /* implicit */ AnyOperation(OpType&& operation) : operation_{std::forward<OpType>(operation)}
+    /* implicit */ AnyOperation(impl::ErasedOperation&& operation) : operation_{std::move(operation)}
     {
     }
 

src/util/async/AnyStopToken.hpp

@@ -26,6 +26,7 @@
 
 #include <memory>
 #include <type_traits>
+#include <utility>
 
 namespace util::async {
 
@@ -41,7 +42,7 @@ class AnyStopToken {
      * @param token The stop token to wrap
      */
     template <SomeStopToken TokenType>
-        requires(not std::is_same_v<std::decay_t<TokenType>, AnyStopToken>)
+        requires NotSameAs<TokenType, AnyStopToken>
     /* implicit */ AnyStopToken(TokenType&& token)
         : pimpl_{std::make_unique<Model<TokenType>>(std::forward<TokenType>(token))}
     {
@@ -142,7 +143,7 @@ class AnyStopToken {
             }
 
             ASSERT(false, "Token type does not support conversion to boost::asio::yield_context");
-            __builtin_unreachable();  // TODO: replace with std::unreachable when C++23 is available
+            std::unreachable();
         }
     };
 

src/util/async/AnyStrand.hpp

@@ -20,6 +20,7 @@
 #pragma once
 
 #include "util/async/AnyStopToken.hpp"
+#include "util/async/Concepts.hpp"
 #include "util/async/impl/ErasedOperation.hpp"
 
 #include <any>
@@ -43,13 +44,14 @@ class AnyStrand {
      * @tparam StrandType The type of the strand to wrap
      * @param strand The strand to wrap
      */
-    template <typename StrandType>
-        requires(not std::is_same_v<std::decay_t<StrandType>, AnyStrand>)
+    template <NotSameAs<AnyStrand> StrandType>
     /* implicit */ AnyStrand(StrandType&& strand)
-        : pimpl_{std::make_unique<Model<StrandType>>(std::forward<StrandType>(strand))}
+        : pimpl_{std::make_shared<Model<StrandType>>(std::forward<StrandType>(strand))}
     {
     }
 
+    AnyStrand(AnyStrand const&) = default;
+    AnyStrand(AnyStrand&&) = default;
     ~AnyStrand() = default;
 
     /**
@@ -163,7 +165,7 @@ class AnyStrand {
     };
 
 private:
-    std::unique_ptr<Concept> pimpl_;
+    std::shared_ptr<Concept> pimpl_;
 };
 
 }  // namespace util::async

src/util/async/Concepts.hpp

@@ -22,7 +22,9 @@
 #include <boost/asio/spawn.hpp>
 
 #include <chrono>
+#include <concepts>
 #include <functional>
+#include <optional>
 #include <type_traits>
 
 namespace util::async {
@@ -141,13 +143,38 @@ concept SomeStdDuration = requires {
     // See https://stackoverflow.com/questions/74383254/concept-that-models-only-the-stdchrono-duration-types
     []<typename Rep, typename Period>( // 
         std::type_identity<std::chrono::duration<Rep, Period>>
-    ) {}(std::type_identity<T>());
+    ) {}(std::type_identity<std::decay_t<T>>());
+};
+
+/**
+ * @brief Specifies that the type must be some std::optional
+ */
+template <typename T>
+concept SomeStdOptional = requires {
+    []<typename Type>( // 
+        std::type_identity<std::optional<Type>>
+    ) {}(std::type_identity<std::decay_t<T>>());
 };
 
 /**
  * @brief Specifies that the type must be some std::duration wrapped in an optional
  */
 template <typename T>
-concept SomeOptStdDuration = requires(T v) { SomeStdDuration<decltype(v.value())>; };
+concept SomeOptStdDuration = SomeStdOptional<T> and SomeStdDuration<decltype(T{}.value())>;
+
+/**
+ * @brief Checks that decayed T s not of the same type as Erased
+ */
+template <typename T, typename Erased>
+concept NotSameAs = not std::is_same_v<std::decay_t<T>, Erased>;
+
+/**
+ * @brief Check that T is an r-value and is not the same type as Erased
+ */
+template <typename T, typename Erased>
+concept RValueNotSameAs = requires(T&& t) {
+    requires std::is_rvalue_reference_v<decltype(t)>;
+    requires NotSameAs<T, Erased>;
+};
 
 }  // namespace util::async

src/util/async/Outcome.hpp

@@ -21,6 +21,7 @@
 
 #include "util/async/context/impl/Cancellation.hpp"
 
+#include <concepts>
 #include <future>
 
 namespace util::async {

src/util/async/context/BasicExecutionContext.hpp

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include "util/Assert.hpp"
 #include "util/async/Concepts.hpp"
 #include "util/async/Error.hpp"
 #include "util/async/Operation.hpp"
@@ -38,6 +39,7 @@
 #include <chrono>
 #include <cstddef>
 #include <expected>
+#include <memory>
 #include <optional>
 #include <type_traits>
 #include <utility>
@@ -59,6 +61,12 @@ struct AsioPoolStrandContext {
     using Executor = boost::asio::strand<boost::asio::thread_pool::executor_type>;
     using Timer = SteadyTimer<Executor>;
 
+    Executor const&
+    getExecutor() const
+    {
+        return executor;
+    }
+
     Executor executor;
 };
 
@@ -67,13 +75,42 @@ struct AsioPoolContext {
     using Timer = SteadyTimer<Executor>;
     using Strand = AsioPoolStrandContext;
 
+    AsioPoolContext(std::size_t numThreads) : executor(std::make_unique<Executor>(numThreads))
+    {
+    }
+
+    AsioPoolContext(AsioPoolContext const&) = delete;
+    AsioPoolContext(AsioPoolContext&&) = default;
+
     Strand
-    makeStrand()
+    makeStrand() const
     {
-        return {boost::asio::make_strand(executor)};
+        ASSERT(executor, "Called after executor was moved from.");
+        return {boost::asio::make_strand(*executor)};
     }
 
-    Executor executor;
+    void
+    stop() const
+    {
+        if (executor)  // don't call if executor was moved from
+            executor->stop();
+    }
+
+    void
+    join() const
+    {
+        if (executor)  // don't call if executor was moved from
+            executor->join();
+    }
+
+    Executor&
+    getExecutor() const
+    {
+        ASSERT(executor, "Called after executor was moved from.");
+        return *executor;
+    }
+
+    std::unique_ptr<Executor> executor;
 };
 
 }  // namespace impl
@@ -151,7 +188,7 @@ class BasicExecutionContext {
         stop();
     }
 
-    BasicExecutionContext(BasicExecutionContext&&) = delete;
+    BasicExecutionContext(BasicExecutionContext&&) = default;
     BasicExecutionContext(BasicExecutionContext const&) = delete;
 
     /**
@@ -323,9 +360,9 @@ class BasicExecutionContext {
      * @brief Stop the execution context as soon as possible
      */
     void
-    stop() noexcept
+    stop() const noexcept
     {
-        context_.executor.stop();
+        context_.stop();
     }
 
     /**
@@ -334,7 +371,7 @@ class BasicExecutionContext {
     void
     join() noexcept
     {
-        context_.executor.join();
+        context_.join();
     }
 };