assertion failed: state.guest_thread.is_none()

[if0ne]

Hi there! I want to link wasm components at runtime. How can I do this with version 39.0.0?

Test Case

Here is the test code:

use std::{collections::HashMap, sync::Arc};

use wasmtime::{
    Config,
    component::{Linker, types::ComponentItem},
};
use wasmtime_wasi::p2::bindings::CommandPre;

const SERVICE: &[u8] = include_bytes!("../cron-service.wasm");
const COMPONENT: &[u8] = include_bytes!("../cron_component.wasm");

struct Ctx {
    id: String,
    ctx: wasmtime_wasi::WasiCtx,
    table: wasmtime_wasi::ResourceTable,
}

impl wasmtime_wasi::WasiView for Ctx {
    fn ctx(&mut self) -> wasmtime_wasi::WasiCtxView<'_> {
        wasmtime_wasi::WasiCtxView {
            ctx: &mut self.ctx,
            table: &mut self.table,
        }
    }
}

pub struct Component {
    component: wasmtime::component::Component,
    linker: Linker<Ctx>,
}

impl Component {
    pub fn new(engine: wasmtime::Engine, bytes: &[u8]) -> Self {
        let component = wasmtime::component::Component::new(&engine, bytes).unwrap();
        let mut linker: Linker<Ctx> = wasmtime::component::Linker::new(&engine);
        wasmtime_wasi::p2::add_to_linker_async(&mut linker).unwrap();
        Self { component, linker }
    }
}

#[tokio::main]
async fn main() {
    let engine = wasmtime::Engine::new(
        &Config::new()
            .async_support(true)
            .wasm_component_model_async(true),
    )
    .unwrap();

    let mut service = Component::new(engine.clone(), SERVICE);
    let mut component = Component::new(engine.clone(), COMPONENT);

    resolve_dependencies(&engine, &mut service, &[&mut component]).await;

    let pre = service.linker.instantiate_pre(&service.component).unwrap();
    let cmd_pre = CommandPre::new(pre).unwrap();

    let mut store = wasmtime::Store::new(
        &engine,
        Ctx {
            id: "TEST".to_string(),
            ctx: wasmtime_wasi::WasiCtx::builder().build(),
            table: wasmtime_wasi::ResourceTable::new(),
        },
    );

    let cmd = cmd_pre.instantiate_async(&mut store).await.unwrap();
    cmd.wasi_cli_run()
        .call_run(&mut store)
        .await
        .unwrap()
        .unwrap();
}

async fn resolve_dependencies(
    engine: &wasmtime::Engine,
    component: &mut Component,
    others: &[&mut Component],
) {
    let mut exported_interfaces = HashMap::new();
    for (idx, other) in others.iter().enumerate() {
        for (export_name, export_item) in other.component.component_type().exports(&engine) {
            if matches!(export_item, ComponentItem::ComponentInstance(_)) {
                exported_interfaces.insert(export_name.to_string(), idx);
            }
        }
    }

    let instance_cache: Arc<tokio::sync::RwLock<Option<(String, wasmtime::component::Instance)>>> =
        Arc::default();

    for (import_name, import_item) in component.component.component_type().imports(engine) {
        if let ComponentItem::ComponentInstance(import_instance) = import_item
            && let Some(exporter_idx) = exported_interfaces.get(import_name).copied()
        {
            let exporter = &others[exporter_idx];

            let (_, export_instance_idx) =
                exporter.component.get_export(None, import_name).unwrap();

            let mut linker_instance = component.linker.instance(import_name).unwrap();

            let pre = exporter
                .linker
                .instantiate_pre(&exporter.component)
                .unwrap();

            for (import_export_name, import_export_ty) in import_instance.exports(&engine) {
                if let ComponentItem::ComponentFunc(_) = import_export_ty {
                    let (_, exported_func_idx) = exporter
                        .component
                        .get_export(Some(&export_instance_idx), import_export_name)
                        .unwrap();

                    let pre = pre.clone();
                    let instance_cache = instance_cache.clone();
                    linker_instance
                        .func_new_async(import_export_name, move |mut store, _, params, results| {
                            let pre = pre.clone();
                            let instance_cache = instance_cache.clone();
                            let id = store.data().id.clone();
                            Box::new(async move {
                                if let Some((store_id, instance)) =
                                    instance_cache.read().await.clone()
                                {
                                    if store_id == id {
                                        let func = instance
                                            .get_func(&mut store, exported_func_idx)
                                            .unwrap();
                                        func.call_async(&mut store, params, results).await?;
                                        func.post_return_async(&mut store).await?;
                                        return Ok(());
                                    }
                                }

                                let new_instance = pre.instantiate_async(&mut store).await?;
                                *instance_cache.write().await = Some((id, new_instance.clone()));

                                let func = new_instance
                                    .get_func(&mut store, exported_func_idx)
                                    .unwrap();

                                func.call_async(&mut store, params, results).await.unwrap();
                                func.post_return_async(&mut store).await.unwrap();

                                Ok(())
                            })
                        })
                        .unwrap();
                }
            }
        }
    }
}

