diff --git a/docs/src/SUMMARY.md b/docs/src/SUMMARY.md index d5cdf76..c632dfa 100644 --- a/docs/src/SUMMARY.md +++ b/docs/src/SUMMARY.md @@ -48,6 +48,7 @@ - [async/promise.hpp](headers/promise.md) - [promise](headers/promise/promise.md) - [future](headers/promise/future.md) + - [async/scope.hpp](headers/scope.md) - [async/post.hpp](headers/post-ack.md) - [post\_ack\_mechanism](headers/post-ack/post_ack_mechanism.md) - [post\_ack\_agent](headers/post-ack/post_ack_agent.md) diff --git a/docs/src/headers/scope.md b/docs/src/headers/scope.md new file mode 100644 index 0000000..7f2a801 --- /dev/null +++ b/docs/src/headers/scope.md @@ -0,0 +1,80 @@ +# async/scope.hpp + +```cpp +#include +``` + +`dynamic_scope` and `dynamic_deferred_scope` are operations that run a number of +senders concurrently, where the number is only known at runtime. The senders are +supplied by a fill function, and the operation only completes once the fill +function has returned and all of the senders have completed. + +## Prototype + +```cpp +template +sender dynamic_scope(Allocator allocator, Fill fill); // (1) + +template +sender dynamic_deferred_scope(Allocator allocator, Fill fill); // (2) +``` + +1. Calls `fill(spawn)`. Each call to `spawn(sender)` starts `sender` +immediately. +2. Calls `fill(spawn)`. The senders passed to `spawn()` are only started after +`fill` returns. + +### Requirements + +`Sender` is a sender that doesn't return any value. `Fill` is invocable with a +reference to a callable that takes a `Sender`. + +`spawn()` may only be called from within `fill`, on the thread that runs +`fill`. It must not be called after `fill` returns and it must not be called +concurrently. + +### Arguments + + - `allocator` - the allocator used to allocate the state of each sender. + - `fill` - the function that supplies the senders. + +### Return value + +This function returns a sender of unspecified type. The sender does not return +any value. + +The senders passed to `spawn()` are started with the environment of the +receiver that the returned sender is connected to. + +With (1), a sender may already have completed by the time that `spawn()` +returns. (2) guarantees that no sender runs before `fill` returns. This is +useful if `fill` runs in a context that the senders must not run in (e.g., while +a lock is held). + +## Examples + +```cpp +async::run(async::dynamic_deferred_scope>(frg::stl_allocator{}, + [] (auto &spawn) { + for(int i = 0; i < 3; i++) { + spawn([] (int i) -> async::result { + std::cout << "Hi " << i << std::endl; + co_return; + }(i)); + std::cout << "Spawned " << i << std::endl; + } + } +)); +std::cout << "Done" << std::endl; +``` + +Output: +``` +Spawned 0 +Spawned 1 +Spawned 2 +Hi 0 +Hi 1 +Hi 2 +Done +``` diff --git a/include/async/scope.hpp b/include/async/scope.hpp new file mode 100644 index 0000000..edb049d --- /dev/null +++ b/include/async/scope.hpp @@ -0,0 +1,141 @@ +#pragma once + +#include +#include + +#include +#include +#include +#include + +namespace async { + +//--------------------------------------------------------------------------------------- +// dynamic_scope() and dynamic_deferred_scope() +//--------------------------------------------------------------------------------------- + +template +struct dynamic_scope_operation { +private: + struct child; + + struct receiver { + void set_value() { + auto self = child_->self_; + frg::destruct(self->allocator_, child_); + self->complete_one_(); + } + + auto get_env() { + return execution::get_env(child_->self_->dr_); + } + + child *child_; + }; + + struct no_hook { }; + + struct child { + child(dynamic_scope_operation *self, Sender s) + : self_{self}, op_{execution::connect(std::move(s), receiver{this})} { } + + dynamic_scope_operation *self_; + [[no_unique_address]] std::conditional_t, no_hook> hook_; + execution::operation_t op_; + }; + +public: + dynamic_scope_operation(Allocator allocator, Fill fill, Receiver dr) + : allocator_{std::move(allocator)}, fill_{std::move(fill)}, dr_{std::move(dr)} { } + + dynamic_scope_operation(const dynamic_scope_operation &) = delete; + + dynamic_scope_operation &operator= (const dynamic_scope_operation &) = delete; + + void start() { + if constexpr (Deferred) { + frg::intrusive_list< + child, + frg::locate_member< + child, + frg::default_list_hook, + &child::hook_ + > + > pending; + size_t n = 0; + + auto spawn = [&] (Sender s) { + pending.push_back(frg::construct(allocator_, this, std::move(s))); + ++n; + }; + fill_(spawn); + + // No child is running yet, so a single RMW is enough. + ctr_.fetch_add(n, std::memory_order_relaxed); + + // Children can complete (and destroy themselves) inline, so unlink them first. + while(!pending.empty()) + execution::start(pending.pop_front()->op_); + } else { + auto spawn = [&] (Sender s) { + ctr_.fetch_add(1, std::memory_order_relaxed); + execution::start(frg::construct(allocator_, this, std::move(s))->op_); + }; + fill_(spawn); + } + + // Drop the reference that prevents completion while start() is still running. + complete_one_(); + } + +private: + void complete_one_() { + auto c = ctr_.fetch_sub(1, std::memory_order_acq_rel); + assert(c > 0); + if(c == 1) + execution::set_value(dr_); + } + + Allocator allocator_; + Fill fill_; + Receiver dr_; + std::atomic ctr_{1}; +}; + +template +struct [[nodiscard]] dynamic_scope_sender { + using value_type = void; + + template Receiver> + friend dynamic_scope_operation + connect(dynamic_scope_sender s, Receiver r) { + return {std::move(s.allocator), std::move(s.fill), std::move(r)}; + } + + Allocator allocator; + Fill fill; +}; + +// Calls fill(spawn). Each spawn(sender) call starts a child immediately. +// Completes once fill() has returned and all children have completed. +template +requires std::is_same_v +dynamic_scope_sender dynamic_scope(Allocator allocator, Fill fill) { + return {std::move(allocator), std::move(fill)}; +} + +// Like dynamic_scope() but children are only started after fill() returns. +template +requires std::is_same_v +dynamic_scope_sender dynamic_deferred_scope(Allocator allocator, Fill fill) { + return {std::move(allocator), std::move(fill)}; +} + +template +sender_awaiter> +operator co_await(dynamic_scope_sender s) { + return {std::move(s)}; +} + +} // namespace async diff --git a/meson.build b/meson.build index f73a95b..ea9df29 100644 --- a/meson.build +++ b/meson.build @@ -25,6 +25,7 @@ if get_option('install_headers') 'include/async/recurring-event.hpp', 'include/async/result.hpp', 'include/async/run-queue.hpp', + 'include/async/scope.hpp', 'include/async/sequenced-event.hpp', 'include/async/wait-group.hpp', 'include/async/generator.hpp', diff --git a/tests/meson.build b/tests/meson.build index 2693f30..39d6282 100644 --- a/tests/meson.build +++ b/tests/meson.build @@ -26,6 +26,7 @@ sources = files( 'with_cancel_cb.cpp', 'generator.cpp', 'run-queue.cpp', + 'scope.cpp', ) exe = executable('gtests', diff --git a/tests/scope.cpp b/tests/scope.cpp new file mode 100644 index 0000000..ac8d16f --- /dev/null +++ b/tests/scope.cpp @@ -0,0 +1,77 @@ +#include +#include +#include +#include +#include + +namespace { + +async::result waitingChild(async::oneshot_event *ev, int *started, int *finished) { + ++*started; + co_await ev->wait(); + ++*finished; +} + +} // namespace + +TEST(DynamicScope, StartsChildrenImmediately) { + async::oneshot_event ev; + int started = 0; + int finished = 0; + bool done = false; + + async::detach(async::dynamic_scope>(frg::stl_allocator{}, + [&] (auto &spawn) { + spawn(waitingChild(&ev, &started, &finished)); + EXPECT_EQ(started, 1); + spawn(waitingChild(&ev, &started, &finished)); + EXPECT_EQ(started, 2); + } + ), [&] { done = true; }); + + ASSERT_FALSE(done); + ev.raise(); + ASSERT_EQ(finished, 2); + ASSERT_TRUE(done); +} + +TEST(DynamicScope, DeferredStartsChildrenAfterFill) { + async::oneshot_event ev; + int started = 0; + int finished = 0; + bool done = false; + + async::detach(async::dynamic_deferred_scope>(frg::stl_allocator{}, + [&] (auto &spawn) { + spawn(waitingChild(&ev, &started, &finished)); + spawn(waitingChild(&ev, &started, &finished)); + EXPECT_EQ(started, 0); + } + ), [&] { done = true; }); + + ASSERT_EQ(started, 2); + ASSERT_FALSE(done); + ev.raise(); + ASSERT_EQ(finished, 2); + ASSERT_TRUE(done); +} + +TEST(DynamicScope, InlineCompletion) { + int n = 0; + bool done = false; + + async::detach(async::dynamic_deferred_scope>(frg::stl_allocator{}, + [&] (auto &spawn) { + for(int i = 0; i < 3; i++) + spawn([] (int *n) -> async::result { ++*n; co_return; }(&n)); + } + ), [&] { done = true; }); + ASSERT_EQ(n, 3); + ASSERT_TRUE(done); + + done = false; + async::detach(async::dynamic_scope>(frg::stl_allocator{}, + [] (auto &) { } + ), [&] { done = true; }); + ASSERT_TRUE(done); +}