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