scope: Add dynamic_scope() and dynamic_deferred_scope() algorithms

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)

docs/src/headers/scope.md

@@ -0,0 +1,80 @@
+# async/scope.hpp
+
+```cpp
+#include <async/scope.hpp>
+```
+
+`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<typename Sender, typename Allocator, typename Fill>
+sender dynamic_scope(Allocator allocator, Fill fill); // (1)
+
+template<typename Sender, typename Allocator, typename Fill>
+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<async::result<void>>(frg::stl_allocator{},
+	[] (auto &spawn) {
+		for(int i = 0; i < 3; i++) {
+			spawn([] (int i) -> async::result<void> {
+				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
+```

include/async/scope.hpp

@@ -0,0 +1,141 @@
+#pragma once
+
+#include <atomic>
+#include <type_traits>
+
+#include <async/basic.hpp>
+#include <async/execution.hpp>
+#include <frg/allocation.hpp>
+#include <frg/list.hpp>
+
+namespace async {
+
+//---------------------------------------------------------------------------------------
+// dynamic_scope() and dynamic_deferred_scope()
+//---------------------------------------------------------------------------------------
+
+template<bool Deferred, typename Receiver, typename Sender, typename Allocator, typename Fill>
+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<Deferred,
+				frg::default_list_hook<child>, no_hook> hook_;
+		execution::operation_t<Sender, receiver> 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>,
+					&child::hook_
+				>
+			> pending;
+			size_t n = 0;
+
+			auto spawn = [&] (Sender s) {
+				pending.push_back(frg::construct<child>(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<child>(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<size_t> ctr_{1};
+};
+
+template<bool Deferred, typename Sender, typename Allocator, typename Fill>
+struct [[nodiscard]] dynamic_scope_sender {
+	using value_type = void;
+
+	template<Receives<value_type> Receiver>
+	friend dynamic_scope_operation<Deferred, Receiver, Sender, Allocator, Fill>
+	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<Sender S, typename Allocator, typename Fill>
+requires std::is_same_v<typename S::value_type, void>
+dynamic_scope_sender<false, S, Allocator, Fill> 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<Sender S, typename Allocator, typename Fill>
+requires std::is_same_v<typename S::value_type, void>
+dynamic_scope_sender<true, S, Allocator, Fill> dynamic_deferred_scope(Allocator allocator, Fill fill) {
+	return {std::move(allocator), std::move(fill)};
+}
+
+template<bool Deferred, typename S, typename Allocator, typename Fill>
+sender_awaiter<dynamic_scope_sender<Deferred, S, Allocator, Fill>>
+operator co_await(dynamic_scope_sender<Deferred, S, Allocator, Fill> s) {
+	return {std::move(s)};
+}
+
+} // namespace async

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',

tests/meson.build

@@ -26,6 +26,7 @@ sources = files(
 	'with_cancel_cb.cpp',
 	'generator.cpp',
 	'run-queue.cpp',
+	'scope.cpp',
 )
 
 exe = executable('gtests',

tests/scope.cpp

@@ -0,0 +1,77 @@
+#include <async/basic.hpp>
+#include <async/oneshot-event.hpp>
+#include <async/result.hpp>
+#include <async/scope.hpp>
+#include <gtest/gtest.h>
+
+namespace {
+
+async::result<void> 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<async::result<void>>(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<async::result<void>>(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<async::result<void>>(frg::stl_allocator{},
+		[&] (auto &spawn) {
+			for(int i = 0; i < 3; i++)
+				spawn([] (int *n) -> async::result<void> { ++*n; co_return; }(&n));
+		}
+	), [&] { done = true; });
+	ASSERT_EQ(n, 3);
+	ASSERT_TRUE(done);
+
+	done = false;
+	async::detach(async::dynamic_scope<async::result<void>>(frg::stl_allocator{},
+		[] (auto &) { }
+	), [&] { done = true; });
+	ASSERT_TRUE(done);
+}