1 #![deny(unsafe_op_in_unsafe_fn)]
2 
3 //! Runtime support for the Component Model Async ABI.
4 //!
5 //! This module and its submodules provide host runtime support for Component
6 //! Model Async features such as async-lifted exports, async-lowered imports,
7 //! streams, futures, and related intrinsics.  See [the Async
8 //! Explainer](https://github.com/WebAssembly/component-model/blob/main/design/mvp/Async.md)
9 //! for a high-level overview.
10 //!
11 //! At the core of this support is an event loop which schedules and switches
12 //! between guest tasks and any host tasks they create.  Each
13 //! `ComponentInstance` will have at most one event loop running at any given
14 //! time, and that loop may be suspended and resumed by the host embedder using
15 //! e.g. `Instance::run_concurrent`.  The `ComponentInstance::poll_until`
16 //! function contains the loop itself, while the
17 //! `ComponentInstance::concurrent_state` field holds its state.
18 //!
19 //! # Public API Overview
20 //!
21 //! ## Top-level API (e.g. kicking off host->guest calls and driving the event loop)
22 //!
23 //! - `[Typed]Func::call_concurrent`: Start a host->guest call to an
24 //! async-lifted or sync-lifted import, creating a guest task.
25 //!
26 //! - `Instance::run_concurrent`: Run the event loop for the specified instance,
27 //! allowing any and all tasks belonging to that instance to make progress.
28 //!
29 //! - `Instance::spawn`: Run a background task as part of the event loop for the
30 //! specified instance.
31 //!
32 //! - `Instance::{future,stream}`: Create a new Component Model `future` or
33 //! `stream`; the read end may be passed to the guest.
34 //!
35 //! - `{Future,Stream}Reader::read` and `{Future,Stream}Writer::write`: read
36 //! from or write to a future or stream, respectively.
37 //!
38 //! ## Host Task API (e.g. implementing concurrent host functions and background tasks)
39 //!
40 //! - `LinkerInstance::func_wrap_concurrent`: Register a concurrent host
41 //! function with the linker.  That function will take an `Accessor` as its
42 //! first parameter, which provides access to the store and instance between
43 //! (but not across) await points.
44 //!
45 //! - `Accessor::with`: Access the store, its associated data, and the current
46 //! instance.
47 //!
48 //! - `Accessor::spawn`: Run a background task as part of the event loop for the
49 //! specified instance.  This is equivalent to `Instance::spawn` but more
50 //! convenient to use in host functions.
51 
52 use crate::component::func::{self, Func, Options};
53 use crate::component::{Component, ComponentInstanceId, HasData, HasSelf, Instance};
54 use crate::fiber::{self, StoreFiber, StoreFiberYield};
55 use crate::store::{StoreInner, StoreOpaque, StoreToken};
56 use crate::vm::component::{CallContext, InstanceFlags, ResourceTables};
57 use crate::vm::{SendSyncPtr, VMFuncRef, VMMemoryDefinition, VMStore};
58 use crate::{AsContext, AsContextMut, StoreContext, StoreContextMut, ValRaw};
59 use anyhow::{Context as _, Result, anyhow, bail};
60 use error_contexts::{GlobalErrorContextRefCount, LocalErrorContextRefCount};
61 use futures::channel::oneshot;
62 use futures::future::{self, Either, FutureExt};
63 use futures::stream::{FuturesUnordered, StreamExt};
64 use futures_and_streams::{FlatAbi, ReturnCode, StreamFutureState, TableIndex, TransmitHandle};
65 use states::StateTable;
66 use std::any::Any;
67 use std::borrow::ToOwned;
68 use std::boxed::Box;
69 use std::cell::UnsafeCell;
70 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
71 use std::fmt;
72 use std::future::Future;
73 use std::marker::PhantomData;
74 use std::mem::{self, MaybeUninit};
75 use std::pin::{Pin, pin};
76 use std::ptr::{self, NonNull};
77 use std::slice;
78 use std::sync::Mutex;
79 use std::task::{Context, Poll, Waker};
80 use std::vec::Vec;
81 use table::{Table, TableDebug, TableError, TableId};
82 use wasmtime_environ::PrimaryMap;
83 use wasmtime_environ::component::{
84     ExportIndex, MAX_FLAT_PARAMS, MAX_FLAT_RESULTS, PREPARE_ASYNC_NO_RESULT,
85     PREPARE_ASYNC_WITH_RESULT, RuntimeComponentInstanceIndex, StringEncoding,
86     TypeComponentGlobalErrorContextTableIndex, TypeComponentLocalErrorContextTableIndex,
87     TypeFutureTableIndex, TypeStreamTableIndex, TypeTupleIndex,
88 };
89 
90 pub use abort::AbortHandle;
91 pub use futures_and_streams::{
92     ErrorContext, FutureReader, FutureWriter, HostFuture, HostStream, ReadBuffer, StreamReader,
93     StreamWriter, VecBuffer, Watch, WriteBuffer,
94 };
95 pub(crate) use futures_and_streams::{
96     ResourcePair, lower_error_context_to_index, lower_future_to_index, lower_stream_to_index,
97 };
98 
99 mod abort;
100 mod error_contexts;
101 mod futures_and_streams;
102 mod states;
103 mod table;
104 mod tls;
105 
106 /// Constant defined in the Component Model spec to indicate that the async
107 /// intrinsic (e.g. `future.write`) has not yet completed.
108 const BLOCKED: u32 = 0xffff_ffff;
109 
110 /// Corresponds to `CallState` in the upstream spec.
111 #[derive(Clone, Copy, Eq, PartialEq, Debug)]
112 pub enum Status {
113     Starting = 0,
114     Started = 1,
115     Returned = 2,
116     StartCancelled = 3,
117     ReturnCancelled = 4,
118 }
119 
120 impl Status {
121     /// Packs this status and the optional `waitable` provided into a 32-bit
122     /// result that the canonical ABI requires.
123     ///
124     /// The low 4 bits are reserved for the status while the upper 28 bits are
125     /// the waitable, if present.
126     pub fn pack(self, waitable: Option<u32>) -> u32 {
127         assert!(matches!(self, Status::Returned) == waitable.is_none());
128         let waitable = waitable.unwrap_or(0);
129         assert!(waitable < (1 << 28));
130         (waitable << 4) | (self as u32)
131     }
132 }
133 
134 /// Corresponds to `EventCode` in the Component Model spec, plus related payload
135 /// data.
136 #[derive(Clone, Copy, Debug)]
137 enum Event {
138     None,
139     Cancelled,
140     Subtask {
141         status: Status,
142     },
143     StreamRead {
144         code: ReturnCode,
145         pending: Option<(TypeStreamTableIndex, u32)>,
146     },
147     StreamWrite {
148         code: ReturnCode,
149         pending: Option<(TypeStreamTableIndex, u32)>,
150     },
151     FutureRead {
152         code: ReturnCode,
153         pending: Option<(TypeFutureTableIndex, u32)>,
154     },
155     FutureWrite {
156         code: ReturnCode,
157         pending: Option<(TypeFutureTableIndex, u32)>,
158     },
159 }
160 
161 impl Event {
162     /// Lower this event to core Wasm integers for delivery to the guest.
163     ///
164     /// Note that the waitable handle, if any, is assumed to be lowered
165     /// separately.
166     fn parts(self) -> (u32, u32) {
167         const EVENT_NONE: u32 = 0;
168         const EVENT_SUBTASK: u32 = 1;
169         const EVENT_STREAM_READ: u32 = 2;
170         const EVENT_STREAM_WRITE: u32 = 3;
171         const EVENT_FUTURE_READ: u32 = 4;
172         const EVENT_FUTURE_WRITE: u32 = 5;
173         const EVENT_CANCELLED: u32 = 6;
174         match self {
175             Event::None => (EVENT_NONE, 0),
176             Event::Cancelled => (EVENT_CANCELLED, 0),
177             Event::Subtask { status } => (EVENT_SUBTASK, status as u32),
178             Event::StreamRead { code, .. } => (EVENT_STREAM_READ, code.encode()),
179             Event::StreamWrite { code, .. } => (EVENT_STREAM_WRITE, code.encode()),
180             Event::FutureRead { code, .. } => (EVENT_FUTURE_READ, code.encode()),
181             Event::FutureWrite { code, .. } => (EVENT_FUTURE_WRITE, code.encode()),
182         }
183     }
184 }
185 
186 /// Corresponds to `CallbackCode` in the spec.
187 mod callback_code {
188     pub const EXIT: u32 = 0;
189     pub const YIELD: u32 = 1;
190     pub const WAIT: u32 = 2;
191     pub const POLL: u32 = 3;
192 }
193 
194 /// A flag indicating that the callee is an async-lowered export.
195 ///
196 /// This may be passed to the `async-start` intrinsic from a fused adapter.
197 const START_FLAG_ASYNC_CALLEE: u32 = wasmtime_environ::component::START_FLAG_ASYNC_CALLEE as u32;
198 
199 /// Provides access to either store data (via the `get` method) or the store
200 /// itself (via [`AsContext`]/[`AsContextMut`]), as well as the component
201 /// instance to which the current host task belongs.
202 ///
203 /// See [`Accessor::with`] for details.
204 pub struct Access<'a, T: 'static, D: HasData = HasSelf<T>> {
205     accessor: &'a Accessor<T, D>,
206     store: StoreContextMut<'a, T>,
207 }
208 
209 impl<'a, T, D> Access<'a, T, D>
210 where
211     D: HasData,
212     T: 'static,
213 {
214     /// Get mutable access to the store data.
215     pub fn data_mut(&mut self) -> &mut T {
216         self.store.data_mut()
217     }
218 
219     /// Get mutable access to the store data.
220     pub fn get(&mut self) -> D::Data<'_> {
221         let get_data = self.accessor.get_data;
222         get_data(self.data_mut())
223     }
224 
225     /// Spawn a background task.
226     ///
227     /// See [`Accessor::spawn`] for details.
228     pub fn spawn(&mut self, task: impl AccessorTask<T, D, Result<()>>) -> AbortHandle
229     where
230         T: 'static,
231     {
232         self.accessor.instance.spawn_with_accessor(
233             self.store.as_context_mut(),
234             self.accessor.clone_for_spawn(),
235             task,
236         )
237     }
238 
239     /// Retrieve the component instance of the caller.
240     pub fn instance(&self) -> Instance {
241         self.accessor.instance()
242     }
243 }
244 
245 impl<'a, T, D> AsContext for Access<'a, T, D>
246 where
247     D: HasData,
248     T: 'static,
249 {
250     type Data = T;
251 
252     fn as_context(&self) -> StoreContext<'_, T> {
253         self.store.as_context()
254     }
255 }
256 
257 impl<'a, T, D> AsContextMut for Access<'a, T, D>
258 where
259     D: HasData,
260     T: 'static,
261 {
262     fn as_context_mut(&mut self) -> StoreContextMut<'_, T> {
263         self.store.as_context_mut()
264     }
265 }
266 
267 /// Provides scoped mutable access to store data in the context of a concurrent
268 /// host task future.
269 ///
270 /// This allows multiple host task futures to execute concurrently and access
271 /// the store between (but not across) `await` points.
272 ///
273 /// # Rationale
274 ///
275 /// This structure is sort of like `&mut T` plus a projection from `&mut T` to
276 /// `D::Data<'_>`. The problem this is solving, however, is that it does not
277 /// literally store these values. The basic problem is that when a concurrent
278 /// host future is being polled it has access to `&mut T` (and the whole
279 /// `Store`) but when it's not being polled it does not have access to these
280 /// values. This reflects how the store is only ever polling one future at a
281 /// time so the store is effectively being passed between futures.
282 ///
283 /// Rust's `Future` trait, however, has no means of passing a `Store`
284 /// temporarily between futures. The [`Context`](std::task::Context) type does
285 /// not have the ability to attach arbitrary information to it at this time.
286 /// This type, [`Accessor`], is used to bridge this expressivity gap.
287 ///
288 /// The [`Accessor`] type here represents the ability to acquire, temporarily in
289 /// a synchronous manner, the current store. The [`Accessor::with`] function
290 /// yields an [`Access`] which can be used to access [`StoreContextMut`], `&mut
291 /// T`, or `D::Data<'_>`. Note though that [`Accessor::with`] intentionally does
292 /// not take an `async` closure as its argument, instead it's a synchronous
293 /// closure which must complete during on run of `Future::poll`. This reflects
294 /// how the store is temporarily made available while a host future is being
295 /// polled.
296 ///
297 /// # Implementation
298 ///
299 /// This type does not actually store `&mut T` nor `StoreContextMut<T>`, and
300 /// this type additionally doesn't even have a lifetime parameter. This is
301 /// instead a representation of proof of the ability to acquire these while a
302 /// future is being polled. Wasmtime will, when it polls a host future,
303 /// configure ambient state such that the `Accessor` that a future closes over
304 /// will work and be able to access the store.
305 ///
306 /// This has a number of implications for users such as:
307 ///
308 /// * It's intentional that `Accessor` cannot be cloned, it needs to stay within
309 ///   the lifetime of a single future.
310 /// * A futures is expected to, however, close over an `Accessor` and keep it
311 ///   alive probably for the duration of the entire future.
312 /// * Different host futures will be given different `Accessor`s, and that's
313 ///   intentional.
314 /// * The `Accessor` type is `Send` and `Sync` irrespective of `T` which
315 ///   alleviates some otherwise required bounds to be written down.
316 ///
317 /// # Using `Accessor` in `Drop`
318 ///
319 /// The methods on `Accessor` are only expected to work in the context of
320 /// `Future::poll` and are not guaranteed to work in `Drop`. This is because a
321 /// host future can be dropped at any time throughout the system and Wasmtime
322 /// store context is not necessarily available at that time. It's recommended to
323 /// not use `Accessor` methods in anything connected to a `Drop` implementation
324 /// as they will panic and have unintended results. If you run into this though
325 /// feel free to file an issue on the Wasmtime repository.
326 pub struct Accessor<T: 'static, D = HasSelf<T>>
327 where
328     D: HasData,
329 {
330     token: StoreToken<T>,
331     get_data: fn(&mut T) -> D::Data<'_>,
332     instance: Instance,
333 }
334 
335 /// A helper trait to take any type of accessor-with-data in functions.
336 ///
337 /// This trait is similar to [`AsContextMut`] except that it's used when
338 /// working with an [`Accessor`] instead of a [`StoreContextMut`]. The
339 /// [`Accessor`] is the main type used in concurrent settings and is passed to
340 /// functions such as [`Func::call_concurrent`] or [`FutureWriter::write`].
341 ///
342 /// This trait is implemented for [`Accessor`] and `&T` where `T` implements
343 /// this trait. This effectively means that regardless of the `D` in
344 /// `Accessor<T, D>` it can still be passed to a function which just needs a
345 /// store accessor.
346 ///
347 /// Acquiring an [`Accessor`] can be done through [`Instance::run_concurrent`]
348 /// for example or in a host function through
349 /// [`Linker::func_wrap_concurrent`](crate::component::Linker::func_wrap_concurrent).
350 pub trait AsAccessor {
351     /// The `T` in `Store<T>` that this accessor refers to.
352     type Data: 'static;
353 
354     /// The `D` in `Accessor<T, D>`, or the projection out of
355     /// `Self::Data`.
356     type AccessorData: HasData;
357 
358     /// Returns the accessor that this is referring to.
359     fn as_accessor(&self) -> &Accessor<Self::Data, Self::AccessorData>;
360 }
361 
362 impl<T: AsAccessor + ?Sized> AsAccessor for &T {
363     type Data = T::Data;
364     type AccessorData = T::AccessorData;
365 
366     fn as_accessor(&self) -> &Accessor<Self::Data, Self::AccessorData> {
367         T::as_accessor(self)
368     }
369 }
370 
371 impl<T, D: HasData> AsAccessor for Accessor<T, D> {
372     type Data = T;
373     type AccessorData = D;
374 
375     fn as_accessor(&self) -> &Accessor<T, D> {
376         self
377     }
378 }
379 
380 // Note that it is intentional at this time that `Accessor` does not actually
381 // store `&mut T` or anything similar. This distinctly enables the `Accessor`
382 // structure to be both `Send` and `Sync` regardless of what `T` is (or `D` for
383 // that matter). This is used to ergonomically simplify bindings where the
384 // majority of the time `Accessor` is closed over in a future which then needs
385 // to be `Send` and `Sync`. To avoid needing to write `T: Send` everywhere (as
386 // you already have to write `T: 'static`...) it helps to avoid this.
387 //
388 // Note as well that `Accessor` doesn't actually store its data at all. Instead
389 // it's more of a "proof" of what can be accessed from TLS. API design around
390 // `Accessor` and functions like `Linker::func_wrap_concurrent` are
391 // intentionally made to ensure that `Accessor` is ideally only used in the
392 // context that TLS variables are actually set. For example host functions are
393 // given `&Accessor`, not `Accessor`, and this prevents them from persisting
394 // the value outside of a future. Within the future the TLS variables are all
395 // guaranteed to be set while the future is being polled.
396 //
397 // Finally though this is not an ironclad guarantee, but nor does it need to be.
398 // The TLS APIs are designed to panic or otherwise model usage where they're
399 // called recursively or similar. It's hoped that code cannot be constructed to
400 // actually hit this at runtime but this is not a safety requirement at this
401 // time.
402 const _: () = {
403     const fn assert<T: Send + Sync>() {}
404     assert::<Accessor<UnsafeCell<u32>>>();
405 };
406 
407 impl<T> Accessor<T> {
408     /// Creates a new `Accessor` backed by the specified functions.
409     ///
410     /// - `get`: used to retrieve the store
411     ///
412     /// - `get_data`: used to "project" from the store's associated data to
413     /// another type (e.g. a field of that data or a wrapper around it).
414     ///
415     /// - `spawn`: used to queue spawned background tasks to be run later
416     ///
417     /// - `instance`: used to access the `Instance` to which this `Accessor`
418     /// (and the future which closes over it) belongs
419     fn new(token: StoreToken<T>, instance: Instance) -> Self {
420         Self {
421             token,
422             get_data: |x| x,
423             instance,
424         }
425     }
426 }
427 
428 impl<T, D> Accessor<T, D>
429 where
430     D: HasData,
431 {
432     /// Run the specified closure, passing it mutable access to the store.
433     ///
434     /// This function is one of the main building blocks of the [`Accessor`]
435     /// type. This yields synchronous, blocking, access to store via an
436     /// [`Access`]. The [`Access`] implements [`AsContextMut`] in addition to
437     /// providing the ability to access `D` via [`Access::get`]. Note that the
438     /// `fun` here is given only temporary access to the store and `T`/`D`
439     /// meaning that the return value `R` here is not allowed to capture borrows
440     /// into the two. If access is needed to data within `T` or `D` outside of
441     /// this closure then it must be `clone`d out, for example.
442     ///
443     /// # Panics
444     ///
445     /// This function will panic if it is call recursively with any other
446     /// accessor already in scope. For example if `with` is called within `fun`,
447     /// then this function will panic. It is up to the embedder to ensure that
448     /// this does not happen.
449     pub fn with<R>(&self, fun: impl FnOnce(Access<'_, T, D>) -> R) -> R {
450         tls::get(|vmstore| {
451             fun(Access {
452                 store: self.token.as_context_mut(vmstore),
453                 accessor: self,
454             })
455         })
456     }
457 
458     /// Changes this accessor to access `D2` instead of the current type
459     /// parameter `D`.
460     ///
461     /// This changes the underlying data access from `T` to `D2::Data<'_>`.
462     ///
463     /// Note that this is not a public or recommended API because it's easy to
464     /// cause panics with this by having two `Accessor` values live at the same
465     /// time. The returned `Accessor` does not refer to this `Accessor` meaning
466     /// that both can be used. You could, for example, call `Accessor::with`
467     /// simultaneously on both. That would cause a panic though.
468     ///
469     /// In short while there's nothing unsafe about this it's a footgun. It's
470     /// here for bindings generation where the provided accessor is transformed
471     /// into a new accessor and then this returned accessor is passed to
472     /// implementations.
473     ///
474     /// Note that one possible fix for this would be a lifetime parameter on
475     /// `Accessor` itself so the returned value could borrow from the original
476     /// value (or this could be `self`-by-value instead of `&mut self`) but in
477     /// attempting that it was found to be a bit too onerous in terms of
478     /// plumbing things around without a whole lot of benefit.
479     ///
480     /// In short, this works, but must be treated with care. The current main
481     /// user, bindings generation, treats this with care.
482     #[doc(hidden)]
483     pub fn with_data<D2: HasData>(&self, get_data: fn(&mut T) -> D2::Data<'_>) -> Accessor<T, D2> {
484         Accessor {
485             token: self.token,
486             get_data,
487             instance: self.instance,
488         }
489     }
490 
491     /// Spawn a background task which will receive an `&Accessor<T, D>` and
492     /// run concurrently with any other tasks in progress for the current
493     /// instance.
494     ///
495     /// This is particularly useful for host functions which return a `stream`
496     /// or `future` such that the code to write to the write end of that
497     /// `stream` or `future` must run after the function returns.
498     ///
499     /// The returned [`AbortHandle`] may be used to cancel the task.
500     ///
501     /// # Panics
502     ///
503     /// Panics if called within a closure provided to the [`Accessor::with`]
504     /// function. This can only be called outside an active invocation of
505     /// [`Accessor::with`].
506     pub fn spawn(&self, task: impl AccessorTask<T, D, Result<()>>) -> AbortHandle
507     where
508         T: 'static,
509     {
510         let instance = self.instance;
511         let accessor = self.clone_for_spawn();
512         self.with(|mut access| {
513             instance.spawn_with_accessor(access.as_context_mut(), accessor, task)
514         })
515     }
516 
517     /// Retrieve the component instance of the caller.
518     pub fn instance(&self) -> Instance {
519         self.instance
520     }
521 
522     fn clone_for_spawn(&self) -> Self {
523         Self {
524             token: self.token,
525             get_data: self.get_data,
526             instance: self.instance,
527         }
528     }
529 }
530 
531 /// Represents a task which may be provided to `Accessor::spawn`,
532 /// `Accessor::forward`, or `Instance::spawn`.
533 // TODO: Replace this with `std::ops::AsyncFnOnce` when that becomes a viable
534 // option.
535 //
536 // `AsyncFnOnce` is still nightly-only in latest stable Rust version as of this
537 // writing (1.84.1), and even with 1.85.0-beta it's not possible to specify
538 // e.g. `Send` and `Sync` bounds on the `Future` type returned by an
539 // `AsyncFnOnce`.  Also, using `F: Future<Output = Result<()>> + Send + Sync,
540 // FN: FnOnce(&Accessor<T>) -> F + Send + Sync + 'static` fails with a type
541 // mismatch error when we try to pass it an async closure (e.g. `async move |_|
542 // { ... }`).  So this seems to be the best we can do for the time being.
543 pub trait AccessorTask<T, D, R>: Send + 'static
544 where
545     D: HasData,
546 {
547     /// Run the task.
548     fn run(self, accessor: &Accessor<T, D>) -> impl Future<Output = R> + Send;
549 }
550 
551 /// Represents the state of a waitable handle.
552 #[derive(Debug)]
553 enum WaitableState {
554     /// Represents a host task handle.
555     HostTask,
556     /// Represents a guest task handle.
557     GuestTask,
558     /// Represents a stream handle.
559     Stream(TypeStreamTableIndex, StreamFutureState),
560     /// Represents a future handle.
561     Future(TypeFutureTableIndex, StreamFutureState),
562     /// Represents a waitable-set handle.
563     Set,
564 }
565 
566 /// Represents parameter and result metadata for the caller side of a
567 /// guest->guest call orchestrated by a fused adapter.
568 enum CallerInfo {
569     /// Metadata for a call to an async-lowered import
570     Async {
571         params: Vec<ValRaw>,
572         has_result: bool,
573     },
574     /// Metadata for a call to an sync-lowered import
575     Sync {
576         params: Vec<ValRaw>,
577         result_count: u32,
578     },
579 }
580 
581 /// Indicates how a guest task is waiting on a waitable set.
582 enum WaitMode {
583     /// The guest task is waiting using `task.wait`
584     Fiber(StoreFiber<'static>),
585     /// The guest task is waiting via a callback declared as part of an
586     /// async-lifted export.
587     Callback(RuntimeComponentInstanceIndex),
588 }
589 
590 /// Represents the reason a fiber is suspending itself.
591 #[derive(Debug)]
592 enum SuspendReason {
593     /// The fiber is waiting for an event to be delivered to the specified
594     /// waitable set or task.
595     Waiting {
596         set: TableId<WaitableSet>,
597         task: TableId<GuestTask>,
598     },
599     /// The fiber has finished handling its most recent work item and is waiting
600     /// for another (or to be dropped if it is no longer needed).
601     NeedWork,
602     /// The fiber is yielding and should be resumed once other tasks have had a
603     /// chance to run.
604     Yielding { task: TableId<GuestTask> },
605 }
606 
607 /// Represents a pending call into guest code for a given guest task.
608 enum GuestCallKind {
609     /// Indicates there's an event to deliver to the task, possibly related to a
610     /// waitable set the task has been waiting on or polling.
611     DeliverEvent {
612         /// The (sub-)component instance in which the task has most recently
613         /// been executing.
614         ///
615         /// Note that this might not be the same as the instance the guest task
616         /// started executing in given that one or more synchronous guest->guest
617         /// calls may have occurred involving multiple instances.
618         instance: RuntimeComponentInstanceIndex,
619         /// The waitable set the event belongs to, if any.
620         ///
621         /// If this is `None` the event will be waiting in the
622         /// `GuestTask::event` field for the task.
623         set: Option<TableId<WaitableSet>>,
624     },
625     /// Indicates that a new guest task call is pending and may be executed
626     /// using the specified closure.
627     Start(Box<dyn FnOnce(&mut dyn VMStore, Instance) -> Result<()> + Send + Sync>),
628 }
629 
630 impl fmt::Debug for GuestCallKind {
631     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
632         match self {
633             Self::DeliverEvent { instance, set } => f
634                 .debug_struct("DeliverEvent")
635                 .field("instance", instance)
636                 .field("set", set)
637                 .finish(),
638             Self::Start(_) => f.debug_tuple("Start").finish(),
639         }
640     }
641 }
642 
643 /// Represents a pending call into guest code for a given guest task.
644 #[derive(Debug)]
645 struct GuestCall {
646     task: TableId<GuestTask>,
647     kind: GuestCallKind,
648 }
649 
650 impl GuestCall {
651     /// Returns whether or not the call is ready to run.
652     ///
653     /// A call will not be ready to run if either:
654     ///
655     /// - the (sub-)component instance to be called has already been entered and
656     /// cannot be reentered until an in-progress call completes
657     ///
658     /// - the call is for a not-yet started task and the (sub-)component
659     /// instance to be called has backpressure enabled
660     fn is_ready(&self, state: &mut ConcurrentState) -> Result<bool> {
661         let task_instance = state.get(self.task)?.instance;
662         let state = state.instance_state(task_instance);
663         let ready = match &self.kind {
664             GuestCallKind::DeliverEvent { .. } => !state.do_not_enter,
665             GuestCallKind::Start(_) => !(state.do_not_enter || state.backpressure),
666         };
667         log::trace!(
668             "call {self:?} ready? {ready} (do_not_enter: {}; backpressure: {})",
669             state.do_not_enter,
670             state.backpressure
671         );
672         Ok(ready)
673     }
674 }
675 
676 /// Represents state related to an in-progress poll operation (e.g. `task.poll`
677 /// or `CallbackCode.POLL`).
678 #[derive(Debug)]
679 struct PollParams {
680     /// Identifies the polling task.
681     task: TableId<GuestTask>,
682     /// The waitable set being polled.
683     set: TableId<WaitableSet>,
684     /// The (sub-)component instance in which the task has most recently been
685     /// executing.
686     ///
687     /// Note that this might not be the same as the instance the guest task
688     /// started executing in given that one or more synchronous guest->guest
689     /// calls may have occurred involving multiple instances.
690     instance: RuntimeComponentInstanceIndex,
691 }
692 
693 /// Represents a pending work item to be handled by the event loop for a given
694 /// component instance.
695 enum WorkItem {
696     /// A host task to be pushed to `ConcurrentState::futures`.
697     PushFuture(Mutex<HostTaskFuture>),
698     /// A fiber to resume.
699     ResumeFiber(StoreFiber<'static>),
700     /// A pending call into guest code for a given guest task.
701     GuestCall(GuestCall),
702     /// A pending `task.poll` or `CallbackCode.POLL` operation.
703     Poll(PollParams),
704 }
705 
706 impl fmt::Debug for WorkItem {
707     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
708         match self {
709             Self::PushFuture(_) => f.debug_tuple("PushFuture").finish(),
710             Self::ResumeFiber(_) => f.debug_tuple("ResumeFiber").finish(),
711             Self::GuestCall(call) => f.debug_tuple("GuestCall").field(call).finish(),
712             Self::Poll(params) => f.debug_tuple("Poll").field(params).finish(),
713         }
714     }
715 }
716 
717 impl ConcurrentState {
718     fn instance_state(&mut self, instance: RuntimeComponentInstanceIndex) -> &mut InstanceState {
719         self.instance_states.entry(instance).or_default()
720     }
721 
722     fn push<V: Send + Sync + 'static>(&mut self, value: V) -> Result<TableId<V>, TableError> {
723         self.table.push(value)
724     }
725 
726     fn get<V: 'static>(&self, id: TableId<V>) -> Result<&V, TableError> {
727         self.table.get(id)
728     }
729 
730     fn get_mut<V: 'static>(&mut self, id: TableId<V>) -> Result<&mut V, TableError> {
731         self.table.get_mut(id)
732     }
733 
734     pub fn add_child<T, U>(
735         &mut self,
736         child: TableId<T>,
737         parent: TableId<U>,
738     ) -> Result<(), TableError> {
739         self.table.add_child(child, parent)
740     }
741 
742     pub fn remove_child<T, U>(
743         &mut self,
744         child: TableId<T>,
745         parent: TableId<U>,
746     ) -> Result<(), TableError> {
747         self.table.remove_child(child, parent)
748     }
749 
750     fn delete<V: 'static>(&mut self, id: TableId<V>) -> Result<V, TableError> {
751         self.table.delete(id)
752     }
753 
754     fn push_future(&mut self, future: HostTaskFuture) {
755         // Note that we can't directly push to `ConcurrentState::futures` here
756         // since this may be called from a future that's being polled inside
757         // `Self::poll_until`, which temporarily removes the `FuturesUnordered`
758         // so it has exclusive access while polling it.  Therefore, we push a
759         // work item to the "high priority" queue, which will actually push to
760         // `ConcurrentState::futures` later.
761         self.push_high_priority(WorkItem::PushFuture(Mutex::new(future)));
762     }
763 
764     fn push_high_priority(&mut self, item: WorkItem) {
765         log::trace!("push high priority: {item:?}");
766         self.high_priority.push(item);
767     }
768 
769     fn push_low_priority(&mut self, item: WorkItem) {
770         log::trace!("push low priority: {item:?}");
771         self.low_priority.push(item);
772     }
773 
774     /// Determine whether the instance associated with the specified guest task
775     /// may be entered (i.e. is not already on the async call stack).
776     ///
777     /// This is an additional check on top of the "may_enter" instance flag;
778     /// it's needed because async-lifted exports with callback functions must
779     /// not call their own instances directly or indirectly, and due to the
780     /// "stackless" nature of callback-enabled guest tasks this may happen even
781     /// if there are no activation records on the stack (i.e. the "may_enter"
782     /// field is `true`) for that instance.
783     fn may_enter(&mut self, mut guest_task: TableId<GuestTask>) -> bool {
784         let guest_instance = self.get(guest_task).unwrap().instance;
785 
786         // Walk the task tree back to the root, looking for potential
787         // reentrance.
788         //
789         // TODO: This could be optimized by maintaining a per-`GuestTask` bitset
790         // such that each bit represents and instance which has been entered by
791         // that task or an ancestor of that task, in which case this would be a
792         // constant time check.
793         loop {
794             match &self.get_mut(guest_task).unwrap().caller {
795                 Caller::Host { .. } => break true,
796                 Caller::Guest { task, instance } => {
797                     if *instance == guest_instance {
798                         break false;
799                     } else {
800                         guest_task = *task;
801                     }
802                 }
803             }
804         }
805     }
806 
807     /// Handle the `CallbackCode` returned from an async-lifted export or its
808     /// callback.
809     ///
810     /// If `initial_call` is `true`, then the code was received from the
811     /// async-lifted export; otherwise, it was received from its callback.
812     fn handle_callback_code(
813         &mut self,
814         guest_task: TableId<GuestTask>,
815         runtime_instance: RuntimeComponentInstanceIndex,
816         code: u32,
817         initial_call: bool,
818     ) -> Result<()> {
819         let (code, set) = unpack_callback_code(code);
820 
821         log::trace!("received callback code from {guest_task:?}: {code} (set: {set})");
822 
823         let task = self.get_mut(guest_task)?;
824 
825         if task.lift_result.is_some() {
826             if code == callback_code::EXIT {
827                 return Err(anyhow!(crate::Trap::NoAsyncResult));
828             }
829             if initial_call {
830                 // Notify any current or future waiters that this subtask has
831                 // started.
832                 Waitable::Guest(guest_task).set_event(
833                     self,
834                     Some(Event::Subtask {
835                         status: Status::Started,
836                     }),
837                 )?;
838             }
839         }
840 
841         let get_set = |instance: &mut Self, handle| {
842             if handle == 0 {
843                 bail!("invalid waitable-set handle");
844             }
845 
846             let (set, WaitableState::Set) =
847                 instance.waitable_tables[runtime_instance].get_mut_by_index(handle)?
848             else {
849                 bail!("invalid waitable-set handle");
850             };
851 
852             Ok(TableId::<WaitableSet>::new(set))
853         };
854 
855         match code {
856             callback_code::EXIT => {
857                 let task = self.get_mut(guest_task)?;
858                 match &task.caller {
859                     Caller::Host {
860                         remove_task_automatically,
861                         ..
862                     } => {
863                         if *remove_task_automatically {
864                             log::trace!("handle_callback_code will delete task {guest_task:?}");
865                             Waitable::Guest(guest_task).delete_from(self)?;
866                         }
867                     }
868                     Caller::Guest { .. } => {
869                         task.exited = true;
870                         task.callback = None;
871                     }
872                 }
873             }
874             callback_code::YIELD => {
875                 // Push this task onto the "low priority" queue so it runs after
876                 // any other tasks have had a chance to run.
877                 let task = self.get_mut(guest_task)?;
878                 assert!(task.event.is_none());
879                 task.event = Some(Event::None);
880                 self.push_low_priority(WorkItem::GuestCall(GuestCall {
881                     task: guest_task,
882                     kind: GuestCallKind::DeliverEvent {
883                         instance: runtime_instance,
884                         set: None,
885                     },
886                 }));
887             }
888             callback_code::WAIT | callback_code::POLL => {
889                 let set = get_set(self, set)?;
890 
891                 if self.get_mut(guest_task)?.event.is_some() || !self.get_mut(set)?.ready.is_empty()
892                 {
893                     // An event is immediately available; deliver it ASAP.
894                     self.push_high_priority(WorkItem::GuestCall(GuestCall {
895                         task: guest_task,
896                         kind: GuestCallKind::DeliverEvent {
897                             instance: runtime_instance,
898                             set: Some(set),
899                         },
900                     }));
901                 } else {
902                     // No event is immediately available.
903                     match code {
904                         callback_code::POLL => {
905                             // We're polling, so just yield and check whether an
906                             // event has arrived after that.
907                             self.push_low_priority(WorkItem::Poll(PollParams {
908                                 task: guest_task,
909                                 instance: runtime_instance,
910                                 set,
911                             }));
912                         }
913                         callback_code::WAIT => {
914                             // We're waiting, so register to be woken up when an
915                             // event is published for this waitable set.
916                             //
917                             // Here we also set `GuestTask::wake_on_cancel`
918                             // which allows `subtask.cancel` to interrupt the
919                             // wait.
920                             let old = self.get_mut(guest_task)?.wake_on_cancel.replace(set);
921                             assert!(old.is_none());
922                             let old = self
923                                 .get_mut(set)?
924                                 .waiting
925                                 .insert(guest_task, WaitMode::Callback(runtime_instance));
926                             assert!(old.is_none());
927                         }
928                         _ => unreachable!(),
929                     }
930                 }
931             }
932             _ => bail!("unsupported callback code: {code}"),
933         }
934 
935         Ok(())
936     }
937 
938     /// Record that we're about to enter a (sub-)component instance which does
939     /// not support more than one concurrent, stackful activation, meaning it
940     /// cannot be entered again until the next call returns.
941     fn enter_instance(&mut self, instance: RuntimeComponentInstanceIndex) {
942         self.instance_state(instance).do_not_enter = true;
943     }
944 
945     /// Record that we've exited a (sub-)component instance previously entered
946     /// with `Self::enter_instance` and then calls `Self::partition_pending`.
947     /// See the documentation for the latter for details.
948     fn exit_instance(&mut self, instance: RuntimeComponentInstanceIndex) -> Result<()> {
949         self.instance_state(instance).do_not_enter = false;
950         self.partition_pending(instance)
951     }
952 
953     /// Iterate over `InstanceState::pending`, moving any ready items into the
954     /// "high priority" work item queue.
955     ///
956     /// See `GuestCall::is_ready` for details.
957     fn partition_pending(&mut self, instance: RuntimeComponentInstanceIndex) -> Result<()> {
958         for (task, kind) in mem::take(&mut self.instance_state(instance).pending).into_iter() {
959             let call = GuestCall { task, kind };
960             if call.is_ready(self)? {
961                 self.push_high_priority(WorkItem::GuestCall(call));
962             } else {
963                 self.instance_state(instance)
964                     .pending
965                     .insert(call.task, call.kind);
966             }
967         }
968 
969         Ok(())
970     }
971 
972     /// Get the next pending event for the specified task and (optional)
973     /// waitable set, along with the waitable handle if applicable.
974     fn get_event(
975         &mut self,
976         guest_task: TableId<GuestTask>,
977         instance: RuntimeComponentInstanceIndex,
978         set: Option<TableId<WaitableSet>>,
979     ) -> Result<Option<(Event, Option<(Waitable, u32)>)>> {
980         Ok(
981             if let Some(event) = self.get_mut(guest_task)?.event.take() {
982                 log::trace!("deliver event {event:?} to {guest_task:?}");
983 
984                 Some((event, None))
985             } else if let Some((set, waitable)) = set
986                 .and_then(|set| {
987                     self.get_mut(set)
988                         .map(|v| v.ready.pop_first().map(|v| (set, v)))
989                         .transpose()
990                 })
991                 .transpose()?
992             {
993                 let event = waitable.common(self)?.event.take().unwrap();
994 
995                 log::trace!(
996                     "deliver event {event:?} to {guest_task:?} for {waitable:?}; set {set:?}"
997                 );
998 
999                 let entry = self.waitable_tables[instance].get_mut_by_rep(waitable.rep());
1000                 let Some((
1001                     handle,
1002                     WaitableState::HostTask
1003                     | WaitableState::GuestTask
1004                     | WaitableState::Stream(..)
1005                     | WaitableState::Future(..),
1006                 )) = entry
1007                 else {
1008                     bail!("handle not found for waitable rep {waitable:?} instance {instance:?}");
1009                 };
1010 
1011                 waitable.on_delivery(self, event);
1012 
1013                 Some((event, Some((waitable, handle))))
1014             } else {
1015                 None
1016             },
1017         )
1018     }
1019 
1020     /// Implements the `backpressure.set` intrinsic.
1021     pub(crate) fn backpressure_set(
1022         &mut self,
1023         caller_instance: RuntimeComponentInstanceIndex,
1024         enabled: u32,
1025     ) -> Result<()> {
1026         let state = self.instance_state(caller_instance);
1027         let old = state.backpressure;
1028         let new = enabled != 0;
1029         state.backpressure = new;
1030 
1031         if old && !new {
1032             // Backpressure was previously enabled and is now disabled; move any
1033             // newly-eligible guest calls to the "high priority" queue.
1034             self.partition_pending(caller_instance)?;
1035         }
1036 
1037         Ok(())
1038     }
1039 
1040     /// Implements the `waitable-set.new` intrinsic.
1041     pub(crate) fn waitable_set_new(
1042         &mut self,
1043         caller_instance: RuntimeComponentInstanceIndex,
1044     ) -> Result<u32> {
1045         let set = self.push(WaitableSet::default())?;
1046         let handle = self.waitable_tables[caller_instance].insert(set.rep(), WaitableState::Set)?;
1047         log::trace!("new waitable set {set:?} (handle {handle})");
1048         Ok(handle)
1049     }
1050 
1051     /// Implements the `waitable-set.drop` intrinsic.
1052     pub(crate) fn waitable_set_drop(
1053         &mut self,
1054         caller_instance: RuntimeComponentInstanceIndex,
1055         set: u32,
1056     ) -> Result<()> {
1057         let (rep, WaitableState::Set) =
1058             self.waitable_tables[caller_instance].remove_by_index(set)?
1059         else {
1060             bail!("invalid waitable-set handle");
1061         };
1062 
1063         log::trace!("drop waitable set {rep} (handle {set})");
1064 
1065         let set = self.delete(TableId::<WaitableSet>::new(rep))?;
1066 
1067         if !set.waiting.is_empty() {
1068             bail!("cannot drop waitable set with waiters");
1069         }
1070 
1071         Ok(())
1072     }
1073 
1074     /// Implements the `waitable.join` intrinsic.
1075     pub(crate) fn waitable_join(
1076         &mut self,
1077         caller_instance: RuntimeComponentInstanceIndex,
1078         waitable_handle: u32,
1079         set_handle: u32,
1080     ) -> Result<()> {
1081         let waitable = Waitable::from_instance(self, caller_instance, waitable_handle)?;
1082 
1083         let set = if set_handle == 0 {
1084             None
1085         } else {
1086             let (set, WaitableState::Set) =
1087                 self.waitable_tables[caller_instance].get_mut_by_index(set_handle)?
1088             else {
1089                 bail!("invalid waitable-set handle");
1090             };
1091 
1092             Some(TableId::<WaitableSet>::new(set))
1093         };
1094 
1095         log::trace!(
1096             "waitable {waitable:?} (handle {waitable_handle}) join set {set:?} (handle {set_handle})",
1097         );
1098 
1099         waitable.join(self, set)
1100     }
1101 
1102     /// Implements the `subtask.drop` intrinsic.
1103     pub(crate) fn subtask_drop(
1104         &mut self,
1105         caller_instance: RuntimeComponentInstanceIndex,
1106         task_id: u32,
1107     ) -> Result<()> {
1108         self.waitable_join(caller_instance, task_id, 0)?;
1109 
1110         let (rep, state) = self.waitable_tables[caller_instance].remove_by_index(task_id)?;
1111 
1112         let (waitable, expected_caller_instance, delete) = match state {
1113             WaitableState::HostTask => {
1114                 let id = TableId::<HostTask>::new(rep);
1115                 let task = self.get(id)?;
1116                 if task.abort_handle.is_some() {
1117                     bail!("cannot drop a subtask which has not yet resolved");
1118                 }
1119                 (Waitable::Host(id), task.caller_instance, true)
1120             }
1121             WaitableState::GuestTask => {
1122                 let id = TableId::<GuestTask>::new(rep);
1123                 let task = self.get(id)?;
1124                 if task.lift_result.is_some() {
1125                     bail!("cannot drop a subtask which has not yet resolved");
1126                 }
1127                 if let Caller::Guest { instance, .. } = &task.caller {
1128                     (Waitable::Guest(id), *instance, task.exited)
1129                 } else {
1130                     unreachable!()
1131                 }
1132             }
1133             _ => bail!("invalid task handle: {task_id}"),
1134         };
1135 
1136         if waitable.take_event(self)?.is_some() {
1137             bail!("cannot drop a subtask with an undelivered event");
1138         }
1139 
1140         if delete {
1141             waitable.delete_from(self)?;
1142         }
1143 
1144         // Since waitables can neither be passed between instances nor forged,
1145         // this should never fail unless there's a bug in Wasmtime, but we check
1146         // here to be sure:
1147         assert_eq!(expected_caller_instance, caller_instance);
1148         log::trace!("subtask_drop {waitable:?} (handle {task_id})");
1149         Ok(())
1150     }
1151 
1152     /// Implements the `context.get` intrinsic.
1153     pub(crate) fn context_get(&mut self, slot: u32) -> Result<u32> {
1154         let task = self.guest_task.unwrap();
1155         let val = self.get(task)?.context[usize::try_from(slot).unwrap()];
1156         log::trace!("context_get {task:?} slot {slot} val {val:#x}");
1157         Ok(val)
1158     }
1159 
1160     /// Implements the `context.set` intrinsic.
1161     pub(crate) fn context_set(&mut self, slot: u32, val: u32) -> Result<()> {
1162         let task = self.guest_task.unwrap();
1163         log::trace!("context_set {task:?} slot {slot} val {val:#x}");
1164         self.get_mut(task)?.context[usize::try_from(slot).unwrap()] = val;
1165         Ok(())
1166     }
1167 }
1168 
1169 impl Instance {
1170     /// Enable or disable concurrent state debugging mode for e.g. integration
1171     /// tests.
1172     ///
1173     /// This will avoid re-using deleted handles, making it easier to catch
1174     /// e.g. "use-after-delete" and "double-delete" errors.  It can also make
1175     /// reading trace output easier since it ensures handles are never
1176     /// repurposed.
1177     #[doc(hidden)]
1178     pub fn enable_concurrent_state_debug(&self, mut store: impl AsContextMut, enable: bool) {
1179         self.id()
1180             .get_mut(store.as_context_mut().0)
1181             .concurrent_state_mut()
1182             .table
1183             .enable_debug(enable);
1184         // TODO: do the same for the tables holding guest-facing handles
1185     }
1186 
1187     /// Assert that all the relevant tables and queues in the concurrent state
1188     /// for this instance are empty.
1189     ///
1190     /// This is for sanity checking in integration tests
1191     /// (e.g. `component-async-tests`) that the relevant state has been cleared
1192     /// after each test concludes.  This should help us catch leaks, e.g. guest
1193     /// tasks which haven't been deleted despite having completed and having
1194     /// been dropped by their supertasks.
1195     #[doc(hidden)]
1196     pub fn assert_concurrent_state_empty(&self, mut store: impl AsContextMut) {
1197         let state = self
1198             .id()
1199             .get_mut(store.as_context_mut().0)
1200             .concurrent_state_mut();
1201         assert!(state.table.is_empty(), "non-empty table: {:?}", state.table);
1202         assert!(state.high_priority.is_empty());
1203         assert!(state.low_priority.is_empty());
1204         assert!(state.guest_task.is_none());
1205         assert!(
1206             state
1207                 .futures
1208                 .get_mut()
1209                 .unwrap()
1210                 .as_ref()
1211                 .unwrap()
1212                 .is_empty()
1213         );
1214         assert!(
1215             state
1216                 .waitable_tables
1217                 .iter()
1218                 .all(|(_, table)| table.is_empty())
1219         );
1220         assert!(
1221             state
1222                 .instance_states
1223                 .iter()
1224                 .all(|(_, state)| state.pending.is_empty())
1225         );
1226         assert!(
1227             state
1228                 .error_context_tables
1229                 .iter()
1230                 .all(|(_, table)| table.is_empty())
1231         );
1232         assert!(state.global_error_context_ref_counts.is_empty());
1233     }
1234 
1235     /// Run the specified closure `fun` to completion as part of this instance's
1236     /// event loop.
1237     ///
1238     /// Like [`Self::run`], this will run `fun` as part of this instance's event
1239     /// loop until it yields a result _or_ there are no more tasks to run.
1240     /// Unlike [`Self::run`], `fun` is provided an [`Accessor`], which provides
1241     /// controlled access to the `Store` and its data.
1242     ///
1243     /// This function can be used to invoke [`Func::call_concurrent`] for
1244     /// example within the async closure provided here.
1245     ///
1246     /// # Example
1247     ///
1248     /// ```
1249     /// # use {
1250     /// #   anyhow::{Result},
1251     /// #   wasmtime::{
1252     /// #     component::{ Component, Linker, Resource, ResourceTable},
1253     /// #     Config, Engine, Store
1254     /// #   },
1255     /// # };
1256     /// #
1257     /// # struct MyResource(u32);
1258     /// # struct Ctx { table: ResourceTable }
1259     /// #
1260     /// # async fn foo() -> Result<()> {
1261     /// # let mut config = Config::new();
1262     /// # let engine = Engine::new(&config)?;
1263     /// # let mut store = Store::new(&engine, Ctx { table: ResourceTable::new() });
1264     /// # let mut linker = Linker::new(&engine);
1265     /// # let component = Component::new(&engine, "")?;
1266     /// # let instance = linker.instantiate_async(&mut store, &component).await?;
1267     /// # let foo = instance.get_typed_func::<(Resource<MyResource>,), (Resource<MyResource>,)>(&mut store, "foo")?;
1268     /// # let bar = instance.get_typed_func::<(u32,), ()>(&mut store, "bar")?;
1269     /// instance.run_concurrent(&mut store, async |accessor| -> wasmtime::Result<_> {
1270     ///    let resource = accessor.with(|mut access| access.get().table.push(MyResource(42)))?;
1271     ///    let (another_resource,) = foo.call_concurrent(accessor, (resource,)).await?;
1272     ///    let value = accessor.with(|mut access| access.get().table.delete(another_resource))?;
1273     ///    bar.call_concurrent(accessor, (value.0,)).await?;
1274     ///    Ok(())
1275     /// }).await??;
1276     /// # Ok(())
1277     /// # }
1278     /// ```
1279     pub async fn run_concurrent<T, R>(
1280         self,
1281         mut store: impl AsContextMut<Data = T>,
1282         fun: impl AsyncFnOnce(&Accessor<T>) -> R,
1283     ) -> Result<R>
1284     where
1285         T: 'static,
1286     {
1287         check_recursive_run();
1288         let mut store = store.as_context_mut();
1289         let token = StoreToken::new(store.as_context_mut());
1290 
1291         self.poll_until(store.as_context_mut(), async move {
1292             let accessor = Accessor::new(token, self);
1293             fun(&accessor).await
1294         })
1295         .await
1296     }
1297 
1298     /// Spawn a background task to run as part of this instance's event loop.
1299     ///
1300     /// The task will receive an `&Accessor<U>` and run concurrently with
1301     /// any other tasks in progress for the instance.
1302     ///
1303     /// Note that the task will only make progress if and when the event loop
1304     /// for this instance is run.
1305     ///
1306     /// The returned [`SpawnHandle`] may be used to cancel the task.
1307     pub fn spawn<U: 'static>(
1308         self,
1309         mut store: impl AsContextMut<Data = U>,
1310         task: impl AccessorTask<U, HasSelf<U>, Result<()>>,
1311     ) -> AbortHandle {
1312         let mut store = store.as_context_mut();
1313         let accessor = Accessor::new(StoreToken::new(store.as_context_mut()), self);
1314         self.spawn_with_accessor(store, accessor, task)
1315     }
1316 
1317     /// Internal implementation of `spawn` functions where a `store` is
1318     /// available along with an `Accessor`.
1319     fn spawn_with_accessor<T, D>(
1320         self,
1321         mut store: StoreContextMut<T>,
1322         accessor: Accessor<T, D>,
1323         task: impl AccessorTask<T, D, Result<()>>,
1324     ) -> AbortHandle
1325     where
1326         T: 'static,
1327         D: HasData,
1328     {
1329         let store = store.as_context_mut();
1330 
1331         // Create an "abortable future" here where internally the future will
1332         // hook calls to poll and possibly spawn more background tasks on each
1333         // iteration.
1334         let (handle, future) =
1335             AbortHandle::run(async move { HostTaskOutput::Result(task.run(&accessor).await) });
1336         self.concurrent_state_mut(store.0)
1337             .push_future(Box::pin(async move {
1338                 future.await.unwrap_or(HostTaskOutput::Result(Ok(())))
1339             }));
1340 
1341         handle
1342     }
1343 
1344     /// Run this instance's event loop.
1345     ///
1346     /// The returned future will resolve when either the specified future
1347     /// completes (in which case we return its result) or no further progress
1348     /// can be made (in which case we trap with `Trap::AsyncDeadlock`).
1349     async fn poll_until<T, R>(
1350         self,
1351         store: StoreContextMut<'_, T>,
1352         future: impl Future<Output = R>,
1353     ) -> Result<R> {
1354         let mut future = pin!(future);
1355 
1356         loop {
1357             // Take `ConcurrentState::futures` out of the instance so we can
1358             // poll it while also safely giving any of the futures inside access
1359             // to `self`.
1360             let mut futures = self
1361                 .concurrent_state_mut(store.0)
1362                 .futures
1363                 .get_mut()
1364                 .unwrap()
1365                 .take()
1366                 .unwrap();
1367             let mut next = pin!(futures.next());
1368 
1369             let result = future::poll_fn(|cx| {
1370                 // First, poll the future we were passed as an argument and
1371                 // return immediately if it's ready.
1372                 if let Poll::Ready(value) = self.set_tls(store.0, || future.as_mut().poll(cx)) {
1373                     return Poll::Ready(Ok(Either::Left(value)));
1374                 }
1375 
1376                 // Next, poll `ConcurrentState::futures` (which includes any
1377                 // pending host tasks and/or background tasks), returning
1378                 // immediately if one of them fails.
1379                 let next = match self.set_tls(store.0, || next.as_mut().poll(cx)) {
1380                     Poll::Ready(Some(output)) => {
1381                         if let Err(e) = output.consume(store.0.traitobj_mut(), self) {
1382                             return Poll::Ready(Err(e));
1383                         }
1384                         Poll::Ready(true)
1385                     }
1386                     Poll::Ready(None) => Poll::Ready(false),
1387                     Poll::Pending => Poll::Pending,
1388                 };
1389 
1390                 let mut instance = self.id().get_mut(store.0);
1391 
1392                 // Next, check the "high priority" work queue and return
1393                 // immediately if it has at least one item.
1394                 let state = instance.as_mut().concurrent_state_mut();
1395                 let ready = mem::take(&mut state.high_priority);
1396                 let ready = if ready.is_empty() {
1397                     // Next, check the "low priority" work queue and return
1398                     // immediately if it has at least one item.
1399                     let ready = mem::take(&mut state.low_priority);
1400                     if ready.is_empty() {
1401                         return match next {
1402                             // In this case, one of the futures in
1403                             // `ConcurrentState::futures` completed
1404                             // successfully, so we return now and continue the
1405                             // outer loop in case there is another one ready to
1406                             // complete.
1407                             Poll::Ready(true) => Poll::Ready(Ok(Either::Right(Vec::new()))),
1408                             // In this case, there are no more pending futures
1409                             // in `ConcurrentState::futures`, there are no
1410                             // remaining work items, _and_ the future we were
1411                             // passed as an argument still hasn't completed,
1412                             // meaning we're stuck, so we return an error.  The
1413                             // underlying assumption is that `future` depends on
1414                             // this component instance making such progress, and
1415                             // thus there's no point in continuing to poll it
1416                             // given we've run out of work to do.
1417                             //
1418                             // Note that we'd also reach this point if the host
1419                             // embedder passed e.g. a `std::future::Pending` to
1420                             // `Instance::run_concurrent`, in which case we'd
1421                             // return a "deadlock" error even when any and all
1422                             // tasks have completed normally.  However, that's
1423                             // not how `Instance::run_concurrent` is intended
1424                             // (and documented) to be used, so it seems
1425                             // reasonable to lump that case in with "real"
1426                             // deadlocks.
1427                             //
1428                             // TODO: Once we've added host APIs for cancelling
1429                             // in-progress tasks, we can return some other,
1430                             // non-error value here, treating it as "normal" and
1431                             // giving the host embedder a chance to intervene by
1432                             // cancelling one or more tasks and/or starting new
1433                             // tasks capable of waking the existing ones.
1434                             Poll::Ready(false) => {
1435                                 Poll::Ready(Err(anyhow!(crate::Trap::AsyncDeadlock)))
1436                             }
1437                             // There is at least one pending future in
1438                             // `ConcurrentState::futures` and we have nothing
1439                             // else to do but wait for now, so we return
1440                             // `Pending`.
1441                             Poll::Pending => Poll::Pending,
1442                         };
1443                     } else {
1444                         ready
1445                     }
1446                 } else {
1447                     ready
1448                 };
1449 
1450                 Poll::Ready(Ok(Either::Right(ready)))
1451             })
1452             .await;
1453 
1454             // Put the `ConcurrentState::futures` back into the instance before
1455             // we return or handle any work items since one or more of those
1456             // items might append more futures.
1457             *self
1458                 .concurrent_state_mut(store.0)
1459                 .futures
1460                 .get_mut()
1461                 .unwrap() = Some(futures);
1462 
1463             match result? {
1464                 // The future we were passed as an argument completed, so we
1465                 // return the result.
1466                 Either::Left(value) => break Ok(value),
1467                 // The future we were passed has not yet completed, so handle
1468                 // any work items and then loop again.
1469                 Either::Right(ready) => {
1470                     for item in ready {
1471                         self.handle_work_item(store.0.traitobj_mut(), item).await?;
1472                     }
1473                 }
1474             }
1475         }
1476     }
1477 
1478     /// Handle the specified work item, possibly resuming a fiber if applicable.
1479     async fn handle_work_item(self, store: &mut StoreOpaque, item: WorkItem) -> Result<()> {
1480         log::trace!("handle work item {item:?}");
1481         match item {
1482             WorkItem::PushFuture(future) => {
1483                 self.concurrent_state_mut(store)
1484                     .futures
1485                     .get_mut()
1486                     .unwrap()
1487                     .as_mut()
1488                     .unwrap()
1489                     .push(future.into_inner().unwrap());
1490             }
1491             WorkItem::ResumeFiber(fiber) => {
1492                 self.resume_fiber(store, fiber).await?;
1493             }
1494             WorkItem::GuestCall(call) => {
1495                 let state = self.concurrent_state_mut(store);
1496                 if call.is_ready(state)? {
1497                     self.run_on_worker(store, call).await?;
1498                 } else {
1499                     let task = state.get_mut(call.task)?;
1500                     if !task.starting_sent {
1501                         task.starting_sent = true;
1502                         if let GuestCallKind::Start(_) = &call.kind {
1503                             Waitable::Guest(call.task).set_event(
1504                                 state,
1505                                 Some(Event::Subtask {
1506                                     status: Status::Starting,
1507                                 }),
1508                             )?;
1509                         }
1510                     }
1511 
1512                     let runtime_instance = state.get(call.task)?.instance;
1513                     state
1514                         .instance_state(runtime_instance)
1515                         .pending
1516                         .insert(call.task, call.kind);
1517                 }
1518             }
1519             WorkItem::Poll(params) => {
1520                 let state = self.concurrent_state_mut(store);
1521                 if state.get_mut(params.task)?.event.is_some()
1522                     || !state.get_mut(params.set)?.ready.is_empty()
1523                 {
1524                     // There's at least one event immediately available; deliver
1525                     // it to the guest ASAP.
1526                     state.push_high_priority(WorkItem::GuestCall(GuestCall {
1527                         task: params.task,
1528                         kind: GuestCallKind::DeliverEvent {
1529                             instance: params.instance,
1530                             set: Some(params.set),
1531                         },
1532                     }));
1533                 } else {
1534                     // There are no events immediately available; deliver
1535                     // `Event::None` to the guest.
1536                     state.get_mut(params.task)?.event = Some(Event::None);
1537                     state.push_high_priority(WorkItem::GuestCall(GuestCall {
1538                         task: params.task,
1539                         kind: GuestCallKind::DeliverEvent {
1540                             instance: params.instance,
1541                             set: Some(params.set),
1542                         },
1543                     }));
1544                 }
1545             }
1546         }
1547 
1548         Ok(())
1549     }
1550 
1551     /// Resume the specified fiber, giving it exclusive access to the specified
1552     /// store.
1553     async fn resume_fiber(self, store: &mut StoreOpaque, fiber: StoreFiber<'static>) -> Result<()> {
1554         let old_task = self.concurrent_state_mut(store).guest_task;
1555         log::trace!("resume_fiber: save current task {old_task:?}");
1556 
1557         let fiber = fiber::resolve_or_release(store, fiber).await?;
1558 
1559         let state = self.concurrent_state_mut(store);
1560 
1561         state.guest_task = old_task;
1562         log::trace!("resume_fiber: restore current task {old_task:?}");
1563 
1564         if let Some(mut fiber) = fiber {
1565             // See the `SuspendReason` documentation for what each case means.
1566             match state.suspend_reason.take().unwrap() {
1567                 SuspendReason::NeedWork => {
1568                     if state.worker.is_none() {
1569                         state.worker = Some(fiber);
1570                     } else {
1571                         fiber.dispose(store);
1572                     }
1573                 }
1574                 SuspendReason::Yielding { .. } => {
1575                     state.push_low_priority(WorkItem::ResumeFiber(fiber));
1576                 }
1577                 SuspendReason::Waiting { set, task } => {
1578                     let old = state
1579                         .get_mut(set)?
1580                         .waiting
1581                         .insert(task, WaitMode::Fiber(fiber));
1582                     assert!(old.is_none());
1583                 }
1584             }
1585         }
1586 
1587         Ok(())
1588     }
1589 
1590     /// Execute the specified guest call on a worker fiber.
1591     async fn run_on_worker(self, store: &mut StoreOpaque, call: GuestCall) -> Result<()> {
1592         let worker = if let Some(fiber) = self.concurrent_state_mut(store).worker.take() {
1593             fiber
1594         } else {
1595             fiber::make_fiber(store.traitobj_mut(), move |store| {
1596                 loop {
1597                     let call = self.concurrent_state_mut(store).guest_call.take().unwrap();
1598                     self.handle_guest_call(store, call)?;
1599 
1600                     self.suspend(store, SuspendReason::NeedWork)?;
1601                 }
1602             })?
1603         };
1604 
1605         let guest_call = &mut self.concurrent_state_mut(store).guest_call;
1606         assert!(guest_call.is_none());
1607         *guest_call = Some(call);
1608 
1609         self.resume_fiber(store, worker).await
1610     }
1611 
1612     /// Execute the specified guest call.
1613     fn handle_guest_call(self, store: &mut dyn VMStore, call: GuestCall) -> Result<()> {
1614         match call.kind {
1615             GuestCallKind::DeliverEvent {
1616                 instance: runtime_instance,
1617                 set,
1618             } => {
1619                 let state = self.concurrent_state_mut(store);
1620                 let (event, waitable) = state.get_event(call.task, runtime_instance, set)?.unwrap();
1621                 let task = state.get_mut(call.task)?;
1622                 let runtime_instance = task.instance;
1623                 let handle = waitable.map(|(_, v)| v).unwrap_or(0);
1624 
1625                 log::trace!(
1626                     "use callback to deliver event {event:?} to {:?} for {waitable:?}",
1627                     call.task,
1628                 );
1629 
1630                 let old_task = state.guest_task.replace(call.task);
1631                 log::trace!(
1632                     "GuestCallKind::DeliverEvent: replaced {old_task:?} with {:?} as current task",
1633                     call.task
1634                 );
1635 
1636                 self.maybe_push_call_context(store.store_opaque_mut(), call.task)?;
1637 
1638                 let state = self.concurrent_state_mut(store);
1639                 state.enter_instance(runtime_instance);
1640 
1641                 let callback = state.get_mut(call.task)?.callback.take().unwrap();
1642 
1643                 let code = callback(store, self, runtime_instance, event, handle)?;
1644 
1645                 let state = self.concurrent_state_mut(store);
1646 
1647                 state.get_mut(call.task)?.callback = Some(callback);
1648 
1649                 state.exit_instance(runtime_instance)?;
1650 
1651                 self.maybe_pop_call_context(store.store_opaque_mut(), call.task)?;
1652 
1653                 let state = self.concurrent_state_mut(store);
1654                 state.handle_callback_code(call.task, runtime_instance, code, false)?;
1655 
1656                 state.guest_task = old_task;
1657                 log::trace!("GuestCallKind::DeliverEvent: restored {old_task:?} as current task");
1658             }
1659             GuestCallKind::Start(fun) => {
1660                 fun(store, self)?;
1661             }
1662         }
1663 
1664         Ok(())
1665     }
1666 
1667     /// Suspend the current fiber, storing the reason in
1668     /// `ConcurrentState::suspend_reason` to indicate the conditions under which
1669     /// it should be resumed.
1670     ///
1671     /// See the `SuspendReason` documentation for details.
1672     fn suspend(self, store: &mut dyn VMStore, reason: SuspendReason) -> Result<()> {
1673         log::trace!("suspend fiber: {reason:?}");
1674 
1675         // If we're yielding or waiting on behalf of a guest task, we'll need to
1676         // pop the call context which manages resource borrows before suspending
1677         // and then push it again once we've resumed.
1678         let task = match &reason {
1679             SuspendReason::Yielding { task } | SuspendReason::Waiting { task, .. } => Some(*task),
1680             SuspendReason::NeedWork => None,
1681         };
1682 
1683         let old_guest_task = if let Some(task) = task {
1684             self.maybe_pop_call_context(store.store_opaque_mut(), task)?;
1685             self.concurrent_state_mut(store).guest_task
1686         } else {
1687             None
1688         };
1689 
1690         let suspend_reason = &mut self.concurrent_state_mut(store).suspend_reason;
1691         assert!(suspend_reason.is_none());
1692         *suspend_reason = Some(reason);
1693 
1694         store.with_blocking(|_, cx| cx.suspend(StoreFiberYield::ReleaseStore))?;
1695 
1696         if let Some(task) = task {
1697             self.concurrent_state_mut(store).guest_task = old_guest_task;
1698             self.maybe_push_call_context(store.store_opaque_mut(), task)?;
1699         }
1700 
1701         Ok(())
1702     }
1703 
1704     /// Push the call context for managing resource borrows for the specified
1705     /// guest task if it has not yet either returned a result or cancelled
1706     /// itself.
1707     fn maybe_push_call_context(
1708         self,
1709         store: &mut StoreOpaque,
1710         guest_task: TableId<GuestTask>,
1711     ) -> Result<()> {
1712         let task = self.concurrent_state_mut(store).get_mut(guest_task)?;
1713         if task.lift_result.is_some() {
1714             log::trace!("push call context for {guest_task:?}");
1715             let call_context = task.call_context.take().unwrap();
1716             store.component_resource_state().0.push(call_context);
1717         }
1718         Ok(())
1719     }
1720 
1721     /// Pop the call context for managing resource borrows for the specified
1722     /// guest task if it has not yet either returned a result or cancelled
1723     /// itself.
1724     fn maybe_pop_call_context(
1725         self,
1726         store: &mut StoreOpaque,
1727         guest_task: TableId<GuestTask>,
1728     ) -> Result<()> {
1729         if self
1730             .concurrent_state_mut(store)
1731             .get(guest_task)?
1732             .lift_result
1733             .is_some()
1734         {
1735             log::trace!("pop call context for {guest_task:?}");
1736             let call_context = Some(store.component_resource_state().0.pop().unwrap());
1737             self.concurrent_state_mut(store)
1738                 .get_mut(guest_task)?
1739                 .call_context = call_context;
1740         }
1741         Ok(())
1742     }
1743 
1744     /// Add the specified guest call to the "high priority" work item queue, to
1745     /// be started as soon as backpressure and/or reentrance rules allow.
1746     ///
1747     /// SAFETY: The raw pointer arguments must be valid references to guest
1748     /// functions (with the appropriate signatures) when the closures queued by
1749     /// this function are called.
1750     unsafe fn queue_call<T: 'static>(
1751         self,
1752         mut store: StoreContextMut<T>,
1753         guest_task: TableId<GuestTask>,
1754         callee: SendSyncPtr<VMFuncRef>,
1755         param_count: usize,
1756         result_count: usize,
1757         flags: Option<InstanceFlags>,
1758         async_: bool,
1759         callback: Option<SendSyncPtr<VMFuncRef>>,
1760         post_return: Option<SendSyncPtr<VMFuncRef>>,
1761     ) -> Result<()> {
1762         /// Return a closure which will call the specified function in the scope
1763         /// of the specified task.
1764         ///
1765         /// This will use `GuestTask::lower_params` to lower the parameters, but
1766         /// will not lift the result; instead, it returns a
1767         /// `[MaybeUninit<ValRaw>; MAX_FLAT_PARAMS]` from which the result, if
1768         /// any, may be lifted.  Note that an async-lifted export will have
1769         /// returned its result using the `task.return` intrinsic (or not
1770         /// returned a result at all, in the case of `task.cancel`), in which
1771         /// case the "result" of this call will either be a callback code or
1772         /// nothing.
1773         ///
1774         /// SAFETY: `callee` must be a valid `*mut VMFuncRef` at the time when
1775         /// the returned closure is called.
1776         unsafe fn make_call<T: 'static>(
1777             store: StoreContextMut<T>,
1778             guest_task: TableId<GuestTask>,
1779             callee: SendSyncPtr<VMFuncRef>,
1780             param_count: usize,
1781             result_count: usize,
1782             flags: Option<InstanceFlags>,
1783         ) -> impl FnOnce(
1784             &mut dyn VMStore,
1785             Instance,
1786         ) -> Result<[MaybeUninit<ValRaw>; MAX_FLAT_PARAMS]>
1787         + Send
1788         + Sync
1789         + 'static
1790         + use<T> {
1791             let token = StoreToken::new(store);
1792             move |store: &mut dyn VMStore, instance: Instance| {
1793                 let mut storage = [MaybeUninit::uninit(); MAX_FLAT_PARAMS];
1794                 let task = instance.concurrent_state_mut(store).get_mut(guest_task)?;
1795                 let may_enter_after_call = task.call_post_return_automatically();
1796                 let lower = task.lower_params.take().unwrap();
1797 
1798                 lower(store, instance, &mut storage[..param_count])?;
1799 
1800                 let mut store = token.as_context_mut(store);
1801 
1802                 // SAFETY: Per the contract documented in `make_call's`
1803                 // documentation, `callee` must be a valid pointer.
1804                 unsafe {
1805                     if let Some(mut flags) = flags {
1806                         flags.set_may_enter(false);
1807                     }
1808                     crate::Func::call_unchecked_raw(
1809                         &mut store,
1810                         callee.as_non_null(),
1811                         NonNull::new(
1812                             &mut storage[..param_count.max(result_count)]
1813                                 as *mut [MaybeUninit<ValRaw>] as _,
1814                         )
1815                         .unwrap(),
1816                     )?;
1817                     if let Some(mut flags) = flags {
1818                         flags.set_may_enter(may_enter_after_call);
1819                     }
1820                 }
1821 
1822                 Ok(storage)
1823             }
1824         }
1825 
1826         // SAFETY: Per the contract described in this function documentation,
1827         // the `callee` pointer which `call` closes over must be valid when
1828         // called by the closure we queue below.
1829         let call = unsafe {
1830             make_call(
1831                 store.as_context_mut(),
1832                 guest_task,
1833                 callee,
1834                 param_count,
1835                 result_count,
1836                 flags,
1837             )
1838         };
1839 
1840         let callee_instance = self.concurrent_state_mut(store.0).get(guest_task)?.instance;
1841         let fun = if callback.is_some() {
1842             assert!(async_);
1843 
1844             Box::new(move |store: &mut dyn VMStore, instance: Instance| {
1845                 let old_task = instance
1846                     .concurrent_state_mut(store)
1847                     .guest_task
1848                     .replace(guest_task);
1849                 log::trace!(
1850                     "stackless call: replaced {old_task:?} with {guest_task:?} as current task"
1851                 );
1852 
1853                 instance.maybe_push_call_context(store.store_opaque_mut(), guest_task)?;
1854 
1855                 instance
1856                     .concurrent_state_mut(store)
1857                     .enter_instance(callee_instance);
1858 
1859                 // SAFETY: See the documentation for `make_call` to review the
1860                 // contract we must uphold for `call` here.
1861                 //
1862                 // Per the contract described in the `queue_call`
1863                 // documentation, the `callee` pointer which `call` closes
1864                 // over must be valid.
1865                 let storage = call(store, instance)?;
1866 
1867                 instance
1868                     .concurrent_state_mut(store)
1869                     .exit_instance(callee_instance)?;
1870 
1871                 instance.maybe_pop_call_context(store.store_opaque_mut(), guest_task)?;
1872 
1873                 let state = instance.concurrent_state_mut(store);
1874                 state.guest_task = old_task;
1875                 log::trace!("stackless call: restored {old_task:?} as current task");
1876 
1877                 // SAFETY: `wasmparser` will have validated that the callback
1878                 // function returns a `i32` result.
1879                 let code = unsafe { storage[0].assume_init() }.get_i32() as u32;
1880 
1881                 state.handle_callback_code(guest_task, callee_instance, code, true)?;
1882 
1883                 Ok(())
1884             })
1885                 as Box<dyn FnOnce(&mut dyn VMStore, Instance) -> Result<()> + Send + Sync>
1886         } else {
1887             let token = StoreToken::new(store.as_context_mut());
1888             Box::new(move |store: &mut dyn VMStore, instance: Instance| {
1889                 let old_task = instance
1890                     .concurrent_state_mut(store)
1891                     .guest_task
1892                     .replace(guest_task);
1893                 log::trace!(
1894                     "stackful call: replaced {old_task:?} with {guest_task:?} as current task",
1895                 );
1896 
1897                 let mut flags = instance.id().get(store).instance_flags(callee_instance);
1898 
1899                 instance.maybe_push_call_context(store.store_opaque_mut(), guest_task)?;
1900 
1901                 // Unless this is a callback-less (i.e. stackful)
1902                 // async-lifted export, we need to record that the instance
1903                 // cannot be entered until the call returns.
1904                 if !async_ {
1905                     instance
1906                         .concurrent_state_mut(store)
1907                         .enter_instance(callee_instance);
1908                 }
1909 
1910                 // SAFETY: See the documentation for `make_call` to review the
1911                 // contract we must uphold for `call` here.
1912                 //
1913                 // Per the contract described in the `queue_call`
1914                 // documentation, the `callee` pointer which `call` closes
1915                 // over must be valid.
1916                 let storage = call(store, instance)?;
1917 
1918                 if async_ {
1919                     // This is a callback-less (i.e. stackful) async-lifted
1920                     // export, so there is no post-return function, and
1921                     // either `task.return` or `task.cancel` should have
1922                     // been called.
1923                     if instance
1924                         .concurrent_state_mut(store)
1925                         .get(guest_task)?
1926                         .lift_result
1927                         .is_some()
1928                     {
1929                         return Err(anyhow!(crate::Trap::NoAsyncResult));
1930                     }
1931                 } else {
1932                     // This is a sync-lifted export, so now is when we lift the
1933                     // result, optionally call the post-return function, if any,
1934                     // and finally notify any current or future waiters that the
1935                     // subtask has returned.
1936 
1937                     let lift = {
1938                         let state = instance.concurrent_state_mut(store);
1939                         state.exit_instance(callee_instance)?;
1940 
1941                         assert!(state.get(guest_task)?.result.is_none());
1942 
1943                         state.get_mut(guest_task)?.lift_result.take().unwrap()
1944                     };
1945 
1946                     // SAFETY: `result_count` represents the number of core Wasm
1947                     // results returned, per `wasmparser`.
1948                     let result = (lift.lift)(store, instance, unsafe {
1949                         mem::transmute::<&[MaybeUninit<ValRaw>], &[ValRaw]>(
1950                             &storage[..result_count],
1951                         )
1952                     })?;
1953 
1954                     let post_return_arg = match result_count {
1955                         0 => ValRaw::i32(0),
1956                         // SAFETY: `result_count` represents the number of
1957                         // core Wasm results returned, per `wasmparser`.
1958                         1 => unsafe { storage[0].assume_init() },
1959                         _ => unreachable!(),
1960                     };
1961 
1962                     if instance
1963                         .concurrent_state_mut(store)
1964                         .get(guest_task)?
1965                         .call_post_return_automatically()
1966                     {
1967                         unsafe { flags.set_needs_post_return(false) }
1968 
1969                         if let Some(func) = post_return {
1970                             let mut store = token.as_context_mut(store);
1971 
1972                             // SAFETY: `func` is a valid `*mut VMFuncRef` from
1973                             // either `wasmtime-cranelift`-generated fused adapter
1974                             // code or `component::Options`.  Per `wasmparser`
1975                             // post-return signature validation, we know it takes a
1976                             // single parameter.
1977                             unsafe {
1978                                 crate::Func::call_unchecked_raw(
1979                                     &mut store,
1980                                     func.as_non_null(),
1981                                     slice::from_ref(&post_return_arg).into(),
1982                                 )?;
1983                             }
1984                         }
1985 
1986                         unsafe { flags.set_may_enter(true) }
1987                     }
1988 
1989                     instance.task_complete(
1990                         store,
1991                         guest_task,
1992                         result,
1993                         Status::Returned,
1994                         post_return_arg,
1995                     )?;
1996                 }
1997 
1998                 instance.maybe_pop_call_context(store.store_opaque_mut(), guest_task)?;
1999 
2000                 let task = instance.concurrent_state_mut(store).get_mut(guest_task)?;
2001 
2002                 match &task.caller {
2003                     Caller::Host {
2004                         remove_task_automatically,
2005                         ..
2006                     } => {
2007                         if *remove_task_automatically {
2008                             Waitable::Guest(guest_task)
2009                                 .delete_from(instance.concurrent_state_mut(store))?;
2010                         }
2011                     }
2012                     Caller::Guest { .. } => {
2013                         task.exited = true;
2014                     }
2015                 }
2016 
2017                 Ok(())
2018             })
2019         };
2020 
2021         self.concurrent_state_mut(store.0)
2022             .push_high_priority(WorkItem::GuestCall(GuestCall {
2023                 task: guest_task,
2024                 kind: GuestCallKind::Start(fun),
2025             }));
2026 
2027         Ok(())
2028     }
2029 
2030     /// Prepare (but do not start) a guest->guest call.
2031     ///
2032     /// This is called from fused adapter code generated in
2033     /// `wasmtime_environ::fact::trampoline::Compiler`.  `start` and `return_`
2034     /// are synthesized Wasm functions which move the parameters from the caller
2035     /// to the callee and the result from the callee to the caller,
2036     /// respectively.  The adapter will call `Self::start_call` immediately
2037     /// after calling this function.
2038     ///
2039     /// SAFETY: All the pointer arguments must be valid pointers to guest
2040     /// entities (and with the expected signatures for the function references
2041     /// -- see `wasmtime_environ::fact::trampoline::Compiler` for details).
2042     unsafe fn prepare_call<T: 'static>(
2043         self,
2044         mut store: StoreContextMut<T>,
2045         start: *mut VMFuncRef,
2046         return_: *mut VMFuncRef,
2047         caller_instance: RuntimeComponentInstanceIndex,
2048         callee_instance: RuntimeComponentInstanceIndex,
2049         task_return_type: TypeTupleIndex,
2050         memory: *mut VMMemoryDefinition,
2051         string_encoding: u8,
2052         caller_info: CallerInfo,
2053     ) -> Result<()> {
2054         enum ResultInfo {
2055             Heap { results: u32 },
2056             Stack { result_count: u32 },
2057         }
2058 
2059         let result_info = match &caller_info {
2060             CallerInfo::Async {
2061                 has_result: true,
2062                 params,
2063             } => ResultInfo::Heap {
2064                 results: params.last().unwrap().get_u32(),
2065             },
2066             CallerInfo::Async {
2067                 has_result: false, ..
2068             } => ResultInfo::Stack { result_count: 0 },
2069             CallerInfo::Sync {
2070                 result_count,
2071                 params,
2072             } if *result_count > u32::try_from(MAX_FLAT_RESULTS).unwrap() => ResultInfo::Heap {
2073                 results: params.last().unwrap().get_u32(),
2074             },
2075             CallerInfo::Sync { result_count, .. } => ResultInfo::Stack {
2076                 result_count: *result_count,
2077             },
2078         };
2079 
2080         let sync_caller = matches!(caller_info, CallerInfo::Sync { .. });
2081 
2082         // Create a new guest task for the call, closing over the `start` and
2083         // `return_` functions to lift the parameters and lower the result,
2084         // respectively.
2085         let start = SendSyncPtr::new(NonNull::new(start).unwrap());
2086         let return_ = SendSyncPtr::new(NonNull::new(return_).unwrap());
2087         let token = StoreToken::new(store.as_context_mut());
2088         let state = self.concurrent_state_mut(store.0);
2089         let old_task = state.guest_task.take();
2090         let new_task = GuestTask::new(
2091             state,
2092             Box::new(move |store, instance, dst| {
2093                 let mut store = token.as_context_mut(store);
2094                 assert!(dst.len() <= MAX_FLAT_PARAMS);
2095                 let mut src = [MaybeUninit::uninit(); MAX_FLAT_PARAMS];
2096                 let count = match caller_info {
2097                     // Async callers, if they have a result, use the last
2098                     // parameter as a return pointer so chop that off if
2099                     // relevant here.
2100                     CallerInfo::Async { params, has_result } => {
2101                         let params = &params[..params.len() - usize::from(has_result)];
2102                         for (param, src) in params.iter().zip(&mut src) {
2103                             src.write(*param);
2104                         }
2105                         params.len()
2106                     }
2107 
2108                     // Sync callers forward everything directly.
2109                     CallerInfo::Sync { params, .. } => {
2110                         for (param, src) in params.iter().zip(&mut src) {
2111                             src.write(*param);
2112                         }
2113                         params.len()
2114                     }
2115                 };
2116                 // SAFETY: `start` is a valid `*mut VMFuncRef` from
2117                 // `wasmtime-cranelift`-generated fused adapter code.  Based on
2118                 // how it was constructed (see
2119                 // `wasmtime_environ::fact::trampoline::Compiler::compile_async_start_adapter`
2120                 // for details) we know it takes count parameters and returns
2121                 // `dst.len()` results.
2122                 unsafe {
2123                     crate::Func::call_unchecked_raw(
2124                         &mut store,
2125                         start.as_non_null(),
2126                         NonNull::new(
2127                             &mut src[..count.max(dst.len())] as *mut [MaybeUninit<ValRaw>] as _,
2128                         )
2129                         .unwrap(),
2130                     )?;
2131                 }
2132                 dst.copy_from_slice(&src[..dst.len()]);
2133                 let state = instance.concurrent_state_mut(store.0);
2134                 let task = state.guest_task.unwrap();
2135                 Waitable::Guest(task).set_event(
2136                     state,
2137                     Some(Event::Subtask {
2138                         status: Status::Started,
2139                     }),
2140                 )?;
2141                 Ok(())
2142             }),
2143             LiftResult {
2144                 lift: Box::new(move |store, instance, src| {
2145                     // SAFETY: See comment in closure passed as `lower_params`
2146                     // parameter above.
2147                     let mut store = token.as_context_mut(store);
2148                     let mut my_src = src.to_owned(); // TODO: use stack to avoid allocation?
2149                     if let ResultInfo::Heap { results } = &result_info {
2150                         my_src.push(ValRaw::u32(*results));
2151                     }
2152                     // SAFETY: `return_` is a valid `*mut VMFuncRef` from
2153                     // `wasmtime-cranelift`-generated fused adapter code.  Based
2154                     // on how it was constructed (see
2155                     // `wasmtime_environ::fact::trampoline::Compiler::compile_async_return_adapter`
2156                     // for details) we know it takes `src.len()` parameters and
2157                     // returns up to 1 result.
2158                     unsafe {
2159                         crate::Func::call_unchecked_raw(
2160                             &mut store,
2161                             return_.as_non_null(),
2162                             my_src.as_mut_slice().into(),
2163                         )?;
2164                     }
2165                     let state = instance.concurrent_state_mut(store.0);
2166                     let task = state.guest_task.unwrap();
2167                     if sync_caller {
2168                         state.get_mut(task)?.sync_result =
2169                             Some(if let ResultInfo::Stack { result_count } = &result_info {
2170                                 match result_count {
2171                                     0 => None,
2172                                     1 => Some(my_src[0]),
2173                                     _ => unreachable!(),
2174                                 }
2175                             } else {
2176                                 None
2177                             });
2178                     }
2179                     Ok(Box::new(DummyResult) as Box<dyn Any + Send + Sync>)
2180                 }),
2181                 ty: task_return_type,
2182                 memory: NonNull::new(memory).map(SendSyncPtr::new),
2183                 string_encoding: StringEncoding::from_u8(string_encoding).unwrap(),
2184             },
2185             Caller::Guest {
2186                 task: old_task.unwrap(),
2187                 instance: caller_instance,
2188             },
2189             None,
2190             callee_instance,
2191         )?;
2192 
2193         let guest_task = state.push(new_task)?;
2194 
2195         if let Some(old_task) = old_task {
2196             if !state.may_enter(guest_task) {
2197                 bail!(crate::Trap::CannotEnterComponent);
2198             }
2199 
2200             state.get_mut(old_task)?.subtasks.insert(guest_task);
2201         };
2202 
2203         // Make the new task the current one so that `Self::start_call` knows
2204         // which one to start.
2205         state.guest_task = Some(guest_task);
2206         log::trace!("pushed {guest_task:?} as current task; old task was {old_task:?}");
2207 
2208         Ok(())
2209     }
2210 
2211     /// Call the specified callback function for an async-lifted export.
2212     ///
2213     /// SAFETY: `function` must be a valid reference to a guest function of the
2214     /// correct signature for a callback.
2215     unsafe fn call_callback<T>(
2216         self,
2217         mut store: StoreContextMut<T>,
2218         callee_instance: RuntimeComponentInstanceIndex,
2219         function: SendSyncPtr<VMFuncRef>,
2220         event: Event,
2221         handle: u32,
2222         may_enter_after_call: bool,
2223     ) -> Result<u32> {
2224         let mut flags = self.id().get(store.0).instance_flags(callee_instance);
2225 
2226         let (ordinal, result) = event.parts();
2227         let params = &mut [
2228             ValRaw::u32(ordinal),
2229             ValRaw::u32(handle),
2230             ValRaw::u32(result),
2231         ];
2232         // SAFETY: `func` is a valid `*mut VMFuncRef` from either
2233         // `wasmtime-cranelift`-generated fused adapter code or
2234         // `component::Options`.  Per `wasmparser` callback signature
2235         // validation, we know it takes three parameters and returns one.
2236         unsafe {
2237             flags.set_may_enter(false);
2238             crate::Func::call_unchecked_raw(
2239                 &mut store,
2240                 function.as_non_null(),
2241                 params.as_mut_slice().into(),
2242             )?;
2243             flags.set_may_enter(may_enter_after_call);
2244         }
2245         Ok(params[0].get_u32())
2246     }
2247 
2248     /// Start a guest->guest call previously prepared using
2249     /// `Self::prepare_call`.
2250     ///
2251     /// This is called from fused adapter code generated in
2252     /// `wasmtime_environ::fact::trampoline::Compiler`.  The adapter will call
2253     /// this function immediately after calling `Self::prepare_call`.
2254     ///
2255     /// SAFETY: The `*mut VMFuncRef` arguments must be valid pointers to guest
2256     /// functions with the appropriate signatures for the current guest task.
2257     /// If this is a call to an async-lowered import, the actual call may be
2258     /// deferred and run after this function returns, in which case the pointer
2259     /// arguments must also be valid when the call happens.
2260     unsafe fn start_call<T: 'static>(
2261         self,
2262         mut store: StoreContextMut<T>,
2263         callback: *mut VMFuncRef,
2264         post_return: *mut VMFuncRef,
2265         callee: *mut VMFuncRef,
2266         param_count: u32,
2267         result_count: u32,
2268         flags: u32,
2269         storage: Option<&mut [MaybeUninit<ValRaw>]>,
2270     ) -> Result<u32> {
2271         let token = StoreToken::new(store.as_context_mut());
2272         let async_caller = storage.is_none();
2273         let state = self.concurrent_state_mut(store.0);
2274         let guest_task = state.guest_task.unwrap();
2275         let may_enter_after_call = state.get(guest_task)?.call_post_return_automatically();
2276         let callee = SendSyncPtr::new(NonNull::new(callee).unwrap());
2277         let param_count = usize::try_from(param_count).unwrap();
2278         assert!(param_count <= MAX_FLAT_PARAMS);
2279         let result_count = usize::try_from(result_count).unwrap();
2280         assert!(result_count <= MAX_FLAT_RESULTS);
2281 
2282         let task = state.get_mut(guest_task)?;
2283         if !callback.is_null() {
2284             // We're calling an async-lifted export with a callback, so store
2285             // the callback and related context as part of the task so we can
2286             // call it later when needed.
2287             let callback = SendSyncPtr::new(NonNull::new(callback).unwrap());
2288             task.callback = Some(Box::new(
2289                 move |store, instance, runtime_instance, event, handle| {
2290                     let store = token.as_context_mut(store);
2291                     unsafe {
2292                         instance.call_callback::<T>(
2293                             store,
2294                             runtime_instance,
2295                             callback,
2296                             event,
2297                             handle,
2298                             may_enter_after_call,
2299                         )
2300                     }
2301                 },
2302             ));
2303         }
2304 
2305         let Caller::Guest {
2306             task: caller,
2307             instance: runtime_instance,
2308         } = &task.caller
2309         else {
2310             // As of this writing, `start_call` is only used for guest->guest
2311             // calls.
2312             unreachable!()
2313         };
2314         let caller = *caller;
2315         let caller_instance = *runtime_instance;
2316 
2317         let callee_instance = task.instance;
2318 
2319         let instance_flags = if callback.is_null() {
2320             None
2321         } else {
2322             Some(self.id().get(store.0).instance_flags(callee_instance))
2323         };
2324 
2325         // Queue the call as a "high priority" work item.
2326         unsafe {
2327             self.queue_call(
2328                 store.as_context_mut(),
2329                 guest_task,
2330                 callee,
2331                 param_count,
2332                 result_count,
2333                 instance_flags,
2334                 (flags & START_FLAG_ASYNC_CALLEE) != 0,
2335                 NonNull::new(callback).map(SendSyncPtr::new),
2336                 NonNull::new(post_return).map(SendSyncPtr::new),
2337             )?;
2338         }
2339 
2340         let state = self.concurrent_state_mut(store.0);
2341 
2342         // Use the caller's `GuestTask::sync_call_set` to register interest in
2343         // the subtask...
2344         let set = state.get_mut(caller)?.sync_call_set;
2345         Waitable::Guest(guest_task).join(state, Some(set))?;
2346 
2347         // ... and suspend this fiber temporarily while we wait for it to start.
2348         //
2349         // Note that we _could_ call the callee directly using the current fiber
2350         // rather than suspend this one, but that would make reasoning about the
2351         // event loop more complicated and is probably only worth doing if
2352         // there's a measurable performance benefit.  In addition, it would mean
2353         // blocking the caller if the callee calls a blocking sync-lowered
2354         // import, and as of this writing the spec says we must not do that.
2355         //
2356         // Alternatively, the fused adapter code could be modified to call the
2357         // callee directly without calling a host-provided intrinsic at all (in
2358         // which case it would need to do its own, inline backpressure checks,
2359         // etc.).  Again, we'd want to see a measurable performance benefit
2360         // before committing to such an optimization.  And again, we'd need to
2361         // update the spec to allow that.
2362         let (status, waitable) = loop {
2363             self.suspend(
2364                 store.0.traitobj_mut(),
2365                 SuspendReason::Waiting { set, task: caller },
2366             )?;
2367 
2368             let state = self.concurrent_state_mut(store.0);
2369 
2370             let event = Waitable::Guest(guest_task).take_event(state)?;
2371             let Some(Event::Subtask { status }) = event else {
2372                 unreachable!();
2373             };
2374 
2375             log::trace!("status {status:?} for {guest_task:?}");
2376 
2377             if status == Status::Returned {
2378                 // It returned, so we can stop waiting.
2379                 break (status, None);
2380             } else if async_caller {
2381                 // It hasn't returned yet, but the caller is calling via an
2382                 // async-lowered import, so we generate a handle for the task
2383                 // waitable and return the status.
2384                 break (
2385                     status,
2386                     Some(
2387                         state.waitable_tables[caller_instance]
2388                             .insert(guest_task.rep(), WaitableState::GuestTask)?,
2389                     ),
2390                 );
2391             } else {
2392                 // The callee hasn't returned yet, and the caller is calling via
2393                 // a sync-lowered import, so we loop and keep waiting until the
2394                 // callee returns.
2395             }
2396         };
2397 
2398         let state = self.concurrent_state_mut(store.0);
2399 
2400         Waitable::Guest(guest_task).join(state, None)?;
2401 
2402         if let Some(storage) = storage {
2403             // The caller used a sync-lowered import to call an async-lifted
2404             // export, in which case the result, if any, has been stashed in
2405             // `GuestTask::sync_result`.
2406             if let Some(result) = state.get_mut(guest_task)?.sync_result.take() {
2407                 if let Some(result) = result {
2408                     storage[0] = MaybeUninit::new(result);
2409                 }
2410 
2411                 Waitable::Guest(guest_task).delete_from(state)?;
2412             } else {
2413                 // This means the callee failed to call either `task.return` or
2414                 // `task.cancel` before exiting.
2415                 return Err(anyhow!(crate::Trap::NoAsyncResult));
2416             }
2417         }
2418 
2419         // Reset the current task to point to the caller as it resumes control.
2420         state.guest_task = Some(caller);
2421         log::trace!("popped current task {guest_task:?}; new task is {caller:?}");
2422 
2423         Ok(status.pack(waitable))
2424     }
2425 
2426     /// Wrap the specified host function in a future which will call it, passing
2427     /// it an `&Accessor<T>`.
2428     ///
2429     /// See the `Accessor` documentation for details.
2430     pub(crate) fn wrap_call<T: 'static, F, R>(
2431         self,
2432         store: StoreContextMut<T>,
2433         closure: F,
2434     ) -> impl Future<Output = Result<R>> + 'static
2435     where
2436         T: 'static,
2437         F: FnOnce(&Accessor<T>) -> Pin<Box<dyn Future<Output = Result<R>> + Send + '_>>
2438             + Send
2439             + Sync
2440             + 'static,
2441         R: Send + Sync + 'static,
2442     {
2443         let token = StoreToken::new(store);
2444         async move {
2445             let mut accessor = Accessor::new(token, self);
2446             closure(&mut accessor).await
2447         }
2448     }
2449 
2450     /// Poll the specified future once on behalf of a guest->host call using an
2451     /// async-lowered import.
2452     ///
2453     /// If it returns `Ready`, return `Ok(None)`.  Otherwise, if it returns
2454     /// `Pending`, add it to the set of futures to be polled as part of this
2455     /// instance's event loop until it completes, and then return
2456     /// `Ok(Some(handle))` where `handle` is the waitable handle to return.
2457     ///
2458     /// Whether the future returns `Ready` immediately or later, the `lower`
2459     /// function will be used to lower the result, if any, into the guest caller's
2460     /// stack and linear memory unless the task has been cancelled.
2461     pub(crate) fn first_poll<T: 'static, R: Send + 'static>(
2462         self,
2463         mut store: StoreContextMut<T>,
2464         future: impl Future<Output = Result<R>> + Send + 'static,
2465         caller_instance: RuntimeComponentInstanceIndex,
2466         lower: impl FnOnce(StoreContextMut<T>, Instance, R) -> Result<()> + Send + 'static,
2467     ) -> Result<Option<u32>> {
2468         let token = StoreToken::new(store.as_context_mut());
2469         let state = self.concurrent_state_mut(store.0);
2470         let caller = state.guest_task.unwrap();
2471 
2472         // Create an abortable future which hooks calls to poll and manages call
2473         // context state for the future.
2474         let (abort_handle, future) = AbortHandle::run(async move {
2475             let mut future = pin!(future);
2476             let mut call_context = None;
2477             future::poll_fn(move |cx| {
2478                 // Push the call context for managing any resource borrows
2479                 // for the task.
2480                 tls::get(|store| {
2481                     if let Some(call_context) = call_context.take() {
2482                         token
2483                             .as_context_mut(store)
2484                             .0
2485                             .component_resource_state()
2486                             .0
2487                             .push(call_context);
2488                     }
2489                 });
2490 
2491                 let result = future.as_mut().poll(cx);
2492 
2493                 if result.is_pending() {
2494                     // Pop the call context for managing any resource
2495                     // borrows for the task.
2496                     tls::get(|store| {
2497                         call_context = Some(
2498                             token
2499                                 .as_context_mut(store)
2500                                 .0
2501                                 .component_resource_state()
2502                                 .0
2503                                 .pop()
2504                                 .unwrap(),
2505                         );
2506                     });
2507                 }
2508                 result
2509             })
2510             .await
2511         });
2512 
2513         // We create a new host task even though it might complete immediately
2514         // (in which case we won't need to pass a waitable back to the guest).
2515         // If it does complete immediately, we'll remove it before we return.
2516         let task = state.push(HostTask::new(caller_instance, Some(abort_handle)))?;
2517 
2518         log::trace!("new host task child of {caller:?}: {task:?}");
2519         let token = StoreToken::new(store.as_context_mut());
2520 
2521         // Map the output of the future to a `HostTaskOutput` responsible for
2522         // lowering the result into the guest's stack and memory, as well as
2523         // notifying any waiters that the task returned.
2524         let mut future = Box::pin(async move {
2525             let result = match future.await {
2526                 Some(result) => result,
2527                 // Task was cancelled; nothing left to do.
2528                 None => return HostTaskOutput::Result(Ok(())),
2529             };
2530             HostTaskOutput::Function(Box::new(move |store, instance| {
2531                 let mut store = token.as_context_mut(store);
2532                 lower(store.as_context_mut(), instance, result?)?;
2533                 let state = instance.concurrent_state_mut(store.0);
2534                 state.get_mut(task)?.abort_handle.take();
2535                 Waitable::Host(task).set_event(
2536                     state,
2537                     Some(Event::Subtask {
2538                         status: Status::Returned,
2539                     }),
2540                 )?;
2541 
2542                 Ok(())
2543             }))
2544         });
2545 
2546         // Finally, poll the future.  We can use a dummy `Waker` here because
2547         // we'll add the future to `ConcurrentState::futures` and poll it
2548         // automatically from the event loop if it doesn't complete immediately
2549         // here.
2550         let poll = self.set_tls(store.0, || {
2551             future
2552                 .as_mut()
2553                 .poll(&mut Context::from_waker(&Waker::noop()))
2554         });
2555 
2556         Ok(match poll {
2557             Poll::Ready(output) => {
2558                 // It finished immediately; lower the result and delete the
2559                 // task.
2560                 output.consume(store.0.traitobj_mut(), self)?;
2561                 log::trace!("delete host task {task:?} (already ready)");
2562                 self.concurrent_state_mut(store.0).delete(task)?;
2563                 None
2564             }
2565             Poll::Pending => {
2566                 // It hasn't finished yet; add the future to
2567                 // `ConcurrentState::futures` so it will be polled by the event
2568                 // loop and allocate a waitable handle to return to the guest.
2569                 let state = self.concurrent_state_mut(store.0);
2570                 state.push_future(future);
2571                 let handle = state.waitable_tables[caller_instance]
2572                     .insert(task.rep(), WaitableState::HostTask)?;
2573                 log::trace!(
2574                     "assign {task:?} handle {handle} for {caller:?} instance {caller_instance:?}"
2575                 );
2576                 Some(handle)
2577             }
2578         })
2579     }
2580 
2581     /// Poll the specified future until it completes on behalf of a guest->host
2582     /// call using a sync-lowered import.
2583     ///
2584     /// This is similar to `Self::first_poll` except it's for sync-lowered
2585     /// imports, meaning we don't need to handle cancellation and we can block
2586     /// the caller until the task completes, at which point the caller can
2587     /// handle lowering the result to the guest's stack and linear memory.
2588     pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
2589         self,
2590         store: &mut dyn VMStore,
2591         future: impl Future<Output = Result<R>> + Send + 'static,
2592         caller_instance: RuntimeComponentInstanceIndex,
2593     ) -> Result<R> {
2594         let state = self.concurrent_state_mut(store);
2595 
2596         // If there is no current guest task set, that means the host function
2597         // was registered using e.g. `LinkerInstance::func_wrap`, in which case
2598         // it should complete immediately.
2599         let Some(caller) = state.guest_task else {
2600             return match pin!(future).poll(&mut Context::from_waker(&Waker::noop())) {
2601                 Poll::Ready(result) => result,
2602                 Poll::Pending => {
2603                     unreachable!()
2604                 }
2605             };
2606         };
2607 
2608         // Save any existing result stashed in `GuestTask::result` so we can
2609         // replace it with the new result.
2610         let old_result = state
2611             .get_mut(caller)
2612             .with_context(|| format!("bad handle: {caller:?}"))?
2613             .result
2614             .take();
2615 
2616         // Add a temporary host task into the table so we can track its
2617         // progress.  Note that we'll never allocate a waitable handle for the
2618         // guest since we're being called synchronously.
2619         let task = state.push(HostTask::new(caller_instance, None))?;
2620 
2621         log::trace!("new host task child of {caller:?}: {task:?}");
2622 
2623         // Map the output of the future to a `HostTaskOutput` which will take
2624         // care of stashing the result in `GuestTask::result` and resuming this
2625         // fiber when the host task completes.
2626         let mut future = Box::pin(future.map(move |result| {
2627             HostTaskOutput::Function(Box::new(move |store, instance| {
2628                 let state = instance.concurrent_state_mut(store);
2629                 state.get_mut(caller)?.result = Some(Box::new(result?) as _);
2630 
2631                 Waitable::Host(task).set_event(
2632                     state,
2633                     Some(Event::Subtask {
2634                         status: Status::Returned,
2635                     }),
2636                 )?;
2637 
2638                 Ok(())
2639             }))
2640         })) as HostTaskFuture;
2641 
2642         // Finally, poll the future.  We can use a dummy `Waker` here because
2643         // we'll add the future to `ConcurrentState::futures` and poll it
2644         // automatically from the event loop if it doesn't complete immediately
2645         // here.
2646         let poll = self.set_tls(store, || {
2647             future
2648                 .as_mut()
2649                 .poll(&mut Context::from_waker(&Waker::noop()))
2650         });
2651 
2652         match poll {
2653             Poll::Ready(output) => {
2654                 // It completed immediately; run the `HostTaskOutput` function
2655                 // to stash the result and delete the task.
2656                 output.consume(store, self)?;
2657                 log::trace!("delete host task {task:?} (already ready)");
2658                 self.concurrent_state_mut(store).delete(task)?;
2659             }
2660             Poll::Pending => {
2661                 // It did not complete immediately; add it to
2662                 // `ConcurrentState::futures` so it will be polled via the event
2663                 // loop, then use `GuestTask::sync_call_set` to wait for the
2664                 // task to complete, suspending the current fiber until it does
2665                 // so.
2666                 let state = self.concurrent_state_mut(store);
2667                 state.push_future(future);
2668 
2669                 let set = state.get_mut(caller)?.sync_call_set;
2670                 Waitable::Host(task).join(state, Some(set))?;
2671 
2672                 self.suspend(store, SuspendReason::Waiting { set, task: caller })?;
2673             }
2674         }
2675 
2676         // Retrieve and return the result.
2677         Ok(*mem::replace(
2678             &mut self.concurrent_state_mut(store).get_mut(caller)?.result,
2679             old_result,
2680         )
2681         .unwrap()
2682         .downcast()
2683         .unwrap())
2684     }
2685 
2686     /// Implements the `task.return` intrinsic, lifting the result for the
2687     /// current guest task.
2688     ///
2689     /// SAFETY: The `memory` and `storage` pointers must be valid, and `storage`
2690     /// must contain at least `storage_len` items.
2691     pub(crate) unsafe fn task_return(
2692         self,
2693         store: &mut dyn VMStore,
2694         ty: TypeTupleIndex,
2695         memory: *mut VMMemoryDefinition,
2696         string_encoding: u8,
2697         storage: *mut ValRaw,
2698         storage_len: usize,
2699     ) -> Result<()> {
2700         // SAFETY: The `wasmtime_cranelift`-generated code that calls this
2701         // method will have ensured that `storage` is a valid pointer containing
2702         // at least `storage_len` items.
2703         let storage = unsafe { std::slice::from_raw_parts(storage, storage_len) };
2704         let state = self.concurrent_state_mut(store);
2705         let guest_task = state.guest_task.unwrap();
2706         let lift = state
2707             .get_mut(guest_task)?
2708             .lift_result
2709             .take()
2710             .ok_or_else(|| {
2711                 anyhow!("`task.return` or `task.cancel` called more than once for current task")
2712             })?;
2713 
2714         if ty != lift.ty
2715             || (!memory.is_null()
2716                 && memory != lift.memory.map(|v| v.as_ptr()).unwrap_or(ptr::null_mut()))
2717             || string_encoding != lift.string_encoding as u8
2718         {
2719             bail!("invalid `task.return` signature and/or options for current task");
2720         }
2721 
2722         assert!(state.get(guest_task)?.result.is_none());
2723 
2724         log::trace!("task.return for {guest_task:?}");
2725 
2726         let result = (lift.lift)(store, self, storage)?;
2727 
2728         self.task_complete(store, guest_task, result, Status::Returned, ValRaw::i32(0))
2729     }
2730 
2731     /// Implements the `task.cancel` intrinsic.
2732     pub(crate) fn task_cancel(
2733         self,
2734         store: &mut dyn VMStore,
2735         _caller_instance: RuntimeComponentInstanceIndex,
2736     ) -> Result<()> {
2737         let state = self.concurrent_state_mut(store);
2738         let guest_task = state.guest_task.unwrap();
2739         let task = state.get_mut(guest_task)?;
2740         if !task.cancel_sent {
2741             bail!("`task.cancel` called by task which has not been cancelled")
2742         }
2743         _ = task.lift_result.take().ok_or_else(|| {
2744             anyhow!("`task.return` or `task.cancel` called more than once for current task")
2745         })?;
2746 
2747         assert!(task.result.is_none());
2748 
2749         log::trace!("task.cancel for {guest_task:?}");
2750 
2751         self.task_complete(
2752             store,
2753             guest_task,
2754             Box::new(DummyResult),
2755             Status::ReturnCancelled,
2756             ValRaw::i32(0),
2757         )
2758     }
2759 
2760     /// Complete the specified guest task (i.e. indicate that it has either
2761     /// returned a (possibly empty) result or cancelled itself).
2762     ///
2763     /// This will return any resource borrows and notify any current or future
2764     /// waiters that the task has completed.
2765     fn task_complete(
2766         self,
2767         store: &mut dyn VMStore,
2768         guest_task: TableId<GuestTask>,
2769         result: Box<dyn Any + Send + Sync>,
2770         status: Status,
2771         post_return_arg: ValRaw,
2772     ) -> Result<()> {
2773         if self
2774             .concurrent_state_mut(store)
2775             .get(guest_task)?
2776             .call_post_return_automatically()
2777         {
2778             let (calls, host_table, _, instance) = store
2779                 .store_opaque_mut()
2780                 .component_resource_state_with_instance(self);
2781             ResourceTables {
2782                 calls,
2783                 host_table: Some(host_table),
2784                 guest: Some(instance.guest_tables()),
2785             }
2786             .exit_call()?;
2787         } else {
2788             // As of this writing, the only scenario where `call_post_return_automatically`
2789             // would be false for a `GuestTask` is for host-to-guest calls using
2790             // `[Typed]Func::call_async`, in which case the `function_index`
2791             // should be a non-`None` value.
2792             let function_index = self
2793                 .concurrent_state_mut(store)
2794                 .get(guest_task)?
2795                 .function_index
2796                 .unwrap();
2797 
2798             self.id()
2799                 .get_mut(store)
2800                 .post_return_arg_set(function_index, post_return_arg);
2801         }
2802 
2803         let state = self.concurrent_state_mut(store);
2804         let task = state.get_mut(guest_task)?;
2805 
2806         if let Caller::Host { tx, .. } = &mut task.caller {
2807             if let Some(tx) = tx.take() {
2808                 _ = tx.send(result);
2809             }
2810         } else {
2811             task.result = Some(result);
2812             Waitable::Guest(guest_task).set_event(state, Some(Event::Subtask { status }))?;
2813         }
2814 
2815         Ok(())
2816     }
2817 
2818     /// Implements the `waitable-set.wait` intrinsic.
2819     pub(crate) fn waitable_set_wait(
2820         self,
2821         store: &mut dyn VMStore,
2822         caller_instance: RuntimeComponentInstanceIndex,
2823         async_: bool,
2824         memory: *mut VMMemoryDefinition,
2825         set: u32,
2826         payload: u32,
2827     ) -> Result<u32> {
2828         let (rep, WaitableState::Set) = self.concurrent_state_mut(store).waitable_tables
2829             [caller_instance]
2830             .get_mut_by_index(set)?
2831         else {
2832             bail!("invalid waitable-set handle");
2833         };
2834 
2835         self.waitable_check(
2836             store,
2837             async_,
2838             WaitableCheck::Wait(WaitableCheckParams {
2839                 set: TableId::new(rep),
2840                 memory,
2841                 payload,
2842                 caller_instance,
2843             }),
2844         )
2845     }
2846 
2847     /// Implements the `waitable-set.poll` intrinsic.
2848     pub(crate) fn waitable_set_poll(
2849         self,
2850         store: &mut dyn VMStore,
2851         caller_instance: RuntimeComponentInstanceIndex,
2852         async_: bool,
2853         memory: *mut VMMemoryDefinition,
2854         set: u32,
2855         payload: u32,
2856     ) -> Result<u32> {
2857         let (rep, WaitableState::Set) = self.concurrent_state_mut(store).waitable_tables
2858             [caller_instance]
2859             .get_mut_by_index(set)?
2860         else {
2861             bail!("invalid waitable-set handle");
2862         };
2863 
2864         self.waitable_check(
2865             store,
2866             async_,
2867             WaitableCheck::Poll(WaitableCheckParams {
2868                 set: TableId::new(rep),
2869                 memory,
2870                 payload,
2871                 caller_instance,
2872             }),
2873         )
2874     }
2875 
2876     /// Implements the `yield` intrinsic.
2877     pub(crate) fn yield_(self, store: &mut dyn VMStore, async_: bool) -> Result<bool> {
2878         self.waitable_check(store, async_, WaitableCheck::Yield)
2879             .map(|_code| {
2880                 // TODO: plumb cancellation to here:
2881                 // https://github.com/bytecodealliance/wasmtime/issues/11191
2882                 false
2883             })
2884     }
2885 
2886     /// Helper function for the `waitable-set.wait`, `waitable-set.poll`, and
2887     /// `yield` intrinsics.
2888     fn waitable_check(
2889         self,
2890         store: &mut dyn VMStore,
2891         async_: bool,
2892         check: WaitableCheck,
2893     ) -> Result<u32> {
2894         if async_ {
2895             bail!(
2896                 "todo: async `waitable-set.wait`, `waitable-set.poll`, and `yield` not yet implemented"
2897             );
2898         }
2899 
2900         let guest_task = self.concurrent_state_mut(store).guest_task.unwrap();
2901 
2902         let (wait, set) = match &check {
2903             WaitableCheck::Wait(params) => (true, Some(params.set)),
2904             WaitableCheck::Poll(params) => (false, Some(params.set)),
2905             WaitableCheck::Yield => (false, None),
2906         };
2907 
2908         // First, suspend this fiber, allowing any other tasks to run.
2909         self.suspend(store, SuspendReason::Yielding { task: guest_task })?;
2910 
2911         log::trace!("waitable check for {guest_task:?}; set {set:?}");
2912 
2913         let state = self.concurrent_state_mut(store);
2914         let task = state.get(guest_task)?;
2915 
2916         if wait && task.callback.is_some() {
2917             bail!("cannot call `task.wait` from async-lifted export with callback");
2918         }
2919 
2920         // If we're waiting, and there are no events immediately available,
2921         // suspend the fiber until that changes.
2922         if wait {
2923             let set = set.unwrap();
2924 
2925             if task.event.is_none() && state.get(set)?.ready.is_empty() {
2926                 let old = state.get_mut(guest_task)?.wake_on_cancel.replace(set);
2927                 assert!(old.is_none());
2928 
2929                 self.suspend(
2930                     store,
2931                     SuspendReason::Waiting {
2932                         set,
2933                         task: guest_task,
2934                     },
2935                 )?;
2936             }
2937         }
2938 
2939         log::trace!("waitable check for {guest_task:?}; set {set:?}, part two");
2940 
2941         let result = match check {
2942             // Deliver any pending events to the guest and return.
2943             WaitableCheck::Wait(params) | WaitableCheck::Poll(params) => {
2944                 let event = self.concurrent_state_mut(store).get_event(
2945                     guest_task,
2946                     params.caller_instance,
2947                     Some(params.set),
2948                 )?;
2949 
2950                 let (ordinal, handle, result) = if wait {
2951                     let (event, waitable) = event.unwrap();
2952                     let handle = waitable.map(|(_, v)| v).unwrap_or(0);
2953                     let (ordinal, result) = event.parts();
2954                     (ordinal, handle, result)
2955                 } else {
2956                     if let Some((event, waitable)) = event {
2957                         let handle = waitable.map(|(_, v)| v).unwrap_or(0);
2958                         let (ordinal, result) = event.parts();
2959                         (ordinal, handle, result)
2960                     } else {
2961                         log::trace!(
2962                             "no events ready to deliver via waitable-set.poll to {guest_task:?}; set {:?}",
2963                             params.set
2964                         );
2965                         let (ordinal, result) = Event::None.parts();
2966                         (ordinal, 0, result)
2967                     }
2968                 };
2969                 let store = store.store_opaque_mut();
2970                 let options = unsafe {
2971                     Options::new(
2972                         store.id(),
2973                         NonNull::new(params.memory),
2974                         None,
2975                         StringEncoding::Utf8,
2976                         true,
2977                         None,
2978                     )
2979                 };
2980                 let ptr = func::validate_inbounds::<(u32, u32)>(
2981                     options.memory_mut(store),
2982                     &ValRaw::u32(params.payload),
2983                 )?;
2984                 options.memory_mut(store)[ptr + 0..][..4].copy_from_slice(&handle.to_le_bytes());
2985                 options.memory_mut(store)[ptr + 4..][..4].copy_from_slice(&result.to_le_bytes());
2986                 Ok(ordinal)
2987             }
2988             // TODO: Check `GuestTask::event` in case it contains
2989             // `Event::Cancelled`, in which case we'll need to return that to
2990             // the guest:
2991             // https://github.com/bytecodealliance/wasmtime/issues/11191
2992             WaitableCheck::Yield => Ok(0),
2993         };
2994 
2995         result
2996     }
2997 
2998     /// Implements the `subtask.cancel` intrinsic.
2999     pub(crate) fn subtask_cancel(
3000         self,
3001         store: &mut dyn VMStore,
3002         caller_instance: RuntimeComponentInstanceIndex,
3003         async_: bool,
3004         task_id: u32,
3005     ) -> Result<u32> {
3006         let concurrent_state = self.concurrent_state_mut(store);
3007         let (rep, state) =
3008             concurrent_state.waitable_tables[caller_instance].get_mut_by_index(task_id)?;
3009         let (waitable, expected_caller_instance) = match state {
3010             WaitableState::HostTask => {
3011                 let id = TableId::<HostTask>::new(rep);
3012                 (
3013                     Waitable::Host(id),
3014                     concurrent_state.get(id)?.caller_instance,
3015                 )
3016             }
3017             WaitableState::GuestTask => {
3018                 let id = TableId::<GuestTask>::new(rep);
3019                 if let Caller::Guest { instance, .. } = &concurrent_state.get(id)?.caller {
3020                     (Waitable::Guest(id), *instance)
3021                 } else {
3022                     unreachable!()
3023                 }
3024             }
3025             _ => bail!("invalid task handle: {task_id}"),
3026         };
3027         // Since waitables can neither be passed between instances nor forged,
3028         // this should never fail unless there's a bug in Wasmtime, but we check
3029         // here to be sure:
3030         assert_eq!(expected_caller_instance, caller_instance);
3031 
3032         log::trace!("subtask_cancel {waitable:?} (handle {task_id})");
3033 
3034         if let Waitable::Host(host_task) = waitable {
3035             if let Some(handle) = concurrent_state.get_mut(host_task)?.abort_handle.take() {
3036                 handle.abort();
3037                 return Ok(Status::ReturnCancelled as u32);
3038             }
3039         } else {
3040             let caller = concurrent_state.guest_task.unwrap();
3041             let guest_task = TableId::<GuestTask>::new(rep);
3042             let task = concurrent_state.get_mut(guest_task)?;
3043             if task.lower_params.is_some() {
3044                 task.lower_params = None;
3045                 task.lift_result = None;
3046 
3047                 // Not yet started; cancel and remove from pending
3048                 let callee_instance = task.instance;
3049 
3050                 let kind = concurrent_state
3051                     .instance_state(callee_instance)
3052                     .pending
3053                     .remove(&guest_task);
3054 
3055                 if kind.is_none() {
3056                     bail!("`subtask.cancel` called after terminal status delivered");
3057                 }
3058 
3059                 return Ok(Status::StartCancelled as u32);
3060             } else if task.lift_result.is_some() {
3061                 // Started, but not yet returned or cancelled; send the
3062                 // `CANCELLED` event
3063                 task.cancel_sent = true;
3064                 // Note that this might overwrite an event that was set earlier
3065                 // (e.g. `Event::None` if the task is yielding, or
3066                 // `Event::Cancelled` if it was already cancelled), but that's
3067                 // okay -- this should supersede the previous state.
3068                 task.event = Some(Event::Cancelled);
3069                 if let Some(set) = task.wake_on_cancel.take() {
3070                     let item = match concurrent_state
3071                         .get_mut(set)?
3072                         .waiting
3073                         .remove(&guest_task)
3074                         .unwrap()
3075                     {
3076                         WaitMode::Fiber(fiber) => WorkItem::ResumeFiber(fiber),
3077                         WaitMode::Callback(instance) => WorkItem::GuestCall(GuestCall {
3078                             task: guest_task,
3079                             kind: GuestCallKind::DeliverEvent {
3080                                 instance,
3081                                 set: None,
3082                             },
3083                         }),
3084                     };
3085                     concurrent_state.push_high_priority(item);
3086 
3087                     self.suspend(store, SuspendReason::Yielding { task: caller })?;
3088                 }
3089 
3090                 let concurrent_state = self.concurrent_state_mut(store);
3091                 let task = concurrent_state.get_mut(guest_task)?;
3092                 if task.lift_result.is_some() {
3093                     // Still not yet returned or cancelled; if `async_`, return
3094                     // `BLOCKED`; otherwise wait
3095                     if async_ {
3096                         return Ok(BLOCKED);
3097                     } else {
3098                         let set = concurrent_state.get_mut(caller)?.sync_call_set;
3099                         Waitable::Guest(guest_task).join(concurrent_state, Some(set))?;
3100 
3101                         self.suspend(store, SuspendReason::Waiting { set, task: caller })?;
3102                     }
3103                 }
3104             }
3105         }
3106 
3107         let event = waitable.take_event(self.concurrent_state_mut(store))?;
3108         if let Some(Event::Subtask {
3109             status: status @ (Status::Returned | Status::ReturnCancelled),
3110         }) = event
3111         {
3112             Ok(status as u32)
3113         } else {
3114             bail!("`subtask.cancel` called after terminal status delivered");
3115         }
3116     }
3117 
3118     /// Configures TLS state so `store` will be available via `tls::get` within
3119     /// the closure `f` provided.
3120     ///
3121     /// This is used to ensure that `Future::poll`, which doesn't take a `store`
3122     /// parameter, is able to get access to the `store` during future poll
3123     /// methods.
3124     fn set_tls<R>(self, store: &mut dyn VMStore, f: impl FnOnce() -> R) -> R {
3125         struct Reset<'a>(&'a mut dyn VMStore, Option<ComponentInstanceId>);
3126 
3127         impl Drop for Reset<'_> {
3128             fn drop(&mut self) {
3129                 self.0.concurrent_async_state_mut().current_instance = self.1;
3130             }
3131         }
3132         let prev = mem::replace(
3133             &mut store.concurrent_async_state_mut().current_instance,
3134             Some(self.id().instance()),
3135         );
3136         let reset = Reset(store, prev);
3137 
3138         tls::set(reset.0, f)
3139     }
3140 
3141     /// Convenience function to reduce boilerplate.
3142     pub(crate) fn concurrent_state_mut<'a>(
3143         &self,
3144         store: &'a mut StoreOpaque,
3145     ) -> &'a mut ConcurrentState {
3146         self.id().get_mut(store).concurrent_state_mut()
3147     }
3148 }
3149 
3150 /// Trait representing component model ABI async intrinsics and fused adapter
3151 /// helper functions.
3152 ///
3153 /// SAFETY (callers): Most of the methods in this trait accept raw pointers,
3154 /// which must be valid for at least the duration of the call (and possibly for
3155 /// as long as the relevant guest task exists, in the case of `*mut VMFuncRef`
3156 /// pointers used for async calls).
3157 pub trait VMComponentAsyncStore {
3158     /// A helper function for fused adapter modules involving calls where the
3159     /// one of the caller or callee is async.
3160     ///
3161     /// This helper is not used when the caller and callee both use the sync
3162     /// ABI, only when at least one is async is this used.
3163     unsafe fn prepare_call(
3164         &mut self,
3165         instance: Instance,
3166         memory: *mut VMMemoryDefinition,
3167         start: *mut VMFuncRef,
3168         return_: *mut VMFuncRef,
3169         caller_instance: RuntimeComponentInstanceIndex,
3170         callee_instance: RuntimeComponentInstanceIndex,
3171         task_return_type: TypeTupleIndex,
3172         string_encoding: u8,
3173         result_count: u32,
3174         storage: *mut ValRaw,
3175         storage_len: usize,
3176     ) -> Result<()>;
3177 
3178     /// A helper function for fused adapter modules involving calls where the
3179     /// caller is sync-lowered but the callee is async-lifted.
3180     unsafe fn sync_start(
3181         &mut self,
3182         instance: Instance,
3183         callback: *mut VMFuncRef,
3184         callee: *mut VMFuncRef,
3185         param_count: u32,
3186         storage: *mut MaybeUninit<ValRaw>,
3187         storage_len: usize,
3188     ) -> Result<()>;
3189 
3190     /// A helper function for fused adapter modules involving calls where the
3191     /// caller is async-lowered.
3192     unsafe fn async_start(
3193         &mut self,
3194         instance: Instance,
3195         callback: *mut VMFuncRef,
3196         post_return: *mut VMFuncRef,
3197         callee: *mut VMFuncRef,
3198         param_count: u32,
3199         result_count: u32,
3200         flags: u32,
3201     ) -> Result<u32>;
3202 
3203     /// The `future.write` intrinsic.
3204     unsafe fn future_write(
3205         &mut self,
3206         instance: Instance,
3207         memory: *mut VMMemoryDefinition,
3208         realloc: *mut VMFuncRef,
3209         string_encoding: u8,
3210         async_: bool,
3211         ty: TypeFutureTableIndex,
3212         future: u32,
3213         address: u32,
3214     ) -> Result<u32>;
3215 
3216     /// The `future.read` intrinsic.
3217     unsafe fn future_read(
3218         &mut self,
3219         instance: Instance,
3220         memory: *mut VMMemoryDefinition,
3221         realloc: *mut VMFuncRef,
3222         string_encoding: u8,
3223         async_: bool,
3224         ty: TypeFutureTableIndex,
3225         future: u32,
3226         address: u32,
3227     ) -> Result<u32>;
3228 
3229     /// The `stream.write` intrinsic.
3230     unsafe fn stream_write(
3231         &mut self,
3232         instance: Instance,
3233         memory: *mut VMMemoryDefinition,
3234         realloc: *mut VMFuncRef,
3235         string_encoding: u8,
3236         async_: bool,
3237         ty: TypeStreamTableIndex,
3238         stream: u32,
3239         address: u32,
3240         count: u32,
3241     ) -> Result<u32>;
3242 
3243     /// The `stream.read` intrinsic.
3244     unsafe fn stream_read(
3245         &mut self,
3246         instance: Instance,
3247         memory: *mut VMMemoryDefinition,
3248         realloc: *mut VMFuncRef,
3249         string_encoding: u8,
3250         async_: bool,
3251         ty: TypeStreamTableIndex,
3252         stream: u32,
3253         address: u32,
3254         count: u32,
3255     ) -> Result<u32>;
3256 
3257     /// The "fast-path" implementation of the `stream.write` intrinsic for
3258     /// "flat" (i.e. memcpy-able) payloads.
3259     unsafe fn flat_stream_write(
3260         &mut self,
3261         instance: Instance,
3262         memory: *mut VMMemoryDefinition,
3263         realloc: *mut VMFuncRef,
3264         async_: bool,
3265         ty: TypeStreamTableIndex,
3266         payload_size: u32,
3267         payload_align: u32,
3268         stream: u32,
3269         address: u32,
3270         count: u32,
3271     ) -> Result<u32>;
3272 
3273     /// The "fast-path" implementation of the `stream.read` intrinsic for "flat"
3274     /// (i.e. memcpy-able) payloads.
3275     unsafe fn flat_stream_read(
3276         &mut self,
3277         instance: Instance,
3278         memory: *mut VMMemoryDefinition,
3279         realloc: *mut VMFuncRef,
3280         async_: bool,
3281         ty: TypeStreamTableIndex,
3282         payload_size: u32,
3283         payload_align: u32,
3284         stream: u32,
3285         address: u32,
3286         count: u32,
3287     ) -> Result<u32>;
3288 
3289     /// The `error-context.debug-message` intrinsic.
3290     unsafe fn error_context_debug_message(
3291         &mut self,
3292         instance: Instance,
3293         memory: *mut VMMemoryDefinition,
3294         realloc: *mut VMFuncRef,
3295         string_encoding: u8,
3296         ty: TypeComponentLocalErrorContextTableIndex,
3297         err_ctx_handle: u32,
3298         debug_msg_address: u32,
3299     ) -> Result<()>;
3300 }
3301 
3302 /// SAFETY: See trait docs.
3303 impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
3304     unsafe fn prepare_call(
3305         &mut self,
3306         instance: Instance,
3307         memory: *mut VMMemoryDefinition,
3308         start: *mut VMFuncRef,
3309         return_: *mut VMFuncRef,
3310         caller_instance: RuntimeComponentInstanceIndex,
3311         callee_instance: RuntimeComponentInstanceIndex,
3312         task_return_type: TypeTupleIndex,
3313         string_encoding: u8,
3314         result_count_or_max_if_async: u32,
3315         storage: *mut ValRaw,
3316         storage_len: usize,
3317     ) -> Result<()> {
3318         // SAFETY: The `wasmtime_cranelift`-generated code that calls
3319         // this method will have ensured that `storage` is a valid
3320         // pointer containing at least `storage_len` items.
3321         let params = unsafe { std::slice::from_raw_parts(storage, storage_len) }.to_vec();
3322 
3323         unsafe {
3324             instance.prepare_call(
3325                 StoreContextMut(self),
3326                 start,
3327                 return_,
3328                 caller_instance,
3329                 callee_instance,
3330                 task_return_type,
3331                 memory,
3332                 string_encoding,
3333                 match result_count_or_max_if_async {
3334                     PREPARE_ASYNC_NO_RESULT => CallerInfo::Async {
3335                         params,
3336                         has_result: false,
3337                     },
3338                     PREPARE_ASYNC_WITH_RESULT => CallerInfo::Async {
3339                         params,
3340                         has_result: true,
3341                     },
3342                     result_count => CallerInfo::Sync {
3343                         params,
3344                         result_count,
3345                     },
3346                 },
3347             )
3348         }
3349     }
3350 
3351     unsafe fn sync_start(
3352         &mut self,
3353         instance: Instance,
3354         callback: *mut VMFuncRef,
3355         callee: *mut VMFuncRef,
3356         param_count: u32,
3357         storage: *mut MaybeUninit<ValRaw>,
3358         storage_len: usize,
3359     ) -> Result<()> {
3360         unsafe {
3361             instance
3362                 .start_call(
3363                     StoreContextMut(self),
3364                     callback,
3365                     ptr::null_mut(),
3366                     callee,
3367                     param_count,
3368                     1,
3369                     START_FLAG_ASYNC_CALLEE,
3370                     // SAFETY: The `wasmtime_cranelift`-generated code that calls
3371                     // this method will have ensured that `storage` is a valid
3372                     // pointer containing at least `storage_len` items.
3373                     Some(std::slice::from_raw_parts_mut(storage, storage_len)),
3374                 )
3375                 .map(drop)
3376         }
3377     }
3378 
3379     unsafe fn async_start(
3380         &mut self,
3381         instance: Instance,
3382         callback: *mut VMFuncRef,
3383         post_return: *mut VMFuncRef,
3384         callee: *mut VMFuncRef,
3385         param_count: u32,
3386         result_count: u32,
3387         flags: u32,
3388     ) -> Result<u32> {
3389         unsafe {
3390             instance.start_call(
3391                 StoreContextMut(self),
3392                 callback,
3393                 post_return,
3394                 callee,
3395                 param_count,
3396                 result_count,
3397                 flags,
3398                 None,
3399             )
3400         }
3401     }
3402 
3403     unsafe fn future_write(
3404         &mut self,
3405         instance: Instance,
3406         memory: *mut VMMemoryDefinition,
3407         realloc: *mut VMFuncRef,
3408         string_encoding: u8,
3409         async_: bool,
3410         ty: TypeFutureTableIndex,
3411         future: u32,
3412         address: u32,
3413     ) -> Result<u32> {
3414         // SAFETY: Per the trait-level documentation for
3415         // `VMComponentAsyncStore`, all raw pointers passed to this function
3416         // must be valid.
3417         unsafe {
3418             instance
3419                 .guest_write(
3420                     StoreContextMut(self),
3421                     memory,
3422                     realloc,
3423                     string_encoding,
3424                     async_,
3425                     TableIndex::Future(ty),
3426                     None,
3427                     future,
3428                     address,
3429                     1,
3430                 )
3431                 .map(|result| result.encode())
3432         }
3433     }
3434 
3435     unsafe fn future_read(
3436         &mut self,
3437         instance: Instance,
3438         memory: *mut VMMemoryDefinition,
3439         realloc: *mut VMFuncRef,
3440         string_encoding: u8,
3441         async_: bool,
3442         ty: TypeFutureTableIndex,
3443         future: u32,
3444         address: u32,
3445     ) -> Result<u32> {
3446         // SAFETY: See corresponding comment in `Self::future_write`.
3447         unsafe {
3448             instance
3449                 .guest_read(
3450                     StoreContextMut(self),
3451                     memory,
3452                     realloc,
3453                     string_encoding,
3454                     async_,
3455                     TableIndex::Future(ty),
3456                     None,
3457                     future,
3458                     address,
3459                     1,
3460                 )
3461                 .map(|result| result.encode())
3462         }
3463     }
3464 
3465     unsafe fn stream_write(
3466         &mut self,
3467         instance: Instance,
3468         memory: *mut VMMemoryDefinition,
3469         realloc: *mut VMFuncRef,
3470         string_encoding: u8,
3471         async_: bool,
3472         ty: TypeStreamTableIndex,
3473         stream: u32,
3474         address: u32,
3475         count: u32,
3476     ) -> Result<u32> {
3477         // SAFETY: See corresponding comment in `Self::future_write`.
3478         unsafe {
3479             instance
3480                 .guest_write(
3481                     StoreContextMut(self),
3482                     memory,
3483                     realloc,
3484                     string_encoding,
3485                     async_,
3486                     TableIndex::Stream(ty),
3487                     None,
3488                     stream,
3489                     address,
3490                     count,
3491                 )
3492                 .map(|result| result.encode())
3493         }
3494     }
3495 
3496     unsafe fn stream_read(
3497         &mut self,
3498         instance: Instance,
3499         memory: *mut VMMemoryDefinition,
3500         realloc: *mut VMFuncRef,
3501         string_encoding: u8,
3502         async_: bool,
3503         ty: TypeStreamTableIndex,
3504         stream: u32,
3505         address: u32,
3506         count: u32,
3507     ) -> Result<u32> {
3508         // SAFETY: See corresponding comment in `Self::future_write`.
3509         unsafe {
3510             instance
3511                 .guest_read(
3512                     StoreContextMut(self),
3513                     memory,
3514                     realloc,
3515                     string_encoding,
3516                     async_,
3517                     TableIndex::Stream(ty),
3518                     None,
3519                     stream,
3520                     address,
3521                     count,
3522                 )
3523                 .map(|result| result.encode())
3524         }
3525     }
3526 
3527     unsafe fn flat_stream_write(
3528         &mut self,
3529         instance: Instance,
3530         memory: *mut VMMemoryDefinition,
3531         realloc: *mut VMFuncRef,
3532         async_: bool,
3533         ty: TypeStreamTableIndex,
3534         payload_size: u32,
3535         payload_align: u32,
3536         stream: u32,
3537         address: u32,
3538         count: u32,
3539     ) -> Result<u32> {
3540         // SAFETY: See corresponding comment in `Self::future_write`.
3541         unsafe {
3542             instance
3543                 .guest_write(
3544                     StoreContextMut(self),
3545                     memory,
3546                     realloc,
3547                     StringEncoding::Utf8 as u8,
3548                     async_,
3549                     TableIndex::Stream(ty),
3550                     Some(FlatAbi {
3551                         size: payload_size,
3552                         align: payload_align,
3553                     }),
3554                     stream,
3555                     address,
3556                     count,
3557                 )
3558                 .map(|result| result.encode())
3559         }
3560     }
3561 
3562     unsafe fn flat_stream_read(
3563         &mut self,
3564         instance: Instance,
3565         memory: *mut VMMemoryDefinition,
3566         realloc: *mut VMFuncRef,
3567         async_: bool,
3568         ty: TypeStreamTableIndex,
3569         payload_size: u32,
3570         payload_align: u32,
3571         stream: u32,
3572         address: u32,
3573         count: u32,
3574     ) -> Result<u32> {
3575         // SAFETY: See corresponding comment in `Self::future_write`.
3576         unsafe {
3577             instance
3578                 .guest_read(
3579                     StoreContextMut(self),
3580                     memory,
3581                     realloc,
3582                     StringEncoding::Utf8 as u8,
3583                     async_,
3584                     TableIndex::Stream(ty),
3585                     Some(FlatAbi {
3586                         size: payload_size,
3587                         align: payload_align,
3588                     }),
3589                     stream,
3590                     address,
3591                     count,
3592                 )
3593                 .map(|result| result.encode())
3594         }
3595     }
3596 
3597     unsafe fn error_context_debug_message(
3598         &mut self,
3599         instance: Instance,
3600         memory: *mut VMMemoryDefinition,
3601         realloc: *mut VMFuncRef,
3602         string_encoding: u8,
3603         ty: TypeComponentLocalErrorContextTableIndex,
3604         err_ctx_handle: u32,
3605         debug_msg_address: u32,
3606     ) -> Result<()> {
3607         // SAFETY: See corresponding comment in `Self::future_write`.
3608         unsafe {
3609             instance.error_context_debug_message(
3610                 StoreContextMut(self),
3611                 memory,
3612                 realloc,
3613                 string_encoding,
3614                 ty,
3615                 err_ctx_handle,
3616                 debug_msg_address,
3617             )
3618         }
3619     }
3620 }
3621 
3622 /// Represents the output of a host task or background task.
3623 enum HostTaskOutput {
3624     /// A plain result
3625     Result(Result<()>),
3626     /// A function to be run after the future completes (e.g. post-processing
3627     /// which requires access to the store and instance).
3628     Function(Box<dyn FnOnce(&mut dyn VMStore, Instance) -> Result<()> + Send>),
3629 }
3630 
3631 impl HostTaskOutput {
3632     /// Retrieve the result of the host or background task, running the
3633     /// post-processing function if present.
3634     fn consume(self, store: &mut dyn VMStore, instance: Instance) -> Result<()> {
3635         match self {
3636             Self::Function(fun) => fun(store, instance),
3637             Self::Result(result) => result,
3638         }
3639     }
3640 }
3641 
3642 type HostTaskFuture = Pin<Box<dyn Future<Output = HostTaskOutput> + Send + 'static>>;
3643 
3644 /// Represents the state of a pending host task.
3645 struct HostTask {
3646     common: WaitableCommon,
3647     caller_instance: RuntimeComponentInstanceIndex,
3648     abort_handle: Option<AbortHandle>,
3649 }
3650 
3651 impl HostTask {
3652     fn new(
3653         caller_instance: RuntimeComponentInstanceIndex,
3654         abort_handle: Option<AbortHandle>,
3655     ) -> Self {
3656         Self {
3657             common: WaitableCommon::default(),
3658             caller_instance,
3659             abort_handle,
3660         }
3661     }
3662 }
3663 
3664 impl TableDebug for HostTask {
3665     fn type_name() -> &'static str {
3666         "HostTask"
3667     }
3668 }
3669 
3670 type CallbackFn = Box<
3671     dyn Fn(&mut dyn VMStore, Instance, RuntimeComponentInstanceIndex, Event, u32) -> Result<u32>
3672         + Send
3673         + Sync
3674         + 'static,
3675 >;
3676 
3677 /// Represents the caller of a given guest task.
3678 enum Caller {
3679     /// The host called the guest task.
3680     Host {
3681         /// If present, may be used to deliver the result.
3682         tx: Option<oneshot::Sender<LiftedResult>>,
3683         /// If true, remove the task from the concurrent state that owns it
3684         /// automatically after it completes.
3685         remove_task_automatically: bool,
3686         /// If true, call `post-return` function (if any) automatically.
3687         call_post_return_automatically: bool,
3688     },
3689     /// Another guest task called the guest task
3690     Guest {
3691         /// The id of the caller
3692         task: TableId<GuestTask>,
3693         /// The instance to use to enforce reentrance rules.
3694         ///
3695         /// Note that this might not be the same as the instance the caller task
3696         /// started executing in given that one or more synchronous guest->guest
3697         /// calls may have occurred involving multiple instances.
3698         instance: RuntimeComponentInstanceIndex,
3699     },
3700 }
3701 
3702 /// Represents a closure and related canonical ABI parameters required to
3703 /// validate a `task.return` call at runtime and lift the result.
3704 struct LiftResult {
3705     lift: RawLift,
3706     ty: TypeTupleIndex,
3707     memory: Option<SendSyncPtr<VMMemoryDefinition>>,
3708     string_encoding: StringEncoding,
3709 }
3710 
3711 /// Represents a pending guest task.
3712 struct GuestTask {
3713     /// See `WaitableCommon`
3714     common: WaitableCommon,
3715     /// Closure to lower the parameters passed to this task.
3716     lower_params: Option<RawLower>,
3717     /// See `LiftResult`
3718     lift_result: Option<LiftResult>,
3719     /// A place to stash the type-erased lifted result if it can't be delivered
3720     /// immediately.
3721     result: Option<LiftedResult>,
3722     /// Closure to call the callback function for an async-lifted export, if
3723     /// provided.
3724     callback: Option<CallbackFn>,
3725     /// See `Caller`
3726     caller: Caller,
3727     /// A place to stash the call context for managing resource borrows while
3728     /// switching between guest tasks.
3729     call_context: Option<CallContext>,
3730     /// A place to stash the lowered result for a sync-to-async call until it
3731     /// can be returned to the caller.
3732     sync_result: Option<Option<ValRaw>>,
3733     /// Whether or not the task has been cancelled (i.e. whether the task is
3734     /// permitted to call `task.cancel`).
3735     cancel_sent: bool,
3736     /// Whether or not we've sent a `Status::Starting` event to any current or
3737     /// future waiters for this waitable.
3738     starting_sent: bool,
3739     /// Context-local state used to implement the `context.{get,set}`
3740     /// intrinsics.
3741     context: [u32; 2],
3742     /// Pending guest subtasks created by this task (directly or indirectly).
3743     ///
3744     /// This is used to re-parent subtasks which are still running when their
3745     /// parent task is disposed.
3746     subtasks: HashSet<TableId<GuestTask>>,
3747     /// Scratch waitable set used to watch subtasks during synchronous calls.
3748     sync_call_set: TableId<WaitableSet>,
3749     /// The instance to which the exported function for this guest task belongs.
3750     ///
3751     /// Note that the task may do a sync->sync call via a fused adapter which
3752     /// results in that task executing code in a different instance, and it may
3753     /// call host functions and intrinsics from that other instance.
3754     instance: RuntimeComponentInstanceIndex,
3755     /// If present, a pending `Event::None` or `Event::Cancelled` to be
3756     /// delivered to this task.
3757     event: Option<Event>,
3758     /// If present, indicates that the task is currently waiting on the
3759     /// specified set but may be cancelled and woken immediately.
3760     wake_on_cancel: Option<TableId<WaitableSet>>,
3761     /// The `ExportIndex` of the guest function being called, if known.
3762     function_index: Option<ExportIndex>,
3763     /// Whether or not the task has exited.
3764     exited: bool,
3765 }
3766 
3767 impl GuestTask {
3768     fn new(
3769         state: &mut ConcurrentState,
3770         lower_params: RawLower,
3771         lift_result: LiftResult,
3772         caller: Caller,
3773         callback: Option<CallbackFn>,
3774         component_instance: RuntimeComponentInstanceIndex,
3775     ) -> Result<Self> {
3776         let sync_call_set = state.push(WaitableSet::default())?;
3777 
3778         Ok(Self {
3779             common: WaitableCommon::default(),
3780             lower_params: Some(lower_params),
3781             lift_result: Some(lift_result),
3782             result: None,
3783             callback,
3784             caller,
3785             call_context: Some(CallContext::default()),
3786             sync_result: None,
3787             cancel_sent: false,
3788             starting_sent: false,
3789             context: [0u32; 2],
3790             subtasks: HashSet::new(),
3791             sync_call_set,
3792             instance: component_instance,
3793             event: None,
3794             wake_on_cancel: None,
3795             function_index: None,
3796             exited: false,
3797         })
3798     }
3799 
3800     /// Dispose of this guest task, reparenting any pending subtasks to the
3801     /// caller.
3802     fn dispose(self, state: &mut ConcurrentState, me: TableId<GuestTask>) -> Result<()> {
3803         // If there are not-yet-delivered completion events for subtasks in
3804         // `self.sync_call_set`, recursively dispose of those subtasks as well.
3805         for waitable in mem::take(&mut state.get_mut(self.sync_call_set)?.ready) {
3806             if let Some(Event::Subtask {
3807                 status: Status::Returned | Status::ReturnCancelled,
3808             }) = waitable.common(state)?.event
3809             {
3810                 waitable.delete_from(state)?;
3811             }
3812         }
3813 
3814         state.delete(self.sync_call_set)?;
3815 
3816         // Reparent any pending subtasks to the caller.
3817         if let Caller::Guest {
3818             task,
3819             instance: runtime_instance,
3820         } = &self.caller
3821         {
3822             let task_mut = state.get_mut(*task)?;
3823             let present = task_mut.subtasks.remove(&me);
3824             assert!(present);
3825 
3826             for subtask in &self.subtasks {
3827                 task_mut.subtasks.insert(*subtask);
3828             }
3829 
3830             for subtask in &self.subtasks {
3831                 state.get_mut(*subtask)?.caller = Caller::Guest {
3832                     task: *task,
3833                     instance: *runtime_instance,
3834                 };
3835             }
3836         } else {
3837             for subtask in &self.subtasks {
3838                 state.get_mut(*subtask)?.caller = Caller::Host {
3839                     tx: None,
3840                     remove_task_automatically: true,
3841                     call_post_return_automatically: true,
3842                 };
3843             }
3844         }
3845 
3846         Ok(())
3847     }
3848 
3849     fn call_post_return_automatically(&self) -> bool {
3850         matches!(
3851             self.caller,
3852             Caller::Guest { .. }
3853                 | Caller::Host {
3854                     call_post_return_automatically: true,
3855                     ..
3856                 }
3857         )
3858     }
3859 }
3860 
3861 impl TableDebug for GuestTask {
3862     fn type_name() -> &'static str {
3863         "GuestTask"
3864     }
3865 }
3866 
3867 /// Represents state common to all kinds of waitables.
3868 #[derive(Default)]
3869 struct WaitableCommon {
3870     /// The currently pending event for this waitable, if any.
3871     event: Option<Event>,
3872     /// The set to which this waitable belongs, if any.
3873     set: Option<TableId<WaitableSet>>,
3874 }
3875 
3876 /// Represents a Component Model Async `waitable`.
3877 #[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
3878 enum Waitable {
3879     /// A host task
3880     Host(TableId<HostTask>),
3881     /// A guest task
3882     Guest(TableId<GuestTask>),
3883     /// The read or write end of a stream or future
3884     Transmit(TableId<TransmitHandle>),
3885 }
3886 
3887 impl Waitable {
3888     /// Retrieve the `Waitable` corresponding to the specified guest-visible
3889     /// handle.
3890     fn from_instance(
3891         state: &mut ConcurrentState,
3892         caller_instance: RuntimeComponentInstanceIndex,
3893         waitable: u32,
3894     ) -> Result<Self> {
3895         let (waitable, state) =
3896             state.waitable_tables[caller_instance].get_mut_by_index(waitable)?;
3897 
3898         Ok(match state {
3899             WaitableState::HostTask => Waitable::Host(TableId::new(waitable)),
3900             WaitableState::GuestTask => Waitable::Guest(TableId::new(waitable)),
3901             WaitableState::Stream(..) | WaitableState::Future(..) => {
3902                 Waitable::Transmit(TableId::new(waitable))
3903             }
3904             _ => bail!("invalid waitable handle"),
3905         })
3906     }
3907 
3908     /// Retrieve the host-visible identifier for this `Waitable`.
3909     fn rep(&self) -> u32 {
3910         match self {
3911             Self::Host(id) => id.rep(),
3912             Self::Guest(id) => id.rep(),
3913             Self::Transmit(id) => id.rep(),
3914         }
3915     }
3916 
3917     /// Move this `Waitable` to the specified set (when `set` is `Some(_)`) or
3918     /// remove it from any set it may currently belong to (when `set` is
3919     /// `None`).
3920     fn join(&self, state: &mut ConcurrentState, set: Option<TableId<WaitableSet>>) -> Result<()> {
3921         let old = mem::replace(&mut self.common(state)?.set, set);
3922 
3923         if let Some(old) = old {
3924             match *self {
3925                 Waitable::Host(id) => state.remove_child(id, old),
3926                 Waitable::Guest(id) => state.remove_child(id, old),
3927                 Waitable::Transmit(id) => state.remove_child(id, old),
3928             }?;
3929 
3930             state.get_mut(old)?.ready.remove(self);
3931         }
3932 
3933         if let Some(set) = set {
3934             match *self {
3935                 Waitable::Host(id) => state.add_child(id, set),
3936                 Waitable::Guest(id) => state.add_child(id, set),
3937                 Waitable::Transmit(id) => state.add_child(id, set),
3938             }?;
3939 
3940             if self.common(state)?.event.is_some() {
3941                 self.mark_ready(state)?;
3942             }
3943         }
3944 
3945         Ok(())
3946     }
3947 
3948     /// Retrieve mutable access to the `WaitableCommon` for this `Waitable`.
3949     fn common<'a>(&self, state: &'a mut ConcurrentState) -> Result<&'a mut WaitableCommon> {
3950         Ok(match self {
3951             Self::Host(id) => &mut state.get_mut(*id)?.common,
3952             Self::Guest(id) => &mut state.get_mut(*id)?.common,
3953             Self::Transmit(id) => &mut state.get_mut(*id)?.common,
3954         })
3955     }
3956 
3957     /// Set or clear the pending event for this waitable and either deliver it
3958     /// to the first waiter, if any, or mark it as ready to be delivered to the
3959     /// next waiter that arrives.
3960     fn set_event(&self, state: &mut ConcurrentState, event: Option<Event>) -> Result<()> {
3961         log::trace!("set event for {self:?}: {event:?}");
3962         self.common(state)?.event = event;
3963         self.mark_ready(state)
3964     }
3965 
3966     /// Take the pending event from this waitable, leaving `None` in its place.
3967     fn take_event(&self, state: &mut ConcurrentState) -> Result<Option<Event>> {
3968         let common = self.common(state)?;
3969         let event = common.event.take();
3970         if let Some(set) = self.common(state)?.set {
3971             state.get_mut(set)?.ready.remove(self);
3972         }
3973         Ok(event)
3974     }
3975 
3976     /// Deliver the current event for this waitable to the first waiter, if any,
3977     /// or else mark it as ready to be delivered to the next waiter that
3978     /// arrives.
3979     fn mark_ready(&self, state: &mut ConcurrentState) -> Result<()> {
3980         if let Some(set) = self.common(state)?.set {
3981             state.get_mut(set)?.ready.insert(*self);
3982             if let Some((task, mode)) = state.get_mut(set)?.waiting.pop_first() {
3983                 let wake_on_cancel = state.get_mut(task)?.wake_on_cancel.take();
3984                 assert!(wake_on_cancel.is_none() || wake_on_cancel == Some(set));
3985 
3986                 let item = match mode {
3987                     WaitMode::Fiber(fiber) => WorkItem::ResumeFiber(fiber),
3988                     WaitMode::Callback(instance) => WorkItem::GuestCall(GuestCall {
3989                         task,
3990                         kind: GuestCallKind::DeliverEvent {
3991                             instance,
3992                             set: Some(set),
3993                         },
3994                     }),
3995                 };
3996                 state.push_high_priority(item);
3997             }
3998         }
3999         Ok(())
4000     }
4001 
4002     /// Handle the imminent delivery of the specified event, e.g. by updating
4003     /// the state of the stream or future.
4004     fn on_delivery(&self, state: &mut ConcurrentState, event: Event) {
4005         match event {
4006             Event::FutureRead {
4007                 pending: Some((ty, handle)),
4008                 ..
4009             }
4010             | Event::FutureWrite {
4011                 pending: Some((ty, handle)),
4012                 ..
4013             } => {
4014                 let runtime_instance = state.component.types()[ty].instance;
4015                 let (rep, WaitableState::Future(actual_ty, state)) = state.waitable_tables
4016                     [runtime_instance]
4017                     .get_mut_by_index(handle)
4018                     .unwrap()
4019                 else {
4020                     unreachable!()
4021                 };
4022                 assert_eq!(*actual_ty, ty);
4023                 assert_eq!(rep, self.rep());
4024                 assert_eq!(*state, StreamFutureState::Busy);
4025                 *state = match event {
4026                     Event::FutureRead { .. } => StreamFutureState::Read { done: false },
4027                     Event::FutureWrite { .. } => StreamFutureState::Write { done: false },
4028                     _ => unreachable!(),
4029                 };
4030             }
4031             Event::StreamRead {
4032                 pending: Some((ty, handle)),
4033                 code,
4034             }
4035             | Event::StreamWrite {
4036                 pending: Some((ty, handle)),
4037                 code,
4038             } => {
4039                 let runtime_instance = state.component.types()[ty].instance;
4040                 let (rep, WaitableState::Stream(actual_ty, state)) = state.waitable_tables
4041                     [runtime_instance]
4042                     .get_mut_by_index(handle)
4043                     .unwrap()
4044                 else {
4045                     unreachable!()
4046                 };
4047                 assert_eq!(*actual_ty, ty);
4048                 assert_eq!(rep, self.rep());
4049                 assert_eq!(*state, StreamFutureState::Busy);
4050                 let done = matches!(code, ReturnCode::Dropped(_));
4051                 *state = match event {
4052                     Event::StreamRead { .. } => StreamFutureState::Read { done },
4053                     Event::StreamWrite { .. } => StreamFutureState::Write { done },
4054                     _ => unreachable!(),
4055                 };
4056             }
4057             _ => {}
4058         }
4059     }
4060 
4061     /// Remove this waitable from the instance's rep table.
4062     fn delete_from(&self, state: &mut ConcurrentState) -> Result<()> {
4063         match self {
4064             Self::Host(task) => {
4065                 log::trace!("delete host task {task:?}");
4066                 state.delete(*task)?;
4067             }
4068             Self::Guest(task) => {
4069                 log::trace!("delete guest task {task:?}");
4070                 state.delete(*task)?.dispose(state, *task)?;
4071             }
4072             Self::Transmit(task) => {
4073                 state.delete(*task)?;
4074             }
4075         }
4076 
4077         Ok(())
4078     }
4079 }
4080 
4081 impl fmt::Debug for Waitable {
4082     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
4083         match self {
4084             Self::Host(id) => write!(f, "{id:?}"),
4085             Self::Guest(id) => write!(f, "{id:?}"),
4086             Self::Transmit(id) => write!(f, "{id:?}"),
4087         }
4088     }
4089 }
4090 
4091 /// Represents a Component Model Async `waitable-set`.
4092 #[derive(Default)]
4093 struct WaitableSet {
4094     /// Which waitables in this set have pending events, if any.
4095     ready: BTreeSet<Waitable>,
4096     /// Which guest tasks are currently waiting on this set, if any.
4097     waiting: BTreeMap<TableId<GuestTask>, WaitMode>,
4098 }
4099 
4100 impl TableDebug for WaitableSet {
4101     fn type_name() -> &'static str {
4102         "WaitableSet"
4103     }
4104 }
4105 
4106 /// Type-erased closure to lower the parameters for a guest task.
4107 type RawLower = Box<
4108     dyn FnOnce(&mut dyn VMStore, Instance, &mut [MaybeUninit<ValRaw>]) -> Result<()> + Send + Sync,
4109 >;
4110 
4111 /// Type-erased closure to lift the result for a guest task.
4112 type RawLift = Box<
4113     dyn FnOnce(&mut dyn VMStore, Instance, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>>
4114         + Send
4115         + Sync,
4116 >;
4117 
4118 /// Type erased result of a guest task which may be downcast to the expected
4119 /// type by a host caller (or simply ignored in the case of a guest caller; see
4120 /// `DummyResult`).
4121 type LiftedResult = Box<dyn Any + Send + Sync>;
4122 
4123 /// Used to return a result from a `LiftFn` when the actual result has already
4124 /// been lowered to a guest task's stack and linear memory.
4125 struct DummyResult;
4126 
4127 /// Represents the state of a currently executing fiber which has been resumed
4128 /// via `self::poll_fn`.
4129 pub(crate) struct AsyncState {
4130     /// The current instance being polled, if any, which is used to perform
4131     /// checks to ensure that futures are always polled within the correct
4132     /// instance.
4133     current_instance: Option<ComponentInstanceId>,
4134 }
4135 
4136 impl Default for AsyncState {
4137     fn default() -> Self {
4138         Self {
4139             current_instance: None,
4140         }
4141     }
4142 }
4143 
4144 /// Represents the Component Model Async state of a (sub-)component instance.
4145 #[derive(Default)]
4146 struct InstanceState {
4147     /// Whether backpressure is set for this instance
4148     backpressure: bool,
4149     /// Whether this instance can be entered
4150     do_not_enter: bool,
4151     /// Pending calls for this instance which require `Self::backpressure` to be
4152     /// `true` and/or `Self::do_not_enter` to be false before they can proceed.
4153     pending: BTreeMap<TableId<GuestTask>, GuestCallKind>,
4154 }
4155 
4156 /// Represents the Component Model Async state of a top-level component instance
4157 /// (i.e. a `super::ComponentInstance`).
4158 pub struct ConcurrentState {
4159     /// The currently running guest task, if any.
4160     guest_task: Option<TableId<GuestTask>>,
4161     /// The set of pending host and background tasks, if any.
4162     ///
4163     /// We must wrap this in a `Mutex` to ensure that `ComponentInstance` and
4164     /// `Store` satisfy a `Sync` bound, but it can't actually be accessed from
4165     /// more than one thread at a time.
4166     ///
4167     /// See `ComponentInstance::poll_until` for where we temporarily take this
4168     /// out, poll it, then put it back to avoid any mutable aliasing hazards.
4169     futures: Mutex<Option<FuturesUnordered<HostTaskFuture>>>,
4170     /// The table of waitables, waitable sets, etc.
4171     table: Table,
4172     /// Per (sub-)component instance states.
4173     ///
4174     /// See `InstanceState` for details and note that this map is lazily
4175     /// populated as needed.
4176     // TODO: this can and should be a `PrimaryMap`
4177     instance_states: HashMap<RuntimeComponentInstanceIndex, InstanceState>,
4178     /// Tables for tracking per-(sub-)component waitable handles and their
4179     /// states.
4180     waitable_tables: PrimaryMap<RuntimeComponentInstanceIndex, StateTable<WaitableState>>,
4181     /// The "high priority" work queue for this instance's event loop.
4182     high_priority: Vec<WorkItem>,
4183     /// The "high priority" work queue for this instance's event loop.
4184     low_priority: Vec<WorkItem>,
4185     /// A place to stash the reason a fiber is suspending so that the code which
4186     /// resumed it will know under what conditions the fiber should be resumed
4187     /// again.
4188     suspend_reason: Option<SuspendReason>,
4189     /// A cached fiber which is waiting for work to do.
4190     ///
4191     /// This helps us avoid creating a new fiber for each `GuestCall` work item.
4192     worker: Option<StoreFiber<'static>>,
4193     /// A place to stash the work item for which we're resuming a worker fiber.
4194     guest_call: Option<GuestCall>,
4195 
4196     /// (Sub)Component specific error context tracking
4197     ///
4198     /// At the component level, only the number of references (`usize`) to a given error context is tracked,
4199     /// with state related to the error context being held at the component model level, in concurrent
4200     /// state.
4201     ///
4202     /// The state tables in the (sub)component local tracking must contain a pointer into the global
4203     /// error context lookups in order to ensure that in contexts where only the local reference is present
4204     /// the global state can still be maintained/updated.
4205     error_context_tables:
4206         PrimaryMap<TypeComponentLocalErrorContextTableIndex, StateTable<LocalErrorContextRefCount>>,
4207 
4208     /// Reference counts for all component error contexts
4209     ///
4210     /// NOTE: it is possible the global ref count to be *greater* than the sum of
4211     /// (sub)component ref counts as tracked by `error_context_tables`, for
4212     /// example when the host holds one or more references to error contexts.
4213     ///
4214     /// The key of this primary map is often referred to as the "rep" (i.e. host-side
4215     /// component-wide representation) of the index into concurrent state for a given
4216     /// stored `ErrorContext`.
4217     ///
4218     /// Stated another way, `TypeComponentGlobalErrorContextTableIndex` is essentially the same
4219     /// as a `TableId<ErrorContextState>`.
4220     global_error_context_ref_counts:
4221         BTreeMap<TypeComponentGlobalErrorContextTableIndex, GlobalErrorContextRefCount>,
4222 
4223     /// Mirror of type information in `ComponentInstance`, placed here for
4224     /// convenience at the cost of an extra `Arc` clone.
4225     component: Component,
4226 }
4227 
4228 impl ConcurrentState {
4229     pub(crate) fn new(component: &Component) -> Self {
4230         let num_waitable_tables = component.env_component().num_runtime_component_instances;
4231         let num_error_context_tables = component.env_component().num_error_context_tables;
4232         let mut waitable_tables =
4233             PrimaryMap::with_capacity(usize::try_from(num_waitable_tables).unwrap());
4234         for _ in 0..num_waitable_tables {
4235             waitable_tables.push(StateTable::default());
4236         }
4237 
4238         let mut error_context_tables = PrimaryMap::<
4239             TypeComponentLocalErrorContextTableIndex,
4240             StateTable<LocalErrorContextRefCount>,
4241         >::with_capacity(num_error_context_tables);
4242         for _ in 0..num_error_context_tables {
4243             error_context_tables.push(StateTable::default());
4244         }
4245 
4246         Self {
4247             guest_task: None,
4248             table: Table::new(),
4249             futures: Mutex::new(Some(FuturesUnordered::new())),
4250             instance_states: HashMap::new(),
4251             waitable_tables,
4252             high_priority: Vec::new(),
4253             low_priority: Vec::new(),
4254             suspend_reason: None,
4255             worker: None,
4256             guest_call: None,
4257             error_context_tables,
4258             global_error_context_ref_counts: BTreeMap::new(),
4259             component: component.clone(),
4260         }
4261     }
4262 
4263     /// Take ownership of any fibers owned by this object.
4264     ///
4265     /// This should be used when disposing of the `Store` containing this object
4266     /// in order to gracefully resolve any and all fibers using
4267     /// `StoreFiber::dispose`.  This is necessary to avoid possible
4268     /// use-after-free bugs due to fibers which may still have access to the
4269     /// `Store`.
4270     ///
4271     /// Note that this will leave the object in an inconsistent and unusable
4272     /// state, so it should only be used just prior to dropping it.
4273     pub(crate) fn take_fibers(&mut self, vec: &mut Vec<StoreFiber<'static>>) {
4274         for entry in mem::take(&mut self.table) {
4275             if let Ok(set) = entry.downcast::<WaitableSet>() {
4276                 for mode in set.waiting.into_values() {
4277                     if let WaitMode::Fiber(fiber) = mode {
4278                         vec.push(fiber);
4279                     }
4280                 }
4281             }
4282         }
4283 
4284         if let Some(fiber) = self.worker.take() {
4285             vec.push(fiber);
4286         }
4287 
4288         let mut take_items = |list| {
4289             for item in mem::take(list) {
4290                 if let WorkItem::ResumeFiber(fiber) = item {
4291                     vec.push(fiber);
4292                 }
4293             }
4294         };
4295 
4296         take_items(&mut self.high_priority);
4297         take_items(&mut self.low_priority);
4298     }
4299 }
4300 
4301 /// Provide a type hint to compiler about the shape of a parameter lower
4302 /// closure.
4303 fn for_any_lower<
4304     F: FnOnce(&mut dyn VMStore, Instance, &mut [MaybeUninit<ValRaw>]) -> Result<()> + Send + Sync,
4305 >(
4306     fun: F,
4307 ) -> F {
4308     fun
4309 }
4310 
4311 /// Provide a type hint to compiler about the shape of a result lift closure.
4312 fn for_any_lift<
4313     F: FnOnce(&mut dyn VMStore, Instance, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>>
4314         + Send
4315         + Sync,
4316 >(
4317     fun: F,
4318 ) -> F {
4319     fun
4320 }
4321 
4322 /// Wrap the specified future in a `poll_fn` which asserts that the future is
4323 /// only polled from the event loop of the specified `Instance`.
4324 ///
4325 /// See `Instance::run_concurrent` for details.
4326 fn checked<F: Future + Send + 'static>(
4327     instance: Instance,
4328     fut: F,
4329 ) -> impl Future<Output = F::Output> + Send + 'static {
4330     async move {
4331         let mut fut = pin!(fut);
4332         future::poll_fn(move |cx| {
4333             let message = "\
4334                 `Future`s which depend on asynchronous component tasks, streams, or \
4335                 futures to complete may only be polled from the event loop of the \
4336                 instance from which they originated.  Please use \
4337                 `Instance::{run_concurrent,spawn}` to poll or await them.\
4338             ";
4339             tls::try_get(|store| {
4340                 let matched = match store {
4341                     tls::TryGet::Some(store) => {
4342                         let a = store.concurrent_async_state_mut().current_instance;
4343                         a == Some(instance.id().instance())
4344                     }
4345                     tls::TryGet::Taken | tls::TryGet::None => false,
4346                 };
4347 
4348                 if !matched {
4349                     panic!("{message}")
4350                 }
4351             });
4352             fut.as_mut().poll(cx)
4353         })
4354         .await
4355     }
4356 }
4357 
4358 /// Assert that `Instance::run_concurrent` has not been called from within an
4359 /// instance's event loop.
4360 fn check_recursive_run() {
4361     tls::try_get(|store| {
4362         if !matches!(store, tls::TryGet::None) {
4363             panic!("Recursive `Instance::run_concurrent` calls not supported")
4364         }
4365     });
4366 }
4367 
4368 fn unpack_callback_code(code: u32) -> (u32, u32) {
4369     (code & 0xF, code >> 4)
4370 }
4371 
4372 /// Helper struct for packaging parameters to be passed to
4373 /// `ComponentInstance::waitable_check` for calls to `waitable-set.wait` or
4374 /// `waitable-set.poll`.
4375 struct WaitableCheckParams {
4376     set: TableId<WaitableSet>,
4377     memory: *mut VMMemoryDefinition,
4378     payload: u32,
4379     caller_instance: RuntimeComponentInstanceIndex,
4380 }
4381 
4382 /// Helper enum for passing parameters to `ComponentInstance::waitable_check`.
4383 enum WaitableCheck {
4384     Wait(WaitableCheckParams),
4385     Poll(WaitableCheckParams),
4386     Yield,
4387 }
4388 
4389 /// Represents a guest task called from the host, prepared using `prepare_call`.
4390 pub(crate) struct PreparedCall<R> {
4391     /// The guest export to be called
4392     handle: Func,
4393     /// The guest task created by `prepare_call`
4394     task: TableId<GuestTask>,
4395     /// The number of lowered core Wasm parameters to pass to the call.
4396     param_count: usize,
4397     /// The `oneshot::Receiver` to which the result of the call will be
4398     /// delivered when it is available.
4399     rx: oneshot::Receiver<LiftedResult>,
4400     _phantom: PhantomData<R>,
4401 }
4402 
4403 impl<R> PreparedCall<R> {
4404     /// Get a copy of the `TaskId` for this `PreparedCall`.
4405     pub(crate) fn task_id(&self) -> TaskId {
4406         TaskId {
4407             handle: self.handle,
4408             task: self.task,
4409         }
4410     }
4411 }
4412 
4413 /// Represents a task created by `prepare_call`.
4414 pub(crate) struct TaskId {
4415     handle: Func,
4416     task: TableId<GuestTask>,
4417 }
4418 
4419 impl TaskId {
4420     /// Remove the specified task from the concurrent state to which it belongs.
4421     ///
4422     /// This must be used with care to avoid use-after-delete or double-delete
4423     /// bugs.  Specifically, it should only be called on tasks created with the
4424     /// `remove_task_automatically` parameter to `prepare_call` set to `false`,
4425     /// which tells the runtime that the caller is responsible for removing the
4426     /// task from the state; otherwise, it will be removed automatically.  Also,
4427     /// it should only be called once for a given task, and only after either
4428     /// the task has completed or the instance has trapped.
4429     pub(crate) fn remove<T>(&self, store: StoreContextMut<T>) -> Result<()> {
4430         Waitable::Guest(self.task).delete_from(self.handle.instance().concurrent_state_mut(store.0))
4431     }
4432 }
4433 
4434 /// Prepare a call to the specified exported Wasm function, providing functions
4435 /// for lowering the parameters and lifting the result.
4436 ///
4437 /// To enqueue the returned `PreparedCall` in the `ComponentInstance`'s event
4438 /// loop, use `queue_call`.
4439 pub(crate) fn prepare_call<T, R>(
4440     mut store: StoreContextMut<T>,
4441     handle: Func,
4442     param_count: usize,
4443     remove_task_automatically: bool,
4444     call_post_return_automatically: bool,
4445     lower_params: impl FnOnce(Func, StoreContextMut<T>, &mut [MaybeUninit<ValRaw>]) -> Result<()>
4446     + Send
4447     + Sync
4448     + 'static,
4449     lift_result: impl FnOnce(Func, &mut StoreOpaque, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>>
4450     + Send
4451     + Sync
4452     + 'static,
4453 ) -> Result<PreparedCall<R>> {
4454     let (options, _flags, ty, raw_options) = handle.abi_info(store.0);
4455 
4456     let instance = handle.instance().id().get(store.0);
4457     let task_return_type = instance.component().types()[ty].results;
4458     let component_instance = raw_options.instance;
4459     let callback = options.callback();
4460     let memory = options.memory_raw().map(SendSyncPtr::new);
4461     let string_encoding = options.string_encoding();
4462     let token = StoreToken::new(store.as_context_mut());
4463     let state = handle.instance().concurrent_state_mut(store.0);
4464 
4465     assert!(state.guest_task.is_none());
4466 
4467     let (tx, rx) = oneshot::channel();
4468 
4469     let mut task = GuestTask::new(
4470         state,
4471         Box::new(for_any_lower(move |store, instance, params| {
4472             debug_assert!(instance.id() == handle.instance().id());
4473             lower_params(handle, token.as_context_mut(store), params)
4474         })),
4475         LiftResult {
4476             lift: Box::new(for_any_lift(move |store, instance, result| {
4477                 debug_assert!(instance.id() == handle.instance().id());
4478                 lift_result(handle, store, result)
4479             })),
4480             ty: task_return_type,
4481             memory,
4482             string_encoding,
4483         },
4484         Caller::Host {
4485             tx: Some(tx),
4486             remove_task_automatically,
4487             call_post_return_automatically,
4488         },
4489         callback.map(|callback| {
4490             let callback = SendSyncPtr::new(callback);
4491             Box::new(
4492                 move |store: &mut dyn VMStore,
4493                       instance: Instance,
4494                       runtime_instance,
4495                       event,
4496                       handle| {
4497                     let store = token.as_context_mut(store);
4498                     // SAFETY: Per the contract of `prepare_call`, the callback
4499                     // will remain valid at least as long is this task exists.
4500                     unsafe {
4501                         instance.call_callback(
4502                             store,
4503                             runtime_instance,
4504                             callback,
4505                             event,
4506                             handle,
4507                             call_post_return_automatically,
4508                         )
4509                     }
4510                 },
4511             ) as CallbackFn
4512         }),
4513         component_instance,
4514     )?;
4515     task.function_index = Some(handle.index());
4516 
4517     let task = state.push(task)?;
4518 
4519     Ok(PreparedCall {
4520         handle,
4521         task,
4522         param_count,
4523         rx,
4524         _phantom: PhantomData,
4525     })
4526 }
4527 
4528 /// Queue a call previously prepared using `prepare_call` to be run as part of
4529 /// the associated `ComponentInstance`'s event loop.
4530 ///
4531 /// The returned future will resolve to the result once it is available, but
4532 /// must only be polled via the instance's event loop. See
4533 /// `Instance::run_concurrent` for details.
4534 pub(crate) fn queue_call<T: 'static, R: Send + 'static>(
4535     mut store: StoreContextMut<T>,
4536     prepared: PreparedCall<R>,
4537 ) -> Result<impl Future<Output = Result<R>> + Send + 'static + use<T, R>> {
4538     let PreparedCall {
4539         handle,
4540         task,
4541         param_count,
4542         rx,
4543         ..
4544     } = prepared;
4545 
4546     queue_call0(store.as_context_mut(), handle, task, param_count)?;
4547 
4548     Ok(checked(
4549         handle.instance(),
4550         rx.map(|result| {
4551             result
4552                 .map(|v| *v.downcast().unwrap())
4553                 .map_err(anyhow::Error::from)
4554         }),
4555     ))
4556 }
4557 
4558 /// Queue a call previously prepared using `prepare_call` to be run as part of
4559 /// the associated `ComponentInstance`'s event loop.
4560 fn queue_call0<T: 'static>(
4561     store: StoreContextMut<T>,
4562     handle: Func,
4563     guest_task: TableId<GuestTask>,
4564     param_count: usize,
4565 ) -> Result<()> {
4566     let (options, flags, _ty, raw_options) = handle.abi_info(store.0);
4567     let is_concurrent = raw_options.async_;
4568     let instance = handle.instance();
4569     let callee = handle.lifted_core_func(store.0);
4570     let callback = options.callback();
4571     let post_return = handle.post_return_core_func(store.0);
4572 
4573     log::trace!("queueing call {guest_task:?}");
4574 
4575     let instance_flags = if callback.is_none() {
4576         None
4577     } else {
4578         Some(flags)
4579     };
4580 
4581     // SAFETY: `callee`, `callback`, and `post_return` are valid pointers
4582     // (with signatures appropriate for this call) and will remain valid as
4583     // long as this instance is valid.
4584     unsafe {
4585         instance.queue_call(
4586             store,
4587             guest_task,
4588             SendSyncPtr::new(callee),
4589             param_count,
4590             1,
4591             instance_flags,
4592             is_concurrent,
4593             callback.map(SendSyncPtr::new),
4594             post_return.map(SendSyncPtr::new),
4595         )
4596     }
4597 }
4598