refactor/rewrite concurrent.rs event loop

The event loop and scheduler in `concurrent.rs` have undergone a lot of changes
over time, reflecting significant changes in the component model spec.  Some
features, such as yielding, backpressure, and waitable sets, were retrofitted
onto the existing code in a somewhat haphazard way, making the code more
convoluted than it needed to be.  In addition, it made control flow hard to
understand and debug, with multiply-nested fibers and no clear division of
responsibility among tasks and the fibers they ran on.

This commit rewrites most of that code to make it more cohesive:

- No nested fibers: we always suspend the current fiber before resuming the next one
- One single function (`poll_until`) runs the event loop (rather than several functions with slightly different semantics, as before)
- The code paths for `{Typed}Func::call_{async,concurrent}` have been unified, reducing redundancy and unsafe code
- We now manage reentrance, yielding, and backpressure by explicitly deferring work as appropriate
    - Work is deferred to either:
        - a high priority queue (work to be done ASAP)
        - a low priority queue (work to be done only once the high priority queue has been drained)
        - a per-instance `pending` bucket (work that can't be done until a current activation for that instance returns and/or backpressure is unset)
- The new code is more `waitable-set`-native, removing obsolete ties between parent and child tasks which no longer apply
- Fixed previous unsoundness due to recursively modifying `ConcurrentState::futures` while polling it
- More types implement `Debug`, making logs easier to follow

Signed-off-by: Joel Dice <joel.dice@fermyon.com>

Author: dicej

crates/test-programs/src/bin/async_cancel_caller.rs

@@ -102,6 +102,11 @@ enum State {
         set: u32,
         waitable: u32,
         params: *mut SleepParams,
+    },
+    S4 {
+        set: u32,
+        waitable: u32,
+        params: *mut SleepParams,
         params2: *mut u64,
     },
 }
@@ -292,17 +297,39 @@ unsafe extern "C" fn callback_run(event0: u32, event1: u32, event2: u32) -> u32
 
                 let result = subtask_cancel_async(waitable);
 
-                // NB: As of this writing, Wasmtime does *not* spawn a new fiber
-                // for async->async guest calls, which means we'll block
-                // synchronously when calling `subtask_cancel_async` even though
-                // we're calling it with the `async` canonical option.  The
-                // following assertion relies on that behavior; it will need to
-                // be changed (and this test case refactored) if/when that
-                // behavior changes.
-                assert_eq!(result, STATUS_RETURN_CANCELLED);
+                // NB: As of this writing, Wasmtime spawns a new fiber for
+                // async->async guest calls, which means the above call should
+                // block asynchronously, giving us back control.  However, the
+                // runtime could alternatively execute the call on the original
+                // fiber, in which case the above call would block synchronously
+                // and return `STATUS_RETURN_CANCELLED`.  If Wasmtime's behavior
+                // changes, this test will need to be modified.
+                assert_eq!(result, BLOCKED);
 
-                waitable_join(waitable, 0);
-                subtask_drop(waitable);
+                waitable_join(waitable, *set);
+
+                let set = *set;
+
+                *state = State::S3 {
+                    set,
+                    waitable,
+                    params: *params,
+                };
+
+                CALLBACK_CODE_WAIT | (set << 4)
+            }
+
+            State::S3 {
+                set,
+                waitable,
+                params,
+            } => {
+                assert_eq!(event0, EVENT_SUBTASK);
+                assert_eq!(event1, *waitable);
+                assert_eq!(event2, STATUS_RETURN_CANCELLED);
+
+                waitable_join(*waitable, 0);
+                subtask_drop(*waitable);
 
                 // Next, call and cancel `sleep::sleep_millis`, which the callee
                 // implements using both an synchronous lift and asynchronous
@@ -327,7 +354,7 @@ unsafe extern "C" fn callback_run(event0: u32, event1: u32, event2: u32) -> u32
 
                 let set = *set;
 
-                *state = State::S3 {
+                *state = State::S4 {
                     set,
                     waitable,
                     params: *params,
@@ -337,7 +364,7 @@ unsafe extern "C" fn callback_run(event0: u32, event1: u32, event2: u32) -> u32
                 CALLBACK_CODE_WAIT | (set << 4)
             }
 
-            State::S3 {
+            State::S4 {
                 set,
                 waitable,
                 params,

crates/test-programs/src/bin/async_poll_stackless.rs

@@ -102,14 +102,17 @@ unsafe extern "C" fn callback_run(event0: u32, event1: u32, event2: u32) -> u32
         }
 
         State::S3 { set, call } => {
-            assert_eq!(event0, EVENT_SUBTASK);
-            assert_eq!(event1, *call);
-            assert_eq!(event2, STATUS_RETURNED);
-
             let set = *set;
-            subtask_drop(*call);
 
-            *state = State::S4 { set };
+            if event0 != EVENT_NONE {
+                assert_eq!(event0, EVENT_SUBTASK);
+                assert_eq!(event1, *call);
+                assert_eq!(event2, STATUS_RETURNED);
+
+                subtask_drop(*call);
+
+                *state = State::S4 { set };
+            }
 
             CALLBACK_CODE_POLL | (set << 4)
         }

crates/test-programs/src/bin/async_round_trip_many_stackful.rs

@@ -100,10 +100,11 @@ unsafe extern "C" fn export_foo(args: *mut u8) {
         let payload = Box::from_raw(payload);
         if event == EVENT_SUBTASK {
             assert_eq!(call, payload[0] as u32);
-            assert_eq!(STATUS_RETURNED, payload[1] as u32);
-            subtask_drop(call);
-            waitable_set_drop(set);
-            status = STATUS_RETURNED;
+            status = payload[1] as u32;
+            if status == STATUS_RETURNED {
+                subtask_drop(call);
+                waitable_set_drop(set);
+            }
         }
     }
     alloc::dealloc(params, layout);

crates/test-programs/src/bin/async_round_trip_stackful.rs

@@ -86,10 +86,11 @@ unsafe extern "C" fn export_foo(ptr: *mut u8, len: usize) {
         let payload = Box::from_raw(payload);
         if event == EVENT_SUBTASK {
             assert_eq!(call, payload[0] as u32);
-            assert_eq!(STATUS_RETURNED, payload[1] as u32);
-            subtask_drop(call);
-            waitable_set_drop(set);
-            status = STATUS_RETURNED;
+            status = payload[1] as u32;
+            if status == STATUS_RETURNED {
+                subtask_drop(call);
+                waitable_set_drop(set);
+            }
         }
     }
     alloc::dealloc(params, layout);