From 4117db201e71b93b70139e341ac717cadc9c88e6 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 29 Jul 2026 09:28:36 -0400 Subject: [PATCH 1/7] Async `Accessor::with` --- crates/fuzzing/src/oracles/component_api.rs | 13 ++ .../src/resource_stream.rs | 18 ++- .../component-async-tests/src/yield_runner.rs | 6 +- .../tests/scenario/backpressure.rs | 11 +- .../tests/scenario/round_trip.rs | 31 ++-- .../tests/scenario/streams.rs | 15 +- .../tests/scenario/transmit.rs | 144 ++++++++++-------- crates/wasi-http/src/handler.rs | 33 +++- crates/wasi-http/src/p3/host/handler.rs | 29 +++- crates/wasi-http/src/p3/proxy.rs | 6 +- crates/wasi-tls/src/p3/host.rs | 56 +++---- crates/wasi/src/p3/clocks/host.rs | 4 +- crates/wasi/src/p3/filesystem/host.rs | 88 ++++++----- .../src/p3/sockets/host/ip_name_lookup.rs | 4 +- crates/wasi/src/p3/sockets/host/types/tcp.rs | 31 ++-- crates/wasi/src/p3/sockets/host/types/udp.rs | 6 +- .../src/runtime/component/concurrent.rs | 15 +- .../src/runtime/component/concurrent/func.rs | 7 +- .../concurrent/futures_and_streams.rs | 38 +++-- crates/wit-bindgen/src/lib.rs | 2 +- tests/all/component_model/async.rs | 49 ++++-- tests/all/component_model/async_dynamic.rs | 2 +- tests/all/component_model/bindgen.rs | 2 +- tests/all/component_model/import.rs | 11 +- tests/component-model | 2 +- 25 files changed, 393 insertions(+), 230 deletions(-) diff --git a/crates/fuzzing/src/oracles/component_api.rs b/crates/fuzzing/src/oracles/component_api.rs index 24ab9a1be79f..1127ec73f729 100644 --- a/crates/fuzzing/src/oracles/component_api.rs +++ b/crates/fuzzing/src/oracles/component_api.rs @@ -10,6 +10,7 @@ use crate::block_on; use crate::generators::{self, CompilerStrategy, InstanceAllocationStrategy}; use crate::oracles::log_wasm; use arbitrary::{Arbitrary, Unstructured}; +use futures::FutureExt as _; use std::any::Any; use std::fmt::Debug; use std::ops::ControlFlow; @@ -293,7 +294,14 @@ where .root() .func_wrap_concurrent(IMPORT_FUNCTION, |a, params| { Box::pin(async move { + // `Accessor::with` is `async` but resolves synchronously, and + // holding its future across an `.await` here trips Send + // inference (the closure's higher-ranked lifetime). Drive it + // with `now_or_never` so no future is held across a suspend + // point; `host_function` itself does no async work. a.with(|mut cx| host_function::(cx.as_context_mut(), params)) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") }) }) .unwrap(); @@ -381,7 +389,12 @@ pub fn dynamic_component_api_target(input: &mut arbitrary::Unstructured) -> arbi .func_new_concurrent(IMPORT_FUNCTION, { move |cx: &Accessor<_, _>, _, params: &[Val], results: &mut [Val]| { Box::pin(async move { + // See the note above: drive the always-ready + // `Accessor::with` future with `now_or_never` rather than + // `.await` to avoid tripping Send inference. cx.with(|mut store| host_function(store.as_context_mut(), params, results)) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") }) } }) diff --git a/crates/misc/component-async-tests/src/resource_stream.rs b/crates/misc/component-async-tests/src/resource_stream.rs index c54b025d6ae6..ea2daae0a8b4 100644 --- a/crates/misc/component-async-tests/src/resource_stream.rs +++ b/crates/misc/component-async-tests/src/resource_stream.rs @@ -38,14 +38,16 @@ impl bindings::local::local::resource_stream::HostWithStore for Ctx { accessor: &Accessor, count: u32, ) -> wasmtime::Result>> { - accessor.with(|mut access| { - let (mut tx, rx) = mpsc::channel(usize::try_from(count).unwrap()); - for _ in 0..count { - tx.try_send(access.get().table.push(ResourceStreamX)?) - .unwrap() - } - StreamReader::new(access, PipeProducer::new(rx)) - }) + accessor + .with(|mut access| { + let (mut tx, rx) = mpsc::channel(usize::try_from(count).unwrap()); + for _ in 0..count { + tx.try_send(access.get().table.push(ResourceStreamX)?) + .unwrap() + } + StreamReader::new(access, PipeProducer::new(rx)) + }) + .await } } diff --git a/crates/misc/component-async-tests/src/yield_runner.rs b/crates/misc/component-async-tests/src/yield_runner.rs index bdeb3217d792..bf13feacd439 100644 --- a/crates/misc/component-async-tests/src/yield_runner.rs +++ b/crates/misc/component-async-tests/src/yield_runner.rs @@ -63,9 +63,9 @@ impl bindings::local::local::ready::HostThingWithStore for Ctx { accessor: &Accessor, thing: Resource, ) -> wasmtime::Result<()> { - let wakers = accessor.with(|mut view| { - Ok::<_, wasmtime::Error>(view.get().table.get(&thing)?.wakers.clone()) - })?; + let wakers = accessor + .with(|mut view| Ok::<_, wasmtime::Error>(view.get().table.get(&thing)?.wakers.clone())) + .await?; future::poll_fn(move |cx| { let mut wakers = wakers.lock().unwrap(); diff --git a/crates/misc/component-async-tests/tests/scenario/backpressure.rs b/crates/misc/component-async-tests/tests/scenario/backpressure.rs index 782706011b31..6912750e201b 100644 --- a/crates/misc/component-async-tests/tests/scenario/backpressure.rs +++ b/crates/misc/component-async-tests/tests/scenario/backpressure.rs @@ -1,4 +1,5 @@ use component_async_tests::Ctx; +use futures::FutureExt as _; use std::{ env, future, pin::Pin, @@ -75,7 +76,15 @@ pub async fn async_backpressure_callee() -> Result<()> { let mut backpressure_is_set = true; future::poll_fn(move |cx| { - let instance_ready = accessor.poll_ready_for_concurrent_call(func, cx).is_ready(); + // `poll_ready_for_concurrent_call` is `async` but resolves + // synchronously (its body just drives `Accessor::with`), so + // extract the `Poll` with `now_or_never`; the real `cx` still + // registers the waker on `Pending`. + let instance_ready = accessor + .poll_ready_for_concurrent_call(func, cx) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") + .is_ready(); let a_ready = is_ready(cx, &mut a); let b_ready = is_ready(cx, &mut b); let c_ready = is_ready(cx, &mut c); diff --git a/crates/misc/component-async-tests/tests/scenario/round_trip.rs b/crates/misc/component-async-tests/tests/scenario/round_trip.rs index bb2c9fee8c84..9ba684d866d2 100644 --- a/crates/misc/component-async-tests/tests/scenario/round_trip.rs +++ b/crates/misc/component-async-tests/tests/scenario/round_trip.rs @@ -271,9 +271,14 @@ async fn test_round_trip_recurse(component: &str, same_instance: bool) -> Result for MyCtx { async fn foo(accessor: &Accessor, s: String) -> wasmtime::Result { - if let Some(instance) = accessor.with(|mut access| access.get().instance.take()) { + if let Some(instance) = accessor + .with(|mut access| access.get().instance.take()) + .await + { run(accessor, &instance).await?; - accessor.with(|mut access| access.get().instance = Some(instance)); + accessor + .with(|mut access| access.get().instance = Some(instance)) + .await; } Ok(format!("{s} - entered host - exited host")) } @@ -282,9 +287,11 @@ async fn test_round_trip_recurse(component: &str, same_instance: bool) -> Result impl component_async_tests::round_trip::bindings::local::local::baz::Host for MyCtx {} async fn run(accessor: &Accessor, instance: &Instance) -> Result<()> { - let round_trip = accessor.with(|mut access| { - component_async_tests::round_trip::bindings::RoundTrip::new(&mut access, &instance) - })?; + let round_trip = accessor + .with(|mut access| { + component_async_tests::round_trip::bindings::RoundTrip::new(&mut access, &instance) + }) + .await?; let input = "hello, world!"; let expected = "hello, world! - entered guest - entered host - exited host - exited guest"; @@ -424,12 +431,14 @@ pub async fn test_round_trip( impl AccessorTask> for Task { async fn run(self, accessor: &Accessor) -> Result<()> { - let round_trip = accessor.with(|mut store| { - component_async_tests::round_trip::bindings::RoundTrip::new( - &mut store, - &self.instance, - ) - })?; + let round_trip = accessor + .with(|mut store| { + component_async_tests::round_trip::bindings::RoundTrip::new( + &mut store, + &self.instance, + ) + }) + .await?; let mut futures = FuturesUnordered::new(); for (input, output) in &self.inputs_and_outputs { diff --git a/crates/misc/component-async-tests/tests/scenario/streams.rs b/crates/misc/component-async-tests/tests/scenario/streams.rs index f54c95ea15eb..f29247acabbf 100644 --- a/crates/misc/component-async-tests/tests/scenario/streams.rs +++ b/crates/misc/component-async-tests/tests/scenario/streams.rs @@ -295,7 +295,9 @@ pub async fn async_closed_stream() -> Result<()> { let stream = guest.local_local_closed_stream().call_get(accessor).await?; let (tx, mut rx) = mpsc::channel(1); - accessor.with(move |store| stream.pipe(store, PipeConsumer::new(tx)))?; + accessor + .with(move |store| stream.pipe(store, PipeConsumer::new(tx))) + .await?; assert!(rx.next().await.is_none()); Ok(()) @@ -498,8 +500,9 @@ async fn test_async_short_reads(delay: bool) -> Result<()> { store .run_concurrent(async |store| { let count = things.len(); - let stream = - store.with(|store| StreamReader::new(store, VecProducer::new(things, delay)))?; + let stream = store + .with(|store| StreamReader::new(store, VecProducer::new(things, delay))) + .await?; let stream = guest .local_local_short_reads() @@ -509,9 +512,9 @@ async fn test_async_short_reads(delay: bool) -> Result<()> { let received_things = Arc::new(Mutex::new(Vec::::with_capacity(count))); // Read just one item at a time from the guest, forcing it to // re-take ownership of any unwritten items. - store.with(|store| { - stream.pipe(store, OneAtATime::new(received_things.clone(), delay)) - })?; + store + .with(|store| stream.pipe(store, OneAtATime::new(received_things.clone(), delay))) + .await?; for i in 0.. { assert!(i < 1000); diff --git a/crates/misc/component-async-tests/tests/scenario/transmit.rs b/crates/misc/component-async-tests/tests/scenario/transmit.rs index d48b9d2d668f..ef93db2ef1d6 100644 --- a/crates/misc/component-async-tests/tests/scenario/transmit.rs +++ b/crates/misc/component-async-tests/tests/scenario/transmit.rs @@ -370,15 +370,17 @@ pub async fn async_readiness() -> Result<()> { .call_start(accessor, rx, expected) .await?; - accessor.with(|access| { - rx.pipe( - access, - DelayedStreamConsumer { - inner: BufferStreamConsumer { expected }, - maybe_yield: yield_times(10).boxed(), - }, - ) - })?; + accessor + .with(|access| { + rx.pipe( + access, + DelayedStreamConsumer { + inner: BufferStreamConsumer { expected }, + maybe_yield: yield_times(10).boxed(), + }, + ) + }) + .await?; Ok(()) }) @@ -630,17 +632,19 @@ impl TransmitTest for DynamicTransmitTest { instance: &'a Self::Instance, params: Self::Params, ) -> Result { - let exchange_function = accessor.with(|mut store| { - let transmit_instance = instance - .get_export_index(store.as_context_mut(), None, "local:local/transmit") - .ok_or_else(|| format_err!("can't find `local:local/transmit` in instance"))?; - let exchange_function = instance - .get_export_index(store.as_context_mut(), Some(&transmit_instance), "exchange") - .ok_or_else(|| format_err!("can't find `exchange` in instance"))?; - instance - .get_func(store.as_context_mut(), exchange_function) - .ok_or_else(|| format_err!("can't find `exchange` in instance")) - })?; + let exchange_function = accessor + .with(|mut store| { + let transmit_instance = instance + .get_export_index(store.as_context_mut(), None, "local:local/transmit") + .ok_or_else(|| format_err!("can't find `local:local/transmit` in instance"))?; + let exchange_function = instance + .get_export_index(store.as_context_mut(), Some(&transmit_instance), "exchange") + .ok_or_else(|| format_err!("can't find `exchange` in instance"))?; + instance + .get_func(store.as_context_mut(), exchange_function) + .ok_or_else(|| format_err!("can't find `exchange` in instance")) + }) + .await?; let mut results = vec![Val::Bool(false)]; exchange_function @@ -776,15 +780,17 @@ async fn test_transmit_with(component: &str) -> Re .boxed(), ); - let params = accessor.with(|s| { - Test::into_params( - s, - control_rx, - caller_stream_rx, - caller_future1_rx, - caller_future2_rx, - ) - }); + let params = accessor + .with(|s| { + Test::into_params( + s, + control_rx, + caller_stream_rx, + caller_future1_rx, + caller_future2_rx, + ) + }) + .await; futures.push( Test::call(accessor, &test, params) @@ -795,19 +801,21 @@ async fn test_transmit_with(component: &str) -> Re while let Some(event) = futures.try_next().await? { match event { Event::Result(result) => { - accessor.with(|mut store| { - let (callee_stream_rx, callee_future1_rx, _) = - Test::from_result(&mut store, result)?; - callee_stream_rx.pipe( - &mut store, - PipeConsumer::new(callee_stream_tx.take().unwrap()), - )?; - callee_future1_rx.pipe( - &mut store, - OneshotConsumer::new(callee_future1_tx.take().unwrap()), - )?; - wasmtime::error::Ok(()) - })?; + accessor + .with(|mut store| { + let (callee_stream_rx, callee_future1_rx, _) = + Test::from_result(&mut store, result)?; + callee_stream_rx.pipe( + &mut store, + PipeConsumer::new(callee_stream_tx.take().unwrap()), + )?; + callee_future1_rx.pipe( + &mut store, + OneshotConsumer::new(callee_future1_tx.take().unwrap()), + )?; + wasmtime::error::Ok(()) + }) + .await?; } Event::ControlWriteA(mut control_tx) => { futures.push( @@ -958,31 +966,33 @@ async fn test_synchronous_transmit(component: &str, procrastinate: bool) -> Resu .call_start(accessor, stream, stream_expected, future, future_expected) .await?; - accessor.with(|mut access| -> wasmtime::Result<_> { - let consumer = DelayedStreamConsumer { - inner: BufferStreamConsumer { - expected: stream_expected, - }, - maybe_yield: yield_times(10).boxed(), - }; - if procrastinate { - stream.pipe(&mut access, ProcrastinatingStreamConsumer(consumer))?; - } else { - stream.pipe(&mut access, consumer)?; - } - let consumer = DelayedFutureConsumer { - inner: ValueFutureConsumer { - expected: future_expected, - }, - maybe_yield: yield_times(10).boxed(), - }; - if procrastinate { - future.pipe(access, ProcrastinatingFutureConsumer(consumer))?; - } else { - future.pipe(access, consumer)?; - } - Ok(()) - })?; + accessor + .with(|mut access| -> wasmtime::Result<_> { + let consumer = DelayedStreamConsumer { + inner: BufferStreamConsumer { + expected: stream_expected, + }, + maybe_yield: yield_times(10).boxed(), + }; + if procrastinate { + stream.pipe(&mut access, ProcrastinatingStreamConsumer(consumer))?; + } else { + stream.pipe(&mut access, consumer)?; + } + let consumer = DelayedFutureConsumer { + inner: ValueFutureConsumer { + expected: future_expected, + }, + maybe_yield: yield_times(10).boxed(), + }; + if procrastinate { + future.pipe(access, ProcrastinatingFutureConsumer(consumer))?; + } else { + future.pipe(access, consumer)?; + } + Ok(()) + }) + .await?; Ok(()) }) diff --git a/crates/wasi-http/src/handler.rs b/crates/wasi-http/src/handler.rs index 35de271162a0..48d32710d3f2 100644 --- a/crates/wasi-http/src/handler.rs +++ b/crates/wasi-http/src/handler.rs @@ -522,7 +522,7 @@ where *status.try_lock().unwrap() = (WorkerStatus::Requests, start_time); futures.push(async move { - let (prepared, expiration) = prepared?; + let (prepared, expiration) = prepared.await?; let sent = prepared.run(accessor, expiration).await?; wasmtime::error::Ok((sent, start_time)) }); @@ -597,7 +597,16 @@ where Poll::Ready(None) | Poll::Pending => {} } - let is_ready = accessor.poll_ready_for_concurrent_call(func, cx).is_ready(); + // `poll_ready_for_concurrent_call` is an `async` method, but + // its body just drives `Accessor::with`, whose future is + // always immediately ready, so `now_or_never` extracts the + // `Poll` synchronously. The real `cx` is threaded into the + // closure, so the waker is still registered on `Pending`. + let is_ready = accessor + .poll_ready_for_concurrent_call(func, cx) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") + .is_ready(); // At this point `futures` is either empty or it's `Pending` // meaning nothing is ready. Note that `Pending` here @@ -656,7 +665,15 @@ where // then we're done with this iteration of `poll`. We'll get // woken up when anything changes, but otherwise it's time // to let something else happen. - if accessor.poll_no_interesting_tasks(cx).is_pending() { + // As with `poll_ready_for_concurrent_call` above, this + // `async` method resolves synchronously, so drive it with + // `now_or_never`; the real `cx` still registers the waker. + if accessor + .poll_no_interesting_tasks(cx) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") + .is_pending() + { break Poll::Pending; } @@ -1078,10 +1095,12 @@ impl<'a, T: Send> Prepared<'a, T> { .await? .0?; - accessor.with(|mut store| { - let response = view(store.get()).table.delete(response)?; - response.into_http_with_getter(&mut store, request_io_result, view) - }) + accessor + .with(|mut store| { + let response = view(store.get()).table.delete(response)?; + response.into_http_with_getter(&mut store, request_io_result, view) + }) + .await }); // TODO: We should also use `oneshot::Sender::poll_close` to be diff --git a/crates/wasi-http/src/p3/host/handler.rs b/crates/wasi-http/src/p3/host/handler.rs index f8d0ab40a638..ffea20899386 100644 --- a/crates/wasi-http/src/p3/host/handler.rs +++ b/crates/wasi-http/src/p3/host/handler.rs @@ -71,10 +71,17 @@ impl HostWithStore for WasiHttp { Box::into_pin(fut).await }), )) - })?; - let (res, io) = Box::into_pin(fut) - .await - .map_err(|e| store.with(|mut store| store.get().error_to_p3(&e)))?; + }) + .await?; + let (res, io) = match Box::into_pin(fut).await { + Ok(res) => res, + Err(e) => { + return Err(store + .with(|mut store| store.get().error_to_p3(&e)) + .await + .into()); + } + }; let ( http::response::Parts { status, headers, .. @@ -83,10 +90,19 @@ impl HostWithStore for WasiHttp { ) = res.into_parts(); let mut io = Box::into_pin(io); - let body = match io.as_mut().poll(&mut Context::from_waker(Waker::noop())) { + // Poll the I/O future in its own statement so the temporary `Context` + // (which is `!Send` due to its `LocalWaker`) is dropped before the + // `.await` in the error arm below. Temporaries in a `match` scrutinee + // otherwise live until the end of the whole `match`, which would hold + // this `!Send` value across the await and make `send`'s future `!Send`. + let io_poll = io.as_mut().poll(&mut Context::from_waker(Waker::noop())); + let body = match io_poll { Poll::Ready(Ok(())) => body, Poll::Ready(Err(e)) => { - return Err(store.with(|mut store| store.get().error_to_p3(&e)).into()); + return Err(store + .with(|mut store| store.get().error_to_p3(&e)) + .await + .into()); } Poll::Pending => { // I/O driver still needs to be polled, spawn a task and send handles to it @@ -118,6 +134,7 @@ impl HostWithStore for WasiHttp { .context("failed to push response to table") .map_err(HttpError::trap) }) + .await } } diff --git a/crates/wasi-http/src/p3/proxy.rs b/crates/wasi-http/src/p3/proxy.rs index 58aa09520841..7d296df1f879 100644 --- a/crates/wasi-http/src/p3/proxy.rs +++ b/crates/wasi-http/src/p3/proxy.rs @@ -18,7 +18,8 @@ impl Service { .table .push(req.into()) .context("failed to push request to table") - })?; + }) + .await?; match self.wasi_http_handler().call_handle(store, req).await? { Ok(res) => { let res = store.with(|mut store| { @@ -28,7 +29,8 @@ impl Service { .table .delete(res) .context("failed to delete response from table") - })?; + }) + .await?; Ok(Ok(res)) } Err(err) => Ok(Err(err)), diff --git a/crates/wasi-tls/src/p3/host.rs b/crates/wasi-tls/src/p3/host.rs index 66c084f415e2..73d79c4a3cf0 100644 --- a/crates/wasi-tls/src/p3/host.rs +++ b/crates/wasi-tls/src/p3/host.rs @@ -184,31 +184,33 @@ impl bindings::tls::client::HostConnectorWithStore for WasiTls { fn connect_err(msg: &'static str) -> BoxFutureTlsStream { Box::pin(async move { Err(Error::msg(msg)) }) } - let (fut, connection) = accessor.with( - move |mut access| -> wasmtime::Result<(BoxFutureTlsStream, _)> { - let WasiTlsCtxView { table, ctx } = access.get(); - let connector = table.delete(this)?; - let connection = connector.connection; - - let Some(ciphertext_writer) = connector.send else { - return Ok(( - connect_err("send() must be called before connect()"), - connection, - )); - }; - let Some(ciphertext_reader) = connector.recv else { - return Ok(( - connect_err("receive() must be called before connect()"), - connection, - )); - }; - - let transport = Box::new(tokio::io::join(ciphertext_reader, ciphertext_writer)); - let fut = ctx.provider.connect(server_name, transport); - - Ok((fut, connection)) - }, - )?; + let (fut, connection) = accessor + .with( + move |mut access| -> wasmtime::Result<(BoxFutureTlsStream, _)> { + let WasiTlsCtxView { table, ctx } = access.get(); + let connector = table.delete(this)?; + let connection = connector.connection; + + let Some(ciphertext_writer) = connector.send else { + return Ok(( + connect_err("send() must be called before connect()"), + connection, + )); + }; + let Some(ciphertext_reader) = connector.recv else { + return Ok(( + connect_err("receive() must be called before connect()"), + connection, + )); + }; + + let transport = Box::new(tokio::io::join(ciphertext_reader, ciphertext_writer)); + let fut = ctx.provider.connect(server_name, transport); + + Ok((fut, connection)) + }, + ) + .await?; match fut.await { Ok(tls_stream) => { @@ -217,7 +219,9 @@ impl bindings::tls::client::HostConnectorWithStore for WasiTls { } Err(e) => { connection.lock().resolve(Box::new(Closed(e.clone()))); - let resource = accessor.with(|mut access| access.get().table.push(e))?; + let resource = accessor + .with(|mut access| access.get().table.push(e)) + .await?; Ok(Err(resource)) } } diff --git a/crates/wasi/src/p3/clocks/host.rs b/crates/wasi/src/p3/clocks/host.rs index a93438e5876e..2cae85dd713a 100644 --- a/crates/wasi/src/p3/clocks/host.rs +++ b/crates/wasi/src/p3/clocks/host.rs @@ -27,7 +27,9 @@ impl monotonic_clock::HostWithStore for WasiClocks { store: &Accessor, when: monotonic_clock::Mark, ) -> wasmtime::Result<()> { - let clock_now = store.with(|mut view| view.get().ctx.monotonic_clock.now()); + let clock_now = store + .with(|mut view| view.get().ctx.monotonic_clock.now()) + .await; if when > clock_now { sleep(Duration::from_nanos(when - clock_now)).await; }; diff --git a/crates/wasi/src/p3/filesystem/host.rs b/crates/wasi/src/p3/filesystem/host.rs index d36589f9c3b4..2dc35e420355 100644 --- a/crates/wasi/src/p3/filesystem/host.rs +++ b/crates/wasi/src/p3/filesystem/host.rs @@ -51,10 +51,10 @@ fn get_dir<'a>( } trait AccessorExt { - fn get_descriptor(&self, fd: &Resource) -> FilesystemResult; - fn get_file(&self, fd: &Resource) -> FilesystemResult; - fn get_dir(&self, fd: &Resource) -> FilesystemResult; - fn get_dir_pair( + async fn get_descriptor(&self, fd: &Resource) -> FilesystemResult; + async fn get_file(&self, fd: &Resource) -> FilesystemResult; + async fn get_dir(&self, fd: &Resource) -> FilesystemResult; + async fn get_dir_pair( &self, a: &Resource, b: &Resource, @@ -62,28 +62,31 @@ trait AccessorExt { } impl AccessorExt for Accessor { - fn get_descriptor(&self, fd: &Resource) -> FilesystemResult { + async fn get_descriptor(&self, fd: &Resource) -> FilesystemResult { self.with(|mut store| { let fd = get_descriptor(store.get().table, fd)?; Ok(fd.clone()) }) + .await } - fn get_file(&self, fd: &Resource) -> FilesystemResult { + async fn get_file(&self, fd: &Resource) -> FilesystemResult { self.with(|mut store| { let file = get_file(store.get().table, fd)?; Ok(file.clone()) }) + .await } - fn get_dir(&self, fd: &Resource) -> FilesystemResult { + async fn get_dir(&self, fd: &Resource) -> FilesystemResult { self.with(|mut store| { let dir = get_dir(store.get().table, fd)?; Ok(dir.clone()) }) + .await } - fn get_dir_pair( + async fn get_dir_pair( &self, a: &Resource, b: &Resource, @@ -94,6 +97,7 @@ impl AccessorExt for Accessor { let b = get_dir(table, b)?; Ok((a.clone(), b.clone())) }) + .await } } @@ -607,7 +611,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { length: Filesize, advice: Advice, ) -> FilesystemResult<()> { - let file = store.get_file(&fd)?; + let file = store.get_file(&fd).await?; file.advise(offset, length, advice.into()).await?; Ok(()) } @@ -616,7 +620,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { store: &Accessor, fd: Resource, ) -> FilesystemResult<()> { - let fd = store.get_descriptor(&fd)?; + let fd = store.get_descriptor(&fd).await?; fd.sync_data().await?; Ok(()) } @@ -625,7 +629,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { store: &Accessor, fd: Resource, ) -> FilesystemResult { - let fd = store.get_descriptor(&fd)?; + let fd = store.get_descriptor(&fd).await?; let flags = fd.get_flags().await?; Ok(flags.into()) } @@ -634,7 +638,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { store: &Accessor, fd: Resource, ) -> FilesystemResult { - let fd = store.get_descriptor(&fd)?; + let fd = store.get_descriptor(&fd).await?; let ty = fd.get_type().await?; Ok(ty.into()) } @@ -644,7 +648,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { fd: Resource, size: Filesize, ) -> FilesystemResult<()> { - let file = store.get_file(&fd)?; + let file = store.get_file(&fd).await?; file.set_size(size).await?; Ok(()) } @@ -655,7 +659,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { data_access_timestamp: NewTimestamp, data_modification_timestamp: NewTimestamp, ) -> FilesystemResult<()> { - let fd = store.get_descriptor(&fd)?; + let fd = store.get_descriptor(&fd).await?; let atim = systemtimespec_from(data_access_timestamp)?; let mtim = systemtimespec_from(data_modification_timestamp)?; fd.set_times(atim, mtim).await?; @@ -707,7 +711,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { } async fn sync(store: &Accessor, fd: Resource) -> FilesystemResult<()> { - let fd = store.get_descriptor(&fd)?; + let fd = store.get_descriptor(&fd).await?; fd.sync().await?; Ok(()) } @@ -717,7 +721,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { fd: Resource, path: String, ) -> FilesystemResult<()> { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; dir.create_directory_at(path).await?; Ok(()) } @@ -726,7 +730,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { store: &Accessor, fd: Resource, ) -> FilesystemResult { - let fd = store.get_descriptor(&fd)?; + let fd = store.get_descriptor(&fd).await?; let stat = fd.stat().await?; Ok(stat.into()) } @@ -737,7 +741,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { path_flags: PathFlags, path: String, ) -> FilesystemResult { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; let stat = dir.stat_at(path_flags.into(), path).await?; Ok(stat.into()) } @@ -750,7 +754,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { data_access_timestamp: NewTimestamp, data_modification_timestamp: NewTimestamp, ) -> FilesystemResult<()> { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; let atim = systemtimespec_from(data_access_timestamp)?; let mtim = systemtimespec_from(data_modification_timestamp)?; dir.set_times_at(path_flags.into(), path, atim, mtim) @@ -766,7 +770,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { new_fd: Resource, new_path: String, ) -> FilesystemResult<()> { - let (old_dir, new_dir) = store.get_dir_pair(&fd, &new_fd)?; + let (old_dir, new_dir) = store.get_dir_pair(&fd, &new_fd).await?; old_dir .link_at(old_path_flags.into(), old_path, &new_dir, new_path) .await?; @@ -781,11 +785,13 @@ impl types::HostDescriptorWithStore for WasiFilesystem { open_flags: OpenFlags, flags: DescriptorFlags, ) -> FilesystemResult> { - let (allow_blocking_current_thread, dir) = store.with(|mut store| { - let store = store.get(); - let dir = get_dir(&store.table, &fd)?; - FilesystemResult::Ok((store.ctx.allow_blocking_current_thread, dir.clone())) - })?; + let (allow_blocking_current_thread, dir) = store + .with(|mut store| { + let store = store.get(); + let dir = get_dir(&store.table, &fd)?; + FilesystemResult::Ok((store.ctx.allow_blocking_current_thread, dir.clone())) + }) + .await?; let fd = dir .open_at( path_flags.into(), @@ -795,7 +801,9 @@ impl types::HostDescriptorWithStore for WasiFilesystem { allow_blocking_current_thread, ) .await?; - let fd = store.with(|mut store| store.get().table.push(fd))?; + let fd = store + .with(|mut store| store.get().table.push(fd)) + .await?; Ok(fd) } @@ -804,7 +812,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { fd: Resource, path: String, ) -> FilesystemResult { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; let path = dir.readlink_at(path).await?; Ok(path) } @@ -814,7 +822,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { fd: Resource, path: String, ) -> FilesystemResult<()> { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; dir.remove_directory_at(path).await?; Ok(()) } @@ -826,7 +834,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { new_fd: Resource, new_path: String, ) -> FilesystemResult<()> { - let (old_dir, new_dir) = store.get_dir_pair(&fd, &new_fd)?; + let (old_dir, new_dir) = store.get_dir_pair(&fd, &new_fd).await?; old_dir.rename_at(old_path, &new_dir, new_path).await?; Ok(()) } @@ -837,7 +845,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { old_path: String, new_path: String, ) -> FilesystemResult<()> { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; dir.symlink_at(old_path, new_path).await?; Ok(()) } @@ -847,7 +855,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { fd: Resource, path: String, ) -> FilesystemResult<()> { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; dir.unlink_file_at(path).await?; Ok(()) } @@ -857,12 +865,14 @@ impl types::HostDescriptorWithStore for WasiFilesystem { fd: Resource, other: Resource, ) -> wasmtime::Result { - let (fd, other) = store.with(|mut store| { - let table = store.get().table; - let fd = get_descriptor(table, &fd)?.clone(); - let other = get_descriptor(table, &other)?.clone(); - wasmtime::error::Ok((fd, other)) - })?; + let (fd, other) = store + .with(|mut store| { + let table = store.get().table; + let fd = get_descriptor(table, &fd)?.clone(); + let other = get_descriptor(table, &other)?.clone(); + wasmtime::error::Ok((fd, other)) + }) + .await?; fd.is_same_object(&other).await } @@ -870,7 +880,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { store: &Accessor, fd: Resource, ) -> FilesystemResult { - let fd = store.get_descriptor(&fd)?; + let fd = store.get_descriptor(&fd).await?; let meta = fd.metadata_hash().await?; Ok(meta.into()) } @@ -881,7 +891,7 @@ impl types::HostDescriptorWithStore for WasiFilesystem { path_flags: PathFlags, path: String, ) -> FilesystemResult { - let dir = store.get_dir(&fd)?; + let dir = store.get_dir(&fd).await?; let meta = dir.metadata_hash_at(path_flags.into(), path).await?; Ok(meta.into()) } diff --git a/crates/wasi/src/p3/sockets/host/ip_name_lookup.rs b/crates/wasi/src/p3/sockets/host/ip_name_lookup.rs index 084e9af317be..634f8b3b4938 100644 --- a/crates/wasi/src/p3/sockets/host/ip_name_lookup.rs +++ b/crates/wasi/src/p3/sockets/host/ip_name_lookup.rs @@ -10,7 +10,9 @@ impl HostWithStore for WasiSockets { store: &Accessor, name: String, ) -> wasmtime::Result, ErrorCode>> { - let fut = store.with(|mut view| resolve_addresses(&view.get().ctx, name)); + let fut = store + .with(|mut view| resolve_addresses(&view.get().ctx, name)) + .await; Ok(match fut.await { Ok(addrs) => Ok(addrs.into_iter().map(|addr| addr.into()).collect()), Err(err) => Err(err.into()), diff --git a/crates/wasi/src/p3/sockets/host/types/tcp.rs b/crates/wasi/src/p3/sockets/host/types/tcp.rs index 02e3051173eb..9b87929ca195 100644 --- a/crates/wasi/src/p3/sockets/host/types/tcp.rs +++ b/crates/wasi/src/p3/sockets/host/types/tcp.rs @@ -9,6 +9,7 @@ use bytes::BytesMut; use core::iter; use core::pin::Pin; use core::task::{Context, Poll}; +use futures::FutureExt as _; use std::net::SocketAddr; use tokio::sync::oneshot; use wasmtime::component::{ @@ -207,17 +208,29 @@ impl HostTcpSocketWithStore for WasiSockets { ) -> SocketResult<()> { let remote_address = SocketAddr::from(remote_address); - store.with(|mut store| { - let socket = get_socket_mut(store.get().table, &socket)?; - let socket = socket.start_connect(remote_address)?; - SocketResult::Ok(socket) - })?; - - std::future::poll_fn(|cx| { - store.with(|mut store| -> Poll> { + store + .with(|mut store| { let socket = get_socket_mut(store.get().table, &socket)?; - socket.poll_finish_connect(cx).map_err(SocketError::from) + let socket = socket.start_connect(remote_address)?; + SocketResult::Ok(socket) }) + .await?; + + std::future::poll_fn(|cx| { + // `Accessor::with` is an `async` method, but its body is fully + // synchronous, so the future it returns is always immediately + // ready. Drive it to completion with `now_or_never` to obtain + // synchronous access to the store from within this poll context. + // The real `cx` is threaded through to `poll_finish_connect`, so + // when the connection is still pending the real waker is registered + // and `Poll::Pending` propagates out of this `poll_fn` as usual. + store + .with(|mut store| -> Poll> { + let socket = get_socket_mut(store.get().table, &socket)?; + socket.poll_finish_connect(cx).map_err(SocketError::from) + }) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") }) .await } diff --git a/crates/wasi/src/p3/sockets/host/types/udp.rs b/crates/wasi/src/p3/sockets/host/types/udp.rs index 2a6981785574..9fbfdcc45f57 100644 --- a/crates/wasi/src/p3/sockets/host/types/udp.rs +++ b/crates/wasi/src/p3/sockets/host/types/udp.rs @@ -39,7 +39,8 @@ impl HostUdpSocketWithStore for WasiSockets { .with(|mut view| -> SocketResult<_> { let socket = get_socket_mut(view.get().table, &socket)?; Ok(socket.send(data, remote_address.map(SocketAddr::from))) - })? + }) + .await? .await?; Ok(()) } @@ -52,7 +53,8 @@ impl HostUdpSocketWithStore for WasiSockets { .with(|mut view| -> SocketResult<_> { let socket = get_socket_mut(view.get().table, &socket)?; Ok(socket.recv()) - })? + }) + .await? .await?; Ok((data, addr.into())) } diff --git a/crates/wasmtime/src/runtime/component/concurrent.rs b/crates/wasmtime/src/runtime/component/concurrent.rs index 079ce49e6e20..1102917f3be5 100644 --- a/crates/wasmtime/src/runtime/component/concurrent.rs +++ b/crates/wasmtime/src/runtime/component/concurrent.rs @@ -461,7 +461,7 @@ where /// accessor already in scope. For example if `with` is called within `fun`, /// then this function will panic. It is up to the embedder to ensure that /// this does not happen. - pub fn with(&self, fun: impl FnOnce(Access<'_, T, D>) -> R) -> R { + pub async fn with(&self, fun: impl FnOnce(Access<'_, T, D>) -> R) -> R { tls::get(|vmstore| { fun(Access { store: self.token.as_context_mut(vmstore), @@ -517,12 +517,13 @@ where /// Panics if called within a closure provided to the [`Accessor::with`] /// function. This can only be called outside an active invocation of /// [`Accessor::with`]. - pub fn spawn(&self, task: impl AccessorTask) -> Result + pub async fn spawn(&self, task: impl AccessorTask) -> Result where T: 'static, { let accessor = self.clone_for_spawn(); self.with(|mut access| access.as_context_mut().spawn_with_accessor(accessor, task)) + .await } fn clone_for_spawn(&self) -> Self { @@ -567,7 +568,7 @@ where /// Note that at this time spawned threads within a task are always /// considered uninteresting. If this function returns ready, then spawned /// threads may still be in the store. - pub fn poll_no_interesting_tasks(&self, cx: &mut Context<'_>) -> Poll<()> { + pub async fn poll_no_interesting_tasks(&self, cx: &mut Context<'_>) -> Poll<()> { self.with(|mut access| { let store = access.as_context_mut().0; let state = store.concurrent_state_mut_without_forcing_current_thread(); @@ -578,6 +579,7 @@ where Poll::Pending } }) + .await } /// Poll to see if the component instance corresponding to the specified @@ -596,7 +598,11 @@ where /// may be notified when _any_ instance becomes callable (i.e. not /// necessarily the last one polled), so this function must be called again /// to determine if the instance of interest is ready. - pub fn poll_ready_for_concurrent_call(&self, func: Func, cx: &mut Context<'_>) -> Poll<()> { + pub async fn poll_ready_for_concurrent_call( + &self, + func: Func, + cx: &mut Context<'_>, + ) -> Poll<()> { self.with(|mut access| { let store = access.as_context_mut().0; let (_, _, _, raw_options) = func.abi_info(store); @@ -611,6 +617,7 @@ where Poll::Pending } }) + .await } } diff --git a/crates/wasmtime/src/runtime/component/concurrent/func.rs b/crates/wasmtime/src/runtime/component/concurrent/func.rs index 371bb8e621e9..e468df35d8f5 100644 --- a/crates/wasmtime/src/runtime/component/concurrent/func.rs +++ b/crates/wasmtime/src/runtime/component/concurrent/func.rs @@ -118,7 +118,9 @@ impl Func { results: &mut [Val], ) -> Result<()> { let accessor = accessor.as_accessor(); - let call = accessor.with(|store| self.start_call_concurrent(store, params, results))?; + let call = accessor + .with(|store| self.start_call_concurrent(store, params, results)) + .await?; self.finish_call_concurrent(accessor, call).await } @@ -348,7 +350,8 @@ where { let call = accessor .as_accessor() - .with(|store| self.start_call_concurrent(store, params))?; + .with(|store| self.start_call_concurrent(store, params)) + .await?; self.finish_call_concurrent(accessor, call).await } diff --git a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs index ff313978bfc3..1b92f2c4f355 100644 --- a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs +++ b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs @@ -1425,8 +1425,11 @@ impl FutureReader { } /// Convenience method around [`Self::close`]. - pub fn close_with(&mut self, accessor: impl AsAccessor) -> Result<()> { - accessor.as_accessor().with(|access| self.close(access)) + pub async fn close_with(&mut self, accessor: impl AsAccessor) -> Result<()> { + accessor + .as_accessor() + .with(|access| self.close(access)) + .await } /// Returns a [`GuardedFutureReader`] which will auto-close this future on @@ -1434,11 +1437,11 @@ impl FutureReader { /// /// Note that the `accessor` provided must own this future and is /// additionally transferred to the `GuardedFutureReader` return value. - pub fn guard(self, accessor: A) -> GuardedFutureReader + pub async fn guard(self, accessor: A) -> GuardedFutureReader where A: AsAccessor, { - GuardedFutureReader::new(accessor, self) + GuardedFutureReader::new(accessor, self).await } /// Attempts to convert this [`FutureReader`] to a [`FutureAny`]. @@ -1607,11 +1610,12 @@ where /// Panics if [`Config::concurrency_support`] is not enabled. /// /// [`Config::concurrency_support`]: crate::Config::concurrency_support - pub fn new(accessor: A, reader: FutureReader) -> Self { + pub async fn new(accessor: A, reader: FutureReader) -> Self { assert!( accessor .as_accessor() .with(|a| a.as_context().0.concurrency_support()) + .await ); Self { reader: Some(reader), @@ -1643,8 +1647,9 @@ where if let Some(reader) = &mut self.reader { // Currently this can only fail if the future is closed twice, which // this guard prevents, so this error shouldn't happen. - let result = reader.close_with(&self.accessor); - debug_assert!(result.is_ok()); + todo!() + // let result = reader.close_with(&self.accessor); + // debug_assert!(result.is_ok()); } } } @@ -1784,8 +1789,11 @@ impl StreamReader { } /// Convenience method around [`Self::close`]. - pub fn close_with(&mut self, accessor: impl AsAccessor) -> Result<()> { - accessor.as_accessor().with(|access| self.close(access)) + pub async fn close_with(&mut self, accessor: impl AsAccessor) -> Result<()> { + accessor + .as_accessor() + .with(|access| self.close(access)) + .await } /// Returns a [`GuardedStreamReader`] which will auto-close this stream on @@ -1793,11 +1801,11 @@ impl StreamReader { /// /// Note that the `accessor` provided must own this future and is /// additionally transferred to the `GuardedStreamReader` return value. - pub fn guard(self, accessor: A) -> GuardedStreamReader + pub async fn guard(self, accessor: A) -> GuardedStreamReader where A: AsAccessor, { - GuardedStreamReader::new(accessor, self) + GuardedStreamReader::new(accessor, self).await } /// Attempts to convert this [`StreamReader`] to a [`StreamAny`]. @@ -1967,11 +1975,12 @@ where /// Panics if [`Config::concurrency_support`] is not enabled. /// /// [`Config::concurrency_support`]: crate::Config::concurrency_support - pub fn new(accessor: A, reader: StreamReader) -> Self { + pub async fn new(accessor: A, reader: StreamReader) -> Self { assert!( accessor .as_accessor() .with(|a| a.as_context().0.concurrency_support()) + .await ); Self { reader: Some(reader), @@ -2003,8 +2012,9 @@ where if let Some(reader) = &mut self.reader { // Currently this can only fail if the future is closed twice, which // this guard prevents, so this error shouldn't happen. - let result = reader.close_with(&self.accessor); - debug_assert!(result.is_ok()); + todo!(); + // let result = reader.close_with(&self.accessor); + // debug_assert!(result.is_ok()); } } } diff --git a/crates/wit-bindgen/src/lib.rs b/crates/wit-bindgen/src/lib.rs index 5154f4491c5f..818efb8a4dad 100644 --- a/crates/wit-bindgen/src/lib.rs +++ b/crates/wit-bindgen/src/lib.rs @@ -2939,7 +2939,7 @@ pub fn add_to_linker( }; let convert = format!("{}::convert_{}", convert_trait, err_name.to_snake_case()); let convert = if func.kind.is_async() { - format!("caller.with(|mut host| {convert}(&mut host_getter(host.get()), e))") + format!("caller.with(|mut host| {convert}(&mut host_getter(host.get()), e)).await") } else if flags.contains(FunctionFlags::STORE) { format!("{convert}(&mut host_getter(caller.data_mut()), e)") } else { diff --git a/tests/all/component_model/async.rs b/tests/all/component_model/async.rs index d7c712e85b2c..44d324fd9bd2 100644 --- a/tests/all/component_model/async.rs +++ b/tests/all/component_model/async.rs @@ -1,4 +1,5 @@ use crate::async_functions::{PollOnce, execute_across_threads}; +use futures::FutureExt as _; use std::pin::Pin; use std::task::{Context, Poll}; use wasmtime::Result; @@ -455,12 +456,19 @@ async fn run_wasm_in_call_async() -> Result<()> { .root() .func_wrap_concurrent("a", |accessor: &Accessor, (): ()| { Box::pin(async move { - let func = accessor.with(|mut access| { - access - .get() - .unwrap() - .get_typed_func::<(), ()>(&mut access, "run") - })?; + // `Accessor::with` is `async` but resolves synchronously; + // holding its future across an `.await` inside this + // `Send`-required concurrent closure trips Send inference, so + // drive it with `now_or_never` instead. + let func = accessor + .with(|mut access| { + access + .get() + .unwrap() + .get_typed_func::<(), ()>(&mut access, "run") + }) + .now_or_never() + .expect("`Accessor::with` resolves synchronously")?; func.call_concurrent(accessor, ()).await?; Ok(()) }) @@ -658,11 +666,14 @@ async fn sync_lower_async_host_does_not_leak() -> Result<()> { // Keep track of the maximum size of the table in // concurrent_state. - accessor.with(|mut s| { - let cur = s.as_context_mut().concurrent_state_table_size(); - let max = s.data_mut(); - *max = (*max).max(cur); - }); + accessor + .with(|mut s| { + let cur = s.as_context_mut().concurrent_state_table_size(); + let max = s.data_mut(); + *max = (*max).max(cur); + }) + .now_or_never() + .expect("`Accessor::with` resolves synchronously"); Ok(()) }) })?; @@ -896,10 +907,18 @@ async fn concurrent_sync_calls_to_async_host() -> Result<()> { .root() .func_wrap_concurrent("await-three-calls", |accessor, (): ()| { Box::pin(async move { - accessor.with(|mut s| { - *s.data_mut() += 1; - }); - while accessor.with(|mut s| *s.data_mut()) < 3 { + accessor + .with(|mut s| { + *s.data_mut() += 1; + }) + .now_or_never() + .expect("`Accessor::with` resolves synchronously"); + while accessor + .with(|mut s| *s.data_mut()) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") + < 3 + { tokio::task::yield_now().await; } Ok(()) diff --git a/tests/all/component_model/async_dynamic.rs b/tests/all/component_model/async_dynamic.rs index 3b0b67435720..0297ae52026e 100644 --- a/tests/all/component_model/async_dynamic.rs +++ b/tests/all/component_model/async_dynamic.rs @@ -271,7 +271,7 @@ async fn stream_any_smoke() -> Result<()> { wasmtime::error::Ok(()) }, async { - store.with(|store| stream.close(store))?; + store.with(|store| stream.close(store)).await?; wasmtime::error::Ok(()) } }?; diff --git a/tests/all/component_model/bindgen.rs b/tests/all/component_model/bindgen.rs index 95bf96dcecc0..e13eec347ea3 100644 --- a/tests/all/component_model/bindgen.rs +++ b/tests/all/component_model/bindgen.rs @@ -274,7 +274,7 @@ mod one_import_concurrent { impl foo::HostWithStore for MyImports { async fn foo(accessor: &Accessor) { - accessor.with(|mut view| view.get().hit = true); + accessor.with(|mut view| view.get().hit = true).await; } } diff --git a/tests/all/component_model/import.rs b/tests/all/component_model/import.rs index f89f10837f0b..ea813e0fcb64 100644 --- a/tests/all/component_model/import.rs +++ b/tests/all/component_model/import.rs @@ -1,6 +1,7 @@ #![cfg(not(miri))] use super::REALLOC_AND_FREE; +use futures::FutureExt as _; use std::ops::Deref; use wasmtime::Result; use wasmtime::component::*; @@ -757,7 +758,10 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> { WasmStr, WasmStr, ),)| { - accessor.with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")); + accessor + .with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")) + .now_or_never() + .expect("`Accessor::with` resolves synchronously"); Box::pin(async { Ok((3u32,)) }) }, )?; @@ -781,7 +785,10 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> { WasmStr, WasmStr, ),)| { - accessor.with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")); + accessor + .with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")) + .now_or_never() + .expect("`Accessor::with` resolves synchronously"); Box::pin(async { Ok(("xyz".to_string(),)) }) }, )?; diff --git a/tests/component-model b/tests/component-model index e8d800509946..87d36cef87c1 160000 --- a/tests/component-model +++ b/tests/component-model @@ -1 +1 @@ -Subproject commit e8d80050994662d150c84e3d52981a9d471a152d +Subproject commit 87d36cef87c1a38338a3a42ccbf423a5ed3e935d From 6bbf5c94d6149a44824729d787927964250aaee9 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 29 Jul 2026 10:01:15 -0400 Subject: [PATCH 2/7] fix submodule --- tests/component-model | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/component-model b/tests/component-model index 87d36cef87c1..e8d800509946 160000 --- a/tests/component-model +++ b/tests/component-model @@ -1 +1 @@ -Subproject commit 87d36cef87c1a38338a3a42ccbf423a5ed3e935d +Subproject commit e8d80050994662d150c84e3d52981a9d471a152d From a3c04d5ecbd96a1d6984454d04c9ab7cdc4e9da8 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 29 Jul 2026 16:04:42 -0400 Subject: [PATCH 3/7] remove now_or_never case --- crates/fuzzing/src/oracles/component_api.rs | 33 ++++++++++++++------- 1 file changed, 22 insertions(+), 11 deletions(-) diff --git a/crates/fuzzing/src/oracles/component_api.rs b/crates/fuzzing/src/oracles/component_api.rs index 1127ec73f729..db53e1cc22cf 100644 --- a/crates/fuzzing/src/oracles/component_api.rs +++ b/crates/fuzzing/src/oracles/component_api.rs @@ -15,7 +15,7 @@ use std::any::Any; use std::fmt::Debug; use std::ops::ControlFlow; use wasmtime::component::{ - self, Accessor, Component, ComponentNamedList, Lift, Linker, Lower, Val, + self, Accessor, AsAccessor, Component, ComponentNamedList, Lift, Linker, Lower, Val, }; use wasmtime::{AsContextMut, Enabled, Engine, Result, Store, StoreContextMut}; use wasmtime_test_util::component_fuzz::{ @@ -289,20 +289,31 @@ where Ok(result.clone()) } + async fn host_function_async( + cx: &Accessor>, + params: P, + ) -> wasmtime::Result + where + P: Debug + PartialEq + 'static, + R: Debug + Clone + 'static, + { + cx.with(|mut a| { + let store = a.as_context_mut(); + log::trace!("received parameters {params:?}"); + let data: &(P, R) = store.data().downcast_ref().unwrap(); + let (expected_params, result) = data; + assert_eq!(params, *expected_params); + log::trace!("returning result {result:?}"); + Ok(result.clone()) + }) + .await + } + if declarations.options.host_async { linker .root() .func_wrap_concurrent(IMPORT_FUNCTION, |a, params| { - Box::pin(async move { - // `Accessor::with` is `async` but resolves synchronously, and - // holding its future across an `.await` here trips Send - // inference (the closure's higher-ranked lifetime). Drive it - // with `now_or_never` so no future is held across a suspend - // point; `host_function` itself does no async work. - a.with(|mut cx| host_function::(cx.as_context_mut(), params)) - .now_or_never() - .expect("`Accessor::with` resolves synchronously") - }) + Box::pin(async move { host_function_async::(a, params).await }) }) .unwrap(); } else { From a50606ef516693070e03fad97974fec1916665fa Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 29 Jul 2026 16:13:49 -0400 Subject: [PATCH 4/7] Revert "remove now_or_never case" This reverts commit a3c04d5ecbd96a1d6984454d04c9ab7cdc4e9da8. --- crates/fuzzing/src/oracles/component_api.rs | 33 +++++++-------------- 1 file changed, 11 insertions(+), 22 deletions(-) diff --git a/crates/fuzzing/src/oracles/component_api.rs b/crates/fuzzing/src/oracles/component_api.rs index db53e1cc22cf..1127ec73f729 100644 --- a/crates/fuzzing/src/oracles/component_api.rs +++ b/crates/fuzzing/src/oracles/component_api.rs @@ -15,7 +15,7 @@ use std::any::Any; use std::fmt::Debug; use std::ops::ControlFlow; use wasmtime::component::{ - self, Accessor, AsAccessor, Component, ComponentNamedList, Lift, Linker, Lower, Val, + self, Accessor, Component, ComponentNamedList, Lift, Linker, Lower, Val, }; use wasmtime::{AsContextMut, Enabled, Engine, Result, Store, StoreContextMut}; use wasmtime_test_util::component_fuzz::{ @@ -289,31 +289,20 @@ where Ok(result.clone()) } - async fn host_function_async( - cx: &Accessor>, - params: P, - ) -> wasmtime::Result - where - P: Debug + PartialEq + 'static, - R: Debug + Clone + 'static, - { - cx.with(|mut a| { - let store = a.as_context_mut(); - log::trace!("received parameters {params:?}"); - let data: &(P, R) = store.data().downcast_ref().unwrap(); - let (expected_params, result) = data; - assert_eq!(params, *expected_params); - log::trace!("returning result {result:?}"); - Ok(result.clone()) - }) - .await - } - if declarations.options.host_async { linker .root() .func_wrap_concurrent(IMPORT_FUNCTION, |a, params| { - Box::pin(async move { host_function_async::(a, params).await }) + Box::pin(async move { + // `Accessor::with` is `async` but resolves synchronously, and + // holding its future across an `.await` here trips Send + // inference (the closure's higher-ranked lifetime). Drive it + // with `now_or_never` so no future is held across a suspend + // point; `host_function` itself does no async work. + a.with(|mut cx| host_function::(cx.as_context_mut(), params)) + .now_or_never() + .expect("`Accessor::with` resolves synchronously") + }) }) .unwrap(); } else { From bf20208464d7caeb1b0d88870502001f63d35ea4 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Thu, 30 Jul 2026 09:54:47 -0400 Subject: [PATCH 5/7] fix fuzzing --- crates/fuzzing/src/oracles/component_api.rs | 27 +++++++-------------- 1 file changed, 9 insertions(+), 18 deletions(-) diff --git a/crates/fuzzing/src/oracles/component_api.rs b/crates/fuzzing/src/oracles/component_api.rs index 1127ec73f729..1eaa41b7ba4d 100644 --- a/crates/fuzzing/src/oracles/component_api.rs +++ b/crates/fuzzing/src/oracles/component_api.rs @@ -10,8 +10,6 @@ use crate::block_on; use crate::generators::{self, CompilerStrategy, InstanceAllocationStrategy}; use crate::oracles::log_wasm; use arbitrary::{Arbitrary, Unstructured}; -use futures::FutureExt as _; -use std::any::Any; use std::fmt::Debug; use std::ops::ControlFlow; use wasmtime::component::{ @@ -265,16 +263,16 @@ where { crate::init_fuzzing(); - let mut store = store::>(input, Box::new(()))?; + let mut store = store::>(input, None)?; let engine = store.engine(); let wat = declarations.make_component(); let wat = wat.as_bytes(); crate::oracles::log_wasm(wat); let component = Component::new(&engine, wat).unwrap(); - let mut linker = Linker::new(&engine); + let mut linker: Linker> = Linker::new(&engine); fn host_function( - cx: StoreContextMut<'_, Box>, + cx: StoreContextMut<'_, Option<(P, R)>>, params: P, ) -> wasmtime::Result where @@ -282,7 +280,7 @@ where R: Debug + Clone + 'static, { log::trace!("received parameters {params:?}"); - let data: &(P, R) = cx.data().downcast_ref().unwrap(); + let data: &(P, R) = cx.data().as_ref().unwrap(); let (expected_params, result) = data; assert_eq!(params, *expected_params); log::trace!("returning result {result:?}"); @@ -292,16 +290,10 @@ where if declarations.options.host_async { linker .root() - .func_wrap_concurrent(IMPORT_FUNCTION, |a, params| { + .func_wrap_concurrent(IMPORT_FUNCTION, |a, params: P| { Box::pin(async move { - // `Accessor::with` is `async` but resolves synchronously, and - // holding its future across an `.await` here trips Send - // inference (the closure's higher-ranked lifetime). Drive it - // with `now_or_never` so no future is held across a suspend - // point; `host_function` itself does no async work. - a.with(|mut cx| host_function::(cx.as_context_mut(), params)) - .now_or_never() - .expect("`Accessor::with` resolves synchronously") + a.with(|mut cx| host_function(cx.as_context_mut(), params)) + .await }) }) .unwrap(); @@ -327,7 +319,7 @@ where while iters.next().is_some() && input.arbitrary()? { let params = input.arbitrary::

()?; let result = input.arbitrary::()?; - *store.data_mut() = Box::new((params.clone(), result.clone())); + *store.data_mut() = Some((params.clone(), result.clone())); log::trace!("passing in parameters {params:?}"); let actual = if declarations.options.guest_caller_async { store @@ -393,8 +385,7 @@ pub fn dynamic_component_api_target(input: &mut arbitrary::Unstructured) -> arbi // `Accessor::with` future with `now_or_never` rather than // `.await` to avoid tripping Send inference. cx.with(|mut store| host_function(store.as_context_mut(), params, results)) - .now_or_never() - .expect("`Accessor::with` resolves synchronously") + .await }) } }) From ea93cdbde19f231f677e38f27d53b345dace2a7c Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Thu, 30 Jul 2026 11:48:02 -0400 Subject: [PATCH 6/7] more cleanup --- crates/wasi-http/src/p3/host/handler.rs | 83 ++++++++++++------------- tests/all/component_model/async.rs | 21 ++----- 2 files changed, 44 insertions(+), 60 deletions(-) diff --git a/crates/wasi-http/src/p3/host/handler.rs b/crates/wasi-http/src/p3/host/handler.rs index ffea20899386..9874712112be 100644 --- a/crates/wasi-http/src/p3/host/handler.rs +++ b/crates/wasi-http/src/p3/host/handler.rs @@ -52,27 +52,28 @@ impl HostWithStore for WasiHttp { let (res_result_tx, res_result_rx) = oneshot::channel(); let getter = store.getter(); - let fut = store.with(|mut store| { - let WasiHttpCtxView { table, .. } = store.get(); - let req = table - .delete(req) - .context("failed to delete request from table") - .map_err(HttpError::trap)?; - let (req, options) = - req.into_http_with_getter(&mut store, io_task_result(io_result_rx), getter)?; - HttpResult::Ok(store.get().hooks.send_request( - req.map(|body| body.with_state(io_task_rx).boxed_unsync()), - options.as_deref().copied(), - Box::new(async { - // Forward the response processing result to `WasiHttpCtx` implementation - let Ok(fut) = res_result_rx.await else { - return Ok(()); - }; - Box::into_pin(fut).await - }), - )) - }) - .await?; + let fut = store + .with(|mut store| { + let WasiHttpCtxView { table, .. } = store.get(); + let req = table + .delete(req) + .context("failed to delete request from table") + .map_err(HttpError::trap)?; + let (req, options) = + req.into_http_with_getter(&mut store, io_task_result(io_result_rx), getter)?; + HttpResult::Ok(store.get().hooks.send_request( + req.map(|body| body.with_state(io_task_rx).boxed_unsync()), + options.as_deref().copied(), + Box::new(async { + // Forward the response processing result to `WasiHttpCtx` implementation + let Ok(fut) = res_result_rx.await else { + return Ok(()); + }; + Box::into_pin(fut).await + }), + )) + }) + .await?; let (res, io) = match Box::into_pin(fut).await { Ok(res) => res, Err(e) => { @@ -90,11 +91,6 @@ impl HostWithStore for WasiHttp { ) = res.into_parts(); let mut io = Box::into_pin(io); - // Poll the I/O future in its own statement so the temporary `Context` - // (which is `!Send` due to its `LocalWaker`) is dropped before the - // `.await` in the error arm below. Temporaries in a `match` scrutinee - // otherwise live until the end of the whole `match`, which would hold - // this `!Send` value across the await and make `send`'s future `!Send`. let io_poll = io.as_mut().poll(&mut Context::from_waker(Waker::noop())); let body = match io_poll { Poll::Ready(Ok(())) => body, @@ -118,23 +114,24 @@ impl HostWithStore for WasiHttp { body.with_state(io).boxed_unsync() } }; - store.with(|mut store| { - let res = Response { - status, - headers: FieldMap::new_immutable(store.get().hooks, headers), - body: Body::Host { - body, - result_tx: res_result_tx, - }, - }; - store - .get() - .table - .push(res) - .context("failed to push response to table") - .map_err(HttpError::trap) - }) - .await + store + .with(|mut store| { + let res = Response { + status, + headers: FieldMap::new_immutable(store.get().hooks, headers), + body: Body::Host { + body, + result_tx: res_result_tx, + }, + }; + store + .get() + .table + .push(res) + .context("failed to push response to table") + .map_err(HttpError::trap) + }) + .await } } diff --git a/tests/all/component_model/async.rs b/tests/all/component_model/async.rs index 44d324fd9bd2..55666089c6b4 100644 --- a/tests/all/component_model/async.rs +++ b/tests/all/component_model/async.rs @@ -1,5 +1,4 @@ use crate::async_functions::{PollOnce, execute_across_threads}; -use futures::FutureExt as _; use std::pin::Pin; use std::task::{Context, Poll}; use wasmtime::Result; @@ -456,10 +455,6 @@ async fn run_wasm_in_call_async() -> Result<()> { .root() .func_wrap_concurrent("a", |accessor: &Accessor, (): ()| { Box::pin(async move { - // `Accessor::with` is `async` but resolves synchronously; - // holding its future across an `.await` inside this - // `Send`-required concurrent closure trips Send inference, so - // drive it with `now_or_never` instead. let func = accessor .with(|mut access| { access @@ -467,8 +462,7 @@ async fn run_wasm_in_call_async() -> Result<()> { .unwrap() .get_typed_func::<(), ()>(&mut access, "run") }) - .now_or_never() - .expect("`Accessor::with` resolves synchronously")?; + .await?; func.call_concurrent(accessor, ()).await?; Ok(()) }) @@ -672,8 +666,7 @@ async fn sync_lower_async_host_does_not_leak() -> Result<()> { let max = s.data_mut(); *max = (*max).max(cur); }) - .now_or_never() - .expect("`Accessor::with` resolves synchronously"); + .await; Ok(()) }) })?; @@ -911,14 +904,8 @@ async fn concurrent_sync_calls_to_async_host() -> Result<()> { .with(|mut s| { *s.data_mut() += 1; }) - .now_or_never() - .expect("`Accessor::with` resolves synchronously"); - while accessor - .with(|mut s| *s.data_mut()) - .now_or_never() - .expect("`Accessor::with` resolves synchronously") - < 3 - { + .await; + while accessor.with(|mut s| *s.data_mut()).await < 3 { tokio::task::yield_now().await; } Ok(()) From 0caa0986d8bf01ff419550406008f85d2a485214 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Thu, 30 Jul 2026 11:52:39 -0400 Subject: [PATCH 7/7] more cleanup --- tests/all/component_model/import.rs | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/tests/all/component_model/import.rs b/tests/all/component_model/import.rs index ea813e0fcb64..df438c1a0ec5 100644 --- a/tests/all/component_model/import.rs +++ b/tests/all/component_model/import.rs @@ -1,7 +1,6 @@ #![cfg(not(miri))] use super::REALLOC_AND_FREE; -use futures::FutureExt as _; use std::ops::Deref; use wasmtime::Result; use wasmtime::component::*; @@ -758,11 +757,12 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> { WasmStr, WasmStr, ),)| { - accessor - .with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")) - .now_or_never() - .expect("`Accessor::with` resolves synchronously"); - Box::pin(async { Ok((3u32,)) }) + Box::pin(async move { + accessor + .with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")) + .await; + Ok((3u32,)) + }) }, )?; linker @@ -785,11 +785,12 @@ async fn test_stack_and_heap_args_and_rets(concurrent: bool) -> Result<()> { WasmStr, WasmStr, ),)| { - accessor - .with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")) - .now_or_never() - .expect("`Accessor::with` resolves synchronously"); - Box::pin(async { Ok(("xyz".to_string(),)) }) + Box::pin(async move { + accessor + .with(|v| assert_eq!(arg.0.to_str(&v).unwrap(), "abc")) + .await; + Ok(("xyz".to_string(),)) + }) }, )?; } else {