Component code:

wit_bindgen::generate!({
    world: "component",
    async: true,
});

struct Component;

impl exports::wasmcloud::example::cron::Guest for Component {
    async fn invoke() -> Result<(), String> {
        eprintln!("Hello from the cron-component!");
        Ok(())
    }
}

export!(Component);

Service code:

wit_bindgen::generate!({
    world: "service",
    async: true,
});

#[tokio::main(flavor = "current_thread")]
async fn main() {
    eprintln!("Starting cron-service with 1 second intervals...");
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        let _ = wasmcloud::example::cron::invoke().await;
    }
}

Shared wit file:

package wasmcloud:example@0.0.1;

interface cron {
    invoke: func() -> result<_, string>;
}

world service {
    import cron;
}

world component {
    export cron;
}

Steps to Reproduce

Expected Results

No panic

Actual Results

Panic:

cargo run
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.09s
     Running `target/debug/test_wt`

thread 'main' panicked at /home/pagafonov/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/wasmtime-39.0.1/src/runtime/component/concurrent.rs:4913:5:
assertion failed: state.guest_thread.is_none()
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

Versions and Environment

tokio = { version = "1.48.0", features = ["full"] }
wasmtime = { version = "39.0.1" }
wasmtime-wasi = "39.0.1"
wit-bindgen = { version = "0.46.0", features = ["async"] }

Extra Info

In version 38, my code works.
If I turn off the component-model-async feature, this code also works.

[alexcrichton]

I'm going to reopen this because I believe this is still an issue with the implementation we need to fix. @if0ne if you'd prefer to not be susbscribed though feel free to unsubscribe!

@dicej and/or @TartanLlama this is the reduction I have for our own test suite:

diff --git a/tests/all/component_model/async.rs b/tests/all/component_model/async.rs
index b1e3bf8d55..015bfa0900 100644
--- a/tests/all/component_model/async.rs
+++ b/tests/all/component_model/async.rs
@@ -767,3 +767,68 @@ async fn cancel_host_future() -> Result<()> {
         }
     }
 }
+
+#[tokio::test]
+#[cfg_attr(miri, ignore)]
+async fn run_wasm_in_call_async() -> Result<()> {
+    let mut config = Config::new();
+    config.async_support(true);
+    config.wasm_component_model_async(true);
+    let engine = Engine::new(&config)?;
+
+    let a = Component::new(
+        &engine,
+        r#"
+(component
+  (type $t (func async))
+  (import "a" (func $f (type $t)))
+  (core func $f (canon lower (func $f)))
+  (core module $a
+    (import "" "f" (func $f))
+    (func (export "run") call $f)
+  )
+  (core instance $a (instantiate $a
+    (with "" (instance (export "f" (func $f))))
+  ))
+  (func (export "run") (type $t)
+    (canon lift (core func $a "run")))
+)
+        "#,
+    )?;
+    let b = Component::new(
+        &engine,
+        r#"
+(component
+  (type $t (func async))
+  (core module $a
+    (func (export "run"))
+  )
+  (core instance $a (instantiate $a))
+  (func (export "run") (type $t)
+    (canon lift (core func $a "run")))
+)
+        "#,
+    )?;
+
+    type State = Option<Instance>;
+
+    let mut linker = Linker::new(&engine);
+    linker
+        .root()
+        .func_wrap_async("a", |mut store: StoreContextMut<'_, State>, (): ()| {
+            Box::new(async move {
+                let instance = store.data().unwrap();
+                let func = instance.get_typed_func::<(), ()>(&mut store, "run")?;
+                func.call_async(&mut store, ()).await?;
+                func.post_return_async(&mut store).await?;
+                Ok(())
+            })
+        })?;
+    let mut store = Store::new(&engine, None);
+    let instance_a = linker.instantiate_async(&mut store, &a).await?;
+    let instance_b = linker.instantiate_async(&mut store, &b).await?;
+    *store.data_mut() = Some(instance_b);
+    let run = instance_a.get_typed_func::<(), ()>(&mut store, "run")?;
+    run.call_async(&mut store, ()).await?;
+    Ok(())
+}

which shows:

$ cargo test --test all run_wasm_in_cal
    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.10s
     Running tests/all/main.rs (target/x86_64-unknown-linux-gnu/debug/deps/all-1c4a0cfe7956e2fe)

running 1 test
test component_model::r#async::run_wasm_in_call_async ... FAILED

failures:

---- component_model::r#async::run_wasm_in_call_async stdout ----

thread 'component_model::r#async::run_wasm_in_call_async' (2825095) panicked at /home/alex/code/wasmtime/crates/wasmtime/src/runtime/component/concurrent.rs:4922:5:
assertion failed: state.guest_thread.is_none()
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace


failures:
    component_model::r#async::run_wasm_in_call_async

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 983 filtered out; finished in 0.01s

error: test failed, to rerun pass `--test all`