Merge pull request #19 from ArsenArsen/waitgroups

wait-group: implement Go-ish wait groups

Author: qookei

docs/src/SUMMARY.md

@@ -23,6 +23,9 @@
     - [spawn](headers/basic/spawn.md)
   - [async/result.hpp](headers/result.md)
   - [async/oneshot.hpp](headers/oneshot-event.md)
+  - [async/wait-group.hpp](headers/wait-group.md)
+    - [wait\_group](headers/wait-group/wait_group.md)
+    - [wait\_in\_group](headers/wait-group/wait_in_group.md)
   - [async/recurring.hpp](headers/recurring-event.md)
   - [async/sequenced.hpp](headers/sequenced-event.md)
   - [async/cancellation.hpp](headers/cancellation.md)

docs/src/headers/wait-group.html

@@ -0,0 +1,97 @@
+# wait\_group
+
+```cpp
+#include <async/wait-group.hpp>
+```
+
+`wait_group` is a synchronization primitive that waits for a counter (that can
+be incremented) to reach zero. It conceptually maps to a group of related work
+being done in parallel, and a few consumers waiting for that work to be done.
+The amount of work is increased by calls to `add()` and decreased by calls to
+`done()`.
+
+This struct also implements
+[BasicLockable](https://en.cppreference.com/w/cpp/named_req/BasicLockable) so
+that it can be used with `std::unique_lock`.
+
+## Prototype
+
+```cpp
+struct wait_group {
+	void done(); // (1)
+	void add(int n); // (2)
+
+	sender wait(cancellation_token ct); // (3)
+	sender wait(); // (4)
+
+	void lock(); // (5)
+	void unlock(); // (6)
+};
+```
+
+1. "Finishes" a work (decrements the work count).
+2. "Adds" more work (increments the work count by `n`).
+3. Returns a sender for the wait operation. The operation waits for the counter
+   to drop to zero.
+4. Same as (3) but it cannot be cancelled.
+5. Equivalent to `add(1)`.
+5. Equivalent to `done()`.
+
+### Arguments
+
+ - `n` - amount of work to "add" to this work group
+ - `ct` - the cancellation token to use to listen for cancellation.
+
+### Return values
+
+1. This method doesn't return any value.
+2. This method doesn't return any value.
+3. This method returns a sender of unspecified type. The sender completes with
+either `true` to indicate success, or `false` to indicate that the wait was cancelled.
+4. Same as (3) except the sender completes without a value.
+
+## Examples
+
+```cpp
+async::wait_group wg { 3 };
+
+([&wg] () -> async::detached {
+	std::cout << "before wait" << std::endl;
+	co_await wg.wait();
+	std::cout << "after wait" << std::endl;
+})();
+
+auto done = [&wg] () {
+	std::cout << "before done" << std::endl;
+	wg.done();
+	std::cout << "after done" << std::endl;
+};
+
+done();
+done();
+std::cout << "before add" << std::endl;
+wg.add(2);
+std::cout << "after add" << std::endl;
+done();
+done();
+done();
+```
+
+Output:
+
+```
+before wait
+before done
+after done
+before done
+after done
+before add
+after add
+before done
+after done
+before done
+after done
+before done
+after wait
+after done
+```

docs/src/headers/wait-group/wait_group.md

@@ -0,0 +1,47 @@
+# wait\_in\_group
+
+`wait_in_group(wg, S)` takes a sender `S` and adds it to the work group `wg`
+(calls `wg.add(1)`) immediately before it's started and marks it as done
+(calls `wg.done()`) immediately after.
+
+## Prototype
+
+```cpp
+template<typename S>
+sender wait_in_group(wait_group &wg, S sender);
+```
+
+## Requirements
+`S` is a sender.
+
+## Arguments
+- `wg` - wait group to wait in
+- `sender` - sender to wrap in the wait group
+
+## Return value
+The value produced by the `sender`.
+
+## Examples
+
+```cpp
+bool should_run() {
+	/* ... */
+}
+async::result<void> handle_conn(tcp_socket conn) {
+	/* ... */
+}
+
+/* ... */
+
+tcp_socket server;
+server.bind(":80");
+server.listen(32);
+async::wait_group handlers { 0 };
+while (should_run()) {
+	auto conn = socket.accept();
+	async::detach(async::wait_in_group(handlers, handle_conn(std::move(conn))));
+}
+
+/* wait for all connections to terminate */
+handlers.wait();
+```

docs/src/headers/wait-group/wait_in_group.md

@@ -0,0 +1,11 @@
+# wait-group
+
+This header includes utilities related to the `wait_group` primitive.
+
+```cpp
+#include <async/wait-group.hpp>
+```
+
+Wait groups are synchronization primitives that wait for a counter to reach
+zero. This counter is conceptually bound to a group of related work that the
+consumer would like to simultaneously wait for (hence the name wait groups).

docs/src/wait-group.md

@@ -1,155 +1,27 @@
 #pragma once
 
 #include <async/algorithm.hpp>
+#include <async/wait-group.hpp>
 #include <async/cancellation.hpp>
 #include <frg/functional.hpp>
 #include <frg/list.hpp>
 
 namespace async {
 
 struct oneshot_event {
-private:
-	struct node {
-		friend struct oneshot_event;
-
-		node() = default;
-
-		node(const node &) = delete;
-
-		node &operator= (const node &) = delete;
-
-		virtual void complete() = 0;
-
-	protected:
-		~node() = default;
-
-	private:
-		// Protected by mutex_.
-		frg::default_list_hook<node> _hook;
-	};
-
-public:
 	void raise() {
-		// Grab all items and mark them as retired while we hold the lock.
-		frg::intrusive_list<
-			node,
-			frg::locate_member<
-				node,
-				frg::default_list_hook<node>,
-				&node::_hook
-			>
-		> items;
-		{
-			frg::unique_lock lock(mutex_);
-			assert(!raised_);
-
-			items.splice(items.end(), queue_);
-			raised_ = true;
-		}
-
-		// Now invoke the individual callbacks.
-		while(!items.empty()) {
-			auto item = items.front();
-			items.pop_front();
-			item->complete();
-		}
+		wg_.done();
 	}
 
-	// ----------------------------------------------------------------------------------
-	// wait() and its boilerplate.
-	// ----------------------------------------------------------------------------------
-
-	template<typename Receiver>
-	struct wait_operation final : private node {
-		wait_operation(oneshot_event *evt, cancellation_token ct, Receiver r)
-		: evt_{evt}, ct_{std::move(ct)}, r_{std::move(r)}, cobs_{this} { }
-
-		bool start_inline() {
-			bool cancelled = false;
-			{
-				frg::unique_lock lock(evt_->mutex_);
-
-				if(!evt_->raised_) {
-					if(!cobs_.try_set(ct_)) {
-						cancelled = true;
-					}else{
-						evt_->queue_.push_back(this);
-						return false;
-					}
-				}
-			}
-
-			execution::set_value_inline(r_, !cancelled);
-			return true;
-		}
-
-	private:
-		void cancel() {
-			bool cancelled = false;
-			{
-				frg::unique_lock lock(evt_->mutex_);
-
-				if(!evt_->raised_) {
-					cancelled = true;
-					auto it = evt_->queue_.iterator_to(this);
-					evt_->queue_.erase(it);
-				}
-			}
-
-			execution::set_value_noinline(r_, !cancelled_);
-		}
-
-		void complete() override {
-			if(cobs_.try_reset())
-				execution::set_value_noinline(r_, true);
-		}
-
-		oneshot_event *evt_;
-		cancellation_token ct_;
-		Receiver r_;
-		cancellation_observer<frg::bound_mem_fn<&wait_operation::cancel>> cobs_;
-		bool cancelled_ = false;
-	};
-
-	struct [[nodiscard]] wait_sender {
-		using value_type = bool;
-
-		template<typename Receiver>
-		friend wait_operation<Receiver> connect(wait_sender s, Receiver r) {
-			return {s.evt, s.ct, std::move(r)};
-		}
-
-		sender_awaiter<wait_sender, bool> operator co_await () {
-			return {*this};
-		}
-
-		oneshot_event *evt;
-		cancellation_token ct;
-	};
-
-	wait_sender wait(cancellation_token ct) {
-		return {this, ct};
+	auto wait(cancellation_token ct) {
+		return wg_.wait(ct);
 	}
 
 	auto wait() {
-		return async::transform(wait(cancellation_token{}), [] (bool waitSuccess) {
-			assert(waitSuccess);
-		});
+		return wg_.wait();
 	}
-
 private:
-	platform::mutex mutex_;
-
-	bool raised_ = false;
-
-	frg::intrusive_list<
-		node,
-		frg::locate_member<
-			node,
-			frg::default_list_hook<node>,
-			&node::_hook
-		>
-	> queue_;
+	wait_group wg_ { 1 };
 };
 
 } // namespace async