commit 4cfc284dace92fa0a825d5adbadd7b02543a303c
author Patrick Williams <patrick@stwcx.xyz> Thu Sep 22 09:53:33 2022 -0500
committer Patrick Williams <patrick@stwcx.xyz> Fri Oct 14 06:35:51 2022 -0500
tree ab00ec9ce66f6de7c03149ef869a6f7c7f2a633d
parent 1b7b54cab81d84e6e19117cd00179a22b46b6a86

async: handle exceptions and stops

Add support in `context` and `scope` to handle exceptions and stop
conditions:

  * When an unhandled_stop occurs, turn it into an UnhandledStop
    exception.
  * When a `scope`-spawned task throws an exception, save it and cause
    `set_error` on the `scope::empty()`'s Receiver.
  * When anything in the `context` completes with `set_error` propagate
    that out to the caller of `context::run`.
  * If multiple exceptions occur within a `scope`, terminate.

Signed-off-by: Patrick Williams <patrick@stwcx.xyz>
Change-Id: I25285f7ece5c0675864489bbe1345fa8e7afa70c

src/async/context.cpp

diff --git a/src/async/context.cpp b/src/async/context.cpp
index a5ed091..4807a6a 100644
--- a/src/async/context.cpp
+++ b/src/async/context.cpp
@@ -131,10 +131,17 @@
     {
         // Now that the workers have been requested to stop, we need to wait
         // until they all drain and then stop the internal tasks.
-        internal_tasks.spawn(pending_tasks.empty() | execution::then([this]() {
-                                 final_stop.request_stop();
-                                 caller_wait.notify_one();
-                                 event_loop.break_run();
+
+        auto complete = [this]() {
+            final_stop.request_stop();
+            caller_wait.notify_one();
+            event_loop.break_run();
+        };
+
+        internal_tasks.spawn(pending_tasks.empty() | execution::then(complete) |
+                             execution::upon_error([=](auto&& e) {
+                                 complete();
+                                 std::rethrow_exception(e);
                              }));
     }
 
@@ -155,8 +162,13 @@
         details::wait_process_completion::wait_once(*this);
     }
 
+    std::exception_ptr pending_exception{};
+
     // Wait for all the internal tasks to complete.
-    std::this_thread::sync_wait(internal_tasks.empty());
+    std::this_thread::sync_wait(internal_tasks.empty() |
+                                execution::upon_error([&](auto&& e) {
+                                    pending_exception = std::move(e);
+                                }));
 
     // Stop has been requested, so finish up the loop.
     loop.finish();
@@ -164,6 +176,12 @@
     {
         worker_thread.join();
     }
+
+    // If there was an exception inside the context, rethrow it.
+    if (pending_exception)
+    {
+        std::rethrow_exception(std::move(pending_exception));
+    }
 }
 
 void context::worker_run(task<> startup)