infra: encapsulate usage of asio co_spawn (#2248)

silkworm/db/kv/api/service_router.cpp

@@ -16,17 +16,15 @@
 
 #include "service_router.hpp"
 
-#include <boost/asio/use_awaitable.hpp>
-
-#include <silkworm/infra/concurrency/co_spawn_sw.hpp>
+#include <silkworm/infra/concurrency/spawn.hpp>
 
 namespace silkworm::db::kv::api {
 
 using namespace boost::asio;
 
 Task<void> StateChangeRunner::run(std::shared_ptr<StateChangeRunner> self) {
     auto run = self->handle_calls();
-    co_await concurrency::co_spawn(self->strand_, std::move(run), use_awaitable);
+    co_await concurrency::spawn_task(self->strand_, std::move(run));
 }
 
 StateChangeRunner::StateChangeRunner(const boost::asio::any_io_executor& executor)

silkworm/db/kv/state_changes_stream.cpp

@@ -16,12 +16,11 @@
 
 #include "state_changes_stream.hpp"
 
-#include <boost/asio/use_future.hpp>
 #include <boost/system/error_code.hpp>
 
 #include <silkworm/infra/common/log.hpp>
-#include <silkworm/infra/concurrency/co_spawn_sw.hpp>
 #include <silkworm/infra/concurrency/shared_service.hpp>
+#include <silkworm/infra/concurrency/spawn.hpp>
 #include <silkworm/infra/grpc/client/call.hpp>
 
 namespace silkworm::db::kv {
@@ -32,7 +31,7 @@ StateChangesStream::StateChangesStream(rpc::ClientContext& context, api::Client&
       cache_(must_use_shared_service<api::StateCache>(scheduler_)) {}
 
 std::future<void> StateChangesStream::open() {
-    return concurrency::co_spawn(scheduler_, run(), boost::asio::use_future);
+    return concurrency::spawn_future(scheduler_, run());
 }
 
 void StateChangesStream::close() {

silkworm/infra/concurrency/awaitable_wait_for_all.hpp

@@ -33,8 +33,8 @@
 #include <boost/asio/this_coro.hpp>
 #include <boost/asio/use_awaitable.hpp>
 
-#include "co_spawn_sw.hpp"
 #include "parallel_group_utils.hpp"
+#include "spawn.hpp"
 
 namespace silkworm::concurrency::awaitable_wait_for_all {
 
@@ -71,8 +71,8 @@ awaitable<void, Executor> operator&&(
 
     auto [order, ex0, ex1] =
         co_await make_parallel_group(
-            co_spawn_sw(ex, std::move(t), deferred),
-            co_spawn_sw(ex, std::move(u), deferred))
+            co_spawn(ex, std::move(t), deferred),
+            co_spawn(ex, std::move(u), deferred))
             .async_wait(
                 wait_for_one_error(),
                 use_awaitable_t<Executor>{});
@@ -93,8 +93,8 @@ awaitable<U, Executor> operator&&(
 
     auto [order, ex0, ex1, r1] =
         co_await make_parallel_group(
-            co_spawn_sw(ex, std::move(t), deferred),
-            co_spawn_sw(ex, detail::awaitable_wrap(std::move(u)), deferred))
+            co_spawn(ex, std::move(t), deferred),
+            co_spawn(ex, detail::awaitable_wrap(std::move(u)), deferred))
             .async_wait(
                 wait_for_one_error(),
                 use_awaitable_t<Executor>{});
@@ -115,8 +115,8 @@ awaitable<T, Executor> operator&&(
 
     auto [order, ex0, r0, ex1] =
         co_await make_parallel_group(
-            co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-            co_spawn_sw(ex, std::move(u), deferred))
+            co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+            co_spawn(ex, std::move(u), deferred))
             .async_wait(
                 wait_for_one_error(),
                 use_awaitable_t<Executor>{});
@@ -137,8 +137,8 @@ awaitable<std::tuple<T, U>, Executor> operator&&(
 
     auto [order, ex0, r0, ex1, r1] =
         co_await make_parallel_group(
-            co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-            co_spawn_sw(ex, detail::awaitable_wrap(std::move(u)), deferred))
+            co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+            co_spawn(ex, detail::awaitable_wrap(std::move(u)), deferred))
             .async_wait(
                 wait_for_one_error(),
                 use_awaitable_t<Executor>{});
@@ -161,8 +161,8 @@ awaitable<std::tuple<T..., std::monostate>, Executor> operator&&(
 
     auto [order, ex0, r0, ex1, r1] =
         co_await make_parallel_group(
-            co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-            co_spawn_sw(ex, std::move(u), deferred))
+            co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+            co_spawn(ex, std::move(u), deferred))
             .async_wait(
                 wait_for_one_error(),
                 use_awaitable_t<Executor>{});
@@ -183,8 +183,8 @@ awaitable<std::tuple<T..., U>, Executor> operator&&(
 
     auto [order, ex0, r0, ex1, r1] =
         co_await make_parallel_group(
-            co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-            co_spawn_sw(ex, detail::awaitable_wrap(std::move(u)), deferred))
+            co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+            co_spawn(ex, detail::awaitable_wrap(std::move(u)), deferred))
             .async_wait(
                 wait_for_one_error(),
                 use_awaitable_t<Executor>{});

silkworm/infra/concurrency/awaitable_wait_for_one.hpp

@@ -34,7 +34,7 @@
 #include <boost/asio/this_coro.hpp>
 #include <boost/asio/use_awaitable.hpp>
 
-#include "co_spawn_sw.hpp"
+#include "spawn.hpp"
 
 namespace silkworm::concurrency::awaitable_wait_for_one {
 
@@ -70,8 +70,8 @@ awaitable<std::variant<std::monostate, std::monostate>, Executor> operator||(awa
     auto ex = co_await this_coro::executor;
 
     auto [order, ex0, ex1] =
-        co_await make_parallel_group(co_spawn_sw(ex, std::move(t), deferred),
-                                     co_spawn_sw(ex, std::move(u), deferred))
+        co_await make_parallel_group(co_spawn(ex, std::move(t), deferred),
+                                     co_spawn(ex, std::move(u), deferred))
             .async_wait(wait_for_one(), use_awaitable_t<Executor>{});
 
     if (order[0] == 0) {
@@ -93,8 +93,8 @@ awaitable<std::variant<std::monostate, U>, Executor> operator||(awaitable<void,
     auto ex = co_await this_coro::executor;
 
     auto [order, ex0, ex1, r1] =
-        co_await make_parallel_group(co_spawn_sw(ex, std::move(t), deferred),
-                                     co_spawn_sw(ex, detail::awaitable_wrap(std::move(u)), deferred))
+        co_await make_parallel_group(co_spawn(ex, std::move(t), deferred),
+                                     co_spawn(ex, detail::awaitable_wrap(std::move(u)), deferred))
             .async_wait(wait_for_one(), use_awaitable_t<Executor>{});
 
     if (order[0] == 0) {
@@ -119,8 +119,8 @@ awaitable<std::variant<T, std::monostate>, Executor> operator||(awaitable<T, Exe
     auto ex = co_await this_coro::executor;
 
     auto [order, ex0, r0, ex1] =
-        co_await make_parallel_group(co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-                                     co_spawn_sw(ex, std::move(u), deferred))
+        co_await make_parallel_group(co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+                                     co_spawn(ex, std::move(u), deferred))
             .async_wait(wait_for_one(), use_awaitable_t<Executor>{});
 
     if (order[0] == 0) {
@@ -145,8 +145,8 @@ awaitable<std::variant<T, U>, Executor> operator||(awaitable<T, Executor> t, awa
     auto ex = co_await this_coro::executor;
 
     auto [order, ex0, r0, ex1, r1] =
-        co_await make_parallel_group(co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-                                     co_spawn_sw(ex, detail::awaitable_wrap(std::move(u)), deferred))
+        co_await make_parallel_group(co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+                                     co_spawn(ex, detail::awaitable_wrap(std::move(u)), deferred))
             .async_wait(wait_for_one(), use_awaitable_t<Executor>{});
 
     if (order[0] == 0) {
@@ -169,8 +169,8 @@ awaitable<std::variant<T..., std::monostate>, Executor> operator||(awaitable<std
     auto ex = co_await this_coro::executor;
 
     auto [order, ex0, r0, ex1] =
-        co_await make_parallel_group(co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-                                     co_spawn_sw(ex, std::move(u), deferred))
+        co_await make_parallel_group(co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+                                     co_spawn(ex, std::move(u), deferred))
             .async_wait(wait_for_one(), use_awaitable_t<Executor>{});
 
     using widen = detail::widen_variant<T..., std::monostate>;
@@ -194,8 +194,8 @@ awaitable<std::variant<T..., U>, Executor> operator||(awaitable<std::variant<T..
     auto ex = co_await this_coro::executor;
 
     auto [order, ex0, r0, ex1, r1] =
-        co_await make_parallel_group(co_spawn_sw(ex, detail::awaitable_wrap(std::move(t)), deferred),
-                                     co_spawn_sw(ex, detail::awaitable_wrap(std::move(u)), deferred))
+        co_await make_parallel_group(co_spawn(ex, detail::awaitable_wrap(std::move(t)), deferred),
+                                     co_spawn(ex, detail::awaitable_wrap(std::move(u)), deferred))
             .async_wait(wait_for_one(), use_awaitable_t<Executor>{});
 
     using widen = detail::widen_variant<T..., U>;

silkworm/infra/concurrency/parallel_group_test.cpp

@@ -24,11 +24,10 @@
 #include <boost/asio/strand.hpp>
 #include <boost/asio/this_coro.hpp>
 #include <boost/asio/use_awaitable.hpp>
-#include <boost/asio/use_future.hpp>
 #include <catch2/catch_test_macros.hpp>
 
 #include <silkworm/infra/concurrency/awaitable_wait_for_all.hpp>
-#include <silkworm/infra/concurrency/co_spawn_sw.hpp>
+#include <silkworm/infra/concurrency/spawn.hpp>
 
 using namespace boost::asio;
 using namespace boost::asio::experimental;
@@ -52,12 +51,12 @@ awaitable<void> throw_op() {
 }
 
 awaitable<void> spawn_throw_op(strand<any_io_executor>& strand) {
-    co_await co_spawn_sw(strand, throw_op(), use_awaitable);
+    co_await spawn_task(strand, throw_op());
 }
 
 awaitable<void> spawn_noop_loop(strand<any_io_executor>& strand) {
     while (true) {
-        co_await co_spawn_sw(strand, noop(), use_awaitable);
+        co_await spawn_task(strand, noop());
     }
 }
 
@@ -74,6 +73,6 @@ awaitable<void> co_spawn_cancellation_handler_bug() {
 
 TEST_CASE("parallel_group.co_spawn_cancellation_handler_bug") {
     io_context context;
-    co_spawn_sw(context, co_spawn_cancellation_handler_bug(), use_future);
+    spawn_future(context, co_spawn_cancellation_handler_bug());
     context.run();
 }

silkworm/infra/concurrency/spawn.hpp

@@ -0,0 +1,69 @@
+/*
+   Copyright 2023 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#pragma once
+
+#include <utility>
+
+#include <silkworm/infra/concurrency/coroutine.hpp>
+
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/deferred.hpp>
+#include <boost/asio/detail/type_traits.hpp>
+#include <boost/asio/use_awaitable.hpp>
+#include <boost/asio/use_future.hpp>
+
+#include <silkworm/infra/concurrency/context_pool.hpp>
+
+namespace silkworm::concurrency {
+
+template <typename Executor>
+concept AsioExecutor = boost::asio::is_executor<Executor>::value || boost::asio::execution::is_executor<Executor>::value;
+
+template <typename ExecutionContext>
+concept AsioExecutionContext = std::is_convertible_v<ExecutionContext&, boost::asio::execution_context&>;
+
+template <AsioExecutor Executor, typename F>
+auto spawn_task(const Executor& ex, F&& f) {
+    return boost::asio::co_spawn(ex, std::forward<F>(f), boost::asio::use_awaitable);
+}
+
+template <AsioExecutionContext ExecutionContext, typename F>
+auto spawn_task(ExecutionContext& ctx, F&& f) {
+    return boost::asio::co_spawn(ctx, std::forward<F>(f), boost::asio::use_awaitable);
+}
+
+template <AsioExecutor Executor, typename F>
+auto spawn_future(const Executor& ex, F&& f) {
+    return boost::asio::co_spawn(ex, std::forward<F>(f), boost::asio::use_future);
+}
+
+template <AsioExecutionContext ExecutionContext, typename F>
+auto spawn_future(ExecutionContext& ctx, F&& f) {
+    return boost::asio::co_spawn(ctx, std::forward<F>(f), boost::asio::use_future);
+}
+
+template <AsioExecutor Executor, typename F>
+auto spawn_future_and_wait(const Executor& ex, F&& f) {
+    return spawn_future(ex, std::forward<F>(f)).get();
+}
+
+template <AsioExecutionContext ExecutionContext, typename F>
+auto spawn_future_and_wait(ExecutionContext& ctx, F&& f) {
+    return spawn_future(ctx, std::forward<F>(f)).get();
+}
+
+}  // namespace silkworm::concurrency