use reference counting for guest tasks

[dicej]

Previously, we relied on GuestTask::ready_to_delete to determine when all references to the task have been released and thus when it can be safely removed from the table and dropped. However, that function did not properly take into account any subtask handle owned by the guest and not yet dropped.

This commit removes GuestTask::ready_to_delete in favor of a reference count field which is incremented for each new reference (e.g. host handles, guest handles, implicit and explicit threads, etc.) and decremented as those references are released.

Fixes #14459

[alexcrichton]

Overall I'm definitely feeling like this is cleaning up a lot of the internals 👍

This diff has two panicking tests, both related to the new Drop impl which I also commented on below, however:

diff --git a/tests/all/component_model/async.rs b/tests/all/component_model/async.rs
index eae7ca0012..b210019bb2 100644
--- a/tests/all/component_model/async.rs
+++ b/tests/all/component_model/async.rs
@@ -1612,3 +1612,73 @@ fn inter_component_stream_is_not_intra_component() -> Result<()> {
 
     Ok(())
 }
+
+const THUNK_COMPONENT: &str = r#"
+    (component
+        (core module $m
+            (func (export "thunk"))
+        )
+        (core instance $i (instantiate $m))
+        (func (export "thunk") async
+            (canon lift (core func $i "thunk"))
+        )
+    )
+"#;
+
+/// Dropping the result of `start_call_concurrent` outside of the store's event
+/// loop should not panic.
+#[tokio::test]
+#[cfg_attr(miri, ignore)]
+async fn drop_start_call_concurrent_outside_event_loop() -> Result<()> {
+    let engine = super::async_engine();
+    let component = Component::new(&engine, THUNK_COMPONENT)?;
+    let mut store = Store::new(&engine, ());
+    let instance = Linker::new(&engine)
+        .instantiate_async(&mut store, &component)
+        .await?;
+    let thunk = instance.get_typed_func::<(), ()>(&mut store, "thunk")?;
+
+    let call = thunk.start_call_concurrent(&mut store, ())?;
+    drop(call);
+
+    Ok(())
+}
+
+/// An error returned from `TaskGroupHook::handle_finish` for a
+/// `call_concurrent` should not panic.
+#[tokio::test]
+#[cfg_attr(miri, ignore)]
+async fn task_group_hook_finish_error_call_concurrent() -> Result<()> {
+    struct FailFinish;
+
+    impl TaskGroupHook for FailFinish {
+        fn handle_start(&mut self, _: TaskGroupId) -> Result<()> {
+            Ok(())
+        }
+        fn handle_enter(&mut self, _: TaskGroupId) -> Result<()> {
+            Ok(())
+        }
+        fn handle_exit(&mut self, _: TaskGroupId) -> Result<()> {
+            Ok(())
+        }
+        fn handle_finish(&mut self, _: TaskGroupId) -> Result<()> {
+            wasmtime::bail!("handle_finish failed")
+        }
+    }
+
+    let engine = super::async_engine();
+    let component = Component::new(&engine, THUNK_COMPONENT)?;
+    let mut store = Store::new(&engine, ());
+    let instance = Linker::new(&engine)
+        .instantiate_async(&mut store, &component)
+        .await?;
+    let thunk = instance.get_typed_func::<(), ()>(&mut store, "thunk")?;
+    store.task_group_hook(FailFinish);
+
+    let result = store
+        .run_concurrent(async |store| thunk.call_concurrent(store, ()).await)
+        .await;
+    assert!(result.is_err() || result.unwrap().is_err());
+
+    Ok(())
+}

[alexcrichton]

This feels extremely dangerous to me because there's nothing keeping this scoped to some sort of Accessor-based closure (e.g. a lifetime or similar). I feel like we've had a lot of historical issues where code like this (with external requirements) was satisfied when first-written but later broken without notice which led to various issues.

Can this be shuffled around or similar to somewhere that's more guaranteed to have the store?

[dicej]

Specifically, the issue here is that {Typed}FuncCallConcurrent (based internally on StagedCall) needs to keep the task alive until it is finished or dropped, at which point it must decrement the ref count. So we either need to document that responsibility (e.g. "you must either call finish_call_concurrent or dispose_call_concurrent before dropping this or the task will leak") or rig up some kind of "deadman switch" based e.g. on a oneshot channel to ensure that the ref count is decremented on the next turn of the event loop.

[alexcrichton]

I don't personally feel too strongly about how to fix things. I'd prioritize not using channels since everything is already too slow, and otherwise beyond that IMO it should basically just not be possible to panic the API. We already require explicit deletion for things like futures/streams/resources, so if this is one more that seems not super consequential

[dicej]

OK, I just pushed an update; let me know what you think.