change the error returned by Instance::run etc. on deadlock

crates/environ/src/trap_encoding.rs

@@ -92,6 +92,11 @@ pub enum Trap {
     /// Async-lifted export failed to produce a result by calling `task.return`
     /// before returning `STATUS_DONE` and/or after all host tasks completed.
     NoAsyncResult,
+
+    /// Async event loop deadlocked; i.e. it cannot make further progress given
+    /// that all host tasks have completed and any/all host-owned stream/future
+    /// handles have been dropped.
+    AsyncDeadlock,
     // if adding a variant here be sure to update the `check!` macro below
 }
 
@@ -129,6 +134,7 @@ impl Trap {
             CastFailure
             CannotEnterComponent
             NoAsyncResult
+            AsyncDeadlock
         }
 
         None
@@ -160,6 +166,7 @@ impl fmt::Display for Trap {
             CastFailure => "cast failure",
             CannotEnterComponent => "cannot enter component instance",
             NoAsyncResult => "async-lifted export failed to produce a result",
+            AsyncDeadlock => "deadlock detected: event loop cannot make further progress",
         };
         write!(f, "wasm trap: {desc}")
     }

crates/wasmtime/src/runtime/component/concurrent.rs

@@ -1710,8 +1710,32 @@ impl ComponentInstance {
                         if ready.is_empty() {
                             return match next {
                                 Poll::Ready(true) => Poll::Ready(Ok(Either::Right(Vec::new()))),
+                                // Here we return an error indicating we can't
+                                // make further progress.  The underlying
+                                // assumption is that `future` depends on this
+                                // component instance making such progress, and
+                                // thus there's no point in continuing to poll
+                                // it given we've run out of work to do.
+                                //
+                                // Note that we'd also reach this point if the
+                                // host embedder passed e.g. a
+                                // `std::future::Pending` to `Instance::run`, in
+                                // which case we'd return a "deadlock" error
+                                // even when any and all tasks have completed
+                                // normally.  However, that's not how
+                                // `Instance::run` is intended (and documented)
+                                // to be used, so it seems reasonable to lump
+                                // that case in with "real" deadlocks.
+                                //
+                                // TODO: Once we've added host APIs for
+                                // cancelling in-progress tasks, we can return
+                                // some other, non-error value here, treating it
+                                // as "normal" and giving the host embedder a
+                                // chance to intervene by cancelling one or more
+                                // tasks and/or starting new tasks capable of
+                                // waking the existing ones.
                                 Poll::Ready(false) => {
-                                    Poll::Ready(Err(anyhow!(crate::Trap::NoAsyncResult)))
+                                    Poll::Ready(Err(anyhow!(crate::Trap::AsyncDeadlock)))
                                 }
                                 Poll::Pending => Poll::Pending,
                             };
@@ -2568,6 +2592,20 @@ impl Instance {
     /// # Ok(())
     /// # }
     /// ```
+    ///
+    /// Note that this function will return a "deadlock" error in either of the
+    /// following scenarios:
+    ///
+    /// - One or more guest tasks are still pending (i.e. have not yet returned,
+    /// or, in the case of async-lifted exports with callbacks, have not yet
+    /// returned `CALLBACK_CODE_EXIT`) even though all host tasks have completed
+    /// all host-owned stream and future handles have been closed, etc.
+    ///
+    /// - Any and all guest tasks complete normally, but the future passed to
+    /// this function continues to return `Pending` when polled.  In that case,
+    /// the future presumably does not depend on any guest task making further
+    /// progress (since no futher progress can be made) and thus is not an
+    /// appropriate future to poll using this function.
     pub async fn run<U: Send, V: Send + Sync + 'static>(
         &self,
         mut store: impl AsContextMut<Data = U>,

tests/misc_testsuite/component-model-async/backpressure-deadlock.wast

@@ -0,0 +1,97 @@
+;;! component_model_async = true
+;;! reference_types = true
+;;! gc_types = true
+;;! multi_memory = true
+
+;; - Component B asks component A to enable backpressure
+;; - Component B makes an async call to component A
+;; - Component B asserts this subtask is in the "STARTING" state
+;; - Component B adds the subtask to a waitable set and calls waitable-set.wait
+;;
+;; This leaves both tasks in a deadlock situation, which, as of this writing,
+;; Wasmtime will handle by trapping.  In the future, once there's a host API for
+;; cancelling tasks, that behavior may change, in which case this test will need
+;; to be updated.
+(component
+
+  (component $A
+    (core func $backpressure.set (canon backpressure.set))
+    (core module $m
+      (import "" "backpressure.set" (func $backpressure.set (param i32)))
+
+      (func (export "f") (result i32) unreachable)
+      (func (export "callback") (param i32 i32 i32) (result i32) unreachable)
+
+      (func (export "turn-on-backpressure")
+        (call $backpressure.set (i32.const 1)))
+    )
+
+    (core instance $i (instantiate $m
+      (with "" (instance
+        (export "backpressure.set" (func $backpressure.set))
+      ))
+    ))
+
+    (func (export "turn-on-backpressure") (canon lift (core func $i "turn-on-backpressure")))
+    (func (export "f")
+      (canon lift (core func $i "f") async (callback (func $i "callback"))))
+  )
+  (instance $A (instantiate $A))
+
+  (core module $libc (memory (export "mem") 1))
+  (core instance $libc (instantiate $libc))
+
+  (core func $f (canon lower (func $A "f") async (memory $libc "mem")))
+  (core func $turn-on-backpressure (canon lower (func $A "turn-on-backpressure")))
+  (core func $waitable-set.new (canon waitable-set.new))
+  (core func $waitable.join (canon waitable.join))
+  (core func $waitable-set.wait (canon waitable-set.wait (memory $libc "mem")))
+
+  (core module $m
+    (import "" "f" (func $f (param i32 i32) (result i32)))
+    (import "" "turn-on-backpressure" (func $turn-on-backpressure))
+    (import "" "waitable-set.new" (func $waitable-set.new (result i32)))
+    (import "" "waitable.join" (func $waitable.join (param i32 i32)))
+    (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32)))
+
+    (func (export "f")
+      (local $status i32)
+      (local $set i32)
+      call $turn-on-backpressure
+
+      (local.set $status (call $f (i32.const 0) (i32.const 0)))
+
+      ;; low 4 bits should be "STARTING == 0"
+      (i32.ne
+        (i32.const 0)
+        (i32.and
+          (local.get $status)
+          (i32.const 0xf)))
+      if unreachable end
+
+      ;; make a new waitable set and join our subtask into it
+      (local.set $set (call $waitable-set.new))
+      (call $waitable.join
+        (i32.shr_u (local.get $status) (i32.const 4))
+        (local.get $set))
+
+      ;; block waiting for our task, which should deadlock (?)
+      (call $waitable-set.wait (local.get $set) (i32.const 0))
+      unreachable
+    )
+  )
+
+  (core instance $i (instantiate $m
+    (with "" (instance
+      (export "f" (func $f))
+      (export "turn-on-backpressure" (func $turn-on-backpressure))
+      (export "waitable-set.new" (func $waitable-set.new))
+      (export "waitable.join" (func $waitable.join))
+      (export "waitable-set.wait" (func $waitable-set.wait))
+    ))
+  ))
+
+  (func (export "f") (canon lift (core func $i "f")))
+)
+
+(assert_trap (invoke "f") "deadlock detected")

tests/misc_testsuite/component-model-async/wait-forever.wast

@@ -53,4 +53,4 @@
     (canon lift (core func $i "run")))
 )
 
