wasmtime/runtime/component/concurrent/func.rs
1use crate::component::concurrent::TaskId;
2use crate::component::concurrent::{self, GuestTaskId, PreparedCall};
3use crate::component::func::LowerContext;
4use crate::component::{AsAccessor, ComponentNamedList, Func, Lift, Lower, TypedFunc, Val};
5use crate::prelude::*;
6use crate::runtime::vm::SendSyncPtr;
7use crate::{AsContextMut, StoreContextMut, ValRaw};
8use core::marker;
9use core::mem::MaybeUninit;
10use core::ptr::NonNull;
11use wasmtime_environ::component::{InterfaceType, MAX_FLAT_PARAMS, MAX_FLAT_RESULTS};
12
13/// Returned from [`Func::start_call_concurrent`] to represent a
14/// pending-but-not-yet-resolved call into wasm.
15pub struct FuncCallConcurrent<'a, T> {
16 call: concurrent::StagedCall<Vec<Val>>,
17 results: &'a mut [Val],
18 _marker: marker::PhantomData<fn(T)>,
19}
20
21impl Func {
22 /// Start a concurrent call to this function.
23 ///
24 /// Concurrency is achieved by relying on the [`Accessor`] argument, which
25 /// can be obtained by calling [`StoreContextMut::run_concurrent`].
26 ///
27 /// Unlike [`Self::call`] and [`Self::call_async`] (both of which require
28 /// exclusive access to the store until the completion of the call), calls
29 /// made using this method may run concurrently with other calls to the same
30 /// instance. In addition, the runtime will call the `post-return` function
31 /// (if any) automatically when the guest task completes.
32 ///
33 /// # Progress
34 ///
35 /// For the wasm task being created in `call_concurrent` to make progress it
36 /// must be run within the scope of [`run_concurrent`]. If there are no
37 /// active calls to [`run_concurrent`] then the wasm task will appear as
38 /// stalled. This is typically not a concern as an [`Accessor`] is bound
39 /// by default to a scope of [`run_concurrent`].
40 ///
41 /// One situation in which this can arise, for example, is that if a
42 /// [`run_concurrent`] computation finishes its async closure before all
43 /// wasm tasks have completed, then there will be no scope of
44 /// [`run_concurrent`] anywhere. In this situation the wasm tasks that have
45 /// not yet completed will not make progress until [`run_concurrent`] is
46 /// called again.
47 ///
48 /// Embedders will need to ensure that this future is `await`'d within the
49 /// scope of [`run_concurrent`] to ensure that the value can be produced
50 /// during the `await` call.
51 ///
52 /// # Cancellation
53 ///
54 /// Cancelling an async task created via `call_concurrent`, at this time, is
55 /// only possible by dropping the store that the computation runs within.
56 /// With [#11833] implemented then it will be possible to request
57 /// cancellation of a task, but that is not yet implemented. Hard-cancelling
58 /// a task will only ever be possible by dropping the entire store and it is
59 /// not possible to remove just one task from a store.
60 ///
61 /// This async function behaves more like a "spawn" than a normal Rust async
62 /// function. When this function is invoked then metadata for the function
63 /// call is recorded in the store connected to the `accessor` argument and
64 /// the wasm invocation is from then on connected to the store. If the
65 /// future created by this function is dropped it does not cancel the
66 /// in-progress execution of the wasm task. Dropping the future
67 /// relinquishes the host's ability to learn about the result of the task
68 /// but the task will still progress and invoke callbacks and such until
69 /// completion.
70 ///
71 /// This function will return an error if [`Config::concurrency_support`] is
72 /// disabled.
73 ///
74 /// [`Config::concurrency_support`]: crate::Config::concurrency_support
75 /// [`run_concurrent`]: crate::Store::run_concurrent
76 /// [#11833]: https://github.com/bytecodealliance/wasmtime/issues/11833
77 /// [`Accessor`]: crate::component::Accessor
78 ///
79 /// # Panics
80 ///
81 /// Panics if the store that the [`Accessor`] is derived from does not own
82 /// this function.
83 ///
84 /// # Example
85 ///
86 /// Using [`StoreContextMut::run_concurrent`] to get an [`Accessor`]:
87 ///
88 /// ```
89 /// # use {
90 /// # wasmtime::{
91 /// # error::{Result},
92 /// # component::{Component, Linker, ResourceTable},
93 /// # Config, Engine, Store
94 /// # },
95 /// # };
96 /// #
97 /// # struct Ctx { table: ResourceTable }
98 /// #
99 /// # async fn foo() -> Result<()> {
100 /// # let mut config = Config::new();
101 /// # let engine = Engine::new(&config)?;
102 /// # let mut store = Store::new(&engine, Ctx { table: ResourceTable::new() });
103 /// # let mut linker = Linker::new(&engine);
104 /// # let component = Component::new(&engine, "")?;
105 /// # let instance = linker.instantiate_async(&mut store, &component).await?;
106 /// let my_func = instance.get_func(&mut store, "my_func").unwrap();
107 /// store.run_concurrent(async |accessor| -> wasmtime::Result<_> {
108 /// my_func.call_concurrent(accessor, &[], &mut Vec::new()).await?;
109 /// Ok(())
110 /// }).await??;
111 /// # Ok(())
112 /// # }
113 /// ```
114 pub async fn call_concurrent(
115 self,
116 accessor: impl AsAccessor<Data: Send>,
117 params: &[Val],
118 results: &mut [Val],
119 ) -> Result<()> {
120 let accessor = accessor.as_accessor();
121 let call = accessor.with(|store| self.start_call_concurrent(store, params, results))?;
122 self.finish_call_concurrent(accessor, call).await
123 }
124
125 /// Performs preparatory work for invoking this function with `params`,
126 /// returning a [`FuncCallConcurrent`]
127 /// which can be passed to [`Func::finish_call_concurrent`] to resolve
128 /// the call.
129 ///
130 /// For more information see [`Func::call_concurrent`].
131 pub fn start_call_concurrent<'a, T: Send + 'static>(
132 self,
133 mut store: impl AsContextMut<Data = T>,
134 params: &'a [Val],
135 results: &'a mut [Val],
136 ) -> Result<FuncCallConcurrent<'a, T>> {
137 self.check_params_results(store.as_context_mut(), params, results)?;
138 let prepared = self.prepare_call_dynamic(store.as_context_mut(), params.to_vec())?;
139 let call = concurrent::StagedCall::new(store.as_context_mut(), prepared)?;
140 Ok(FuncCallConcurrent {
141 call,
142 results,
143 _marker: marker::PhantomData,
144 })
145 }
146
147 /// Completes a call that was initiated via
148 /// [`Func::start_call_concurrent`].
149 pub async fn finish_call_concurrent<T: Send>(
150 self,
151 accessor: impl AsAccessor<Data = T>,
152 call: FuncCallConcurrent<'_, T>,
153 ) -> Result<()> {
154 // Intentionally not used today, but left here for future API
155 // compatibility with using this.
156 let _ = accessor;
157 let FuncCallConcurrent { call, results, .. } = call;
158 let run_results = call.await?;
159 assert_eq!(run_results.len(), results.len());
160 for (result, slot) in run_results.into_iter().zip(results) {
161 *slot = result;
162 }
163 Ok(())
164 }
165
166 /// Calls `concurrent::prepare_call` with monomorphized functions for
167 /// lowering the parameters and lifting the result.
168 fn prepare_call_dynamic<'a, T: Send + 'static>(
169 self,
170 mut store: StoreContextMut<'a, T>,
171 params: Vec<Val>,
172 ) -> Result<PreparedCall<Vec<Val>>> {
173 let store = store.as_context_mut();
174 let (options, flags, ty, raw_options) = self.abi_info(store.0);
175 let async_ = raw_options.async_;
176 let instance = self.instance();
177
178 concurrent::prepare_call(
179 store,
180 self,
181 MAX_FLAT_PARAMS,
182 false,
183 move |store, params_out| {
184 Func::with_lower_context(instance, store, options, flags, ty, |cx, ty| {
185 Self::lower_args(cx, ¶ms, ty, params_out)
186 })
187 },
188 move |store, results| {
189 let max_flat = if async_ {
190 MAX_FLAT_PARAMS
191 } else {
192 MAX_FLAT_RESULTS
193 };
194 let results = Func::with_lift_context(instance, store, options, ty, |cx, ty| {
195 Self::lift_results(cx, ty, results, max_flat)?.collect::<Result<Vec<_>>>()
196 })?;
197 Ok(Box::new(results))
198 },
199 )
200 }
201}
202
203impl<T> FuncCallConcurrent<'_, T> {
204 /// Returns the task that this invocation corresponds to.
205 ///
206 /// This can be later correlated with [`StoreContextMut::async_call_stack`]
207 /// for example.
208 pub fn task(&self) -> GuestTaskId {
209 self.call.task()
210 }
211}
212
213/// Returned from [`TypedFunc::start_call_concurrent`] to represent a
214/// pending-but-not-yet-resolved call into wasm.
215pub struct TypedFuncCallConcurrent<T, P, R> {
216 call: concurrent::StagedCall<R>,
217 _marker: marker::PhantomData<fn(T, P)>,
218}
219
220impl<Params, Return> TypedFunc<Params, Return>
221where
222 Params: ComponentNamedList + Lower,
223 Return: ComponentNamedList + Lift,
224{
225 pub(crate) async fn call_async_concurrent(
226 &self,
227 mut store: impl AsContextMut<Data: Send>,
228 params: Params,
229 ) -> Result<Return>
230 where
231 Return: 'static,
232 {
233 let mut store = store.as_context_mut();
234 let ptr = SendSyncPtr::from(NonNull::from(¶ms).cast::<u8>());
235 let prepared = self.prepare_call(store.as_context_mut(), true, move |cx, ty, dst| {
236 // SAFETY: The goal here is to get `Params`, a non-`'static`
237 // value, to live long enough to the lowering of the
238 // parameters. We're guaranteed that `Params` lives in the
239 // future of the outer function (we're in an `async fn`) so it'll
240 // stay alive as long as the future itself. That is distinct,
241 // for example, from the signature of `call_concurrent` below.
242 //
243 // Here a pointer to `Params` is smuggled to this location
244 // through a `SendSyncPtr<u8>` to thwart the `'static` check
245 // of rustc and the signature of `prepare_call`.
246 //
247 // Note the use of `SignalOnDrop` in the code that follows
248 // this closure, which ensures that the task will be removed
249 // from the concurrent state to which it belongs when the
250 // containing `Future` is dropped, so long as the parameters
251 // have not yet been lowered. Since this closure is removed from
252 // the task after the parameters are lowered, it will never be called
253 // after the containing `Future` is dropped.
254 let params = unsafe { ptr.cast::<Params>().as_ref() };
255 Self::lower_args(cx, ty, dst, params)
256 })?;
257
258 struct SignalOnDrop<'a, T: 'static> {
259 store: StoreContextMut<'a, T>,
260 task: TaskId,
261 }
262
263 impl<'a, T> Drop for SignalOnDrop<'a, T> {
264 fn drop(&mut self) {
265 self.task.host_future_dropped(self.store.0).unwrap();
266 }
267 }
268
269 let mut wrapper = SignalOnDrop {
270 store,
271 task: prepared.task_id(),
272 };
273
274 let result = concurrent::StagedCall::new(wrapper.store.as_context_mut(), prepared)?;
275 wrapper
276 .store
277 .as_context_mut()
278 .run_concurrent_trap_on_idle(async |_| Ok(result.await?))
279 .await?
280 }
281
282 /// Start a concurrent call to this function.
283 ///
284 /// Concurrency is achieved by relying on the [`Accessor`] argument, which
285 /// can be obtained by calling [`StoreContextMut::run_concurrent`].
286 ///
287 /// Unlike [`Self::call`] and [`Self::call_async`] (both of which require
288 /// exclusive access to the store until the completion of the call), calls
289 /// made using this method may run concurrently with other calls to the same
290 /// instance. In addition, the runtime will call the `post-return` function
291 /// (if any) automatically when the guest task completes.
292 ///
293 /// This function will return an error if [`Config::concurrency_support`] is
294 /// disabled.
295 ///
296 /// [`Config::concurrency_support`]: crate::Config::concurrency_support
297 ///
298 /// # Progress and Cancellation
299 ///
300 /// For more information about how to make progress on the wasm task or how
301 /// to cancel the wasm task see the documentation for
302 /// [`Func::call_concurrent`].
303 ///
304 /// [`Func::call_concurrent`]: crate::component::Func::call_concurrent
305 ///
306 /// # Panics
307 ///
308 /// Panics if the store that the [`Accessor`] is derived from does not own
309 /// this function.
310 ///
311 /// [`Accessor`]: crate::component::Accessor
312 ///
313 /// # Example
314 ///
315 /// Using [`StoreContextMut::run_concurrent`] to get an [`Accessor`]:
316 ///
317 /// ```
318 /// # use {
319 /// # wasmtime::{
320 /// # error::{Result},
321 /// # component::{Component, Linker, ResourceTable},
322 /// # Config, Engine, Store
323 /// # },
324 /// # };
325 /// #
326 /// # struct Ctx { table: ResourceTable }
327 /// #
328 /// # async fn foo() -> Result<()> {
329 /// # let mut config = Config::new();
330 /// # let engine = Engine::new(&config)?;
331 /// # let mut store = Store::new(&engine, Ctx { table: ResourceTable::new() });
332 /// # let mut linker = Linker::new(&engine);
333 /// # let component = Component::new(&engine, "")?;
334 /// # let instance = linker.instantiate_async(&mut store, &component).await?;
335 /// let my_typed_func = instance.get_typed_func::<(), ()>(&mut store, "my_typed_func")?;
336 /// store.run_concurrent(async |accessor| -> wasmtime::Result<_> {
337 /// my_typed_func.call_concurrent(accessor, ()).await?;
338 /// Ok(())
339 /// }).await??;
340 /// # Ok(())
341 /// # }
342 /// ```
343 pub async fn call_concurrent(
344 self,
345 accessor: impl AsAccessor<Data: Send>,
346 params: Params,
347 ) -> Result<Return>
348 where
349 Params: 'static,
350 Return: 'static,
351 {
352 let call = accessor
353 .as_accessor()
354 .with(|store| self.start_call_concurrent(store, params))?;
355 self.finish_call_concurrent(accessor, call).await
356 }
357
358 /// Performs preparatory work for invoking this function with `params`,
359 /// returning a [`TypedFuncCallConcurrent`]
360 /// which can be passed to [`TypedFunc::finish_call_concurrent`] to resolve
361 /// the call.
362 ///
363 /// For more information see [`TypedFunc::call_concurrent`].
364 pub fn start_call_concurrent<T>(
365 self,
366 mut store: impl AsContextMut<Data = T>,
367 params: Params,
368 ) -> Result<TypedFuncCallConcurrent<T, Params, Return>>
369 where
370 T: Send + 'static,
371 Params: 'static,
372 Return: 'static,
373 {
374 let mut store = store.as_context_mut();
375 let mut store = store.as_context_mut();
376 ensure!(
377 store.0.concurrency_support(),
378 "cannot use `call_concurrent` Config::concurrency_support disabled",
379 );
380
381 let prepared = self.prepare_call(store.as_context_mut(), false, move |cx, ty, dst| {
382 Self::lower_args(cx, ty, dst, ¶ms)
383 })?;
384 let call = concurrent::StagedCall::new(store, prepared)?;
385 Ok(TypedFuncCallConcurrent {
386 call,
387 _marker: marker::PhantomData,
388 })
389 }
390
391 /// Completes a call that was initiated via
392 /// [`TypedFunc::start_call_concurrent`].
393 pub async fn finish_call_concurrent<T>(
394 self,
395 accessor: impl AsAccessor<Data = T>,
396 call: TypedFuncCallConcurrent<T, Params, Return>,
397 ) -> Result<Return>
398 where
399 T: Send + 'static,
400 Params: 'static,
401 Return: 'static,
402 {
403 // This is intentionally part of the public API but not used yet.
404 // This'll likely want to be used in future refactorings.
405 let _ = accessor;
406 call.call.await
407 }
408
409 /// Calls `concurrent::prepare_call` with monomorphized functions for
410 /// lowering the parameters and lifting the result according to the number
411 /// of core Wasm parameters and results in the signature of the function to
412 /// be called.
413 fn prepare_call<T>(
414 self,
415 store: StoreContextMut<'_, T>,
416 host_future_present: bool,
417 lower: impl FnOnce(
418 &mut LowerContext<T>,
419 InterfaceType,
420 &mut [MaybeUninit<ValRaw>],
421 ) -> Result<()>
422 + Send
423 + Sync
424 + 'static,
425 ) -> Result<PreparedCall<Return>>
426 where
427 Return: 'static,
428 {
429 use crate::component::storage::slice_to_storage;
430 debug_assert!(store.0.concurrency_support());
431
432 let param_count = if Params::flatten_count() <= MAX_FLAT_PARAMS {
433 Params::flatten_count()
434 } else {
435 1
436 };
437 let (options, flags, ty, raw_options) = self.func().abi_info(store.0);
438 let instance = self.func().instance();
439 let max_results = if raw_options.async_ {
440 MAX_FLAT_PARAMS
441 } else {
442 MAX_FLAT_RESULTS
443 };
444
445 concurrent::prepare_call(
446 store,
447 *self.func(),
448 param_count,
449 host_future_present,
450 move |store, params_out| {
451 Func::with_lower_context(instance, store, options, flags, ty, |cx, ty| {
452 lower(cx, ty, params_out)
453 })
454 },
455 move |store, results| {
456 let result = if Return::flatten_count() <= max_results {
457 Func::with_lift_context(instance, store, options, ty, |cx, ty| {
458 // SAFETY: Per the safety requirements documented for the
459 // `ComponentType` trait, `Return::Lower` must be
460 // compatible at the binary level with a `[ValRaw; N]`,
461 // where `N` is `mem::size_of::<Return::Lower>() /
462 // mem::size_of::<ValRaw>()`. And since this function
463 // is only used when `Return::flatten_count() <=
464 // MAX_FLAT_RESULTS` and `MAX_FLAT_RESULTS == 1`, `N`
465 // can only either be 0 or 1.
466 //
467 // See `ComponentInstance::exit_call` for where we use
468 // the result count passed from
469 // `wasmtime_environ::fact::trampoline`-generated code
470 // to ensure the slice has the correct length, and also
471 // `concurrent::start_call` for where we conservatively
472 // use a slice length of 1 unconditionally. Also note
473 // that, as of this writing `slice_to_storage`
474 // double-checks the slice length is sufficient.
475 let results: &Return::Lower = unsafe { slice_to_storage(results) };
476 Self::lift_stack_result(cx, ty, results)
477 })?
478 } else {
479 Func::with_lift_context(instance, store, options, ty, |cx, ty| {
480 Self::lift_heap_result(cx, ty, &results[0])
481 })?
482 };
483 Ok(Box::new(result))
484 },
485 )
486 }
487}
488
489impl<T, P, R> TypedFuncCallConcurrent<T, P, R> {
490 /// Returns the task that this invocation corresponds to.
491 ///
492 /// This can be later correlated with [`StoreContextMut::async_call_stack`]
493 /// for example.
494 pub fn task(&self) -> GuestTaskId {
495 self.call.task()
496 }
497}