Skip to content

WIP: Async Accessor::with #14019

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
adamrk wants to merge 7 commits into
base: main
Choose a base branch
from
Draft

WIP: Async Accessor::with #14019

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4117db2
Async `Accessor::with`
Jul 29, 2026
6bbf5c9
fix submodule
Jul 29, 2026
a3c04d5
remove now_or_never case
Jul 29, 2026
a50606e
Revert "remove now_or_never case"
Jul 29, 2026
bf20208
fix fuzzing
Jul 30, 2026
ea93cdb
more cleanup
Jul 30, 2026
0caa098
more cleanup
Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions crates/fuzzing/src/oracles/component_api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ use crate::block_on;
use crate::generators::{self, CompilerStrategy, InstanceAllocationStrategy};
use crate::oracles::log_wasm;
use arbitrary::{Arbitrary, Unstructured};
use std::any::Any;
use std::fmt::Debug;
use std::ops::ControlFlow;
use wasmtime::component::{
Expand Down Expand Up @@ -264,24 +263,24 @@ where
{
crate::init_fuzzing();

let mut store = store::<Box<dyn Any + Send>>(input, Box::new(()))?;
let mut store = store::<Option<(P, R)>>(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<Option<(P, R)>> = Linker::new(&engine);

fn host_function<P, R>(
cx: StoreContextMut<'_, Box<dyn Any + Send>>,
cx: StoreContextMut<'_, Option<(P, R)>>,
params: P,
) -> wasmtime::Result<R>
where
P: Debug + PartialEq + 'static,
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:?}");
Expand All @@ -291,9 +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 {
a.with(|mut cx| host_function::<P, R>(cx.as_context_mut(), params))
a.with(|mut cx| host_function(cx.as_context_mut(), params))
.await
})
})
.unwrap();
Expand All @@ -319,7 +319,7 @@ where
while iters.next().is_some() && input.arbitrary()? {
let params = input.arbitrary::<P>()?;
let result = input.arbitrary::<R>()?;
*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
Expand Down Expand Up @@ -381,7 +381,11 @@ 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))
.await
})
}
})
Expand Down
18 changes: 10 additions & 8 deletions crates/misc/component-async-tests/src/resource_stream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,14 +38,16 @@ impl<T> bindings::local::local::resource_stream::HostWithStore<T> for Ctx {
accessor: &Accessor<T, Self>,
count: u32,
) -> wasmtime::Result<StreamReader<Resource<ResourceStreamX>>> {
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
}
}

Expand Down
6 changes: 3 additions & 3 deletions crates/misc/component-async-tests/src/yield_runner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,9 +63,9 @@ impl<T> bindings::local::local::ready::HostThingWithStore<T> for Ctx {
accessor: &Accessor<T, Self>,
thing: Resource<Thing>,
) -> 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();
Expand Down
11 changes: 10 additions & 1 deletion crates/misc/component-async-tests/tests/scenario/backpressure.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use component_async_tests::Ctx;
use futures::FutureExt as _;
use std::{
env, future,
pin::Pin,
Expand Down Expand Up @@ -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);
Expand Down
31 changes: 20 additions & 11 deletions crates/misc/component-async-tests/tests/scenario/round_trip.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,9 +271,14 @@ async fn test_round_trip_recurse(component: &str, same_instance: bool) -> Result
for MyCtx
{
async fn foo(accessor: &Accessor<T, Self>, s: String) -> wasmtime::Result<String> {
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"))
}
Expand All @@ -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<T: Send>(accessor: &Accessor<T, MyCtx>, 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";
Expand Down Expand Up @@ -424,12 +431,14 @@ pub async fn test_round_trip(

impl AccessorTask<Ctx, HasSelf<Ctx>> for Task {
async fn run(self, accessor: &Accessor<Ctx>) -> 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 {
Expand Down
15 changes: 9 additions & 6 deletions crates/misc/component-async-tests/tests/scenario/streams.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
Expand Down Expand Up @@ -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()
Expand All @@ -509,9 +512,9 @@ async fn test_async_short_reads(delay: bool) -> Result<()> {
let received_things = Arc::new(Mutex::new(Vec::<Thing>::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);
Expand Down
144 changes: 77 additions & 67 deletions crates/misc/component-async-tests/tests/scenario/transmit.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(())
})
Expand Down Expand Up @@ -630,17 +632,19 @@ impl TransmitTest for DynamicTransmitTest {
instance: &'a Self::Instance,
params: Self::Params,
) -> Result<Self::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
Expand Down Expand Up @@ -776,15 +780,17 @@ async fn test_transmit_with<Test: TransmitTest + 'static>(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)
Expand All @@ -795,19 +801,21 @@ async fn test_transmit_with<Test: TransmitTest + 'static>(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(
Expand Down Expand Up @@ -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(())
})
Expand Down
Loading
Loading