-(assert_trap (invoke "run") "async-lifted export failed to produce a result")
+(assert_trap (invoke "run") "deadlock detected")

tests/misc_testsuite/component-model-async/wait-forever2.wast

@@ -0,0 +1,76 @@
+;;! component_model_async = true
+;;! reference_types = true
+;;! gc_types = true
+;;! multi_memory = true
+
+(component
+  (component $C
+    (core module $Memory (memory (export "mem") 1))
+    (core instance $memory (instantiate $Memory))
+    (core module $CM
+      (import "" "mem" (memory 1))
+      (import "" "waitable-set.new" (func $waitable-set.new (result i32)))
+
+      (func (export "f") (result i32)
+        (local $ws i32)
+        ;; return WAIT on an empty waitable set
+        (local.set $ws (call $waitable-set.new))
+        (i32.or (i32.const 2 (; WAIT ;)) (i32.shl (local.get $ws) (i32.const 4)))
+      )
+      (func (export "cb") (param $event_code i32) (param $index i32) (param $payload i32) (result i32)
+        unreachable
+      )
+    )
+    (canon waitable-set.new (core func $waitable-set.new))
+    (core instance $cm (instantiate $CM (with "" (instance
+      (export "mem" (memory $memory "mem"))
+      (export "waitable-set.new" (func $waitable-set.new))
+    ))))
+    (func (export "f") (result u32) (canon lift
+      (core func $cm "f")
+      async (memory $memory "mem") (callback (func $cm "cb"))
+    ))
+  )
+
+  (component $D
+    (import "f" (func $f (result u32)))
+
+    (core module $Memory (memory (export "mem") 1))
+    (core instance $memory (instantiate $Memory))
+    (core module $DM
+      (import "" "mem" (memory 1))
+      (import "" "waitable.join" (func $waitable.join (param i32 i32)))
+      (import "" "waitable-set.new" (func $waitable-set.new (result i32)))
+      (import "" "waitable-set.wait" (func $waitable-set.wait (param i32 i32) (result i32)))
+      (import "" "f" (func $f (param i32 i32) (result i32)))
+
+      (func (export "g") (result i32)
+        (local $ws i32) (local $ret i32) (local $subtaski i32)
+        (local.set $ws (call $waitable-set.new))
+        (local.set $ret (call $f (i32.const 0) (i32.const 0)))
+        (local.set $subtaski (i32.shr_u (local.get $ret) (i32.const 4)))
+        (call $waitable.join (local.get $subtaski) (local.get $ws))
+        (call $waitable-set.wait (local.get $ws) (i32.const 0))
+        unreachable
+      )
+    )
+    (canon waitable.join (core func $waitable.join))
+    (canon waitable-set.new (core func $waitable-set.new))
+    (canon waitable-set.wait (memory $memory "mem") (core func $waitable-set.wait))
+    (canon lower (func $f) async (memory $memory "mem") (core func $f'))
+    (core instance $dm (instantiate $DM (with "" (instance
+      (export "mem" (memory $memory "mem"))
+      (export "waitable.join" (func $waitable.join))
+      (export "waitable-set.new" (func $waitable-set.new))
+      (export "waitable-set.wait" (func $waitable-set.wait))
+      (export "f" (func $f'))
+    ))))
+    (func (export "f") (result u32) (canon lift (core func $dm "g")))
+  )
+
+  (instance $c (instantiate $C))
+  (instance $d (instantiate $D (with "f" (func $c "f"))))
+  (func (export "f") (alias export $d "f"))
+)
+
+(assert_trap (invoke "f") "deadlock detected")