Merge pull request #143 from bytecodealliance/dicej/refactor-concurrent-rs

refactor/rewrite `concurrent.rs` event loop

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);