Skip to main content

wasmtime/runtime/component/
concurrent.rs

1//! Runtime support for the Component Model Async ABI.
2//!
3//! This module and its submodules provide host runtime support for Component
4//! Model Async features such as async-lifted exports, async-lowered imports,
5//! streams, futures, and related intrinsics.  See [the Async
6//! Explainer](https://github.com/WebAssembly/component-model/blob/main/design/mvp/Concurrency.md)
7//! for a high-level overview.
8//!
9//! At the core of this support is an event loop which schedules and switches
10//! between guest tasks and any host tasks they create.  Each
11//! `Store` will have at most one event loop running at any given
12//! time, and that loop may be suspended and resumed by the host embedder using
13//! e.g. `StoreContextMut::run_concurrent`.  The `StoreContextMut::poll_until`
14//! function contains the loop itself, while the
15//! `StoreOpaque::concurrent_state` field holds its state.
16//!
17//! # Public API Overview
18//!
19//! ## Top-level API (e.g. kicking off host->guest calls and driving the event loop)
20//!
21//! - `[Typed]Func::call_concurrent`: Start a host->guest call to an
22//! async-lifted or sync-lifted import, creating a guest task.
23//!
24//! - `StoreContextMut::run_concurrent`: Run the event loop for the specified
25//! instance, allowing any and all tasks belonging to that instance to make
26//! progress.
27//!
28//! - `StoreContextMut::spawn`: Run a background task as part of the event loop
29//! for the specified instance.
30//!
31//! - `{Future,Stream}Reader::new`: Create a new Component Model `future` or
32//! `stream` which may be passed to the guest.  This takes a
33//! `{Future,Stream}Producer` implementation which will be polled for items when
34//! the consumer requests them.
35//!
36//! - `{Future,Stream}Reader::pipe`: Consume a `future` or `stream` by
37//! connecting it to a `{Future,Stream}Consumer` which will consume any items
38//! produced by the write end.
39//!
40//! ## Host Task API (e.g. implementing concurrent host functions and background tasks)
41//!
42//! - `LinkerInstance::func_wrap_concurrent`: Register a concurrent host
43//! function with the linker.  That function will take an `Accessor` as its
44//! first parameter, which provides access to the store between (but not across)
45//! await points.
46//!
47//! - `Accessor::with`: Access the store and its associated data.
48//!
49//! - `Accessor::spawn`: Run a background task as part of the event loop for the
50//! store.  This is equivalent to `StoreContextMut::spawn` but more convenient to use
51//! in host functions.
52
53use self::error_contexts::GlobalErrorContextRefCount;
54use crate::component::func::{Func, call_post_return};
55use crate::component::{
56    HasData, HasSelf, Instance, Resource, ResourceTable, ResourceTableError, RuntimeInstance,
57};
58use crate::fiber::{self, StoreFiber, StoreFiberYield};
59use crate::hash_set::HashSet;
60#[cfg(feature = "gc")]
61use crate::module::ModuleRegistry;
62use crate::prelude::*;
63use crate::store::{Store, StoreId, StoreInner, StoreOpaque, StoreToken};
64#[cfg(feature = "gc")]
65use crate::vm::GcRootsList;
66use crate::vm::component::{CallContext, ComponentInstance, CurrentScope, InstanceState, Scope};
67use crate::vm::{AlwaysMut, SendSyncPtr, VMFuncRef, VMLazyThread, VMMemoryDefinition, VMStore};
68use crate::{
69    AsContext, AsContextMut, FuncType, Result, StoreContext, StoreContextMut, ValRaw, ValType, bail,
70};
71use crate::{Instance as ModuleInstance, bail_bug};
72use alloc::borrow::ToOwned;
73use alloc::collections::{BTreeMap, BTreeSet, VecDeque};
74use core::any::Any;
75use core::cell::UnsafeCell;
76use core::fmt;
77use core::future;
78use core::future::Future;
79use core::marker::PhantomData;
80use core::mem::{self, ManuallyDrop, MaybeUninit};
81use core::ops::DerefMut;
82use core::pin::{Pin, pin};
83use core::ptr::{self, NonNull};
84use core::task::{Context, Poll, Waker};
85use futures::channel::oneshot;
86use futures::stream::{FuturesUnordered, StreamExt};
87use futures_and_streams::{FlatAbi, ReturnCode, TransmitHandle, TransmitIndex};
88use table::{TableDebug, TableId};
89use wasmtime_environ::component::{
90    CanonicalAbiInfo, CanonicalOptions, CanonicalOptionsDataModel, MAX_FLAT_PARAMS,
91    MAX_FLAT_RESULTS, OptionsIndex, PREPARE_ASYNC_NO_RESULT, PREPARE_ASYNC_WITH_RESULT,
92    RuntimeComponentInstanceIndex, RuntimeTableIndex, StringEncoding,
93    TypeComponentGlobalErrorContextTableIndex, TypeComponentLocalErrorContextTableIndex,
94    TypeFuncIndex, TypeFutureTableIndex, TypeStreamTableIndex, TypeTupleIndex,
95};
96use wasmtime_environ::packed_option::ReservedValue;
97use wasmtime_environ::{NUM_COMPONENT_CONTEXT_SLOTS, Trap};
98#[cfg(feature = "gc")]
99use wasmtime_unwinder::Unwind;
100
101pub use abort::JoinHandle;
102pub use func::{FuncCallConcurrent, TypedFuncCallConcurrent};
103pub use future_stream_any::{FutureAny, StreamAny};
104pub use futures_and_streams::{
105    Destination, DirectDestination, DirectSource, ErrorContext, FutureConsumer, FutureProducer,
106    FutureReader, GuardedFutureReader, GuardedStreamReader, ReadBuffer, Source, StreamConsumer,
107    StreamProducer, StreamReader, StreamResult, VecBuffer, WriteBuffer,
108};
109pub(crate) use futures_and_streams::{ResourcePair, lower_error_context_to_index};
110
111mod abort;
112mod error_contexts;
113mod func;
114mod future_stream_any;
115mod futures_and_streams;
116pub(crate) mod table;
117pub(crate) mod tls;
118
119/// Constant defined in the Component Model spec to indicate that the async
120/// intrinsic (e.g. `future.write`) has not yet completed.
121const BLOCKED: u32 = 0xffff_ffff;
122
123/// Corresponds to `CallState` in the upstream spec.
124#[derive(Clone, Copy, Eq, PartialEq, Debug)]
125pub enum Status {
126    Starting = 0,
127    Started = 1,
128    Returned = 2,
129    StartCancelled = 3,
130    ReturnCancelled = 4,
131}
132
133impl Status {
134    /// Packs this status and the optional `waitable` provided into a 32-bit
135    /// result that the canonical ABI requires.
136    ///
137    /// The low 4 bits are reserved for the status while the upper 28 bits are
138    /// the waitable, if present.
139    pub fn pack(self, waitable: Option<u32>) -> u32 {
140        assert!(matches!(self, Status::Returned) == waitable.is_none());
141        let waitable = waitable.unwrap_or(0);
142        assert!(waitable < (1 << 28));
143        (waitable << 4) | (self as u32)
144    }
145}
146
147/// Corresponds to `EventCode` in the Component Model spec, plus related payload
148/// data.
149#[derive(Clone, Copy, Debug)]
150enum Event {
151    None,
152    Subtask {
153        status: Status,
154    },
155    StreamRead {
156        code: ReturnCode,
157        pending: Option<(TypeStreamTableIndex, u32)>,
158    },
159    StreamWrite {
160        code: ReturnCode,
161        pending: Option<(TypeStreamTableIndex, u32)>,
162    },
163    FutureRead {
164        code: ReturnCode,
165        pending: Option<(TypeFutureTableIndex, u32)>,
166    },
167    FutureWrite {
168        code: ReturnCode,
169        pending: Option<(TypeFutureTableIndex, u32)>,
170    },
171    Cancelled,
172}
173
174impl Event {
175    /// Lower this event to core Wasm integers for delivery to the guest.
176    ///
177    /// Note that the waitable handle, if any, is assumed to be lowered
178    /// separately.
179    fn parts(self) -> (u32, u32) {
180        const EVENT_NONE: u32 = 0;
181        const EVENT_SUBTASK: u32 = 1;
182        const EVENT_STREAM_READ: u32 = 2;
183        const EVENT_STREAM_WRITE: u32 = 3;
184        const EVENT_FUTURE_READ: u32 = 4;
185        const EVENT_FUTURE_WRITE: u32 = 5;
186        const EVENT_CANCELLED: u32 = 6;
187        match self {
188            Event::None => (EVENT_NONE, 0),
189            Event::Cancelled => (EVENT_CANCELLED, 0),
190            Event::Subtask { status } => (EVENT_SUBTASK, status as u32),
191            Event::StreamRead { code, .. } => (EVENT_STREAM_READ, code.encode()),
192            Event::StreamWrite { code, .. } => (EVENT_STREAM_WRITE, code.encode()),
193            Event::FutureRead { code, .. } => (EVENT_FUTURE_READ, code.encode()),
194            Event::FutureWrite { code, .. } => (EVENT_FUTURE_WRITE, code.encode()),
195        }
196    }
197}
198
199/// Corresponds to `CallbackCode` in the spec.
200mod callback_code {
201    pub const EXIT: u32 = 0;
202    pub const YIELD: u32 = 1;
203    pub const WAIT: u32 = 2;
204}
205
206/// A flag indicating that the callee is an async-lowered export.
207///
208/// This may be passed to the `async-start` intrinsic from a fused adapter.
209const START_FLAG_ASYNC_CALLEE: u32 = wasmtime_environ::component::START_FLAG_ASYNC_CALLEE as u32;
210
211/// Provides access to either store data (via the `get` method) or the store
212/// itself (via [`AsContext`]/[`AsContextMut`]), as well as the component
213/// instance to which the current host task belongs.
214///
215/// See [`Accessor::with`] for details.
216pub struct Access<'a, T: 'static, D: HasData + ?Sized = HasSelf<T>> {
217    store: StoreContextMut<'a, T>,
218    get_data: fn(&mut T) -> D::Data<'_>,
219}
220
221impl<'a, T, D> Access<'a, T, D>
222where
223    D: HasData + ?Sized,
224    T: 'static,
225{
226    /// Creates a new [`Access`] from its component parts.
227    pub fn new(store: StoreContextMut<'a, T>, get_data: fn(&mut T) -> D::Data<'_>) -> Self {
228        Self { store, get_data }
229    }
230
231    /// Get mutable access to the store data.
232    pub fn data_mut(&mut self) -> &mut T {
233        self.store.data_mut()
234    }
235
236    /// Get mutable access to the store data.
237    pub fn get(&mut self) -> D::Data<'_> {
238        (self.get_data)(self.data_mut())
239    }
240
241    /// Spawn a background task.
242    ///
243    /// See [`Accessor::spawn`] for details.
244    pub fn spawn(&mut self, task: impl AccessorTask<T, D>) -> Result<JoinHandle>
245    where
246        T: 'static,
247    {
248        let accessor = Accessor {
249            get_data: self.get_data,
250            token: StoreToken::new(self.store.as_context_mut()),
251        };
252        self.store
253            .as_context_mut()
254            .spawn_with_accessor(accessor, task)
255    }
256
257    /// Returns the getter this accessor is using to project from `T` into
258    /// `D::Data`.
259    pub fn getter(&self) -> fn(&mut T) -> D::Data<'_> {
260        self.get_data
261    }
262}
263
264impl<'a, T, D> AsContext for Access<'a, T, D>
265where
266    D: HasData + ?Sized,
267    T: 'static,
268{
269    type Data = T;
270
271    fn as_context(&self) -> StoreContext<'_, T> {
272        self.store.as_context()
273    }
274}
275
276impl<'a, T, D> AsContextMut for Access<'a, T, D>
277where
278    D: HasData + ?Sized,
279    T: 'static,
280{
281    fn as_context_mut(&mut self) -> StoreContextMut<'_, T> {
282        self.store.as_context_mut()
283    }
284}
285
286/// Provides scoped mutable access to store data in the context of a concurrent
287/// host task future.
288///
289/// This allows multiple host task futures to execute concurrently and access
290/// the store between (but not across) `await` points.
291///
292/// # Rationale
293///
294/// This structure is sort of like `&mut T` plus a projection from `&mut T` to
295/// `D::Data<'_>`. The problem this is solving, however, is that it does not
296/// literally store these values. The basic problem is that when a concurrent
297/// host future is being polled it has access to `&mut T` (and the whole
298/// `Store`) but when it's not being polled it does not have access to these
299/// values. This reflects how the store is only ever polling one future at a
300/// time so the store is effectively being passed between futures.
301///
302/// Rust's `Future` trait, however, has no means of passing a `Store`
303/// temporarily between futures. The [`Context`](core::task::Context) type does
304/// not have the ability to attach arbitrary information to it at this time.
305/// This type, [`Accessor`], is used to bridge this expressivity gap.
306///
307/// The [`Accessor`] type here represents the ability to acquire, temporarily in
308/// a synchronous manner, the current store. The [`Accessor::with`] function
309/// yields an [`Access`] which can be used to access [`StoreContextMut`], `&mut
310/// T`, or `D::Data<'_>`. Note though that [`Accessor::with`] intentionally does
311/// not take an `async` closure as its argument, instead it's a synchronous
312/// closure which must complete during on run of `Future::poll`. This reflects
313/// how the store is temporarily made available while a host future is being
314/// polled.
315///
316/// # Implementation
317///
318/// This type does not actually store `&mut T` nor `StoreContextMut<T>`, and
319/// this type additionally doesn't even have a lifetime parameter. This is
320/// instead a representation of proof of the ability to acquire these while a
321/// future is being polled. Wasmtime will, when it polls a host future,
322/// configure ambient state such that the `Accessor` that a future closes over
323/// will work and be able to access the store.
324///
325/// This has a number of implications for users such as:
326///
327/// * It's intentional that `Accessor` cannot be cloned, it needs to stay within
328///   the lifetime of a single future.
329/// * A future is expected to, however, close over an `Accessor` and keep it
330///   alive probably for the duration of the entire future.
331/// * Different host futures will be given different `Accessor`s, and that's
332///   intentional.
333/// * The `Accessor` type is `Send` and `Sync` irrespective of `T` which
334///   alleviates some otherwise required bounds to be written down.
335///
336/// # Using `Accessor` in `Drop`
337///
338/// The methods on `Accessor` are only expected to work in the context of
339/// `Future::poll` and are not guaranteed to work in `Drop`. This is because a
340/// host future can be dropped at any time throughout the system and Wasmtime
341/// store context is not necessarily available at that time. It's recommended to
342/// not use `Accessor` methods in anything connected to a `Drop` implementation
343/// as they will panic and have unintended results. If you run into this though
344/// feel free to file an issue on the Wasmtime repository.
345pub struct Accessor<T: 'static, D = HasSelf<T>>
346where
347    D: HasData + ?Sized,
348{
349    token: StoreToken<T>,
350    get_data: fn(&mut T) -> D::Data<'_>,
351}
352
353/// A helper trait to take any type of accessor-with-data in functions.
354///
355/// This trait is similar to [`AsContextMut`] except that it's used when
356/// working with an [`Accessor`] instead of a [`StoreContextMut`]. The
357/// [`Accessor`] is the main type used in concurrent settings and is passed to
358/// functions such as [`Func::call_concurrent`].
359///
360/// This trait is implemented for [`Accessor`] and `&T` where `T` implements
361/// this trait. This effectively means that regardless of the `D` in
362/// `Accessor<T, D>` it can still be passed to a function which just needs a
363/// store accessor.
364///
365/// Acquiring an [`Accessor`] can be done through
366/// [`StoreContextMut::run_concurrent`] for example or in a host function
367/// through
368/// [`Linker::func_wrap_concurrent`](crate::component::LinkerInstance::func_wrap_concurrent).
369pub trait AsAccessor {
370    /// The `T` in `Store<T>` that this accessor refers to.
371    type Data: 'static;
372
373    /// The `D` in `Accessor<T, D>`, or the projection out of
374    /// `Self::Data`.
375    type AccessorData: HasData + ?Sized;
376
377    /// Returns the accessor that this is referring to.
378    fn as_accessor(&self) -> &Accessor<Self::Data, Self::AccessorData>;
379}
380
381impl<T: AsAccessor + ?Sized> AsAccessor for &T {
382    type Data = T::Data;
383    type AccessorData = T::AccessorData;
384
385    fn as_accessor(&self) -> &Accessor<Self::Data, Self::AccessorData> {
386        T::as_accessor(self)
387    }
388}
389
390impl<T, D: HasData + ?Sized> AsAccessor for Accessor<T, D> {
391    type Data = T;
392    type AccessorData = D;
393
394    fn as_accessor(&self) -> &Accessor<T, D> {
395        self
396    }
397}
398
399// Note that it is intentional at this time that `Accessor` does not actually
400// store `&mut T` or anything similar. This distinctly enables the `Accessor`
401// structure to be both `Send` and `Sync` regardless of what `T` is (or `D` for
402// that matter). This is used to ergonomically simplify bindings where the
403// majority of the time `Accessor` is closed over in a future which then needs
404// to be `Send` and `Sync`. To avoid needing to write `T: Send` everywhere (as
405// you already have to write `T: 'static`...) it helps to avoid this.
406//
407// Note as well that `Accessor` doesn't actually store its data at all. Instead
408// it's more of a "proof" of what can be accessed from TLS. API design around
409// `Accessor` and functions like `Linker::func_wrap_concurrent` are
410// intentionally made to ensure that `Accessor` is ideally only used in the
411// context that TLS variables are actually set. For example host functions are
412// given `&Accessor`, not `Accessor`, and this prevents them from persisting
413// the value outside of a future. Within the future the TLS variables are all
414// guaranteed to be set while the future is being polled.
415//
416// Finally though this is not an ironclad guarantee, but nor does it need to be.
417// The TLS APIs are designed to panic or otherwise model usage where they're
418// called recursively or similar. It's hoped that code cannot be constructed to
419// actually hit this at runtime but this is not a safety requirement at this
420// time.
421const _: () = {
422    const fn assert<T: Send + Sync>() {}
423    assert::<Accessor<UnsafeCell<u32>>>();
424};
425
426impl<T> Accessor<T> {
427    /// Creates a new `Accessor` backed by the specified functions.
428    ///
429    /// - `get`: used to retrieve the store
430    ///
431    /// - `get_data`: used to "project" from the store's associated data to
432    /// another type (e.g. a field of that data or a wrapper around it).
433    ///
434    /// - `spawn`: used to queue spawned background tasks to be run later
435    pub(crate) fn new(token: StoreToken<T>) -> Self {
436        Self {
437            token,
438            get_data: |x| x,
439        }
440    }
441}
442
443impl<T, D> Accessor<T, D>
444where
445    D: HasData + ?Sized,
446{
447    /// Run the specified closure, passing it mutable access to the store.
448    ///
449    /// This function is one of the main building blocks of the [`Accessor`]
450    /// type. This yields synchronous, blocking, access to the store via an
451    /// [`Access`]. The [`Access`] implements [`AsContextMut`] in addition to
452    /// providing the ability to access `D` via [`Access::get`]. Note that the
453    /// `fun` here is given only temporary access to the store and `T`/`D`
454    /// meaning that the return value `R` here is not allowed to capture borrows
455    /// into the two. If access is needed to data within `T` or `D` outside of
456    /// this closure then it must be `clone`d out, for example.
457    ///
458    /// # Panics
459    ///
460    /// This function will panic if it is call recursively with any other
461    /// accessor already in scope. For example if `with` is called within `fun`,
462    /// then this function will panic. It is up to the embedder to ensure that
463    /// this does not happen.
464    pub fn with<R>(&self, fun: impl FnOnce(Access<'_, T, D>) -> R) -> R {
465        tls::get(|vmstore| {
466            fun(Access {
467                store: self.token.as_context_mut(vmstore),
468                get_data: self.get_data,
469            })
470        })
471    }
472
473    /// Returns the getter this accessor is using to project from `T` into
474    /// `D::Data`.
475    pub fn getter(&self) -> fn(&mut T) -> D::Data<'_> {
476        self.get_data
477    }
478
479    /// Changes this accessor to access `D2` instead of the current type
480    /// parameter `D`.
481    ///
482    /// This changes the underlying data access from `T` to `D2::Data<'_>`.
483    ///
484    /// # Panics
485    ///
486    /// When using this API the returned value is disconnected from `&self` and
487    /// the lifetime binding the `self` argument. An `Accessor` only works
488    /// within the context of the closure or async closure that it was
489    /// originally given to, however. This means that due to the fact that the
490    /// returned value has no lifetime connection it's possible to use the
491    /// accessor outside of `&self`, the original accessor, and panic.
492    ///
493    /// The returned value should only be used within the scope of the original
494    /// `Accessor` that `self` refers to.
495    pub fn with_getter<D2: HasData>(
496        &self,
497        get_data: fn(&mut T) -> D2::Data<'_>,
498    ) -> Accessor<T, D2> {
499        Accessor {
500            token: self.token,
501            get_data,
502        }
503    }
504
505    /// Spawn a background task which will receive an `&Accessor<T, D>` and
506    /// run concurrently with any other tasks in progress for the current
507    /// store.
508    ///
509    /// This is particularly useful for host functions which return a `stream`
510    /// or `future` such that the code to write to the write end of that
511    /// `stream` or `future` must run after the function returns.
512    ///
513    /// The returned [`JoinHandle`] may be used to cancel the task.
514    ///
515    /// # Panics
516    ///
517    /// Panics if called within a closure provided to the [`Accessor::with`]
518    /// function. This can only be called outside an active invocation of
519    /// [`Accessor::with`].
520    pub fn spawn(&self, task: impl AccessorTask<T, D>) -> Result<JoinHandle>
521    where
522        T: 'static,
523    {
524        let accessor = self.clone_for_spawn();
525        self.with(|mut access| access.as_context_mut().spawn_with_accessor(accessor, task))
526    }
527
528    fn clone_for_spawn(&self) -> Self {
529        Self {
530            token: self.token,
531            get_data: self.get_data,
532        }
533    }
534
535    /// Polls to see if this store contains any "interesting" tasks still within
536    /// it.
537    ///
538    /// Returns `Poll::Ready(())` if there are no more interesting tasks, and
539    /// otherwise returns `Poll::Pending`. If pending is returned then whenever
540    /// the last remaining "interesting" task has exited the provided context's
541    /// waker will be notified. Note that only the waker passed to the last call
542    /// to `poll_no_interesting_tasks` for the store will be notified, so this
543    /// is only appropriate to use once-at-a-time per store.
544    ///
545    /// The component model specification, as of this current date, does not
546    /// have a distinction between "interesting" tasks and not. The current
547    /// intention is that in a future revision of the component model this will
548    /// be distinguished at the component ABI level where tasks will be able to
549    /// flag themselves as "interesting" optionally. Additionally extra work can
550    /// be opted-in to being "interesting".
551    ///
552    /// For now what this means is that all component model tasks within this
553    /// store are considered interesting. This specifically includes the entire
554    /// duration of a task, so even all of the time after a task has returned
555    /// but before it has exited. This means that this function is, today,
556    /// effectively a proxy for "are there any more tasks still running in this
557    /// store". This can be used by embedders to determine whether there's any
558    /// more work going on, even in the background, for a particular guest.
559    /// Hosts can use this as a signal that the guest wants to stay alive a
560    /// little longer, even after a task has returned.
561    ///
562    /// In the future this predicate won't include all tasks in this store. Some
563    /// tasks will be able to flag themselves as not interesting, meaning that
564    /// when this returns ready it'd be possible that there are still tasks
565    /// remaining in the store.
566    ///
567    /// Note that at this time spawned threads within a task are always
568    /// considered uninteresting. If this function returns ready, then spawned
569    /// threads may still be in the store.
570    pub fn poll_no_interesting_tasks(&self, cx: &mut Context<'_>) -> Poll<()> {
571        self.with(|mut access| {
572            let store = access.as_context_mut().0;
573            let state = store.concurrent_state_mut_without_forcing_current_thread();
574            if state.interesting_tasks == 0 {
575                Poll::Ready(())
576            } else {
577                state.interesting_tasks_empty_waker = Some(cx.waker().clone());
578                Poll::Pending
579            }
580        })
581    }
582
583    /// Poll to see if the component instance corresponding to the specified
584    /// function is ready to run a concurrent call without queuing it (i.e. does
585    /// not have backpressure enabled and does not have a sync call in
586    /// progress).
587    ///
588    /// Returns `Poll::Ready(())` if the component instance is ready to run a
589    /// concurrent call, and otherwise returns `Poll::Pending`.  If pending is
590    /// returned then whenever the instance becomes ready for a call the
591    /// provided context's waker will be notified.  Note that only the waker
592    /// passed to the last call to `poll_ready_for_concurrent_call` for the
593    /// store will be notified (regardless of whether the same or different
594    /// `Func` is specified relative to earlier calls), so this is only
595    /// appropriate to use once-at-a-time per store.  Also note that the waker
596    /// may be notified when _any_ instance becomes callable (i.e. not
597    /// necessarily the last one polled), so this function must be called again
598    /// to determine if the instance of interest is ready.
599    pub fn poll_ready_for_concurrent_call(&self, func: Func, cx: &mut Context<'_>) -> Poll<()> {
600        self.with(|mut access| {
601            let store = access.as_context_mut().0;
602            let (_, _, _, raw_options) = func.abi_info(store);
603            let instance = func.instance().runtime_instance(raw_options.instance);
604            let state = store.instance_state(instance).concurrent_state();
605            if state.backpressure == 0 {
606                Poll::Ready(())
607            } else {
608                store
609                    .concurrent_state_mut_without_forcing_current_thread()
610                    .ready_for_concurrent_call_waker = Some(cx.waker().clone());
611                Poll::Pending
612            }
613        })
614    }
615}
616
617/// Represents a task which may be provided to `Accessor::spawn`,
618/// `Accessor::forward`, or `StorecContextMut::spawn`.
619// TODO: Replace this with `core::ops::AsyncFnOnce` when that becomes a viable
620// option.
621//
622// As of this writing, it's not possible to specify e.g. `Send` and `Sync`
623// bounds on the `Future` type returned by an `AsyncFnOnce`.  Also, using `F:
624// Future<Output = Result<()>> + Send + Sync, FN: FnOnce(&Accessor<T>) -> F +
625// Send + Sync + 'static` fails with a type mismatch error when we try to pass
626// it an async closure (e.g. `async move |_| { ... }`).  So this seems to be the
627// best we can do for the time being.
628pub trait AccessorTask<T, D = HasSelf<T>>: Send + 'static
629where
630    D: HasData + ?Sized,
631{
632    /// Run the task.
633    fn run(self, accessor: &Accessor<T, D>) -> impl Future<Output = Result<()>> + Send;
634}
635
636/// Represents parameter and result metadata for the caller side of a
637/// guest->guest call orchestrated by a fused adapter.
638enum CallerInfo {
639    /// Metadata for a call to an async-lowered import
640    Async {
641        params: Vec<ValRaw>,
642        has_result: bool,
643    },
644    /// Metadata for a call to an sync-lowered import
645    Sync {
646        params: Vec<ValRaw>,
647        result_count: u32,
648    },
649}
650
651/// Indicates how a guest task is waiting on a waitable set.
652enum WaitMode {
653    /// The guest task is waiting using `task.wait`
654    Fiber(StoreFiber<'static>),
655    /// The guest task is waiting via a callback declared as part of an
656    /// async-lifted export.
657    Callback(Instance),
658    Caller {
659        fiber: StoreFiber<'static>,
660        callee: TableId<GuestTask>,
661    },
662}
663
664#[derive(Debug)]
665enum WaitReason {
666    GuestSubtask(TableId<GuestTask>),
667    Other,
668}
669
670/// Represents the reason a fiber is suspending itself.
671#[derive(Debug)]
672enum SuspendReason {
673    /// The fiber is waiting for an event to be delivered to the specified
674    /// waitable set or task.
675    Waiting {
676        set: TableId<WaitableSet>,
677        thread: QualifiedThreadId,
678    },
679    WaitingForGuestSubtask {
680        caller: QualifiedThreadId,
681        callee: TableId<GuestTask>,
682    },
683    /// The fiber has finished handling its most recent work item and is waiting
684    /// for another (or to be dropped if it is no longer needed).
685    NeedWork,
686    /// The fiber is yielding and should be resumed once other tasks have had a
687    /// chance to run.
688    Yielding { thread: QualifiedThreadId },
689    /// The fiber was explicitly suspended with a call to `thread.suspend` or `thread.switch-to`.
690    ExplicitlySuspending { thread: QualifiedThreadId },
691}
692
693/// Represents a pending call into guest code for a given guest task.
694enum GuestCallKind {
695    /// Indicates there's an event to deliver to the task, possibly related to a
696    /// waitable set the task has been waiting on or polling.
697    DeliverEvent {
698        /// The instance to which the task belongs.
699        instance: Instance,
700        /// The waitable set the event belongs to, if any.
701        ///
702        /// If this is `None` the event will be waiting in the
703        /// `GuestTask::event` field for the task.
704        set: Option<TableId<WaitableSet>>,
705    },
706    /// Indicates that a new guest task call is pending and may be executed
707    /// using the specified closure.
708    ///
709    /// If the closure returns `Ok(Some(call))`, the `call` should be run
710    /// immediately using `handle_guest_call`.
711    StartImplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
712    StartExplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
713}
714
715impl fmt::Debug for GuestCallKind {
716    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
717        match self {
718            Self::DeliverEvent { instance, set } => f
719                .debug_struct("DeliverEvent")
720                .field("instance", instance)
721                .field("set", set)
722                .finish(),
723            Self::StartImplicit(_) => f.debug_tuple("StartImplicit").finish(),
724            Self::StartExplicit(_) => f.debug_tuple("StartExplicit").finish(),
725        }
726    }
727}
728
729/// The target of a suspension intrinsic.
730#[derive(Copy, Clone, Debug)]
731pub enum SuspensionTarget {
732    Resume(u32),
733    Promote(u32),
734    None,
735}
736
737/// Behavior for `resume_thread`.
738#[derive(Copy, Clone, Debug)]
739pub enum ResumeThread {
740    Promote,
741    Resume,
742    ResumeLater,
743}
744
745/// Represents a pending call into guest code for a given guest thread.
746#[derive(Debug)]
747struct GuestCall {
748    thread: QualifiedThreadId,
749    kind: GuestCallKind,
750}
751
752impl GuestCall {
753    /// Returns whether or not the call is ready to run.
754    ///
755    /// A call will not be ready to run if either:
756    ///
757    /// - the (sub-)component instance to be called has already been entered and
758    /// cannot be reentered until an in-progress call completes
759    ///
760    /// - the call is for a not-yet started task and the (sub-)component
761    /// instance to be called has backpressure enabled
762    fn is_ready(&self, store: &mut StoreOpaque) -> Result<bool> {
763        let task = store.concurrent_state_mut()?.get_mut(self.thread.task)?;
764        let async_typed = task.async_typed;
765        let instance = task.instance;
766        let state = store.instance_state(instance).concurrent_state();
767
768        let ready = match &self.kind {
769            GuestCallKind::DeliverEvent { .. } => !state.do_not_enter,
770            GuestCallKind::StartImplicit(_) => {
771                !async_typed || !(state.do_not_enter || state.backpressure > 0)
772            }
773            GuestCallKind::StartExplicit(_) => true,
774        };
775        log::trace!(
776            "call {self:?} ready? {ready} (do_not_enter: {}; backpressure: {})",
777            state.do_not_enter,
778            state.backpressure
779        );
780        Ok(ready)
781    }
782}
783
784/// Job to be run on a worker fiber.
785enum WorkerItem {
786    GuestCall(GuestCall),
787    Function(AlwaysMut<Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send>>),
788}
789
790/// Represents a pending work item to be handled by the event loop for a given
791/// component instance.
792enum WorkItem {
793    /// A host task to be pushed to `ConcurrentState::futures`.
794    PushFuture(AlwaysMut<HostTaskFuture>),
795    /// A fiber to resume.
796    ResumeFiber {
797        instance: RuntimeInstance,
798        thread: QualifiedThreadId,
799        fiber: StoreFiber<'static>,
800    },
801    /// A thread to resume.
802    ResumeThread {
803        instance: RuntimeInstance,
804        thread: QualifiedThreadId,
805    },
806    /// A pending call into guest code for a given guest task.
807    GuestCall {
808        instance: RuntimeInstance,
809        call: GuestCall,
810    },
811    /// A job to run on a worker fiber.
812    WorkerFunction(AlwaysMut<Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send>>),
813}
814
815impl fmt::Debug for WorkItem {
816    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
817        match self {
818            Self::PushFuture(_) => f.debug_tuple("PushFuture").finish(),
819            Self::ResumeFiber {
820                instance, thread, ..
821            } => f
822                .debug_struct("ResumeFiber")
823                .field("instance", instance)
824                .field("thread", thread)
825                .finish(),
826            Self::ResumeThread { instance, thread } => f
827                .debug_struct("ResumeThread")
828                .field("instance", instance)
829                .field("thread", thread)
830                .finish(),
831            Self::GuestCall { instance, call } => f
832                .debug_struct("GuestCall")
833                .field("instance", instance)
834                .field("call", call)
835                .finish(),
836            Self::WorkerFunction(_) => f.debug_tuple("WorkerFunction").finish(),
837        }
838    }
839}
840
841/// Whether a suspension intrinsic was cancelled or completed
842#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
843pub(crate) enum WaitResult {
844    Cancelled,
845    Completed,
846}
847
848/// Poll the specified future until it completes on behalf of a guest->host call
849/// using a sync-lowered import.
850///
851/// This is similar to `Instance::first_poll` except it's for sync-lowered
852/// imports, meaning we don't need to handle cancellation and we can block the
853/// caller until the task completes, at which point the caller can handle
854/// lowering the result to the guest's stack and linear memory.
855pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
856    store: &mut dyn VMStore,
857    host_task: EnteredHostTask,
858    future: impl Future<Output = Result<R>> + Send + 'static,
859) -> Result<R> {
860    // Poll the future once before creating a host task. The host task will be
861    // created lazily if it's needed during the poll and otherwise will be
862    // created if the future suspends.  We can use a dummy `Waker` here because
863    // we'll add the future to `ConcurrentState::futures` and poll it
864    // automatically from the event loop if it doesn't complete immediately
865    // here.
866    let mut future = Box::pin(future);
867    let poll = tls::set(store, || {
868        future
869            .as_mut()
870            .poll(&mut Context::from_waker(&Waker::noop()))
871    });
872
873    let caller = match host_task {
874        Some(caller) => caller,
875        None => bail_bug!("host task wasn't created but should have been"),
876    };
877
878    let task = match poll {
879        // It completed immediately, so no persistent host task is needed.
880        Poll::Ready(result) => return result,
881
882        // It did not complete immediately; create the host task and add it to
883        // `ConcurrentState::futures` so it will be polled via the event loop;
884        // then use `GuestThread::sync_call_set` to wait for the task to
885        // complete, suspending the current fiber until it does so.
886        Poll::Pending => {
887            let Some(task) = store.materialize_host_task_id()? else {
888                bail_bug!("current thread is not a host thread")
889            };
890
891            // Wrap the future in a closure which will stash its result in the
892            // host task and resume this fiber when it completes.
893            let future = Box::pin(async move {
894                let result = run_with_host_task_set(task, future).await??;
895                tls::get(move |store| {
896                    let state = store.concurrent_state_mut()?;
897                    let host_state = &mut state.get_mut(task)?.state;
898                    assert!(matches!(host_state, HostTaskState::CalleeStarted));
899                    *host_state = HostTaskState::CalleeFinished(Box::new(result));
900
901                    Waitable::Host(task).set_event(
902                        state,
903                        Some(Event::Subtask {
904                            status: Status::Returned,
905                        }),
906                    )?;
907
908                    Ok(())
909                })
910            }) as HostTaskFuture;
911
912            let caller_instance = store.concurrent_state_mut()?.get_mut(caller.task)?.instance;
913            store.switch_or_trap_if_may_not_suspend(caller_instance)?;
914
915            let state = store.concurrent_state_mut()?;
916            state.push_future(future);
917
918            let set = state.get_mut(caller.thread)?.sync_call_set;
919            Waitable::Host(task).join(state, Some(set))?;
920
921            store.suspend(SuspendReason::Waiting {
922                set,
923                thread: caller,
924            })?;
925
926            // Remove the `task` from the `sync_call_set` to ensure that when
927            // this function returns and the task is deleted that there are no
928            // more lingering references to this host task.
929            Waitable::Host(task).join(store.concurrent_state_mut()?, None)?;
930            task
931        }
932    };
933
934    // Retrieve and return the result.
935    let host_state = &mut store.concurrent_state_mut()?.get_mut(task)?.state;
936    match mem::replace(host_state, HostTaskState::CalleeDone { cancelled: false }) {
937        HostTaskState::CalleeFinished(result) => Ok(match result.downcast() {
938            Ok(result) => *result,
939            Err(_) => bail_bug!("host task finished with wrong type of result"),
940        }),
941        _ => bail_bug!("unexpected host task state after completion"),
942    }
943}
944
945/// Execute the specified guest call.
946fn handle_guest_call(store: &mut dyn VMStore, call: GuestCall) -> Result<()> {
947    match call.kind {
948        GuestCallKind::DeliverEvent { instance, set } => {
949            let (event, waitable) = match instance.get_event(store, call.thread.task, set, true)? {
950                Some(pair) => pair,
951                None => bail_bug!("delivering non-present event"),
952            };
953            let state = store.concurrent_state_mut()?;
954            let task = state.get_mut(call.thread.task)?;
955            let runtime_instance = task.instance;
956            let handle = waitable.map(|(_, v)| v).unwrap_or(0);
957
958            log::trace!(
959                "use callback to deliver event {event:?} to {:?} for {waitable:?}",
960                call.thread,
961            );
962
963            let old_thread = store.set_thread(call.thread)?;
964            log::trace!(
965                "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread",
966                call.thread
967            );
968
969            store.enter_instance(runtime_instance);
970
971            let Some(callback) = store
972                .concurrent_state_mut()?
973                .get_mut(call.thread.task)?
974                .callback
975                .take()
976            else {
977                bail_bug!("guest task callback field not present")
978            };
979
980            let code = callback(store, event, handle)?;
981
982            store
983                .concurrent_state_mut()?
984                .get_mut(call.thread.task)?
985                .callback = Some(callback);
986
987            store.exit_instance(runtime_instance)?;
988
989            store.set_thread(old_thread)?;
990
991            instance.handle_callback_code(store, call.thread, runtime_instance.index, code)?;
992
993            log::trace!("GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread");
994        }
995        GuestCallKind::StartImplicit(fun) => {
996            fun(store)?;
997        }
998        GuestCallKind::StartExplicit(fun) => {
999            fun(store)?;
1000        }
1001    }
1002
1003    Ok(())
1004}
1005
1006impl<T> Store<T> {
1007    /// Convenience wrapper for [`StoreContextMut::run_concurrent`].
1008    pub async fn run_concurrent<R>(&mut self, fun: impl AsyncFnOnce(&Accessor<T>) -> R) -> Result<R>
1009    where
1010        T: Send + 'static,
1011    {
1012        ensure!(
1013            self.as_context().0.concurrency_support(),
1014            "cannot use `run_concurrent` when Config::concurrency_support disabled",
1015        );
1016        self.as_context_mut().run_concurrent(fun).await
1017    }
1018
1019    #[doc(hidden)]
1020    pub fn assert_concurrent_state_empty(&mut self) {
1021        self.as_context_mut().assert_concurrent_state_empty();
1022    }
1023
1024    #[doc(hidden)]
1025    pub fn concurrent_state_table_size(&mut self) -> usize {
1026        self.as_context_mut().concurrent_state_table_size()
1027    }
1028
1029    /// Convenience wrapper for [`StoreContextMut::spawn`].
1030    pub fn spawn(&mut self, task: impl AccessorTask<T, HasSelf<T>>) -> Result<JoinHandle>
1031    where
1032        T: 'static,
1033    {
1034        self.as_context_mut().spawn(task)
1035    }
1036}
1037
1038impl<T> StoreContextMut<'_, T> {
1039    /// Assert that all the relevant tables and queues in the concurrent state
1040    /// for this store are empty.
1041    ///
1042    /// This is for sanity checking in integration tests
1043    /// (e.g. `component-async-tests`) that the relevant state has been cleared
1044    /// after each test concludes.  This should help us catch leaks, e.g. guest
1045    /// tasks which haven't been deleted despite having completed and having
1046    /// been dropped by their supertasks.
1047    ///
1048    /// Only intended for use in Wasmtime's own testing.
1049    #[doc(hidden)]
1050    pub fn assert_concurrent_state_empty(self) {
1051        let store = self.0;
1052        store
1053            .store_data_mut()
1054            .components
1055            .assert_instance_states_empty();
1056        let state = store.concurrent_state_mut().unwrap();
1057        assert!(
1058            state.table.get_mut().is_empty(),
1059            "non-empty table: {:?}",
1060            state.table.get_mut()
1061        );
1062        assert!(state.switch_item.is_none());
1063        assert!(state.high_priority.is_empty());
1064        assert!(state.low_priority.is_empty());
1065        assert!(state.unforced_current_thread.is_none());
1066        assert!(state.deferred_host_call_context.is_none());
1067        assert!(state.futures_mut().unwrap().is_empty());
1068        assert!(state.global_error_context_ref_counts.is_empty());
1069    }
1070
1071    /// Helper function to perform tests over the size of the concurrent state
1072    /// table which can be useful for detecting leaks.
1073    ///
1074    /// Only intended for use in Wasmtime's own testing.
1075    #[doc(hidden)]
1076    pub fn concurrent_state_table_size(&mut self) -> usize {
1077        self.0
1078            .concurrent_state_mut()
1079            .unwrap()
1080            .table
1081            .get_mut()
1082            .iter_mut()
1083            .count()
1084    }
1085
1086    /// Spawn a background task to run as part of this instance's event loop.
1087    ///
1088    /// The task will receive an `&Accessor<U>` and run concurrently with
1089    /// any other tasks in progress for the instance.
1090    ///
1091    /// Note that the task will only make progress if and when the event loop
1092    /// for this instance is run.
1093    ///
1094    /// The returned [`JoinHandle`] may be used to cancel the task.
1095    pub fn spawn(mut self, task: impl AccessorTask<T>) -> Result<JoinHandle>
1096    where
1097        T: 'static,
1098    {
1099        let accessor = Accessor::new(StoreToken::new(self.as_context_mut()));
1100        self.spawn_with_accessor(accessor, task)
1101    }
1102
1103    /// Internal implementation of `spawn` functions where a `store` is
1104    /// available along with an `Accessor`.
1105    fn spawn_with_accessor<D>(
1106        self,
1107        accessor: Accessor<T, D>,
1108        task: impl AccessorTask<T, D>,
1109    ) -> Result<JoinHandle>
1110    where
1111        T: 'static,
1112        D: HasData + ?Sized,
1113    {
1114        // Create an "abortable future" here where internally the future will
1115        // hook calls to poll and possibly spawn more background tasks on each
1116        // iteration.
1117        let (handle, future) = JoinHandle::run(async move { task.run(&accessor).await });
1118        self.0
1119            .concurrent_state_mut()?
1120            .push_future(Box::pin(async move { future.await.unwrap_or(Ok(())) }));
1121        Ok(handle)
1122    }
1123
1124    /// Run the specified closure `fun` to completion as part of this store's
1125    /// event loop.
1126    ///
1127    /// This will run `fun` as part of this store's event loop until it
1128    /// yields a result.  `fun` is provided an [`Accessor`], which provides
1129    /// controlled access to the store and its data.
1130    ///
1131    /// This function can be used to invoke [`Func::call_concurrent`] for
1132    /// example within the async closure provided here.
1133    ///
1134    /// This function will unconditionally return an error if
1135    /// [`Config::concurrency_support`] is disabled.
1136    ///
1137    /// [`Config::concurrency_support`]: crate::Config::concurrency_support
1138    ///
1139    /// # Store-blocking behavior
1140    ///
1141    /// At this time there are certain situations in which the `Future` returned
1142    /// by the `AsyncFnOnce` passed to this function will not be polled for an
1143    /// extended period of time, despite one or more `Waker::wake` events having
1144    /// occurred for the task to which it belongs.  This can manifest as the
1145    /// `Future` seeming to be "blocked" or "locked up", but is actually due to
1146    /// the `Store` being held by e.g. a blocking host function, preventing the
1147    /// `Future` from being polled. A canonical example of this is when the
1148    /// `fun` provided to this function attempts to set a timeout for an
1149    /// invocation of a wasm function. In this situation the async closure is
1150    /// waiting both on (a) the wasm computation to finish, and (b) the timeout
1151    /// to elapse. At this time this setup will not always work and the timeout
1152    /// may not reliably fire.
1153    ///
1154    /// This function will not block the current thread and as such is always
1155    /// suitable to run in an `async` context, but the current implementation of
1156    /// Wasmtime can lead to situations where a certain wasm computation is
1157    /// required to make progress the closure to make progress. This is an
1158    /// artifact of Wasmtime's historical implementation of `async` functions
1159    /// and is the topic of [#11869] and [#11870]. In the timeout example from
1160    /// above it means that Wasmtime can get "wedged" for a bit where (a) must
1161    /// progress for a readiness notification of (b) to get delivered.
1162    ///
1163    /// This effectively means that it's not possible to reliably perform a
1164    /// "select" operation within the `fun` closure, which timeouts for example
1165    /// are based on. Fixing this requires some relatively major refactoring
1166    /// work within Wasmtime itself. This is a known pitfall otherwise and one
1167    /// that is intended to be fixed one day. In the meantime it's recommended
1168    /// to apply timeouts or such to the entire `run_concurrent` call itself
1169    /// rather than internally.
1170    ///
1171    /// [#11869]: https://github.com/bytecodealliance/wasmtime/issues/11869
1172    /// [#11870]: https://github.com/bytecodealliance/wasmtime/issues/11870
1173    ///
1174    /// # Example
1175    ///
1176    /// ```
1177    /// # use {
1178    /// #   wasmtime::{
1179    /// #     error::{Result},
1180    /// #     component::{ Component, Linker, Resource, ResourceTable},
1181    /// #     Config, Engine, Store
1182    /// #   },
1183    /// # };
1184    /// #
1185    /// # struct MyResource(u32);
1186    /// # struct Ctx { table: ResourceTable }
1187    /// #
1188    /// # async fn foo() -> Result<()> {
1189    /// # let mut config = Config::new();
1190    /// # let engine = Engine::new(&config)?;
1191    /// # let mut store = Store::new(&engine, Ctx { table: ResourceTable::new() });
1192    /// # let mut linker = Linker::new(&engine);
1193    /// # let component = Component::new(&engine, "")?;
1194    /// # let instance = linker.instantiate_async(&mut store, &component).await?;
1195    /// # let foo = instance.get_typed_func::<(Resource<MyResource>,), (Resource<MyResource>,)>(&mut store, "foo")?;
1196    /// # let bar = instance.get_typed_func::<(u32,), ()>(&mut store, "bar")?;
1197    /// store.run_concurrent(async |accessor| -> wasmtime::Result<_> {
1198    ///    let resource = accessor.with(|mut access| access.get().table.push(MyResource(42)))?;
1199    ///    let (another_resource,) = foo.call_concurrent(accessor, (resource,)).await?;
1200    ///    let value = accessor.with(|mut access| access.get().table.delete(another_resource))?;
1201    ///    bar.call_concurrent(accessor, (value.0,)).await?;
1202    ///    Ok(())
1203    /// }).await??;
1204    /// # Ok(())
1205    /// # }
1206    /// ```
1207    pub async fn run_concurrent<R>(self, fun: impl AsyncFnOnce(&Accessor<T>) -> R) -> Result<R>
1208    where
1209        T: Send + 'static,
1210    {
1211        ensure!(
1212            self.0.concurrency_support(),
1213            "cannot use `run_concurrent` when Config::concurrency_support disabled",
1214        );
1215        self.do_run_concurrent(fun, false).await
1216    }
1217
1218    pub(super) async fn run_concurrent_trap_on_idle<R>(
1219        self,
1220        fun: impl AsyncFnOnce(&Accessor<T>) -> R,
1221    ) -> Result<R> {
1222        self.do_run_concurrent(fun, true).await
1223    }
1224
1225    async fn do_run_concurrent<R>(
1226        mut self,
1227        fun: impl AsyncFnOnce(&Accessor<T>) -> R,
1228        trap_on_idle: bool,
1229    ) -> Result<R> {
1230        debug_assert!(self.0.concurrency_support());
1231        let already_running = self
1232            .0
1233            .concurrent_state_mut_already_forced_current_thread()
1234            .event_loop_running;
1235        if already_running {
1236            bail!("Recursive `StoreContextMut::run_concurrent` calls not supported")
1237        }
1238        let token = StoreToken::new(self.as_context_mut());
1239
1240        struct Dropper<'a, T: 'static, V> {
1241            store: StoreContextMut<'a, T>,
1242            value: ManuallyDrop<V>,
1243        }
1244
1245        impl<'a, T, V> Drop for Dropper<'a, T, V> {
1246            fn drop(&mut self) {
1247                self.store
1248                    .0
1249                    .concurrent_state_mut_already_forced_current_thread()
1250                    .event_loop_running = false;
1251
1252                tls::set(self.store.0, || {
1253                    // SAFETY: Here we drop the value without moving it for the
1254                    // first and only time -- per the contract for `Drop::drop`,
1255                    // this code won't run again, and the `value` field will no
1256                    // longer be accessible.
1257                    unsafe { ManuallyDrop::drop(&mut self.value) }
1258                });
1259            }
1260        }
1261
1262        let accessor = &Accessor::new(token);
1263        self.0
1264            .concurrent_state_mut_already_forced_current_thread()
1265            .event_loop_running = true;
1266        let dropper = &mut Dropper {
1267            store: self,
1268            value: ManuallyDrop::new(fun(accessor)),
1269        };
1270        // SAFETY: We never move `dropper` nor its `value` field.
1271        let future = unsafe { Pin::new_unchecked(dropper.value.deref_mut()) };
1272
1273        dropper
1274            .store
1275            .as_context_mut()
1276            .poll_until(future, trap_on_idle)
1277            .await
1278    }
1279
1280    /// Run this store's event loop.
1281    ///
1282    /// The returned future will resolve when the specified future completes or,
1283    /// if `trap_on_idle` is true, when the event loop can't make further
1284    /// progress.
1285    async fn poll_until<R>(
1286        mut self,
1287        mut future: Pin<&mut impl Future<Output = R>>,
1288        trap_on_idle: bool,
1289    ) -> Result<R> {
1290        struct Reset<'a, T: 'static> {
1291            store: StoreContextMut<'a, T>,
1292            futures: Option<FuturesUnordered<HostTaskFuture>>,
1293        }
1294
1295        impl<'a, T> Drop for Reset<'a, T> {
1296            fn drop(&mut self) {
1297                if let Some(futures) = self.futures.take() {
1298                    *self
1299                        .store
1300                        .0
1301                        .concurrent_state_mut_already_forced_current_thread()
1302                        .futures
1303                        .get_mut() = Some(futures);
1304                }
1305            }
1306        }
1307
1308        loop {
1309            // Take `ConcurrentState::futures` out of the store so we can poll
1310            // it while also safely giving any of the futures inside access to
1311            // `self`.
1312            let futures = self.0.concurrent_state_mut()?.futures.get_mut().take();
1313            let mut reset = Reset {
1314                store: self.as_context_mut(),
1315                futures,
1316            };
1317            let mut next = match reset.futures.as_mut() {
1318                Some(f) => pin!(f.next()),
1319                None => bail_bug!("concurrent state missing futures field"),
1320            };
1321
1322            enum PollResult<R> {
1323                Complete(R),
1324                ProcessWork {
1325                    ready: Option<WorkItem>,
1326                    low_priority: bool,
1327                },
1328            }
1329
1330            let result = future::poll_fn(|cx| {
1331                // First, poll the future we were passed as an argument and
1332                // return immediately if it's ready.
1333                if let Poll::Ready(value) = tls::set(reset.store.0, || future.as_mut().poll(cx)) {
1334                    return Poll::Ready(Ok(PollResult::Complete(value)));
1335                }
1336
1337                // Next, poll `ConcurrentState::futures` (which includes any
1338                // pending host tasks and/or background tasks), returning
1339                // immediately if one of them fails.
1340                let next = match tls::set(reset.store.0, || next.as_mut().poll(cx)) {
1341                    Poll::Ready(Some(output)) => {
1342                        match output {
1343                            Err(e) => return Poll::Ready(Err(e)),
1344                            Ok(()) => {}
1345                        }
1346                        Poll::Ready(true)
1347                    }
1348                    Poll::Ready(None) => Poll::Ready(false),
1349                    Poll::Pending => Poll::Pending,
1350                };
1351
1352                // Next, identify the next work item to process, if any, using
1353                // the following priority order:
1354                //
1355                // - `switch_item`: Represents the guest thread we _must_ switch
1356                // to before any other thread runs per the determinism
1357                // requirements in the Component Model spec.
1358                //
1359                // - `high_priority`: "Urgent" work items, e.g. async calls have
1360                // become freshly unblocked due to backpressure clearing or
1361                // similar, stream or future state updates, etc.
1362                //
1363                // - `low_priority`: Work items such as resuming a fiber after
1364                // it yields, in which case the point is to let other items run
1365                // first.
1366                let state = reset.store.0.concurrent_state_mut()?;
1367                let mut ready = state.switch_item.take();
1368                let mut low_priority = false;
1369                if ready.is_none() {
1370                    ready = state.high_priority.pop_back();
1371                    if ready.is_none() {
1372                        ready = state.low_priority.pop_back();
1373                        low_priority = true;
1374                    }
1375                }
1376                if ready.is_some() {
1377                    return Poll::Ready(Ok(PollResult::ProcessWork {
1378                        ready,
1379                        low_priority,
1380                    }));
1381                }
1382
1383                // Finally, if we have nothing else to do right now, determine what to do
1384                // based on whether there are any pending futures in
1385                // `ConcurrentState::futures`.
1386                return match next {
1387                    Poll::Ready(true) => {
1388                        // In this case, one of the futures in
1389                        // `ConcurrentState::futures` completed
1390                        // successfully, so we return now and continue
1391                        // the outer loop in case there is another one
1392                        // ready to complete.
1393                        Poll::Ready(Ok(PollResult::ProcessWork {
1394                            ready: None,
1395                            low_priority: false,
1396                        }))
1397                    }
1398                    Poll::Ready(false) => {
1399                        // Poll the future we were passed one last time
1400                        // in case one of `ConcurrentState::futures` had
1401                        // the side effect of unblocking it.
1402                        if let Poll::Ready(value) =
1403                            tls::set(reset.store.0, || future.as_mut().poll(cx))
1404                        {
1405                            Poll::Ready(Ok(PollResult::Complete(value)))
1406                        } else {
1407                            // In this case, there are no more pending
1408                            // futures in `ConcurrentState::futures`,
1409                            // there are no remaining work items, _and_
1410                            // the future we were passed as an argument
1411                            // still hasn't completed.
1412                            if trap_on_idle {
1413                                // `trap_on_idle` is true, so we exit
1414                                // immediately.
1415
1416                                // If there are any tasks belonging to
1417                                // an instance which may not suspend, trap with
1418                                // `CannotBlockSyncTask`:
1419                                Poll::Ready(Err(if reset.store.0.any_may_not_suspend()? {
1420                                    Trap::CannotBlockSyncTask.into()
1421                                } else {
1422                                    // Otherwise, trap with `AsyncDeadlock`:
1423                                    Trap::AsyncDeadlock.into()
1424                                }))
1425                            } else {
1426                                // `trap_on_idle` is false, so we assume
1427                                // that future will wake up and give us
1428                                // more work to do when it's ready to.
1429                                Poll::Pending
1430                            }
1431                        }
1432                    }
1433                    // There is at least one pending future in
1434                    // `ConcurrentState::futures` and we have nothing
1435                    // else to do but wait for now, so we return
1436                    // `Pending`.
1437                    Poll::Pending => Poll::Pending,
1438                };
1439            })
1440            .await;
1441
1442            // Put the `ConcurrentState::futures` back into the store before we
1443            // return or handle any work items since one or more of those items
1444            // might append more futures.
1445            drop(reset);
1446
1447            match result? {
1448                // The future we were passed as an argument completed, so we
1449                // return the result.
1450                PollResult::Complete(value) => break Ok(value),
1451                // The future we were passed has not yet completed, so handle
1452                // any work items and then loop again.
1453                PollResult::ProcessWork {
1454                    ready,
1455                    low_priority,
1456                } => {
1457                    struct Dispose<'a, T: 'static> {
1458                        store: StoreContextMut<'a, T>,
1459                        ready: Option<WorkItem>,
1460                    }
1461
1462                    impl<'a, T> Drop for Dispose<'a, T> {
1463                        fn drop(&mut self) {
1464                            if let Some(item) = self.ready.take() {
1465                                match item {
1466                                    WorkItem::ResumeFiber { mut fiber, .. } => {
1467                                        fiber.dispose(self.store.0)
1468                                    }
1469                                    WorkItem::PushFuture(future) => {
1470                                        tls::set(self.store.0, move || drop(future))
1471                                    }
1472                                    _ => {}
1473                                }
1474                            }
1475                        }
1476                    }
1477
1478                    let mut dispose = Dispose {
1479                        store: self.as_context_mut(),
1480                        ready,
1481                    };
1482
1483                    // If we're about to run a low-priority task, first yield to
1484                    // the executor.  This ensures that it won't be starved of
1485                    // the ability to e.g. update the readiness of sockets,
1486                    // etc. which the guest may be using `thread.yield` along
1487                    // with `waitable-set.poll` to monitor in a CPU-heavy loop.
1488                    //
1489                    // This works for e.g. `thread.yield` and callbacks which
1490                    // return `CALLBACK_CODE_YIELD` because we queue a low
1491                    // priority item to resume the task (i.e. resume the thread
1492                    // or call the callback, respectively) just prior to
1493                    // suspending it.  Indeed, as of this writing those are the
1494                    // _only_ situations we queue low-priority tasks.
1495                    // Therefore, we interpret the guest's request to yield as
1496                    // meaning "yield to other guest tasks _and_/_or_ host
1497                    // operations such as updating socket readiness", the latter
1498                    // being the async runtime's responsibility.
1499                    //
1500                    // In the future, if this ends up causing measurable
1501                    // performance issues, this could be optimized such that we
1502                    // only yield periodically (e.g. for batches of low priority
1503                    // items) and not for each and every idividual item.
1504                    if low_priority {
1505                        dispose.store.0.yield_now().await
1506                    }
1507
1508                    if let Some(item) = dispose.ready.take() {
1509                        dispose
1510                            .store
1511                            .as_context_mut()
1512                            .handle_work_item(item)
1513                            .await?;
1514                    }
1515                }
1516            }
1517        }
1518    }
1519
1520    /// Handle the specified work item, possibly resuming a fiber if applicable.
1521    async fn handle_work_item(self, item: WorkItem) -> Result<()> {
1522        log::trace!("handle work item {item:?}");
1523        match item {
1524            WorkItem::PushFuture(future) => {
1525                self.0
1526                    .concurrent_state_mut()?
1527                    .futures_mut()?
1528                    .push(future.into_inner());
1529            }
1530            WorkItem::ResumeFiber { fiber, .. } => {
1531                self.0.resume_fiber(fiber).await?;
1532            }
1533            WorkItem::ResumeThread { thread, .. } => {
1534                if let GuestThreadState::Ready { fiber, .. } = mem::replace(
1535                    &mut self.0.concurrent_state_mut()?.get_mut(thread.thread)?.state,
1536                    GuestThreadState::Running,
1537                ) {
1538                    self.0.resume_fiber(fiber).await?;
1539                } else {
1540                    bail_bug!("cannot resume non-pending thread {thread:?}");
1541                }
1542            }
1543            WorkItem::GuestCall { call, .. } => {
1544                if call.is_ready(self.0)? {
1545                    self.run_on_worker(WorkerItem::GuestCall(call)).await?;
1546                } else {
1547                    let state = self.0.concurrent_state_mut()?;
1548                    let task = state.get_mut(call.thread.task)?;
1549                    if !task.starting_sent {
1550                        task.starting_sent = true;
1551                        if let GuestCallKind::StartImplicit(_) = &call.kind {
1552                            Waitable::Guest(call.thread.task).set_event(
1553                                state,
1554                                Some(Event::Subtask {
1555                                    status: Status::Starting,
1556                                }),
1557                            )?;
1558                        }
1559                    }
1560
1561                    let instance = state.get_mut(call.thread.task)?.instance;
1562                    self.0
1563                        .instance_state(instance)
1564                        .concurrent_state()
1565                        .pending
1566                        .insert(call.thread, call.kind);
1567                }
1568            }
1569            WorkItem::WorkerFunction(fun) => {
1570                self.run_on_worker(WorkerItem::Function(fun)).await?;
1571            }
1572        }
1573
1574        Ok(())
1575    }
1576
1577    /// Execute the specified guest call on a worker fiber.
1578    async fn run_on_worker(self, item: WorkerItem) -> Result<()> {
1579        let worker = if let Some(fiber) = self.0.concurrent_state_mut()?.worker.take() {
1580            fiber
1581        } else {
1582            // SAFETY: the `make_fiber_unchecked` function is unsafe because the
1583            // returned fiber is unconditionally `Send` as opposed to being
1584            // conditionally send depending on the argument (in this case
1585            // `self.0`). This `async` function, however, is conditionally
1586            // `Send` depending on `self`, in this case `StoreContextMut<T>`,
1587            // which is already going to be conditionally `Send` depending on
1588            // `T`.
1589            //
1590            // The returned fiber is possibly stored within the `Store<T>` as
1591            // well. If `T: Send` then that's fine and everything's dandy. If
1592            // `T: !Send`, however, then the store is already not-`Send` meaning
1593            // that putting more actually-not-`Send` things inside of it isn't
1594            // an issue.
1595            //
1596            // The main issue here is that the returned fiber effectively can't
1597            // get transferred outside the context of the store. That's an
1598            // implementation detail we'll have to rely on, but is currently
1599            // true.
1600            unsafe {
1601                fiber::make_fiber_unchecked(self.0, move |store| {
1602                    loop {
1603                        let Some(item) = store.concurrent_state_mut()?.worker_item.take() else {
1604                            bail_bug!("worker_item not present when resuming fiber")
1605                        };
1606                        match item {
1607                            WorkerItem::GuestCall(call) => handle_guest_call(store, call)?,
1608                            WorkerItem::Function(fun) => fun.into_inner()(store)?,
1609                        }
1610
1611                        store.suspend(SuspendReason::NeedWork)?;
1612                    }
1613                })?
1614            }
1615        };
1616
1617        let worker_item = &mut self.0.concurrent_state_mut()?.worker_item;
1618        assert!(worker_item.is_none());
1619        *worker_item = Some(item);
1620
1621        self.0.resume_fiber(worker).await
1622    }
1623
1624    /// Wrap the specified host function in a future which will call it, passing
1625    /// it an `&Accessor<T>`.
1626    ///
1627    /// See the `Accessor` documentation for details.
1628    pub(crate) fn wrap_call<F, R>(self, closure: F) -> impl Future<Output = Result<R>> + 'static
1629    where
1630        T: 'static,
1631        F: FnOnce(&Accessor<T>) -> Pin<Box<dyn Future<Output = Result<R>> + Send + '_>>
1632            + Send
1633            + Sync
1634            + 'static,
1635        R: Send + Sync + 'static,
1636    {
1637        let token = StoreToken::new(self);
1638        async move {
1639            let mut accessor = Accessor::new(token);
1640            closure(&mut accessor).await
1641        }
1642    }
1643
1644    /// Returns an iterator over the current async call stack defined by the
1645    /// component model.
1646    ///
1647    /// This can be used, for example to correlate a host import call with which
1648    /// root export task originally called it.
1649    ///
1650    /// Tasks are yielded "youngest first" where the first item in the iterator
1651    /// is the current task, and the last item in the iterator is the original
1652    /// call.
1653    ///
1654    /// # Omitted Tasks
1655    ///
1656    /// This stack might not contain an entry for every guest-to-guest component
1657    /// call that is currently executing: if those guest-to-guest calls are not
1658    /// asynchronous and provably cannot access async task state, then Wasmtime
1659    /// can avoid materializing the guest task state for that call, and such
1660    /// tasks will not have a `GuestTaskId` and will not appear in this stack
1661    /// trace.
1662    ///
1663    /// The `GuestTaskId`s for host-to-guest and guest-to-host calls, however,
1664    /// will always appear in this stack trace.
1665    pub fn async_call_stack(&mut self) -> Result<impl Iterator<Item = GuestTaskId>> {
1666        let mut cur = Some(self.0.current_thread()?);
1667        let state = self.0.concurrent_state_mut()?;
1668        Ok(core::iter::from_fn(move || {
1669            while let Some(t) = cur {
1670                cur = state.parent(t);
1671                if let Some(task) = t.guest_task() {
1672                    return Some(GuestTaskId(task));
1673                }
1674            }
1675
1676            None
1677        }))
1678    }
1679
1680    pub(crate) async fn start_instance(
1681        &mut self,
1682        instance: ModuleInstance,
1683    ) -> Result<ModuleInstance> {
1684        let (tx, rx) = oneshot::channel();
1685        let token = StoreToken::new(self.as_context_mut());
1686        self.0.queue_task(move |store| {
1687            _ = tx.send(
1688                instance
1689                    .start_raw(&mut token.as_context_mut(store))
1690                    .map(|()| instance),
1691            );
1692            Ok(())
1693        })?;
1694        self.as_context_mut()
1695            .run_concurrent_trap_on_idle(async |_| {
1696                rx.await
1697                    .map_err(|_| format_err!("oneshot channel canceled"))
1698            })
1699            .await??
1700    }
1701}
1702
1703/// Return value of [`StoreOpaque::host_task_create`].
1704///
1705/// This is an `Option` to handle the dynamic `store.concurrency_support()`
1706/// property. When present this records the guest thread to restore when the
1707/// host call exits. The corresponding [`HostTask`] will need to be lazily
1708/// created if needed via [`StoreOpaque::materialize_host_task_id`].
1709pub type EnteredHostTask = Option<QualifiedThreadId>;
1710
1711impl StoreOpaque {
1712    /// Returns the currently-running thread, promoting any deferred lazy guest
1713    /// thread into a fully-materialized `CurrentThread`. Deferred [`HostTask`]s
1714    /// are not materialized.
1715    #[inline]
1716    pub(crate) fn current_thread(&mut self) -> Result<CurrentThread> {
1717        // Without concurrency support there is nothing to force.
1718        if !self.concurrency_support() {
1719            return Ok(CurrentThread::None);
1720        }
1721
1722        // If the JIT-visible current thread isn't a deferred thread then
1723        // `ConcurrentState` is already up to date.
1724        if !self
1725            .vm_store_context_mut()
1726            .current_thread_mut()
1727            .is_deferred()
1728        {
1729            return Ok(self
1730                .concurrent_state_mut_already_forced_current_thread()
1731                .unforced_current_thread);
1732        }
1733
1734        self.force_deferred_current_thread()
1735    }
1736
1737    /// Slow path of [`Self::current_thread`]: promote the deferred lazy
1738    /// thread into a fully-materialized `CurrentThread`.
1739    #[cold]
1740    fn force_deferred_current_thread(&mut self) -> Result<CurrentThread> {
1741        // The component instance whose adapters pushed the deferred frames; all
1742        // frames in a guest-to-guest, sync-to-sync call chain of fused adapters
1743        // live within a single `wasmtime::component::Instance` (because
1744        // cross-`wasmtime::component::Instance` calls don't go through fused
1745        // adapters), and guest code only ever runs as a guest thread, so the
1746        // chain's base thread is already materialized in `ConcurrentState` and
1747        // we can get the `ComponentInstanceId` shared by the whole chain from
1748        // here.
1749        let state = self.concurrent_state_mut_without_forcing_current_thread();
1750        let id = match state.unforced_current_thread.guest_task() {
1751            Some(task) => state.get_mut(task)?.instance.instance,
1752            None => bail_bug!("deferred component-model thread with non-guest base"),
1753        };
1754
1755        // Collect the deferred frames pushed inline by fused adapters, walking
1756        // the `parent` chain from innermost to the base.
1757        let mut frames = Vec::new();
1758        let mut cur = *self.vm_store_context_mut().current_thread_mut();
1759        while let Some(ptr) = cur.as_deferred() {
1760            // SAFETY: `ptr` points at a `VMDeferredThread` living in a fused
1761            // adapter's stack frame that is suspended below us on the stack
1762            // (mid-call, waiting for this nested call to return), so the
1763            // referent is still valid and exclusively ours to read.
1764            let deferred = unsafe { ptr.as_non_null().as_ref() };
1765            frames.push((
1766                deferred.callee_async != 0,
1767                deferred.callee_instance,
1768                deferred.saved_context,
1769            ));
1770            cur = deferred.parent;
1771        }
1772
1773        // Mark the current thread forced *before* replaying so that any
1774        // reentrant `force_current_thread` call short-circuits via the
1775        // non-deferred path above.
1776        *self.vm_store_context_mut().current_thread_mut() = VMLazyThread::forced();
1777
1778        // Save the current context, as we need to overwrite it while replaying
1779        // below.
1780        let current_context = *self.vm_store_context_mut().component_context_mut();
1781
1782        // Replay the deferred `enter_guest_sync_call`s outermost-first so that
1783        // the resulting `ConcurrentState` matches what the non-deferred path
1784        // would have otherwise produced.
1785        for (callee_async, callee_instance, saved_context) in frames.into_iter().rev() {
1786            // Restore the caller's context slots so that we save the correct
1787            // values into the caller's thread, exactly as the non-deferred path
1788            // would have on entry.
1789            *self.vm_store_context_mut().component_context_mut() = saved_context;
1790            let callee = RuntimeInstance {
1791                instance: id,
1792                index: RuntimeComponentInstanceIndex::from_u32(callee_instance),
1793            };
1794            self.enter_guest_sync_call(callee_async, callee)?;
1795        }
1796
1797        // Replaying done; restore the current context.
1798        *self.vm_store_context_mut().component_context_mut() = current_context;
1799
1800        Ok(self
1801            .concurrent_state_mut_without_forcing_current_thread()
1802            .unforced_current_thread)
1803    }
1804
1805    fn current_guest_thread(&mut self) -> Result<QualifiedThreadId> {
1806        match self.current_thread()?.guest() {
1807            Some(id) => Ok(*id),
1808            None => bail_bug!("current thread is not a guest thread"),
1809        }
1810    }
1811
1812    // A result of `None` may indicate that this is either the top-level event
1813    // loop, a deferred host task, or concurrency support is disabled. In all
1814    // cases we don't have an ID for the task.
1815    pub(crate) fn current_materialized_host_task(&mut self) -> Result<Option<TableId<HostTask>>> {
1816        match self.current_thread()? {
1817            CurrentThread::Host(id) => Ok(Some(id)),
1818            CurrentThread::DeferredHost(_) | CurrentThread::None => Ok(None),
1819            _ => bail_bug!("current thread is not a host thread"),
1820        }
1821    }
1822
1823    /// Returns the current host task ID, materializing a deferred host task if
1824    /// one is active. `None` represents a call from the top-level host.
1825    fn materialize_host_task_id(&mut self) -> Result<Option<TableId<HostTask>>> {
1826        Ok(self
1827            .concurrent_state_mut()?
1828            .materialize_current_host_task_id()?)
1829    }
1830
1831    fn enter_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> {
1832        log::trace!("enter sync-typed call {callee:?}");
1833        let state = self.instance_state(callee).concurrent_state();
1834        let old_do_not_suspend = state.do_not_suspend;
1835        state.do_not_suspend = true;
1836
1837        let thread = self.current_guest_thread()?;
1838        let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?;
1839        if thread.old_do_not_suspend.is_some() {
1840            bail_bug!("current thread already has `old_do_not_suspend` value");
1841        }
1842
1843        thread.old_do_not_suspend = Some(old_do_not_suspend);
1844
1845        Ok(())
1846    }
1847
1848    fn exit_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> {
1849        log::trace!("exit sync-typed call {callee:?}");
1850        let thread = self.current_guest_thread()?;
1851        let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?;
1852        let Some(old_do_not_suspend) = thread.old_do_not_suspend.take() else {
1853            bail_bug!("current thread missing `old_do_not_suspend` value");
1854        };
1855        let state = self.instance_state(callee).concurrent_state();
1856        state.do_not_suspend = old_do_not_suspend;
1857        Ok(())
1858    }
1859
1860    /// Push a `GuestTask` onto the task stack for either a sync-to-sync,
1861    /// guest-to-guest call or a sync host-to-guest call.
1862    ///
1863    /// This task will only be used for the purpose of handling calls to
1864    /// intrinsic functions; both parameter lowering and result lifting are
1865    /// assumed to be taken care of elsewhere.
1866    ///
1867    /// NB: for sync-to-sync, guest-to-guest calls we delay task construction in
1868    /// fused adapters, see `StoreOpaque::current_thread`, `VMDeferredThread`,
1869    /// and `lower_fact_enter_sync_call`. Make sure all this stuff stays in
1870    /// sync!
1871    pub(crate) fn enter_guest_sync_call(
1872        &mut self,
1873        callee_async_typed: bool,
1874        callee: RuntimeInstance,
1875    ) -> Result<()> {
1876        log::trace!("enter sync-lifted call {callee:?}");
1877        if !self.concurrency_support() {
1878            return self.enter_call_not_concurrent();
1879        }
1880
1881        let thread = self.current_thread()?;
1882        let caller = if let Some(thread) = thread.guest() {
1883            Caller::Guest { thread: *thread }
1884        } else {
1885            Caller::Host {
1886                tx: None,
1887                host_future_present: false,
1888                caller: self.materialize_host_task_id()?,
1889            }
1890        };
1891        let state = self.concurrent_state_mut()?;
1892        let guest_thread = GuestTask::new(
1893            state,
1894            Box::new(move |_, _| bail_bug!("cannot lower params in sync call")),
1895            LiftResult {
1896                lift: Box::new(move |_, _| bail_bug!("cannot lift result in sync call")),
1897                ty: TypeTupleIndex::reserved_value(),
1898                memory: None,
1899                string_encoding: StringEncoding::Utf8,
1900            },
1901            caller,
1902            None,
1903            callee,
1904            callee_async_typed,
1905            true,
1906        )?;
1907
1908        Instance::from_wasmtime(self, callee.instance).add_guest_thread_to_instance_table(
1909            guest_thread.thread,
1910            self,
1911            callee.index,
1912        )?;
1913        self.set_thread(guest_thread)?;
1914
1915        if !callee_async_typed {
1916            self.enter_sync_call(callee)?;
1917        }
1918
1919        Ok(())
1920    }
1921
1922    /// Pop a `GuestTask` previously pushed using `enter_guest_sync_call`.
1923    ///
1924    /// NB: for sync-to-sync, guest-to-guest calls we delay task construction in
1925    /// fused adapters and then when the call returns we check to see if the
1926    /// task's contruction was forced and if not avoid calling out of the JIT
1927    /// code to this function. See `lower_fact_exit_sync_call`. Make sure all
1928    /// this stuff stays in sync!
1929    pub(crate) fn exit_guest_sync_call(&mut self) -> Result<()> {
1930        if !self.concurrency_support() {
1931            return Ok(self.exit_call_not_concurrent());
1932        }
1933
1934        let thread = match self.current_thread()?.guest() {
1935            Some(t) => *t,
1936            None => bail_bug!("expected task when exiting"),
1937        };
1938        let task = self.concurrent_state_mut()?.get_mut(thread.task)?;
1939        let instance = task.instance;
1940
1941        let caller = match &task.caller {
1942            &Caller::Guest { thread } => thread.into(),
1943            &Caller::Host { caller, .. } => caller
1944                .map(CurrentThread::Host)
1945                .unwrap_or(CurrentThread::None),
1946        };
1947        task.lift_result = None;
1948        task.exited = true;
1949        let async_typed = task.async_typed;
1950
1951        if !async_typed {
1952            self.exit_sync_call(instance)?;
1953        }
1954
1955        self.set_thread(caller)?;
1956
1957        log::trace!("exit sync-lifted call {instance:?}");
1958
1959        if async_typed {
1960            // If we're async-typed, returning control to our caller won't help
1961            // resolve any outstanding sync-typed call which might be in
1962            // progress, so we may need to switch or trap before exiting this
1963            // thread:
1964            self.switch_or_trap_if_may_not_suspend(instance)?;
1965        }
1966
1967        self.cleanup_thread(thread, instance, CleanupTask::Yes)?;
1968
1969        Ok(())
1970    }
1971
1972    /// Similar to `enter_guest_sync_call` except for when the guest makes a
1973    /// transition to the host.
1974    ///
1975    /// This initially records a deferred host call. A full [`HostTask`] should
1976    /// be allocated later if needed via
1977    /// [`StoreOpaque::materialize_host_task_id`].
1978    pub(crate) fn host_task_create(&mut self) -> Result<EnteredHostTask> {
1979        if !self.concurrency_support() {
1980            self.enter_call_not_concurrent()?;
1981            return Ok(None);
1982        }
1983        let caller = self.current_guest_thread()?;
1984        log::trace!("new deferred host task with caller {caller:?}");
1985        self.set_thread(CurrentThread::DeferredHost(caller))?;
1986        let state = self.concurrent_state_mut()?;
1987        debug_assert!(state.deferred_host_call_context.is_none());
1988        state.deferred_host_call_context = Some(CallContext::default());
1989        state.debug_assert_deferred_host_invariant();
1990        Ok(Some(caller))
1991    }
1992
1993    /// Dual of `host_task_create` and signifies that the host has finished and
1994    /// will be cleaned up.
1995    ///
1996    /// Note that this isn't invoked when the host is invoked asynchronously and
1997    /// the host isn't complete yet. In that situation the host task persists
1998    /// and will be cleaned up separately in `subtask_drop`
1999    pub(crate) fn host_task_delete(
2000        &mut self,
2001        original_task: EnteredHostTask,
2002        materialized_task: Option<TableId<HostTask>>,
2003    ) -> Result<()> {
2004        match original_task {
2005            Some(caller) => {
2006                self.set_thread(caller)?;
2007                if materialized_task.is_none() {
2008                    let state = self.concurrent_state_mut()?;
2009                    let context = state
2010                        .deferred_host_call_context
2011                        .take()
2012                        .expect("deferred host call context should be present");
2013                    debug_assert!(context.is_empty());
2014                    state.debug_assert_deferred_host_invariant();
2015                }
2016                log::trace!(
2017                    "delete host task with caller {original_task:?} and materialized as {materialized_task:?}"
2018                );
2019                if let Some(task) = materialized_task {
2020                    self.concurrent_state_mut()?.delete(task)?;
2021                }
2022            }
2023            None => {
2024                debug_assert!(materialized_task.is_none());
2025                self.exit_call_not_concurrent();
2026            }
2027        }
2028        Ok(())
2029    }
2030
2031    /// Helper function to retrieve the `InstanceState` for the
2032    /// specified instance.
2033    fn instance_state(&mut self, instance: RuntimeInstance) -> &mut InstanceState {
2034        self.component_instance_mut(instance.instance)
2035            .instance_state(instance.index)
2036    }
2037
2038    /// Configure the currently running `thread`.
2039    ///
2040    /// This will save off any state necessary for the previous thread, if
2041    /// applicable, and then it'll additionally update state for `thread` if
2042    /// needed too.
2043    fn set_thread(&mut self, thread: impl Into<CurrentThread>) -> Result<CurrentThread> {
2044        let thread = thread.into();
2045        let state = self.concurrent_state_mut()?;
2046        state.debug_assert_deferred_host_invariant();
2047        let old_thread = mem::replace(&mut state.unforced_current_thread, thread);
2048
2049        // First thing to do after swapping threads is updating the context
2050        // slots for this thread within the store. This restores the behavior of
2051        // `context.{get,set}`. This involves taking the old state out of the
2052        // store, saving it in the thread that's being swapped from, and doing
2053        // the inverse for the new thread. When debug assertions are enabled
2054        // this also leaves behind sentinel values to try to uncover bugs where
2055        // this may be forgotten.
2056        if let Some(old_thread) = old_thread.guest() {
2057            let old_context = *self.vm_store_context_mut().component_context_mut();
2058            self.concurrent_state_mut()?
2059                .get_mut(old_thread.thread)?
2060                .context = old_context;
2061        }
2062        if cfg!(debug_assertions) {
2063            *self.vm_store_context_mut().component_context_mut() =
2064                [u32::MAX; NUM_COMPONENT_CONTEXT_SLOTS];
2065        }
2066        if let Some(thread) = thread.guest() {
2067            let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?;
2068            let context = thread.context;
2069            if cfg!(debug_assertions) {
2070                thread.context = [u32::MAX; NUM_COMPONENT_CONTEXT_SLOTS];
2071            }
2072            *self.vm_store_context_mut().component_context_mut() = context;
2073        }
2074
2075        // Keep the JIT-visible current-thread pointer in sync.
2076        *self.vm_store_context_mut().current_thread_mut() = if thread.is_none() {
2077            VMLazyThread::none()
2078        } else {
2079            VMLazyThread::forced()
2080        };
2081
2082        Ok(old_thread)
2083    }
2084
2085    /// Call `switch_if_may_not_suspend` and trap if it returns `false`.
2086    fn switch_or_trap_if_may_not_suspend(&mut self, instance: RuntimeInstance) -> Result<()> {
2087        if self.switch_if_may_not_suspend(instance)? {
2088            Ok(())
2089        } else {
2090            Err(Trap::CannotBlockSyncTask.into())
2091        }
2092    }
2093
2094    /// Check if the specified instance has a sync-typed call in progress; if so
2095    /// attempt to switch to another ready thread for that instance, and if no
2096    /// such thread exists, return false.
2097    fn switch_if_may_not_suspend(&mut self, instance: RuntimeInstance) -> Result<bool> {
2098        // Call this for the side effect of forcing any deferred task creation,
2099        // which may influence the value of `ConcurrentState::do_not_suspend`
2100        // below:
2101        self.concurrent_state_mut()?;
2102
2103        Ok(!self.concurrency_support()
2104            || !self
2105                .instance_state(instance)
2106                .concurrent_state()
2107                .do_not_suspend
2108            || self
2109                .concurrent_state_mut()?
2110                .promote_instance_local_thread_work_item(instance)?)
2111    }
2112
2113    /// Record that we're about to enter a (sub-)component instance which does
2114    /// not support more than one concurrent, stackful activation, meaning it
2115    /// cannot be entered again until the next call returns.
2116    fn enter_instance(&mut self, instance: RuntimeInstance) {
2117        log::trace!("enter {instance:?}");
2118        self.instance_state(instance)
2119            .concurrent_state()
2120            .do_not_enter = true;
2121    }
2122
2123    /// Record that we've exited a (sub-)component instance previously entered
2124    /// with `Self::enter_instance` and then calls `Self::partition_pending`.
2125    /// See the documentation for the latter for details.
2126    fn exit_instance(&mut self, instance: RuntimeInstance) -> Result<()> {
2127        log::trace!("exit {instance:?}");
2128        self.instance_state(instance)
2129            .concurrent_state()
2130            .do_not_enter = false;
2131        self.partition_pending(instance)
2132    }
2133
2134    /// Iterate over `InstanceState::pending`, moving any ready items into the
2135    /// "high priority" work item queue.
2136    ///
2137    /// Also, notify `ConcurrentState::ready_for_concurrent_call_waker` if
2138    /// present.
2139    ///
2140    /// See `GuestCall::is_ready` for details.
2141    fn partition_pending(&mut self, instance: RuntimeInstance) -> Result<()> {
2142        for (thread, kind) in
2143            mem::take(&mut self.instance_state(instance).concurrent_state().pending).into_iter()
2144        {
2145            let call = GuestCall { thread, kind };
2146            if call.is_ready(self)? {
2147                self.concurrent_state_mut()?
2148                    .push_high_priority(WorkItem::GuestCall { instance, call });
2149            } else {
2150                self.instance_state(instance)
2151                    .concurrent_state()
2152                    .pending
2153                    .insert(call.thread, call.kind);
2154            }
2155        }
2156
2157        if let Some(waker) = self
2158            .concurrent_state_mut()?
2159            .ready_for_concurrent_call_waker
2160            .take()
2161        {
2162            waker.wake();
2163        }
2164
2165        Ok(())
2166    }
2167
2168    /// Implements the `backpressure.{inc,dec}` intrinsics.
2169    pub(crate) fn backpressure_modify(
2170        &mut self,
2171        caller_instance: RuntimeInstance,
2172        modify: impl FnOnce(u16) -> Option<u16>,
2173    ) -> Result<()> {
2174        let state = self.instance_state(caller_instance).concurrent_state();
2175        let old = state.backpressure;
2176        let new = modify(old).ok_or_else(|| Trap::BackpressureOverflow)?;
2177        state.backpressure = new;
2178
2179        if old > 0 && new == 0 {
2180            // Backpressure was previously enabled and is now disabled; move any
2181            // newly-eligible guest calls to the "high priority" queue.
2182            self.partition_pending(caller_instance)?;
2183        }
2184
2185        Ok(())
2186    }
2187
2188    /// Resume the specified fiber, giving it exclusive access to the specified
2189    /// store.
2190    async fn resume_fiber(&mut self, fiber: StoreFiber<'static>) -> Result<()> {
2191        let old_thread = self.current_thread()?;
2192        log::trace!("resume_fiber: save current thread {old_thread:?}");
2193
2194        let fiber = fiber::resolve_or_release(self, fiber).await?;
2195
2196        self.set_thread(old_thread)?;
2197
2198        let state = self.concurrent_state_mut()?;
2199
2200        if let Some(ot) = old_thread.guest() {
2201            state.get_mut(ot.thread)?.state = GuestThreadState::Running;
2202        }
2203        log::trace!("resume_fiber: restore current thread {old_thread:?}");
2204
2205        if let Some(mut fiber) = fiber {
2206            log::trace!("resume_fiber: suspend reason {:?}", &state.suspend_reason);
2207            // See the `SuspendReason` documentation for what each case means.
2208            let reason = match state.suspend_reason.take() {
2209                Some(r) => r,
2210                None => bail_bug!("suspend reason missing when resuming fiber"),
2211            };
2212            match reason {
2213                SuspendReason::NeedWork => {
2214                    if state.worker.is_none() {
2215                        state.worker = Some(fiber);
2216                    } else {
2217                        fiber.dispose(self);
2218                    }
2219                }
2220                SuspendReason::Yielding { thread } => {
2221                    state.get_mut(thread.thread)?.state = GuestThreadState::Ready { fiber };
2222                    let instance = state.get_mut(thread.task)?.instance;
2223                    state.push_low_priority(WorkItem::ResumeThread { instance, thread });
2224                }
2225                SuspendReason::ExplicitlySuspending { thread } => {
2226                    state.get_mut(thread.thread)?.state = GuestThreadState::Suspended(fiber);
2227                }
2228                SuspendReason::Waiting { set, thread } => {
2229                    let old = state
2230                        .get_mut(set)?
2231                        .waiting
2232                        .insert(thread, WaitMode::Fiber(fiber));
2233                    assert!(old.is_none());
2234                }
2235                SuspendReason::WaitingForGuestSubtask { caller, callee } => {
2236                    let set = state.get_mut(caller.thread)?.sync_call_set;
2237                    let old = state
2238                        .get_mut(set)?
2239                        .waiting
2240                        .insert(caller, WaitMode::Caller { fiber, callee });
2241                    assert!(old.is_none());
2242                }
2243            };
2244        } else {
2245            log::trace!("resume_fiber: fiber has exited");
2246        }
2247
2248        Ok(())
2249    }
2250
2251    /// Suspend the current fiber, storing the reason in
2252    /// `ConcurrentState::suspend_reason` to indicate the conditions under which
2253    /// it should be resumed.
2254    ///
2255    /// See the `SuspendReason` documentation for details.
2256    fn suspend(&mut self, reason: SuspendReason) -> Result<()> {
2257        log::trace!("suspend fiber: {reason:?}");
2258
2259        // If we're yielding or waiting on behalf of a guest thread, we'll need to
2260        // pop the call context which manages resource borrows before suspending
2261        // and then push it again once we've resumed.
2262        let task = match &reason {
2263            SuspendReason::Yielding { thread, .. }
2264            | SuspendReason::Waiting { thread, .. }
2265            | SuspendReason::WaitingForGuestSubtask { caller: thread, .. }
2266            | SuspendReason::ExplicitlySuspending { thread } => Some(thread.task),
2267            SuspendReason::NeedWork => None,
2268        };
2269
2270        let old_guest_thread = if let Some(task) = task {
2271            // If we haven't set `ConcurrentState::switch_item` yet (e.g. if
2272            // we're not calling a subtask), and we're running in a task that
2273            // has a subtask status update for its caller, this is a good time
2274            // to deliver that update (and in fact we are _required_ to do so by
2275            // the CM spec).
2276            let state = self.concurrent_state_mut()?;
2277            if state.switch_item.is_none() {
2278                if let Some(item) = state.get_mut(task)?.switch_item.take() {
2279                    state.set_switch_item(item)?;
2280                }
2281            }
2282
2283            self.current_thread()?
2284        } else {
2285            CurrentThread::None
2286        };
2287
2288        let suspend_reason = &mut self.concurrent_state_mut()?.suspend_reason;
2289        assert!(suspend_reason.is_none());
2290        *suspend_reason = Some(reason);
2291
2292        // We'll panic if we call `Self::with_blocking` when the fiber is being
2293        // disposed, so check for that and exit ASAP if appropriate.
2294        if !self.fiber_async_state_mut().can_block() {
2295            return Err(format_err!("future dropped"));
2296        }
2297
2298        self.with_blocking(|_, cx| cx.suspend(StoreFiberYield::ReleaseStore))?;
2299
2300        if task.is_some() {
2301            self.set_thread(old_guest_thread)?;
2302        }
2303
2304        Ok(())
2305    }
2306
2307    fn wait_for_event(
2308        &mut self,
2309        caller_instance: RuntimeInstance,
2310        waitable: Waitable,
2311        reason: WaitReason,
2312    ) -> Result<()> {
2313        let caller = self.current_guest_thread()?;
2314        let state = self.concurrent_state_mut()?;
2315
2316        waitable.trap_if_in_waitable_set(state)?;
2317
2318        let set = state.get_mut(caller.thread)?.sync_call_set;
2319        waitable.join(state, Some(set))?;
2320
2321        self.switch_or_trap_if_may_not_suspend(caller_instance)?;
2322
2323        self.suspend(match reason {
2324            WaitReason::GuestSubtask(callee) => {
2325                SuspendReason::WaitingForGuestSubtask { caller, callee }
2326            }
2327            WaitReason::Other => SuspendReason::Waiting {
2328                set,
2329                thread: caller,
2330            },
2331        })?;
2332        let state = self.concurrent_state_mut()?;
2333        waitable.join(state, None)
2334    }
2335
2336    /// Cleans up the data structures backing the `guest_thread` specified,
2337    /// removing it from the internal tables of `runtime_instance` as well.
2338    ///
2339    /// This function is used whenever a guest thread has fully exited and
2340    /// completed. This'll clean up the associated `GuestThread` structure and
2341    /// related resources it contains.
2342    ///
2343    /// Other functionality that this implements is:
2344    ///
2345    /// * This will perform conditional cleanup of the `GuestTask` that owns
2346    ///   this thread if `cleanup_task` is `CleanupTask::Yes`.
2347    /// * If there are no more threads in the `GuestTask` that this thread is
2348    ///   associated with, and if the task hasn't produced a result (e.g. it's not
2349    ///   returned or cancelled), then a trap will be raised that a result
2350    ///   wasn't ever produced.
2351    /// * If this task is finished, meaning the top-level thread exited and
2352    ///   additionally it's been returned or cancelled, then this will handle
2353    ///   management of the store's "active interesting tasks" counter.
2354    ///
2355    /// Effectively this is intended to be a "narrow waist" through which many
2356    /// destruction operations are funneled through.
2357    fn cleanup_thread(
2358        &mut self,
2359        guest_thread: QualifiedThreadId,
2360        runtime_instance: RuntimeInstance,
2361        cleanup_task: CleanupTask,
2362    ) -> Result<()> {
2363        let state = self.concurrent_state_mut()?;
2364        // If we never suspended, we never had a chance to deliver a subtask
2365        // status update, if any, to our caller, so we do that here:
2366        if let Some(item) = state.get_mut(guest_thread.task)?.switch_item.take() {
2367            state.set_switch_item(item)?;
2368        }
2369        let thread_data = state.get_mut(guest_thread.thread)?;
2370        let sync_call_set = thread_data.sync_call_set;
2371        if let Some(guest_id) = thread_data.instance_rep {
2372            self.instance_state(runtime_instance)
2373                .thread_handle_table()
2374                .guest_thread_remove(guest_id)?;
2375        }
2376        let state = self.concurrent_state_mut()?;
2377
2378        // Clean up any pending subtasks in the sync_call_set
2379        for waitable in mem::take(&mut state.get_mut(sync_call_set)?.ready) {
2380            if let Some(Event::Subtask {
2381                status: Status::Returned | Status::ReturnCancelled,
2382            }) = waitable.common(state)?.event
2383            {
2384                waitable.delete_from(state)?;
2385            }
2386        }
2387
2388        state.delete(guest_thread.thread)?;
2389        state.delete(sync_call_set)?;
2390        let task = state.get_mut(guest_thread.task)?;
2391        task.threads.remove(&guest_thread.thread);
2392
2393        if task.threads.is_empty() && !task.returned_or_cancelled() {
2394            bail!(Trap::NoAsyncResult);
2395        }
2396        let ready_to_delete = task.ready_to_delete();
2397
2398        if !task.decremented_interesting_task_count && task.exited && task.returned_or_cancelled() {
2399            task.decremented_interesting_task_count = true;
2400
2401            debug_assert!(state.interesting_tasks > 0);
2402            state.interesting_tasks -= 1;
2403            if state.interesting_tasks == 0
2404                && let Some(waker) = state.interesting_tasks_empty_waker.take()
2405            {
2406                waker.wake();
2407            }
2408        }
2409
2410        match cleanup_task {
2411            CleanupTask::Yes => {
2412                if ready_to_delete {
2413                    Waitable::Guest(guest_thread.task).delete_from(state)?;
2414                }
2415            }
2416            CleanupTask::No => {}
2417        }
2418
2419        Ok(())
2420    }
2421
2422    /// Performs cancellation of the `guest_task` specified with the
2423    /// precondition that the task hasn't lowered its parameters.
2424    ///
2425    /// In this situation the task hasn't ever been started meaning it hasn't
2426    /// actually run any wasm code yet. This requires cleaning up metadata such
2427    /// as thread information attached to the task.
2428    ///
2429    /// The main two entrypoints for this function are:
2430    ///
2431    /// * Task cancellation via `subtask.cancel`, the intrinsic.
2432    /// * Dropping a host `call_async` future which needs to cancel the task
2433    ///   because it cannot reference its parameters any more.
2434    fn cancel_guest_subtask_without_lowered_parameters(
2435        &mut self,
2436        caller_instance: RuntimeInstance,
2437        guest_task: TableId<GuestTask>,
2438    ) -> Result<()> {
2439        let concurrent_state = self.concurrent_state_mut()?;
2440        let task = concurrent_state.get_mut(guest_task)?;
2441        assert!(!task.already_lowered_parameters());
2442        // The task is in a `starting` state, meaning it hasn't run at
2443        // all yet.  Here we update its fields to indicate that it is
2444        // ready to delete immediately once `subtask.drop` is called.
2445        task.lower_params = None;
2446        task.lift_result = None;
2447        task.exited = true;
2448        let instance = task.instance;
2449
2450        // Clean up the thread within this task as it's now never going
2451        // to run.
2452        assert_eq!(1, task.threads.len());
2453        let thread = *task.threads.iter().next().unwrap();
2454        self.cleanup_thread(
2455            QualifiedThreadId {
2456                task: guest_task,
2457                thread,
2458            },
2459            caller_instance,
2460            CleanupTask::No,
2461        )?;
2462
2463        // Not yet started; cancel and remove from pending
2464        let pending = &mut self.instance_state(instance).concurrent_state().pending;
2465        let pending_count = pending.len();
2466        pending.retain(|thread, _| thread.task != guest_task);
2467        // If there were no pending threads for this task, we're in an error state
2468        if pending.len() == pending_count {
2469            bail!(Trap::SubtaskCancelAfterTerminal);
2470        }
2471        Ok(())
2472    }
2473
2474    /// Used by `ResourceTables` to record the scope of a borrow to get undone
2475    /// in the future.
2476    pub(crate) fn current_scope(&mut self) -> Result<Option<CurrentScope>> {
2477        if !self.concurrency_support() {
2478            return Ok(self
2479                .current_scope_id_not_concurrent()?
2480                .map(|id| CurrentScope::Id(Scope::Id(id))));
2481        }
2482
2483        Ok(match self.current_thread()? {
2484            CurrentThread::Guest(id) => Some(CurrentScope::Id(Scope::Id(id.task.rep()))),
2485            CurrentThread::GuestTask(id) => Some(CurrentScope::Id(Scope::Id(id.rep()))),
2486            CurrentThread::Host(id) => Some(CurrentScope::Id(Scope::HostId(id.rep()))),
2487            CurrentThread::DeferredHost(_) => Some(CurrentScope::DeferredHost),
2488            CurrentThread::None => return Ok(None),
2489        })
2490    }
2491
2492    fn queue_task(
2493        &mut self,
2494        task: impl FnOnce(&mut dyn VMStore) -> Result<()> + Send + 'static,
2495    ) -> Result<()> {
2496        self.concurrent_state_mut()?
2497            .push_high_priority(WorkItem::WorkerFunction(AlwaysMut::new(Box::new(task))));
2498        Ok(())
2499    }
2500
2501    /// Used in `poll_until` just prior to trapping due to a "deadlock"
2502    /// condition.
2503    ///
2504    /// This helps us distinguish between a simple deadlock condition (where no
2505    /// work is available for the event loop to do, nor is there any way for new
2506    /// work to be added) and a "cannot block sync task" condition where at
2507    /// least once instance has an outstanding sync-typed task running, in which
2508    /// case we'll trap with a different error message.
2509    fn any_may_not_suspend(&mut self) -> Result<bool> {
2510        // Note that this currently requires a linear search across the whole
2511        // `ConcurrentState::table`.  We _could_ optimize that, but since (1)
2512        // this function is only used when trapping, (2) the only thing you can
2513        // really do with a store that's been poisoned by a trap is drop it, and
2514        // (3) we must do a linear search through the table when dropping a
2515        // store anyway to dispose of fibers, it's reasonable for us to also do
2516        // a linear search here.
2517        Ok(self
2518            .concurrent_state_mut()?
2519            .table
2520            .get_mut()
2521            .iter_mut()
2522            .filter_map(|entry| {
2523                if let Some(task) = entry.downcast_ref::<GuestTask>() {
2524                    Some(task.instance)
2525                } else {
2526                    None
2527                }
2528            })
2529            .collect::<Vec<_>>()
2530            .into_iter()
2531            .any(|instance| {
2532                self.instance_state(instance)
2533                    .concurrent_state()
2534                    .do_not_suspend
2535            }))
2536    }
2537}
2538
2539enum CleanupTask {
2540    Yes,
2541    No,
2542}
2543
2544impl Instance {
2545    /// Get the next pending event for the specified task and (optional)
2546    /// waitable set, along with the waitable handle if applicable.
2547    fn get_event(
2548        self,
2549        store: &mut StoreOpaque,
2550        guest_task: TableId<GuestTask>,
2551        set: Option<TableId<WaitableSet>>,
2552        cancellable: bool,
2553    ) -> Result<Option<(Event, Option<(Waitable, u32)>)>> {
2554        let state = store.concurrent_state_mut()?;
2555
2556        let event = &mut state.get_mut(guest_task)?.event;
2557        if let Some(ev) = event
2558            && (cancellable || !matches!(ev, Event::Cancelled))
2559        {
2560            log::trace!("deliver event {ev:?} to {guest_task:?}");
2561            let ev = *ev;
2562            *event = None;
2563            return Ok(Some((ev, None)));
2564        }
2565
2566        let set = match set {
2567            Some(set) => set,
2568            None => return Ok(None),
2569        };
2570        let waitable = match state.get_mut(set)?.ready.pop_first() {
2571            Some(v) => v,
2572            None => return Ok(None),
2573        };
2574
2575        let common = waitable.common(state)?;
2576        let handle = match common.handle {
2577            Some(h) => h,
2578            None => bail_bug!("handle not set when delivering event"),
2579        };
2580        let event = match common.event.take() {
2581            Some(e) => e,
2582            None => bail_bug!("event not set when delivering event"),
2583        };
2584
2585        log::trace!(
2586            "deliver event {event:?} to {guest_task:?} for {waitable:?} (handle {handle}); set {set:?}"
2587        );
2588
2589        waitable.on_delivery(store, self, event)?;
2590
2591        Ok(Some((event, Some((waitable, handle)))))
2592    }
2593
2594    /// Handle the `CallbackCode` returned from an async-lifted export or its
2595    /// callback.
2596    ///
2597    /// If this returns `Ok(Some(call))`, then `call` should be run immediately
2598    /// using `handle_guest_call`.
2599    fn handle_callback_code(
2600        self,
2601        store: &mut StoreOpaque,
2602        guest_thread: QualifiedThreadId,
2603        runtime_instance: RuntimeComponentInstanceIndex,
2604        code: u32,
2605    ) -> Result<()> {
2606        let (code, set) = unpack_callback_code(code);
2607
2608        log::trace!("received callback code from {guest_thread:?}: {code} (set: {set})");
2609
2610        let state = store.concurrent_state_mut()?;
2611
2612        if let Some(item) = state.get_mut(guest_thread.task)?.switch_item.take() {
2613            state.set_switch_item(item)?;
2614        }
2615
2616        let get_set = |store: &mut StoreOpaque, handle| -> Result<_> {
2617            let set = store
2618                .instance_state(self.runtime_instance(runtime_instance))
2619                .handle_table()
2620                .waitable_set_rep(handle)?;
2621
2622            Ok(TableId::<WaitableSet>::new(set))
2623        };
2624
2625        match code {
2626            callback_code::EXIT => {
2627                log::trace!("implicit thread {guest_thread:?} completed");
2628                let task = store.concurrent_state_mut()?.get_mut(guest_thread.task)?;
2629                task.exited = true;
2630                task.callback = None;
2631
2632                let runtime_instance = self.runtime_instance(runtime_instance);
2633
2634                // Since we're async-typed, returning control to our caller
2635                // won't help resolve any outstanding sync-typed call which
2636                // might be in progress, so we may need to switch or trap before
2637                // exiting this thread:
2638                store.switch_or_trap_if_may_not_suspend(runtime_instance)?;
2639
2640                store.cleanup_thread(guest_thread, runtime_instance, CleanupTask::Yes)?;
2641            }
2642            callback_code::YIELD => {
2643                let task = state.get_mut(guest_thread.task)?;
2644                // If an `Event::Cancelled` is pending, we'll deliver that;
2645                // otherwise, we'll deliver `Event::None`.  Note that
2646                // `GuestTask::event` is only ever set to one of those two
2647                // `Event` variants.
2648                if let Some(event) = task.event {
2649                    assert!(matches!(event, Event::None | Event::Cancelled));
2650                } else {
2651                    task.event = Some(Event::None);
2652                }
2653                let call = GuestCall {
2654                    thread: guest_thread,
2655                    kind: GuestCallKind::DeliverEvent {
2656                        instance: self,
2657                        set: None,
2658                    },
2659                };
2660                // Push this thread onto the "low priority" queue so it runs
2661                // after any other threads have had a chance to run.
2662                state.push_low_priority(WorkItem::GuestCall {
2663                    instance: self.runtime_instance(runtime_instance),
2664                    call,
2665                });
2666            }
2667            callback_code::WAIT => {
2668                let set = get_set(store, set)?;
2669                let state = store.concurrent_state_mut()?;
2670
2671                if state.get_mut(guest_thread.task)?.event.is_some()
2672                    || !state.get_mut(set)?.ready.is_empty()
2673                {
2674                    // An event is immediately available; deliver it ASAP.
2675                    state.push_high_priority(WorkItem::GuestCall {
2676                        instance: self.runtime_instance(runtime_instance),
2677                        call: GuestCall {
2678                            thread: guest_thread,
2679                            kind: GuestCallKind::DeliverEvent {
2680                                instance: self,
2681                                set: Some(set),
2682                            },
2683                        },
2684                    });
2685                } else {
2686                    // No event is immediately available.
2687                    //
2688                    // We're waiting, so register to be woken up when an event
2689                    // is published for this waitable set.
2690                    //
2691                    // Here we also set `GuestTask::wake_on_cancel` which allows
2692                    // `subtask.cancel` to interrupt the wait.
2693                    let old = state
2694                        .get_mut(guest_thread.thread)?
2695                        .wake_on_cancel
2696                        .replace(set);
2697                    if !old.is_none() {
2698                        bail_bug!("thread unexpectedly had wake_on_cancel set");
2699                    }
2700                    let old = state
2701                        .get_mut(set)?
2702                        .waiting
2703                        .insert(guest_thread, WaitMode::Callback(self));
2704                    if !old.is_none() {
2705                        bail_bug!("set's waiting set already had this thread registered");
2706                    }
2707                }
2708            }
2709            _ => bail!(Trap::UnsupportedCallbackCode),
2710        }
2711
2712        Ok(())
2713    }
2714
2715    /// Stage the specified guest call as the work item to run next, to be
2716    /// started as soon as backpressure and/or reentrance rules allow.
2717    ///
2718    /// SAFETY: The raw pointer arguments must be valid references to guest
2719    /// functions (with the appropriate signatures) when the closures staged by
2720    /// this function are called.
2721    unsafe fn stage_call<T: 'static>(
2722        self,
2723        mut store: StoreContextMut<T>,
2724        guest_thread: QualifiedThreadId,
2725        callee: SendSyncPtr<VMFuncRef>,
2726        param_count: usize,
2727        result_count: usize,
2728        async_: bool,
2729        callback: Option<SendSyncPtr<VMFuncRef>>,
2730        post_return: Option<SendSyncPtr<VMFuncRef>>,
2731        host_caller: bool,
2732    ) -> Result<()> {
2733        /// Return a closure which will call the specified function in the scope
2734        /// of the specified task.
2735        ///
2736        /// This will use `GuestTask::lower_params` to lower the parameters, but
2737        /// will not lift the result; instead, it returns a
2738        /// `[MaybeUninit<ValRaw>; MAX_FLAT_PARAMS]` from which the result, if
2739        /// any, may be lifted.  Note that an async-lifted export will have
2740        /// returned its result using the `task.return` intrinsic (or not
2741        /// returned a result at all, in the case of `task.cancel`), in which
2742        /// case the "result" of this call will either be a callback code or
2743        /// nothing.
2744        ///
2745        /// SAFETY: `callee` must be a valid `*mut VMFuncRef` at the time when
2746        /// the returned closure is called.
2747        unsafe fn make_call<T: 'static>(
2748            store: StoreContextMut<T>,
2749            guest_thread: QualifiedThreadId,
2750            callee: SendSyncPtr<VMFuncRef>,
2751            param_count: usize,
2752            result_count: usize,
2753        ) -> impl FnOnce(&mut dyn VMStore) -> Result<[MaybeUninit<ValRaw>; MAX_FLAT_PARAMS]>
2754        + Send
2755        + Sync
2756        + 'static
2757        + use<T> {
2758            let token = StoreToken::new(store);
2759            move |store: &mut dyn VMStore| {
2760                let mut storage = [MaybeUninit::uninit(); MAX_FLAT_PARAMS];
2761
2762                store
2763                    .concurrent_state_mut()?
2764                    .get_mut(guest_thread.thread)?
2765                    .state = GuestThreadState::Running;
2766                let task = store.concurrent_state_mut()?.get_mut(guest_thread.task)?;
2767                let lower = match task.lower_params.take() {
2768                    Some(l) => l,
2769                    None => bail_bug!("lower_params missing"),
2770                };
2771
2772                lower(store, &mut storage[..param_count])?;
2773
2774                let mut store = token.as_context_mut(store);
2775
2776                // SAFETY: Per the contract documented in `make_call's`
2777                // documentation, `callee` must be a valid pointer.
2778                unsafe {
2779                    crate::Func::call_unchecked_raw(
2780                        &mut store,
2781                        callee.as_non_null(),
2782                        NonNull::new(
2783                            &mut storage[..param_count.max(result_count)]
2784                                as *mut [MaybeUninit<ValRaw>] as _,
2785                        )
2786                        .unwrap(),
2787                    )?;
2788                }
2789
2790                Ok(storage)
2791            }
2792        }
2793
2794        // SAFETY: Per the contract described in this function documentation,
2795        // the `callee` pointer which `call` closes over must be valid when
2796        // called by the closure we queue below.
2797        let call = unsafe {
2798            make_call(
2799                store.as_context_mut(),
2800                guest_thread,
2801                callee,
2802                param_count,
2803                result_count,
2804            )
2805        };
2806
2807        let callee_instance = store
2808            .0
2809            .concurrent_state_mut()?
2810            .get_mut(guest_thread.task)?
2811            .instance;
2812
2813        let fun = if callback.is_some() {
2814            assert!(async_);
2815
2816            Box::new(move |store: &mut dyn VMStore| {
2817                self.add_guest_thread_to_instance_table(
2818                    guest_thread.thread,
2819                    store,
2820                    callee_instance.index,
2821                )?;
2822                let old_thread = store.set_thread(guest_thread)?;
2823                log::trace!(
2824                    "stackless call: replaced {old_thread:?} with {guest_thread:?} as current thread"
2825                );
2826
2827                store.enter_instance(callee_instance);
2828
2829                // SAFETY: See the documentation for `make_call` to review the
2830                // contract we must uphold for `call` here.
2831                //
2832                // Per the contract described in the `stage_call`
2833                // documentation, the `callee` pointer which `call` closes
2834                // over must be valid.
2835                let storage = call(store)?;
2836
2837                store.exit_instance(callee_instance)?;
2838
2839                store.set_thread(old_thread)?;
2840                let state = store.concurrent_state_mut()?;
2841                if let Some(t) = old_thread.guest() {
2842                    state.get_mut(t.thread)?.state = GuestThreadState::Running;
2843                }
2844                log::trace!("stackless call: restored {old_thread:?} as current thread");
2845
2846                // SAFETY: `wasmparser` will have validated that the callback
2847                // function returns a `i32` result.
2848                let code = unsafe { storage[0].assume_init() }.get_i32() as u32;
2849
2850                self.handle_callback_code(store, guest_thread, callee_instance.index, code)
2851            }) as Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>
2852        } else {
2853            let token = StoreToken::new(store.as_context_mut());
2854            Box::new(move |store: &mut dyn VMStore| {
2855                self.add_guest_thread_to_instance_table(
2856                    guest_thread.thread,
2857                    store,
2858                    callee_instance.index,
2859                )?;
2860                let old_thread = store.set_thread(guest_thread)?;
2861                log::trace!(
2862                    "sync/async-stackful call: replaced {old_thread:?} with {guest_thread:?} as current thread",
2863                );
2864                let flags = self.id().get(store).instance_flags(callee_instance.index);
2865
2866                let callee_async_typed = store
2867                    .concurrent_state_mut()?
2868                    .get_mut(guest_thread.task)?
2869                    .async_typed;
2870
2871                // Unless this is a callback-less (i.e. stackful) async-lifted
2872                // or sync-typed export, we need to record that the instance
2873                // cannot be entered until the call returns.
2874                if !async_ && callee_async_typed {
2875                    store.enter_instance(callee_instance);
2876                }
2877
2878                if !callee_async_typed {
2879                    store.enter_sync_call(callee_instance)?;
2880                }
2881
2882                // SAFETY: See the documentation for `make_call` to review the
2883                // contract we must uphold for `call` here.
2884                //
2885                // Per the contract described in the `stage_call`
2886                // documentation, the `callee` pointer which `call` closes
2887                // over must be valid.
2888                let storage = call(store)?;
2889
2890                if !callee_async_typed {
2891                    store.exit_sync_call(callee_instance)?;
2892                }
2893
2894                if !async_ {
2895                    // This is a sync-lifted export, so now is when we lift the
2896                    // result, optionally call the post-return function, if any,
2897                    // and finally notify any current or future waiters that the
2898                    // subtask has returned.
2899
2900                    if callee_async_typed {
2901                        store.exit_instance(callee_instance)?;
2902                    }
2903
2904                    let lift = {
2905                        let state = store.concurrent_state_mut()?;
2906                        if !state.get_mut(guest_thread.task)?.result.is_none() {
2907                            bail_bug!("task has already produced a result");
2908                        }
2909
2910                        match state.get_mut(guest_thread.task)?.lift_result.take() {
2911                            Some(lift) => lift,
2912                            None => bail_bug!("lift_result field is missing"),
2913                        }
2914                    };
2915
2916                    // SAFETY: `result_count` represents the number of core Wasm
2917                    // results returned, per `wasmparser`.
2918                    let result = (lift.lift)(store, unsafe {
2919                        mem::transmute::<&[MaybeUninit<ValRaw>], &[ValRaw]>(
2920                            &storage[..result_count],
2921                        )
2922                    })?;
2923
2924                    let post_return_arg = match result_count {
2925                        0 => ValRaw::i32(0),
2926                        // SAFETY: `result_count` represents the number of
2927                        // core Wasm results returned, per `wasmparser`.
2928                        1 => unsafe { storage[0].assume_init() },
2929                        _ => unreachable!(),
2930                    };
2931
2932                    unsafe {
2933                        call_post_return(
2934                            token.as_context_mut(store),
2935                            post_return.map(|v| v.as_non_null()),
2936                            post_return_arg,
2937                            flags,
2938                        )?;
2939                    }
2940
2941                    self.task_complete(store, guest_thread.task, result, Status::Returned)?;
2942                }
2943
2944                store.set_thread(old_thread)?;
2945
2946                store
2947                    .concurrent_state_mut()?
2948                    .get_mut(guest_thread.task)?
2949                    .exited = true;
2950
2951                log::trace!(
2952                    "clean up thread; async lifted? {async_} async typed? {callee_async_typed}"
2953                );
2954
2955                if callee_async_typed {
2956                    // If we're async-typed, returning control to our caller
2957                    // won't help resolve any outstanding sync-typed call which
2958                    // might be in progress, so we may need to switch or trap
2959                    // before exiting this thread:
2960                    store.switch_or_trap_if_may_not_suspend(callee_instance)?;
2961                }
2962
2963                // This is a callback-less call, so the implicit thread has now completed
2964                store.cleanup_thread(guest_thread, callee_instance, CleanupTask::Yes)?;
2965                Ok(())
2966            })
2967        };
2968
2969        store.0.concurrent_state_mut()?.push_work_item(
2970            WorkItem::GuestCall {
2971                instance: callee_instance,
2972                call: GuestCall {
2973                    thread: guest_thread,
2974                    kind: GuestCallKind::StartImplicit(fun),
2975                },
2976            },
2977            if host_caller {
2978                Priority::High
2979            } else {
2980                Priority::Switch
2981            },
2982        )?;
2983
2984        Ok(())
2985    }
2986
2987    /// Prepare (but do not start) a guest->guest call.
2988    ///
2989    /// This is called from fused adapter code generated in
2990    /// `wasmtime_environ::fact::trampoline::Compiler`.  `start` and `return_`
2991    /// are synthesized Wasm functions which move the parameters from the caller
2992    /// to the callee and the result from the callee to the caller,
2993    /// respectively.  The adapter will call `Self::start_call` immediately
2994    /// after calling this function.
2995    ///
2996    /// SAFETY: All the pointer arguments must be valid pointers to guest
2997    /// entities (and with the expected signatures for the function references
2998    /// -- see `wasmtime_environ::fact::trampoline::Compiler` for details).
2999    unsafe fn prepare_call<T: 'static>(
3000        self,
3001        mut store: StoreContextMut<T>,
3002        start: NonNull<VMFuncRef>,
3003        return_: NonNull<VMFuncRef>,
3004        caller_instance: RuntimeComponentInstanceIndex,
3005        callee_instance: RuntimeComponentInstanceIndex,
3006        task_return_type: TypeTupleIndex,
3007        callee_async_typed: bool,
3008        memory: *mut VMMemoryDefinition,
3009        string_encoding: StringEncoding,
3010        caller_info: CallerInfo,
3011    ) -> Result<()> {
3012        enum ResultInfo {
3013            Heap { results: u32 },
3014            Stack { result_count: u32 },
3015        }
3016
3017        let result_info = match &caller_info {
3018            CallerInfo::Async {
3019                has_result: true,
3020                params,
3021            } => ResultInfo::Heap {
3022                results: match params.last() {
3023                    Some(r) => r.get_u32(),
3024                    None => bail_bug!("retptr missing"),
3025                },
3026            },
3027            CallerInfo::Async {
3028                has_result: false, ..
3029            } => ResultInfo::Stack { result_count: 0 },
3030            CallerInfo::Sync {
3031                result_count,
3032                params,
3033            } if *result_count > u32::try_from(MAX_FLAT_RESULTS)? => ResultInfo::Heap {
3034                results: match params.last() {
3035                    Some(r) => r.get_u32(),
3036                    None => bail_bug!("arg ptr missing"),
3037                },
3038            },
3039            CallerInfo::Sync { result_count, .. } => ResultInfo::Stack {
3040                result_count: *result_count,
3041            },
3042        };
3043
3044        let sync_caller = matches!(caller_info, CallerInfo::Sync { .. });
3045
3046        // Create a new guest task for the call, closing over the `start` and
3047        // `return_` functions to lift the parameters and lower the result,
3048        // respectively.
3049        let start = SendSyncPtr::new(start);
3050        let return_ = SendSyncPtr::new(return_);
3051        let token = StoreToken::new(store.as_context_mut());
3052        let old_thread = store.0.current_guest_thread()?;
3053        let state = store.0.concurrent_state_mut()?;
3054
3055        debug_assert_eq!(
3056            state.get_mut(old_thread.task)?.instance,
3057            self.runtime_instance(caller_instance)
3058        );
3059
3060        let guest_thread = GuestTask::new(
3061            state,
3062            Box::new(move |store, dst| {
3063                let mut store = token.as_context_mut(store);
3064                assert!(dst.len() <= MAX_FLAT_PARAMS);
3065                // The `+ 1` here accounts for the return pointer, if any:
3066                let mut src = [MaybeUninit::uninit(); MAX_FLAT_PARAMS + 1];
3067                let count = match caller_info {
3068                    // Async callers, if they have a result, use the last
3069                    // parameter as a return pointer so chop that off if
3070                    // relevant here.
3071                    CallerInfo::Async { params, has_result } => {
3072                        let params = &params[..params.len() - usize::from(has_result)];
3073                        for (param, src) in params.iter().zip(&mut src) {
3074                            src.write(*param);
3075                        }
3076                        params.len()
3077                    }
3078
3079                    // Sync callers forward everything directly.
3080                    CallerInfo::Sync { params, .. } => {
3081                        for (param, src) in params.iter().zip(&mut src) {
3082                            src.write(*param);
3083                        }
3084                        params.len()
3085                    }
3086                };
3087                // SAFETY: `start` is a valid `*mut VMFuncRef` from
3088                // `wasmtime-cranelift`-generated fused adapter code.  Based on
3089                // how it was constructed (see
3090                // `wasmtime_environ::fact::trampoline::Compiler::compile_async_start_adapter`
3091                // for details) we know it takes count parameters and returns
3092                // `dst.len()` results.
3093                unsafe {
3094                    crate::Func::call_unchecked_raw(
3095                        &mut store,
3096                        start.as_non_null(),
3097                        NonNull::new(
3098                            &mut src[..count.max(dst.len())] as *mut [MaybeUninit<ValRaw>] as _,
3099                        )
3100                        .unwrap(),
3101                    )?;
3102                }
3103                dst.copy_from_slice(&src[..dst.len()]);
3104                let task = store.0.current_guest_thread()?.task;
3105                let state = store.0.concurrent_state_mut()?;
3106                Waitable::Guest(task).set_event(
3107                    state,
3108                    Some(Event::Subtask {
3109                        status: Status::Started,
3110                    }),
3111                )?;
3112                Ok(())
3113            }),
3114            LiftResult {
3115                lift: Box::new(move |store, src| {
3116                    // SAFETY: See comment in closure passed as `lower_params`
3117                    // parameter above.
3118                    let mut store = token.as_context_mut(store);
3119                    let mut my_src = src.to_owned(); // TODO: use stack to avoid allocation?
3120                    if let ResultInfo::Heap { results } = &result_info {
3121                        my_src.push(ValRaw::u32(*results));
3122                    }
3123
3124                    // SAFETY: `return_` is a valid `*mut VMFuncRef` from
3125                    // `wasmtime-cranelift`-generated fused adapter code.  Based
3126                    // on how it was constructed (see
3127                    // `wasmtime_environ::fact::trampoline::Compiler::compile_async_return_adapter`
3128                    // for details) we know it takes `src.len()` parameters and
3129                    // returns up to 1 result.
3130                    unsafe {
3131                        crate::Func::call_unchecked_raw(
3132                            &mut store,
3133                            return_.as_non_null(),
3134                            my_src.as_mut_slice().into(),
3135                        )?;
3136                    }
3137
3138                    let thread = store.0.current_guest_thread()?;
3139                    let state = store.0.concurrent_state_mut()?;
3140                    if sync_caller {
3141                        state.get_mut(thread.task)?.sync_result = SyncResult::Produced(
3142                            if let ResultInfo::Stack { result_count } = &result_info {
3143                                match result_count {
3144                                    0 => None,
3145                                    1 => Some(my_src[0]),
3146                                    _ => unreachable!(),
3147                                }
3148                            } else {
3149                                None
3150                            },
3151                        );
3152                    }
3153                    Ok(Box::new(DummyResult) as Box<dyn Any + Send + Sync>)
3154                }),
3155                ty: task_return_type,
3156                memory: NonNull::new(memory).map(SendSyncPtr::new),
3157                string_encoding,
3158            },
3159            Caller::Guest { thread: old_thread },
3160            None,
3161            self.runtime_instance(callee_instance),
3162            callee_async_typed,
3163            // We don't know whether the callee export was lifted sync or async
3164            // yet, but we'll update this in `start_call`:
3165            false,
3166        )?;
3167
3168        // Make the new thread the current one so that `Self::start_call` knows
3169        // which one to start.
3170        store.0.set_thread(guest_thread)?;
3171        log::trace!("pushed {guest_thread:?} as current thread; old thread was {old_thread:?}");
3172
3173        Ok(())
3174    }
3175
3176    /// Call the specified callback function for an async-lifted export.
3177    ///
3178    /// SAFETY: `function` must be a valid reference to a guest function of the
3179    /// correct signature for a callback.
3180    unsafe fn call_callback<T>(
3181        self,
3182        mut store: StoreContextMut<T>,
3183        function: SendSyncPtr<VMFuncRef>,
3184        event: Event,
3185        handle: u32,
3186    ) -> Result<u32> {
3187        let (ordinal, result) = event.parts();
3188        let params = &mut [
3189            ValRaw::u32(ordinal),
3190            ValRaw::u32(handle),
3191            ValRaw::u32(result),
3192        ];
3193        // SAFETY: `func` is a valid `*mut VMFuncRef` from either
3194        // `wasmtime-cranelift`-generated fused adapter code or
3195        // `component::Options`.  Per `wasmparser` callback signature
3196        // validation, we know it takes three parameters and returns one.
3197        unsafe {
3198            crate::Func::call_unchecked_raw(
3199                &mut store,
3200                function.as_non_null(),
3201                params.as_mut_slice().into(),
3202            )?;
3203        }
3204        Ok(params[0].get_u32())
3205    }
3206
3207    /// Start a guest->guest call previously prepared using
3208    /// `Self::prepare_call`.
3209    ///
3210    /// This is called from fused adapter code generated in
3211    /// `wasmtime_environ::fact::trampoline::Compiler`.  The adapter will call
3212    /// this function immediately after calling `Self::prepare_call`.
3213    ///
3214    /// SAFETY: The `*mut VMFuncRef` arguments must be valid pointers to guest
3215    /// functions with the appropriate signatures for the current guest task.
3216    /// If this is a call to an async-lowered import, the actual call may be
3217    /// deferred and run after this function returns, in which case the pointer
3218    /// arguments must also be valid when the call happens.
3219    unsafe fn start_call<T: 'static>(
3220        self,
3221        mut store: StoreContextMut<T>,
3222        callback: *mut VMFuncRef,
3223        post_return: *mut VMFuncRef,
3224        callee: NonNull<VMFuncRef>,
3225        param_count: u32,
3226        result_count: u32,
3227        flags: u32,
3228        storage: Option<&mut [MaybeUninit<ValRaw>]>,
3229    ) -> Result<u32> {
3230        let token = StoreToken::new(store.as_context_mut());
3231        let async_caller = storage.is_none();
3232        let guest_thread = store.0.current_guest_thread()?;
3233        let state = store.0.concurrent_state_mut()?;
3234
3235        if !state.event_loop_running {
3236            bail_bug!("Instance::start_call called without a running event loop");
3237        }
3238
3239        let callee = SendSyncPtr::new(callee);
3240        let param_count = usize::try_from(param_count)?;
3241        assert!(param_count <= MAX_FLAT_PARAMS);
3242        let result_count = usize::try_from(result_count)?;
3243        assert!(result_count <= MAX_FLAT_RESULTS);
3244
3245        let task = state.get_mut(guest_thread.task)?;
3246        let callee_async_typed = task.async_typed;
3247        let callee_instance = task.instance;
3248
3249        task.async_lifted = (flags & START_FLAG_ASYNC_CALLEE) != 0;
3250
3251        if let Some(callback) = NonNull::new(callback) {
3252            // We're calling an async-lifted export with a callback, so store
3253            // the callback and related context as part of the task so we can
3254            // call it later when needed.
3255            let callback = SendSyncPtr::new(callback);
3256            task.callback = Some(Box::new(move |store, event, handle| {
3257                let store = token.as_context_mut(store);
3258                unsafe { self.call_callback::<T>(store, callback, event, handle) }
3259            }));
3260        }
3261
3262        let Caller::Guest { thread: caller } = &task.caller else {
3263            // As of this writing, `start_call` is only used for guest->guest
3264            // calls.
3265            bail_bug!("start_call unexpectedly invoked for host->guest call");
3266        };
3267        let caller = *caller;
3268        let caller_instance = state.get_mut(caller.task)?.instance;
3269
3270        // Stage the call as the work item to run next, modulo backpressure, etc.
3271        unsafe {
3272            self.stage_call(
3273                store.as_context_mut(),
3274                guest_thread,
3275                callee,
3276                param_count,
3277                result_count,
3278                (flags & START_FLAG_ASYNC_CALLEE) != 0,
3279                NonNull::new(callback).map(SendSyncPtr::new),
3280                NonNull::new(post_return).map(SendSyncPtr::new),
3281                false,
3282            )?;
3283        }
3284
3285        let old_do_not_suspend = if callee_async_typed {
3286            // If we're starting an async-typed call, it is permitted to suspend
3287            // since that will just return control back to the caller, which may
3288            // be able to avoid blocking if needed.
3289            //
3290            // We'll restore the old value after the call either returns or
3291            // suspends.
3292            let state = store.0.instance_state(callee_instance).concurrent_state();
3293            let old_do_not_suspend = state.do_not_suspend;
3294            state.do_not_suspend = false;
3295            Some(old_do_not_suspend)
3296        } else {
3297            None
3298        };
3299
3300        let state = store.0.concurrent_state_mut()?;
3301
3302        // Use the caller's `GuestThread::sync_call_set` to register interest in
3303        // the subtask...
3304        let guest_waitable = Waitable::Guest(guest_thread.task);
3305        let old_set = guest_waitable.common(state)?.set;
3306        let set = state.get_mut(caller.thread)?.sync_call_set;
3307        guest_waitable.join(state, Some(set))?;
3308
3309        store.0.set_thread(CurrentThread::None)?;
3310
3311        // ... and suspend this fiber temporarily while we wait for it to start.
3312        //
3313        // Note that we _could_ call the callee directly using the current fiber
3314        // rather than suspend this one, but that would make reasoning about the
3315        // event loop more complicated and is probably only worth doing if
3316        // there's a measurable performance benefit.  In addition, it would mean
3317        // blocking the caller if the callee calls a blocking sync-lowered
3318        // import, and as of this writing the spec says we must not do that.
3319        //
3320        // Alternatively, the fused adapter code could be modified to call the
3321        // callee directly without calling a host-provided intrinsic at all (in
3322        // which case it would need to do its own, inline backpressure checks,
3323        // etc.).  Again, we'd want to see a measurable performance benefit
3324        // before committing to such an optimization.  And again, we'd need to
3325        // update the spec to allow that.
3326        let (status, waitable) = loop {
3327            store.0.suspend(SuspendReason::WaitingForGuestSubtask {
3328                caller,
3329                callee: guest_thread.task,
3330            })?;
3331
3332            if let Some(old_do_not_suspend) = old_do_not_suspend {
3333                store
3334                    .0
3335                    .instance_state(callee_instance)
3336                    .concurrent_state()
3337                    .do_not_suspend = old_do_not_suspend;
3338            }
3339
3340            let state = store.0.concurrent_state_mut()?;
3341
3342            log::trace!("taking event for {:?}", guest_thread.task);
3343            let event = guest_waitable.take_event(state)?;
3344            let Some(Event::Subtask { status }) = event else {
3345                bail_bug!("subtasks should only get subtask events, got {event:?}")
3346            };
3347
3348            log::trace!("status {status:?} for {:?}", guest_thread.task);
3349
3350            if status == Status::Returned {
3351                // It returned, so we can stop waiting.
3352                break (status, None);
3353            } else if async_caller {
3354                // It hasn't returned yet, but the caller is calling via an
3355                // async-lowered import, so we generate a handle for the task
3356                // waitable and return the status.
3357                let handle = store
3358                    .0
3359                    .instance_state(caller_instance)
3360                    .handle_table()
3361                    .subtask_insert_guest(guest_thread.task.rep())?;
3362                store
3363                    .0
3364                    .concurrent_state_mut()?
3365                    .get_mut(guest_thread.task)?
3366                    .common
3367                    .handle = Some(handle);
3368                break (status, Some(handle));
3369            } else {
3370                // The callee hasn't returned yet, and the caller is calling via
3371                // a sync-lowered import, so we loop and keep waiting until the
3372                // callee returns.
3373                store.0.switch_or_trap_if_may_not_suspend(caller_instance)?;
3374            }
3375        };
3376
3377        guest_waitable.join(store.0.concurrent_state_mut()?, old_set)?;
3378
3379        // Reset the current thread to point to the caller as it resumes control.
3380        store.0.set_thread(caller)?;
3381        store
3382            .0
3383            .concurrent_state_mut()?
3384            .get_mut(caller.thread)?
3385            .state = GuestThreadState::Running;
3386        log::trace!("popped current thread {guest_thread:?}; new thread is {caller:?}");
3387
3388        if let Some(storage) = storage {
3389            // The caller used a sync-lowered import to call an async-lifted
3390            // export, in which case the result, if any, has been stashed in
3391            // `GuestTask::sync_result`.
3392            let state = store.0.concurrent_state_mut()?;
3393            let task = state.get_mut(guest_thread.task)?;
3394            if let Some(result) = task.sync_result.take()? {
3395                if let Some(result) = result {
3396                    storage[0] = MaybeUninit::new(result);
3397                }
3398
3399                if task.exited && task.ready_to_delete() {
3400                    Waitable::Guest(guest_thread.task).delete_from(state)?;
3401                }
3402            }
3403        }
3404
3405        Ok(status.pack(waitable))
3406    }
3407
3408    /// Poll the specified future once on behalf of a guest->host call using an
3409    /// async-lowered import.
3410    ///
3411    /// If it returns `Ready`, return `Ok(None)`.  Otherwise, if it returns
3412    /// `Pending`, add it to the set of futures to be polled as part of this
3413    /// instance's event loop until it completes, and then return
3414    /// `Ok(Some(handle))` where `handle` is the waitable handle to return.
3415    ///
3416    /// Whether the future returns `Ready` immediately or later, the `lower`
3417    /// function will be used to lower the result, if any, into the guest caller's
3418    /// stack and linear memory. The `lower` function is invoked with the
3419    /// `Option<R>` param `None` if the future is cancelled. The
3420    /// `Option<TableId<HostTask>>` is passed as `Some` if the host task was
3421    /// materialized during execution and allows `lower` to delete the task if
3422    /// needed.
3423    pub(crate) fn first_poll<T: 'static, R: Send + 'static>(
3424        self,
3425        mut store: StoreContextMut<'_, T>,
3426        host_task: EnteredHostTask,
3427        future: impl Future<Output = Result<R>> + Send + 'static,
3428        lower: impl FnOnce(StoreContextMut<T>, Option<R>, bool, Option<TableId<HostTask>>) -> Result<()>
3429        + Send
3430        + 'static,
3431    ) -> Result<u32> {
3432        let token = StoreToken::new(store.as_context_mut());
3433
3434        // Create an abortable future which hooks calls to poll and manages call
3435        // context state for the future.
3436        let (join_handle, future) = JoinHandle::run(future);
3437        let mut future = Box::pin(future);
3438
3439        // Finally, poll the future.  We can use a dummy `Waker` here because
3440        // we'll add the future to `ConcurrentState::futures` and poll it
3441        // automatically from the event loop if it doesn't complete immediately
3442        // here.
3443        let poll = tls::set(store.0, || {
3444            future
3445                .as_mut()
3446                .poll(&mut Context::from_waker(&Waker::noop()))
3447        });
3448
3449        match poll {
3450            // It finished immediately; lower the result and delete the task.
3451            Poll::Ready(result) => {
3452                let result = result.transpose()?;
3453                // Check if the host task was materialized so that it can be
3454                // deleted in `lower`.
3455                let task = store.0.current_materialized_host_task()?;
3456                lower(store.as_context_mut(), result, true, task)?;
3457                return Ok(Status::Returned.pack(None));
3458            }
3459
3460            // Future isn't ready yet, so fall through.
3461            Poll::Pending => {}
3462        }
3463
3464        // The future will outlive this call frame, so materialize the deferred
3465        // host task and attach its cancellation handle before publishing it to
3466        // the event loop.
3467        let Some(task) = store.0.materialize_host_task_id()? else {
3468            bail_bug!("current thread is not a host thread")
3469        };
3470        {
3471            let state = &mut store.0.concurrent_state_mut()?.get_mut(task)?.state;
3472            assert!(matches!(state, HostTaskState::CalleeStarted));
3473            *state = HostTaskState::CalleeRunning(join_handle);
3474        }
3475
3476        // It hasn't finished yet; add the future to
3477        // `ConcurrentState::futures` so it will be polled by the event
3478        // loop and allocate a waitable handle to return to the guest.
3479
3480        // Wrap the future in a closure responsible for lowering the result into
3481        // the guest's stack and memory, as well as notifying any waiters that
3482        // the task returned.
3483        let future = Box::pin(async move {
3484            let result = match run_with_host_task_set(task, future).await? {
3485                Some(result) => Some(result?),
3486                None => None,
3487            };
3488            let on_complete = move |store: &mut dyn VMStore| {
3489                // Restore the `current_thread` to be the host so `lower` knows
3490                // how to manipulate borrows and knows which scope of borrows
3491                // to check.
3492                let mut store = token.as_context_mut(store);
3493                let old = store.0.set_thread(task)?;
3494
3495                let status = if result.is_some() {
3496                    Status::Returned
3497                } else {
3498                    Status::ReturnCancelled
3499                };
3500
3501                lower(store.as_context_mut(), result, false, Some(task))?;
3502                let state = store.0.concurrent_state_mut()?;
3503                match &mut state.get_mut(task)?.state {
3504                    // The task is already flagged as finished because it was
3505                    // cancelled. No need to transition further.
3506                    HostTaskState::CalleeDone { .. } => {}
3507
3508                    // Otherwise transition this task to the done state.
3509                    other => *other = HostTaskState::CalleeDone { cancelled: false },
3510                }
3511                Waitable::Host(task).set_event(state, Some(Event::Subtask { status }))?;
3512
3513                store.0.set_thread(old)?;
3514                Ok(())
3515            };
3516
3517            // Here we schedule a task to run on a worker fiber to do the
3518            // lowering since it may involve a call to the guest's realloc
3519            // function. This is necessary because calling the guest while
3520            // there are host embedder frames on the stack is unsound.
3521            tls::get(move |store| {
3522                store
3523                    .concurrent_state_mut()?
3524                    .push_high_priority(WorkItem::WorkerFunction(AlwaysMut::new(Box::new(
3525                        on_complete,
3526                    ))));
3527                Ok(())
3528            })
3529        });
3530
3531        // Make this task visible to the guest and then record what it
3532        // was made visible as.
3533        let caller = match host_task {
3534            Some(caller) => caller,
3535            None => bail_bug!("host task wasn't created but should have been"),
3536        };
3537        let state = store.0.concurrent_state_mut()?;
3538        state.push_future(future);
3539        let instance = state.get_mut(caller.task)?.instance;
3540        let handle = store
3541            .0
3542            .instance_state(instance)
3543            .handle_table()
3544            .subtask_insert_host(task.rep())?;
3545        store.0.concurrent_state_mut()?.get_mut(task)?.common.handle = Some(handle);
3546        log::trace!("assign {task:?} handle {handle} for {caller:?} instance {instance:?}");
3547
3548        // Restore the currently running thread to this host task's
3549        // caller. Note that the host task isn't deallocated as it's
3550        // within the store and will get deallocated later.
3551        store.0.set_thread(caller)?;
3552        Ok(Status::Started.pack(Some(handle)))
3553    }
3554
3555    /// Implements the `task.return` intrinsic, lifting the result for the
3556    /// current guest task.
3557    pub(crate) fn task_return(
3558        self,
3559        store: &mut dyn VMStore,
3560        ty: TypeTupleIndex,
3561        options: OptionsIndex,
3562        storage: &[ValRaw],
3563    ) -> Result<()> {
3564        let guest_thread = store.current_guest_thread()?;
3565        let state = store.concurrent_state_mut()?;
3566        let lift = state
3567            .get_mut(guest_thread.task)?
3568            .lift_result
3569            .take()
3570            .ok_or_else(|| Trap::TaskCancelOrReturnTwice)?;
3571        if !state.get_mut(guest_thread.task)?.result.is_none() {
3572            bail_bug!("task result unexpectedly already set");
3573        }
3574
3575        let CanonicalOptions {
3576            string_encoding,
3577            data_model,
3578            ..
3579        } = &self.id().get(store).component().env_component().options[options];
3580
3581        let invalid = ty != lift.ty
3582            || string_encoding != &lift.string_encoding
3583            || match data_model {
3584                CanonicalOptionsDataModel::LinearMemory(opts) => match opts.memory {
3585                    Some(memory) => {
3586                        let expected = lift.memory.map(|v| v.as_ptr()).unwrap_or(ptr::null_mut());
3587                        let actual = self.id().get(store).runtime_memory(memory);
3588                        expected != actual.as_ptr()
3589                    }
3590                    // Memory not specified, meaning it didn't need to be
3591                    // specified per validation, so not invalid.
3592                    None => false,
3593                },
3594                // Always invalid as this isn't supported.
3595                CanonicalOptionsDataModel::Gc { .. } => true,
3596            };
3597
3598        if invalid {
3599            bail!(Trap::TaskReturnInvalid);
3600        }
3601
3602        log::trace!("task.return for {guest_thread:?}");
3603
3604        let result = (lift.lift)(store, storage)?;
3605        self.task_complete(store, guest_thread.task, result, Status::Returned)
3606    }
3607
3608    /// Implements the `task.cancel` intrinsic.
3609    pub(crate) fn task_cancel(self, store: &mut StoreOpaque) -> Result<()> {
3610        let guest_thread = store.current_guest_thread()?;
3611        let state = store.concurrent_state_mut()?;
3612        let task = state.get_mut(guest_thread.task)?;
3613        if !task.cancel_sent {
3614            bail!(Trap::TaskCancelNotCancelled);
3615        }
3616        _ = task
3617            .lift_result
3618            .take()
3619            .ok_or_else(|| Trap::TaskCancelOrReturnTwice)?;
3620
3621        if !task.result.is_none() {
3622            bail_bug!("task result should not bet set yet");
3623        }
3624
3625        log::trace!("task.cancel for {guest_thread:?}");
3626
3627        self.task_complete(
3628            store,
3629            guest_thread.task,
3630            Box::new(DummyResult),
3631            Status::ReturnCancelled,
3632        )
3633    }
3634
3635    /// Complete the specified guest task (i.e. indicate that it has either
3636    /// returned a (possibly empty) result or cancelled itself).
3637    ///
3638    /// This will return any resource borrows and notify any current or future
3639    /// waiters that the task has completed.
3640    fn task_complete(
3641        self,
3642        store: &mut StoreOpaque,
3643        guest_task: TableId<GuestTask>,
3644        result: Box<dyn Any + Send + Sync>,
3645        status: Status,
3646    ) -> Result<()> {
3647        store
3648            .component_resource_tables(Some(self))?
3649            .validate_scope_exit()?;
3650
3651        let state = store.concurrent_state_mut()?;
3652        let task = state.get_mut(guest_task)?;
3653
3654        if let Caller::Host { tx, .. } = &mut task.caller {
3655            if let Some(tx) = tx.take() {
3656                _ = tx.send(result);
3657            }
3658        } else {
3659            task.result = Some(result);
3660            Waitable::Guest(guest_task).set_event(state, Some(Event::Subtask { status }))?;
3661        }
3662
3663        Ok(())
3664    }
3665
3666    /// Implements the `waitable-set.new` intrinsic.
3667    pub(crate) fn waitable_set_new(
3668        self,
3669        store: &mut StoreOpaque,
3670        caller_instance: RuntimeComponentInstanceIndex,
3671    ) -> Result<u32> {
3672        let set = store.concurrent_state_mut()?.push(WaitableSet::default())?;
3673        let handle = store
3674            .instance_state(self.runtime_instance(caller_instance))
3675            .handle_table()
3676            .waitable_set_insert(set.rep())?;
3677        log::trace!("new waitable set {set:?} (handle {handle})");
3678        Ok(handle)
3679    }
3680
3681    /// Implements the `waitable-set.drop` intrinsic.
3682    pub(crate) fn waitable_set_drop(
3683        self,
3684        store: &mut StoreOpaque,
3685        caller_instance: RuntimeComponentInstanceIndex,
3686        set: u32,
3687    ) -> Result<()> {
3688        let rep = store
3689            .instance_state(self.runtime_instance(caller_instance))
3690            .handle_table()
3691            .waitable_set_remove(set)?;
3692
3693        log::trace!("drop waitable set {rep} (handle {set})");
3694
3695        // Note that we're careful to check for waiters _before_ deleting the
3696        // set to avoid dropping any waiters in `WaitMode::Fiber(_)`, which
3697        // would panic.  See `drop-waitable-set-with-waiters.wast` for details.
3698        if !store
3699            .concurrent_state_mut()?
3700            .get_mut(TableId::<WaitableSet>::new(rep))?
3701            .waiting
3702            .is_empty()
3703        {
3704            bail!(Trap::WaitableSetDropHasWaiters);
3705        }
3706
3707        store
3708            .concurrent_state_mut()?
3709            .delete(TableId::<WaitableSet>::new(rep))?;
3710
3711        Ok(())
3712    }
3713
3714    /// Implements the `waitable.join` intrinsic.
3715    pub(crate) fn waitable_join(
3716        self,
3717        store: &mut StoreOpaque,
3718        caller_instance: RuntimeComponentInstanceIndex,
3719        waitable_handle: u32,
3720        set_handle: u32,
3721    ) -> Result<()> {
3722        let mut instance = self.id().get_mut(store);
3723        let waitable =
3724            Waitable::from_instance(instance.as_mut(), caller_instance, waitable_handle)?;
3725
3726        let set = if set_handle == 0 {
3727            None
3728        } else {
3729            let set = instance.instance_states().0[caller_instance]
3730                .handle_table()
3731                .waitable_set_rep(set_handle)?;
3732
3733            let state = store.concurrent_state_mut()?;
3734            if let Some(old) = waitable.common(state)?.set
3735                && state.get_mut(old)?.is_sync_call_set
3736            {
3737                bail!(Trap::WaitableSyncAndAsync);
3738            }
3739
3740            Some(TableId::<WaitableSet>::new(set))
3741        };
3742
3743        log::trace!(
3744            "waitable {waitable:?} (handle {waitable_handle}) join set {set:?} (handle {set_handle})",
3745        );
3746
3747        waitable.join(store.concurrent_state_mut()?, set)
3748    }
3749
3750    /// Implements the `subtask.drop` intrinsic.
3751    pub(crate) fn subtask_drop(
3752        self,
3753        store: &mut StoreOpaque,
3754        caller_instance: RuntimeComponentInstanceIndex,
3755        task_id: u32,
3756    ) -> Result<()> {
3757        self.waitable_join(store, caller_instance, task_id, 0)?;
3758
3759        let (rep, is_host) = store
3760            .instance_state(self.runtime_instance(caller_instance))
3761            .handle_table()
3762            .subtask_remove(task_id)?;
3763
3764        let concurrent_state = store.concurrent_state_mut()?;
3765        let (waitable, delete) = if is_host {
3766            let id = TableId::<HostTask>::new(rep);
3767            let task = concurrent_state.get_mut(id)?;
3768            match &task.state {
3769                HostTaskState::CalleeRunning(_) => bail!(Trap::SubtaskDropNotResolved),
3770                HostTaskState::CalleeDone { .. } => {}
3771                HostTaskState::CalleeStarted | HostTaskState::CalleeFinished(_) => {
3772                    bail_bug!("invalid state for callee in `subtask.drop`")
3773                }
3774            }
3775            (Waitable::Host(id), true)
3776        } else {
3777            let id = TableId::<GuestTask>::new(rep);
3778            let task = concurrent_state.get_mut(id)?;
3779            if task.lift_result.is_some() {
3780                bail!(Trap::SubtaskDropNotResolved);
3781            }
3782            (
3783                Waitable::Guest(id),
3784                concurrent_state.get_mut(id)?.ready_to_delete(),
3785            )
3786        };
3787
3788        waitable.common(concurrent_state)?.handle = None;
3789
3790        // If this subtask has an event that means that the terminal status of
3791        // this subtask wasn't yet received so it can't be dropped yet.
3792        if waitable.take_event(concurrent_state)?.is_some() {
3793            bail!(Trap::SubtaskDropNotResolved);
3794        }
3795
3796        if delete {
3797            waitable.delete_from(concurrent_state)?;
3798        }
3799
3800        log::trace!("subtask_drop {waitable:?} (handle {task_id})");
3801        Ok(())
3802    }
3803
3804    /// Implements the `waitable-set.wait` intrinsic.
3805    pub(crate) fn waitable_set_wait(
3806        self,
3807        store: &mut StoreOpaque,
3808        options: OptionsIndex,
3809        set: u32,
3810        payload: u32,
3811    ) -> Result<u32> {
3812        let &CanonicalOptions {
3813            instance: caller_instance,
3814            ..
3815        } = &self.id().get(store).component().env_component().options[options];
3816        let caller = self.runtime_instance(caller_instance);
3817        let rep = store
3818            .instance_state(self.runtime_instance(caller_instance))
3819            .handle_table()
3820            .waitable_set_rep(set)?;
3821
3822        self.waitable_check(
3823            store,
3824            caller,
3825            WaitableCheck::Wait,
3826            WaitableCheckParams {
3827                set: TableId::new(rep),
3828                options,
3829                payload,
3830            },
3831        )
3832    }
3833
3834    /// Implements the `waitable-set.poll` intrinsic.
3835    pub(crate) fn waitable_set_poll(
3836        self,
3837        store: &mut StoreOpaque,
3838        options: OptionsIndex,
3839        set: u32,
3840        payload: u32,
3841    ) -> Result<u32> {
3842        let &CanonicalOptions {
3843            instance: caller_instance,
3844            ..
3845        } = &self.id().get(store).component().env_component().options[options];
3846        let caller = self.runtime_instance(caller_instance);
3847        let rep = store
3848            .instance_state(caller)
3849            .handle_table()
3850            .waitable_set_rep(set)?;
3851
3852        self.waitable_check(
3853            store,
3854            caller,
3855            WaitableCheck::Poll,
3856            WaitableCheckParams {
3857                set: TableId::new(rep),
3858                options,
3859                payload,
3860            },
3861        )
3862    }
3863
3864    /// Implements the `thread.index` intrinsic.
3865    pub(crate) fn thread_index(&self, store: &mut dyn VMStore) -> Result<u32> {
3866        let thread_id = store.current_guest_thread()?.thread;
3867        match store
3868            .concurrent_state_mut()?
3869            .get_mut(thread_id)?
3870            .instance_rep
3871        {
3872            Some(r) => Ok(r),
3873            None => bail_bug!("thread should have instance_rep by now"),
3874        }
3875    }
3876
3877    /// Implements the `thread.new-indirect` intrinsic.
3878    pub(crate) fn thread_new_indirect<T: 'static>(
3879        self,
3880        mut store: StoreContextMut<T>,
3881        runtime_instance: RuntimeComponentInstanceIndex,
3882        _func_ty_idx: TypeFuncIndex, // currently unused
3883        start_func_table_idx: RuntimeTableIndex,
3884        start_func_idx: u32,
3885        context: i32,
3886    ) -> Result<u32> {
3887        log::trace!("creating new thread");
3888
3889        let start_func_ty = FuncType::new(store.engine(), [ValType::I32], []);
3890        let (instance, registry) = self.id().get_mut_and_registry(store.0);
3891        let callee = instance
3892            .index_runtime_func_table(registry, start_func_table_idx, start_func_idx as u64)?
3893            .ok_or_else(|| Trap::ThreadNewIndirectUninitialized)?;
3894        if callee.type_index(store.0) != start_func_ty.type_index() {
3895            bail!(Trap::ThreadNewIndirectInvalidType);
3896        }
3897
3898        let token = StoreToken::new(store.as_context_mut());
3899        let start_func = Box::new(
3900            move |store: &mut dyn VMStore, guest_thread: QualifiedThreadId| -> Result<()> {
3901                let old_thread = store.set_thread(guest_thread)?;
3902                log::trace!(
3903                    "thread start: replaced {old_thread:?} with {guest_thread:?} as current thread"
3904                );
3905
3906                let mut store = token.as_context_mut(store);
3907                let mut params = [ValRaw::i32(context)];
3908                // Use call_unchecked rather than call or call_async, as we don't want to run the function
3909                // on a separate fiber if we're running in an async store.
3910                unsafe { callee.call_unchecked(store.as_context_mut(), &mut params)? };
3911
3912                store.0.set_thread(old_thread)?;
3913
3914                let runtime_instance = self.runtime_instance(runtime_instance);
3915
3916                // We're not returning to any caller, so we may need to switch
3917                // or trap if the instance has an outstanding sync-typed call:
3918                store
3919                    .0
3920                    .switch_or_trap_if_may_not_suspend(runtime_instance)?;
3921
3922                store
3923                    .0
3924                    .cleanup_thread(guest_thread, runtime_instance, CleanupTask::Yes)?;
3925
3926                log::trace!("explicit thread {guest_thread:?} completed");
3927                let state = store.0.concurrent_state_mut()?;
3928                if let Some(t) = old_thread.guest() {
3929                    state.get_mut(t.thread)?.state = GuestThreadState::Running;
3930                }
3931                log::trace!("thread start: restored {old_thread:?} as current thread");
3932
3933                Ok(())
3934            },
3935        );
3936
3937        let current_thread = store.0.current_guest_thread()?;
3938        let state = store.0.concurrent_state_mut()?;
3939        let parent_task = current_thread.task;
3940
3941        let new_thread = GuestThread::new_explicit(state, parent_task, start_func)?;
3942        let thread_id = state.push(new_thread)?;
3943        state.get_mut(parent_task)?.threads.insert(thread_id);
3944
3945        log::trace!("new thread with id {thread_id:?} created");
3946
3947        self.add_guest_thread_to_instance_table(thread_id, store.0, runtime_instance)
3948    }
3949
3950    pub(crate) fn resume_thread(
3951        self,
3952        store: &mut StoreOpaque,
3953        runtime_instance: RuntimeComponentInstanceIndex,
3954        thread_idx: u32,
3955        how: ResumeThread,
3956    ) -> Result<bool> {
3957        let thread_id =
3958            GuestThread::from_instance(self.id().get_mut(store), runtime_instance, thread_idx)?;
3959        let state = store.concurrent_state_mut()?;
3960        let guest_thread = QualifiedThreadId::qualify(state, thread_id)?;
3961
3962        if store.current_guest_thread()? == guest_thread {
3963            bail!(Trap::CannotResumeThread);
3964        }
3965
3966        let state = store.concurrent_state_mut()?;
3967        let thread = state.get_mut(guest_thread.thread)?;
3968        let priority = match how {
3969            ResumeThread::Promote | ResumeThread::Resume => Priority::Switch,
3970            ResumeThread::ResumeLater => Priority::Low,
3971        };
3972
3973        match (&how, &thread.state) {
3974            // Promotion is a noop unless the thread is in a ready state.
3975            (ResumeThread::Promote, GuestThreadState::Ready { .. }) => {}
3976            (ResumeThread::Promote, _) => return Ok(false),
3977
3978            // When resuming a thread it must be in a suspended state otherwise
3979            // this operation is a trap.
3980            (
3981                ResumeThread::Resume | ResumeThread::ResumeLater,
3982                GuestThreadState::NotStartedExplicit(_) | GuestThreadState::Suspended(_),
3983            ) => {}
3984            (ResumeThread::Resume | ResumeThread::ResumeLater, _) => {
3985                bail!(Trap::CannotResumeThread)
3986            }
3987        }
3988
3989        match mem::replace(&mut thread.state, GuestThreadState::Running) {
3990            GuestThreadState::NotStartedExplicit(start_func) => {
3991                log::trace!("starting thread {guest_thread:?}");
3992                let guest_call = WorkItem::GuestCall {
3993                    instance: self.runtime_instance(runtime_instance),
3994                    call: GuestCall {
3995                        thread: guest_thread,
3996                        kind: GuestCallKind::StartExplicit(Box::new(move |store| {
3997                            start_func(store, guest_thread)
3998                        })),
3999                    },
4000                };
4001                store
4002                    .concurrent_state_mut()?
4003                    .push_work_item(guest_call, priority)?;
4004            }
4005            GuestThreadState::Suspended(fiber) => {
4006                log::trace!("resuming thread {thread_id:?} that was suspended");
4007                store.concurrent_state_mut()?.push_work_item(
4008                    WorkItem::ResumeFiber {
4009                        instance: self.runtime_instance(runtime_instance),
4010                        thread: guest_thread,
4011                        fiber,
4012                    },
4013                    priority,
4014                )?;
4015            }
4016            GuestThreadState::Ready { fiber } => {
4017                log::trace!("resuming thread {thread_id:?} that was ready");
4018                thread.state = GuestThreadState::Ready { fiber };
4019                store
4020                    .concurrent_state_mut()?
4021                    .promote_thread_work_item(guest_thread)?;
4022            }
4023            other @ (GuestThreadState::NotStartedImplicit
4024            | GuestThreadState::Running
4025            | GuestThreadState::Completed) => {
4026                thread.state = other;
4027            }
4028        }
4029        Ok(true)
4030    }
4031
4032    fn add_guest_thread_to_instance_table(
4033        self,
4034        thread_id: TableId<GuestThread>,
4035        store: &mut StoreOpaque,
4036        runtime_instance: RuntimeComponentInstanceIndex,
4037    ) -> Result<u32> {
4038        let guest_id = store
4039            .instance_state(self.runtime_instance(runtime_instance))
4040            .thread_handle_table()
4041            .guest_thread_insert(thread_id.rep())?;
4042        store
4043            .concurrent_state_mut()?
4044            .get_mut(thread_id)?
4045            .instance_rep = Some(guest_id);
4046        Ok(guest_id)
4047    }
4048
4049    /// Helper function for the `thread.yield`, thread.suspend`,
4050    /// `thread.suspend-then-resume`, `thread.suspend-then-promote`,
4051    /// `thread.yield-then-resume`, and `thread.yield-then-promote` intrinsics.
4052    pub(crate) fn suspension_intrinsic(
4053        self,
4054        store: &mut StoreOpaque,
4055        caller: RuntimeComponentInstanceIndex,
4056        yielding: bool,
4057        to_thread: SuspensionTarget,
4058    ) -> Result<WaitResult> {
4059        let check_suspend = match to_thread {
4060            SuspensionTarget::Promote(thread) => {
4061                !self.resume_thread(store, caller, thread, ResumeThread::Promote)?
4062            }
4063            SuspensionTarget::Resume(thread) => {
4064                if !self.resume_thread(store, caller, thread, ResumeThread::Resume)? {
4065                    bail_bug!(
4066                        "`resume_thread` should only ever return false \
4067                         when `ResumeThread::Promote` is passed to it"
4068                    );
4069                }
4070                false
4071            }
4072            SuspensionTarget::None => true,
4073        };
4074
4075        if check_suspend && !store.switch_if_may_not_suspend(self.runtime_instance(caller))? {
4076            return if yielding {
4077                Ok(WaitResult::Completed)
4078            } else {
4079                Err(Trap::CannotBlockSyncTask.into())
4080            };
4081        }
4082
4083        let guest_thread = store.current_guest_thread()?;
4084
4085        let reason = if yielding {
4086            SuspendReason::Yielding {
4087                thread: guest_thread,
4088            }
4089        } else {
4090            SuspendReason::ExplicitlySuspending {
4091                thread: guest_thread,
4092            }
4093        };
4094
4095        store.suspend(reason)?;
4096
4097        Ok(WaitResult::Completed)
4098    }
4099
4100    /// Helper function for the `waitable-set.wait` and `waitable-set.poll` intrinsics.
4101    fn waitable_check(
4102        self,
4103        store: &mut StoreOpaque,
4104        caller: RuntimeInstance,
4105        check: WaitableCheck,
4106        params: WaitableCheckParams,
4107    ) -> Result<u32> {
4108        let guest_thread = store.current_guest_thread()?;
4109
4110        log::trace!("waitable check for {guest_thread:?}; set {:?}", params.set);
4111
4112        let state = store.concurrent_state_mut()?;
4113        let task = state.get_mut(guest_thread.task)?;
4114
4115        // If we're waiting, and there are no events immediately available,
4116        // suspend the fiber until that changes.
4117        match &check {
4118            WaitableCheck::Wait => {
4119                let set = params.set;
4120
4121                if (task.event.is_none() || matches!(task.event, Some(Event::Cancelled)))
4122                    && state.get_mut(set)?.ready.is_empty()
4123                {
4124                    store.switch_or_trap_if_may_not_suspend(caller)?;
4125
4126                    store.suspend(SuspendReason::Waiting {
4127                        set,
4128                        thread: guest_thread,
4129                    })?;
4130                }
4131            }
4132            WaitableCheck::Poll => {}
4133        }
4134
4135        log::trace!(
4136            "waitable check for {guest_thread:?}; set {:?}, part two",
4137            params.set
4138        );
4139
4140        // Deliver any pending events to the guest and return.
4141        let event = self.get_event(store, guest_thread.task, Some(params.set), false)?;
4142
4143        let (ordinal, handle, result) = match &check {
4144            WaitableCheck::Wait => {
4145                let (event, waitable) = match event {
4146                    Some(p) => p,
4147                    None => bail_bug!("event expected to be present"),
4148                };
4149                let handle = waitable.map(|(_, v)| v).unwrap_or(0);
4150                let (ordinal, result) = event.parts();
4151                (ordinal, handle, result)
4152            }
4153            WaitableCheck::Poll => {
4154                if let Some((event, waitable)) = event {
4155                    let handle = waitable.map(|(_, v)| v).unwrap_or(0);
4156                    let (ordinal, result) = event.parts();
4157                    (ordinal, handle, result)
4158                } else {
4159                    log::trace!(
4160                        "no events ready to deliver via waitable-set.poll to {:?}; set {:?}",
4161                        guest_thread.task,
4162                        params.set
4163                    );
4164                    let (ordinal, result) = Event::None.parts();
4165                    (ordinal, 0, result)
4166                }
4167            }
4168        };
4169        let memory = self.options_memory_mut(store, params.options);
4170        let ptr = crate::component::func::validate_inbounds_dynamic(
4171            &CanonicalAbiInfo::POINTER_PAIR,
4172            memory,
4173            &ValRaw::u32(params.payload),
4174        )?;
4175        memory[ptr + 0..][..4].copy_from_slice(&handle.to_le_bytes());
4176        memory[ptr + 4..][..4].copy_from_slice(&result.to_le_bytes());
4177        Ok(ordinal)
4178    }
4179
4180    /// Implements the `subtask.cancel` intrinsic.
4181    pub(crate) fn subtask_cancel(
4182        self,
4183        store: &mut StoreOpaque,
4184        caller_instance: RuntimeComponentInstanceIndex,
4185        async_: bool,
4186        task_id: u32,
4187    ) -> Result<u32> {
4188        let (rep, is_host) = store
4189            .instance_state(self.runtime_instance(caller_instance))
4190            .handle_table()
4191            .subtask_rep(task_id)?;
4192        let waitable = if is_host {
4193            Waitable::Host(TableId::<HostTask>::new(rep))
4194        } else {
4195            Waitable::Guest(TableId::<GuestTask>::new(rep))
4196        };
4197        let concurrent_state = store.concurrent_state_mut()?;
4198
4199        log::trace!("subtask_cancel {waitable:?} (handle {task_id}; async {async_})");
4200
4201        waitable.trap_if_in_waitable_set(concurrent_state)?;
4202
4203        let needs_block;
4204        if let Waitable::Host(host_task) = waitable {
4205            let state = &mut concurrent_state.get_mut(host_task)?.state;
4206            match mem::replace(state, HostTaskState::CalleeDone { cancelled: true }) {
4207                // If the callee is still running, signal an abort is requested.
4208                //
4209                // After cancelling this falls through to block waiting for the
4210                // host task to actually finish assuming that `async_` is false.
4211                // This blocking behavior resolves the race of `handle.abort()`
4212                // with the task actually getting cancelled or finishing.
4213                HostTaskState::CalleeRunning(handle) => {
4214                    handle.abort();
4215                    needs_block = true;
4216                }
4217
4218                // Cancellation was already requested, so fail as the task can't
4219                // be cancelled twice.
4220                HostTaskState::CalleeDone { cancelled } => {
4221                    if cancelled {
4222                        bail!(Trap::SubtaskCancelAfterTerminal);
4223                    } else {
4224                        // The callee is already done so there's no need to
4225                        // block further for an event.
4226                        needs_block = false;
4227                    }
4228                }
4229
4230                // These states should not be possible for a subtask that's
4231                // visible from the guest, so trap here.
4232                HostTaskState::CalleeStarted | HostTaskState::CalleeFinished(_) => {
4233                    bail_bug!("invalid states for host callee")
4234                }
4235            }
4236        } else {
4237            let guest_task = TableId::<GuestTask>::new(rep);
4238            let task = concurrent_state.get_mut(guest_task)?;
4239            if !task.already_lowered_parameters() {
4240                store.cancel_guest_subtask_without_lowered_parameters(
4241                    self.runtime_instance(caller_instance),
4242                    guest_task,
4243                )?;
4244                return Ok(Status::StartCancelled as u32);
4245            } else if !task.returned_or_cancelled() {
4246                // Started, but not yet returned or cancelled; send the
4247                // `CANCELLED` event
4248                task.cancel_sent = true;
4249                // Note that this might overwrite an event that was set earlier
4250                // (e.g. `Event::None` if the task is yielding, or
4251                // `Event::Cancelled` if it was already cancelled), but that's
4252                // okay -- this should supersede the previous state.
4253                task.event = Some(Event::Cancelled);
4254                let runtime_instance = task.instance;
4255                for thread in task.threads.clone() {
4256                    let thread = QualifiedThreadId {
4257                        task: guest_task,
4258                        thread,
4259                    };
4260                    let thread_mut = concurrent_state.get_mut(thread.thread)?;
4261
4262                    let yield_ = |store: &mut StoreOpaque| {
4263                        // While we're yielding, temporarily set
4264                        // `do_not_suspend` to false on the subtask's instance
4265                        // since we'll be getting control back if it does
4266                        // suspend.
4267                        let state = store.instance_state(runtime_instance).concurrent_state();
4268                        let old_do_not_suspend = state.do_not_suspend;
4269                        state.do_not_suspend = false;
4270
4271                        let caller = store.current_guest_thread()?;
4272
4273                        // Temporarily add the waitable to the caller's
4274                        // `sync_call_set` to ensure that (1) it isn't already
4275                        // part of a different set and (2) it can't be added to
4276                        // a different set while we yield to the subtask.
4277                        let state = store.concurrent_state_mut()?;
4278                        let set = state.get_mut(caller.thread)?.sync_call_set;
4279                        waitable.join(state, Some(set))?;
4280
4281                        store.suspend(SuspendReason::Yielding { thread: caller })?;
4282
4283                        let state = store.concurrent_state_mut()?;
4284                        waitable.join(state, None)?;
4285
4286                        store
4287                            .instance_state(runtime_instance)
4288                            .concurrent_state()
4289                            .do_not_suspend = old_do_not_suspend;
4290
4291                        Ok::<(), crate::Error>(())
4292                    };
4293
4294                    if let Some(set) = thread_mut.wake_on_cancel.take() {
4295                        // The thread is in a cancellable wait, so wake it up:
4296                        let item = match concurrent_state.get_mut(set)?.waiting.remove(&thread) {
4297                            Some(WaitMode::Fiber(fiber)) => WorkItem::ResumeFiber {
4298                                instance: runtime_instance,
4299                                thread,
4300                                fiber,
4301                            },
4302                            Some(WaitMode::Callback(instance)) => WorkItem::GuestCall {
4303                                instance: runtime_instance,
4304                                call: GuestCall {
4305                                    thread,
4306                                    kind: GuestCallKind::DeliverEvent {
4307                                        instance,
4308                                        set: None,
4309                                    },
4310                                },
4311                            },
4312                            Some(WaitMode::Caller { .. }) => {
4313                                bail_bug!("unexpected `WaitMode::Caller` in wake_on_cancel set")
4314                            }
4315                            None => bail_bug!("thread not present in wake_on_cancel set"),
4316                        };
4317                        concurrent_state.set_switch_item(item)?;
4318
4319                        yield_(store)?;
4320
4321                        break;
4322                    }
4323                }
4324
4325                // Guest tasks need to block if they have not yet returned or
4326                // cancelled, even as a result of the event delivery above.
4327                needs_block = !store
4328                    .concurrent_state_mut()?
4329                    .get_mut(guest_task)?
4330                    .returned_or_cancelled()
4331            } else {
4332                needs_block = false;
4333            }
4334        };
4335
4336        // If we need to block waiting on the terminal status of this subtask
4337        // then return immediately in `async` mode, or otherwise wait for the
4338        // event to get signaled through the store.
4339        if needs_block {
4340            if async_ {
4341                return Ok(BLOCKED);
4342            }
4343
4344            // Wait for this waitable to get signaled with its terminal
4345            // status. Once that's done fall through to the shared code.
4346            store.wait_for_event(
4347                self.runtime_instance(caller_instance),
4348                waitable,
4349                if is_host {
4350                    WaitReason::Other
4351                } else {
4352                    WaitReason::GuestSubtask(TableId::<GuestTask>::new(rep))
4353                },
4354            )?;
4355
4356            // .. fall through to determine what event's in store for us.
4357        }
4358
4359        let event = waitable.take_event(store.concurrent_state_mut()?)?;
4360        if let Some(Event::Subtask {
4361            status: status @ (Status::Returned | Status::ReturnCancelled),
4362        }) = event
4363        {
4364            Ok(status as u32)
4365        } else {
4366            bail!(Trap::SubtaskCancelAfterTerminal);
4367        }
4368    }
4369}
4370
4371/// Trait representing component model ABI async intrinsics and fused adapter
4372/// helper functions.
4373///
4374/// SAFETY (callers): Most of the methods in this trait accept raw pointers,
4375/// which must be valid for at least the duration of the call (and possibly for
4376/// as long as the relevant guest task exists, in the case of `*mut VMFuncRef`
4377/// pointers used for async calls).
4378pub trait VMComponentAsyncStore {
4379    /// A helper function for fused adapter modules involving calls where the
4380    /// one of the caller or callee is async.
4381    ///
4382    /// This helper is not used when the caller and callee both use the sync
4383    /// ABI, only when at least one is async is this used.
4384    unsafe fn prepare_call(
4385        &mut self,
4386        instance: Instance,
4387        memory: *mut VMMemoryDefinition,
4388        start: NonNull<VMFuncRef>,
4389        return_: NonNull<VMFuncRef>,
4390        caller_instance: RuntimeComponentInstanceIndex,
4391        callee_instance: RuntimeComponentInstanceIndex,
4392        task_return_type: TypeTupleIndex,
4393        callee_async: bool,
4394        string_encoding: StringEncoding,
4395        result_count: u32,
4396        storage: *mut ValRaw,
4397        storage_len: usize,
4398    ) -> Result<()>;
4399
4400    /// A helper function for fused adapter modules involving calls where the
4401    /// caller is sync-lowered but the callee is async-lifted.
4402    unsafe fn sync_start(
4403        &mut self,
4404        instance: Instance,
4405        callback: *mut VMFuncRef,
4406        callee: NonNull<VMFuncRef>,
4407        param_count: u32,
4408        storage: *mut MaybeUninit<ValRaw>,
4409        storage_len: usize,
4410    ) -> Result<()>;
4411
4412    /// A helper function for fused adapter modules involving calls where the
4413    /// caller is async-lowered.
4414    unsafe fn async_start(
4415        &mut self,
4416        instance: Instance,
4417        callback: *mut VMFuncRef,
4418        post_return: *mut VMFuncRef,
4419        callee: NonNull<VMFuncRef>,
4420        param_count: u32,
4421        result_count: u32,
4422        flags: u32,
4423    ) -> Result<u32>;
4424
4425    /// The `future.write` intrinsic.
4426    fn future_write(
4427        &mut self,
4428        instance: Instance,
4429        caller: RuntimeComponentInstanceIndex,
4430        ty: TypeFutureTableIndex,
4431        options: OptionsIndex,
4432        future: u32,
4433        address: u32,
4434    ) -> Result<u32>;
4435
4436    /// The `future.read` intrinsic.
4437    fn future_read(
4438        &mut self,
4439        instance: Instance,
4440        caller: RuntimeComponentInstanceIndex,
4441        ty: TypeFutureTableIndex,
4442        options: OptionsIndex,
4443        future: u32,
4444        address: u32,
4445    ) -> Result<u32>;
4446
4447    /// The `future.drop-writable` intrinsic.
4448    fn future_drop_writable(
4449        &mut self,
4450        instance: Instance,
4451        ty: TypeFutureTableIndex,
4452        writer: u32,
4453    ) -> Result<()>;
4454
4455    /// The `stream.write` intrinsic.
4456    fn stream_write(
4457        &mut self,
4458        instance: Instance,
4459        caller: RuntimeComponentInstanceIndex,
4460        ty: TypeStreamTableIndex,
4461        options: OptionsIndex,
4462        stream: u32,
4463        address: u32,
4464        count: u32,
4465    ) -> Result<u32>;
4466
4467    /// The `stream.read` intrinsic.
4468    fn stream_read(
4469        &mut self,
4470        instance: Instance,
4471        caller: RuntimeComponentInstanceIndex,
4472        ty: TypeStreamTableIndex,
4473        options: OptionsIndex,
4474        stream: u32,
4475        address: u32,
4476        count: u32,
4477    ) -> Result<u32>;
4478
4479    /// The "fast-path" implementation of the `stream.write` intrinsic for
4480    /// "flat" (i.e. memcpy-able) payloads.
4481    fn flat_stream_write(
4482        &mut self,
4483        instance: Instance,
4484        caller: RuntimeComponentInstanceIndex,
4485        ty: TypeStreamTableIndex,
4486        options: OptionsIndex,
4487        payload_size: u32,
4488        payload_align: u32,
4489        stream: u32,
4490        address: u32,
4491        count: u32,
4492    ) -> Result<u32>;
4493
4494    /// The "fast-path" implementation of the `stream.read` intrinsic for "flat"
4495    /// (i.e. memcpy-able) payloads.
4496    fn flat_stream_read(
4497        &mut self,
4498        instance: Instance,
4499        caller: RuntimeComponentInstanceIndex,
4500        ty: TypeStreamTableIndex,
4501        options: OptionsIndex,
4502        payload_size: u32,
4503        payload_align: u32,
4504        stream: u32,
4505        address: u32,
4506        count: u32,
4507    ) -> Result<u32>;
4508
4509    /// The `stream.drop-writable` intrinsic.
4510    fn stream_drop_writable(
4511        &mut self,
4512        instance: Instance,
4513        ty: TypeStreamTableIndex,
4514        writer: u32,
4515    ) -> Result<()>;
4516
4517    /// The `error-context.debug-message` intrinsic.
4518    fn error_context_debug_message(
4519        &mut self,
4520        instance: Instance,
4521        ty: TypeComponentLocalErrorContextTableIndex,
4522        options: OptionsIndex,
4523        err_ctx_handle: u32,
4524        debug_msg_address: u32,
4525    ) -> Result<()>;
4526
4527    /// The `thread.new-indirect` intrinsic
4528    fn thread_new_indirect(
4529        &mut self,
4530        instance: Instance,
4531        caller: RuntimeComponentInstanceIndex,
4532        func_ty_idx: TypeFuncIndex,
4533        start_func_table_idx: RuntimeTableIndex,
4534        start_func_idx: u32,
4535        context: i32,
4536    ) -> Result<u32>;
4537}
4538
4539/// SAFETY: See trait docs.
4540impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
4541    unsafe fn prepare_call(
4542        &mut self,
4543        instance: Instance,
4544        memory: *mut VMMemoryDefinition,
4545        start: NonNull<VMFuncRef>,
4546        return_: NonNull<VMFuncRef>,
4547        caller_instance: RuntimeComponentInstanceIndex,
4548        callee_instance: RuntimeComponentInstanceIndex,
4549        task_return_type: TypeTupleIndex,
4550        callee_async: bool,
4551        string_encoding: StringEncoding,
4552        result_count_or_max_if_async: u32,
4553        storage: *mut ValRaw,
4554        storage_len: usize,
4555    ) -> Result<()> {
4556        // SAFETY: The `wasmtime_cranelift`-generated code that calls
4557        // this method will have ensured that `storage` is a valid
4558        // pointer containing at least `storage_len` items.
4559        let params = unsafe { core::slice::from_raw_parts(storage, storage_len) }.to_vec();
4560
4561        unsafe {
4562            instance.prepare_call(
4563                StoreContextMut(self),
4564                start,
4565                return_,
4566                caller_instance,
4567                callee_instance,
4568                task_return_type,
4569                callee_async,
4570                memory,
4571                string_encoding,
4572                match result_count_or_max_if_async {
4573                    PREPARE_ASYNC_NO_RESULT => CallerInfo::Async {
4574                        params,
4575                        has_result: false,
4576                    },
4577                    PREPARE_ASYNC_WITH_RESULT => CallerInfo::Async {
4578                        params,
4579                        has_result: true,
4580                    },
4581                    result_count => CallerInfo::Sync {
4582                        params,
4583                        result_count,
4584                    },
4585                },
4586            )
4587        }
4588    }
4589
4590    unsafe fn sync_start(
4591        &mut self,
4592        instance: Instance,
4593        callback: *mut VMFuncRef,
4594        callee: NonNull<VMFuncRef>,
4595        param_count: u32,
4596        storage: *mut MaybeUninit<ValRaw>,
4597        storage_len: usize,
4598    ) -> Result<()> {
4599        unsafe {
4600            instance
4601                .start_call(
4602                    StoreContextMut(self),
4603                    callback,
4604                    ptr::null_mut(),
4605                    callee,
4606                    param_count,
4607                    1,
4608                    START_FLAG_ASYNC_CALLEE,
4609                    // SAFETY: The `wasmtime_cranelift`-generated code that calls
4610                    // this method will have ensured that `storage` is a valid
4611                    // pointer containing at least `storage_len` items.
4612                    Some(core::slice::from_raw_parts_mut(storage, storage_len)),
4613                )
4614                .map(drop)
4615        }
4616    }
4617
4618    unsafe fn async_start(
4619        &mut self,
4620        instance: Instance,
4621        callback: *mut VMFuncRef,
4622        post_return: *mut VMFuncRef,
4623        callee: NonNull<VMFuncRef>,
4624        param_count: u32,
4625        result_count: u32,
4626        flags: u32,
4627    ) -> Result<u32> {
4628        unsafe {
4629            instance.start_call(
4630                StoreContextMut(self),
4631                callback,
4632                post_return,
4633                callee,
4634                param_count,
4635                result_count,
4636                flags,
4637                None,
4638            )
4639        }
4640    }
4641
4642    fn future_write(
4643        &mut self,
4644        instance: Instance,
4645        caller: RuntimeComponentInstanceIndex,
4646        ty: TypeFutureTableIndex,
4647        options: OptionsIndex,
4648        future: u32,
4649        address: u32,
4650    ) -> Result<u32> {
4651        instance
4652            .guest_write(
4653                StoreContextMut(self),
4654                caller,
4655                TransmitIndex::Future(ty),
4656                options,
4657                None,
4658                future,
4659                address,
4660                1,
4661            )
4662            .map(|result| result.encode())
4663    }
4664
4665    fn future_read(
4666        &mut self,
4667        instance: Instance,
4668        caller: RuntimeComponentInstanceIndex,
4669        ty: TypeFutureTableIndex,
4670        options: OptionsIndex,
4671        future: u32,
4672        address: u32,
4673    ) -> Result<u32> {
4674        instance
4675            .guest_read(
4676                StoreContextMut(self),
4677                caller,
4678                TransmitIndex::Future(ty),
4679                options,
4680                None,
4681                future,
4682                address,
4683                1,
4684            )
4685            .map(|result| result.encode())
4686    }
4687
4688    fn stream_write(
4689        &mut self,
4690        instance: Instance,
4691        caller: RuntimeComponentInstanceIndex,
4692        ty: TypeStreamTableIndex,
4693        options: OptionsIndex,
4694        stream: u32,
4695        address: u32,
4696        count: u32,
4697    ) -> Result<u32> {
4698        instance
4699            .guest_write(
4700                StoreContextMut(self),
4701                caller,
4702                TransmitIndex::Stream(ty),
4703                options,
4704                None,
4705                stream,
4706                address,
4707                count,
4708            )
4709            .map(|result| result.encode())
4710    }
4711
4712    fn stream_read(
4713        &mut self,
4714        instance: Instance,
4715        caller: RuntimeComponentInstanceIndex,
4716        ty: TypeStreamTableIndex,
4717        options: OptionsIndex,
4718        stream: u32,
4719        address: u32,
4720        count: u32,
4721    ) -> Result<u32> {
4722        instance
4723            .guest_read(
4724                StoreContextMut(self),
4725                caller,
4726                TransmitIndex::Stream(ty),
4727                options,
4728                None,
4729                stream,
4730                address,
4731                count,
4732            )
4733            .map(|result| result.encode())
4734    }
4735
4736    fn future_drop_writable(
4737        &mut self,
4738        instance: Instance,
4739        ty: TypeFutureTableIndex,
4740        writer: u32,
4741    ) -> Result<()> {
4742        instance.guest_drop_writable(self, TransmitIndex::Future(ty), writer)
4743    }
4744
4745    fn flat_stream_write(
4746        &mut self,
4747        instance: Instance,
4748        caller: RuntimeComponentInstanceIndex,
4749        ty: TypeStreamTableIndex,
4750        options: OptionsIndex,
4751        payload_size: u32,
4752        payload_align: u32,
4753        stream: u32,
4754        address: u32,
4755        count: u32,
4756    ) -> Result<u32> {
4757        instance
4758            .guest_write(
4759                StoreContextMut(self),
4760                caller,
4761                TransmitIndex::Stream(ty),
4762                options,
4763                Some(FlatAbi {
4764                    size: payload_size,
4765                    align: payload_align,
4766                }),
4767                stream,
4768                address,
4769                count,
4770            )
4771            .map(|result| result.encode())
4772    }
4773
4774    fn flat_stream_read(
4775        &mut self,
4776        instance: Instance,
4777        caller: RuntimeComponentInstanceIndex,
4778        ty: TypeStreamTableIndex,
4779        options: OptionsIndex,
4780        payload_size: u32,
4781        payload_align: u32,
4782        stream: u32,
4783        address: u32,
4784        count: u32,
4785    ) -> Result<u32> {
4786        instance
4787            .guest_read(
4788                StoreContextMut(self),
4789                caller,
4790                TransmitIndex::Stream(ty),
4791                options,
4792                Some(FlatAbi {
4793                    size: payload_size,
4794                    align: payload_align,
4795                }),
4796                stream,
4797                address,
4798                count,
4799            )
4800            .map(|result| result.encode())
4801    }
4802
4803    fn stream_drop_writable(
4804        &mut self,
4805        instance: Instance,
4806        ty: TypeStreamTableIndex,
4807        writer: u32,
4808    ) -> Result<()> {
4809        instance.guest_drop_writable(self, TransmitIndex::Stream(ty), writer)
4810    }
4811
4812    fn error_context_debug_message(
4813        &mut self,
4814        instance: Instance,
4815        ty: TypeComponentLocalErrorContextTableIndex,
4816        options: OptionsIndex,
4817        err_ctx_handle: u32,
4818        debug_msg_address: u32,
4819    ) -> Result<()> {
4820        instance.error_context_debug_message(
4821            StoreContextMut(self),
4822            ty,
4823            options,
4824            err_ctx_handle,
4825            debug_msg_address,
4826        )
4827    }
4828
4829    fn thread_new_indirect(
4830        &mut self,
4831        instance: Instance,
4832        caller: RuntimeComponentInstanceIndex,
4833        func_ty_idx: TypeFuncIndex,
4834        start_func_table_idx: RuntimeTableIndex,
4835        start_func_idx: u32,
4836        context: i32,
4837    ) -> Result<u32> {
4838        instance.thread_new_indirect(
4839            StoreContextMut(self),
4840            caller,
4841            func_ty_idx,
4842            start_func_table_idx,
4843            start_func_idx,
4844            context,
4845        )
4846    }
4847}
4848
4849type HostTaskFuture = Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>;
4850
4851/// Runs the given future with the current thread set to `task` each time it is
4852/// polled.
4853async fn run_with_host_task_set<F>(task: TableId<HostTask>, future: F) -> Result<F::Output>
4854where
4855    F: Future,
4856{
4857    let mut future = pin!(future);
4858    future::poll_fn(|cx| {
4859        let old_thread = match tls::get(|store| store.set_thread(task)) {
4860            Ok(thread) => thread,
4861            Err(error) => return Poll::Ready(Err(error)),
4862        };
4863        let result = future.as_mut().poll(cx);
4864        match tls::get(|store| store.set_thread(old_thread)) {
4865            Ok(_) => result.map(Ok),
4866            Err(error) => Poll::Ready(Err(error)),
4867        }
4868    })
4869    .await
4870}
4871
4872/// Represents the state of a pending host task.
4873///
4874/// This is used to represent tasks when the guest calls into the host.
4875pub(crate) struct HostTask {
4876    common: WaitableCommon,
4877
4878    /// The calling guest task of this host task. Note that this is only used
4879    /// for backtrace purposes and is additionally not updated when the guest
4880    /// task finishes.
4881    ///
4882    /// TODO: this field should probably get deleted entirely and the
4883    /// backtrace-purposes here should move to an auxiliary table.
4884    caller: TableId<GuestTask>,
4885
4886    /// State of borrows/etc the host needs to track. Used when the guest passes
4887    /// borrows to the host, for example.
4888    call_context: CallContext,
4889
4890    state: HostTaskState,
4891}
4892
4893enum HostTaskState {
4894    /// A host task has been created and it's considered "started".
4895    ///
4896    /// The host task has yet to enter `first_poll` or `poll_and_block` which
4897    /// is where this will get updated further.
4898    CalleeStarted,
4899
4900    /// State used for tasks in `first_poll` meaning that the guest did an async
4901    /// lower of a host async function which is blocked. The specified handle is
4902    /// linked to the future in the main `FuturesUnordered` of a store which is
4903    /// used to cancel it if the guest requests cancellation.
4904    CalleeRunning(JoinHandle),
4905
4906    /// Terminal state used for tasks in `poll_and_block` to store the result of
4907    /// their computation. Note that this state is not used for tasks in
4908    /// `first_poll`.
4909    CalleeFinished(LiftedResult),
4910
4911    /// Terminal state for host tasks meaning that the task was cancelled or the
4912    /// result was taken.
4913    CalleeDone { cancelled: bool },
4914}
4915
4916impl HostTask {
4917    fn new(caller: TableId<GuestTask>, state: HostTaskState) -> Self {
4918        Self {
4919            common: WaitableCommon::default(),
4920            call_context: CallContext::default(),
4921            caller,
4922            state,
4923        }
4924    }
4925}
4926
4927impl TableDebug for HostTask {
4928    fn type_name() -> &'static str {
4929        "HostTask"
4930    }
4931}
4932
4933type CallbackFn = Box<dyn Fn(&mut dyn VMStore, Event, u32) -> Result<u32> + Send + Sync + 'static>;
4934
4935/// Represents the caller of a given guest task.
4936enum Caller {
4937    /// The host called the guest task.
4938    Host {
4939        /// If present, may be used to deliver the result.
4940        tx: Option<oneshot::Sender<LiftedResult>>,
4941        /// If true, there's a host future that must be dropped before the task
4942        /// can be deleted.
4943        host_future_present: bool,
4944        /// The host task which called into the guest, or `None` for a call from
4945        /// the top-level host. The task may belong to an entirely unrelated
4946        /// top-level component instance than the one the host called into.
4947        caller: Option<TableId<HostTask>>,
4948    },
4949    /// Another guest thread called the guest task
4950    Guest {
4951        /// The id of the caller
4952        thread: QualifiedThreadId,
4953    },
4954}
4955
4956/// Represents a closure and related canonical ABI parameters required to
4957/// validate a `task.return` call at runtime and lift the result.
4958struct LiftResult {
4959    lift: RawLift,
4960    ty: TypeTupleIndex,
4961    memory: Option<SendSyncPtr<VMMemoryDefinition>>,
4962    string_encoding: StringEncoding,
4963}
4964
4965/// The table ID for a guest thread, qualified by the task to which it belongs.
4966///
4967/// This exists to minimize table lookups and the necessity to pass stores around mutably
4968/// for the common case of identifying the task to which a thread belongs.
4969#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
4970pub(crate) struct QualifiedThreadId {
4971    task: TableId<GuestTask>,
4972    thread: TableId<GuestThread>,
4973}
4974
4975impl QualifiedThreadId {
4976    fn qualify(
4977        state: &mut ConcurrentState,
4978        thread: TableId<GuestThread>,
4979    ) -> Result<QualifiedThreadId> {
4980        Ok(QualifiedThreadId {
4981            task: state.get_mut(thread)?.parent_task,
4982            thread,
4983        })
4984    }
4985}
4986
4987impl fmt::Debug for QualifiedThreadId {
4988    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
4989        f.debug_tuple("QualifiedThreadId")
4990            .field(&self.task.rep())
4991            .field(&self.thread.rep())
4992            .finish()
4993    }
4994}
4995
4996enum GuestThreadState {
4997    NotStartedImplicit,
4998    NotStartedExplicit(
4999        Box<dyn FnOnce(&mut dyn VMStore, QualifiedThreadId) -> Result<()> + Send + Sync>,
5000    ),
5001    Running,
5002    Suspended(StoreFiber<'static>),
5003    Ready {
5004        fiber: StoreFiber<'static>,
5005    },
5006    Completed,
5007}
5008
5009impl fmt::Debug for GuestThreadState {
5010    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
5011        match self {
5012            Self::NotStartedImplicit => f.debug_tuple("NotStartedImplicit").finish(),
5013            Self::NotStartedExplicit(_) => f.debug_tuple("NotStartedExplicit").finish(),
5014            Self::Running => f.debug_tuple("Running").finish(),
5015            Self::Suspended(_) => f.debug_tuple("Suspended").finish(),
5016            Self::Ready { .. } => f.debug_struct("Ready").finish(),
5017            Self::Completed => f.debug_tuple("Completed").finish(),
5018        }
5019    }
5020}
5021
5022pub struct GuestThread {
5023    /// Context-local state used to implement the `context.{get,set}`
5024    /// intrinsics.
5025    context: [u32; NUM_COMPONENT_CONTEXT_SLOTS],
5026    /// The owning guest task.
5027    parent_task: TableId<GuestTask>,
5028    /// If present, indicates that the thread is currently waiting on the
5029    /// specified set but may be cancelled and woken immediately.
5030    wake_on_cancel: Option<TableId<WaitableSet>>,
5031    /// The execution state of this guest thread
5032    state: GuestThreadState,
5033    /// The index of this thread in the component instance's handle table.
5034    /// This must always be `Some` after initialization.
5035    instance_rep: Option<u32>,
5036    /// Scratch waitable set used to watch subtasks during synchronous calls.
5037    sync_call_set: TableId<WaitableSet>,
5038    /// The old value of `do_not_suspend` prior to the sync-typed task for which
5039    /// this thread was created, if relevant.
5040    old_do_not_suspend: Option<bool>,
5041}
5042
5043impl GuestThread {
5044    /// Retrieve the `GuestThread` corresponding to the specified guest-visible
5045    /// handle.
5046    fn from_instance(
5047        state: Pin<&mut ComponentInstance>,
5048        caller_instance: RuntimeComponentInstanceIndex,
5049        guest_thread: u32,
5050    ) -> Result<TableId<Self>> {
5051        let rep = state.instance_states().0[caller_instance]
5052            .thread_handle_table()
5053            .guest_thread_rep(guest_thread)?;
5054        Ok(TableId::new(rep))
5055    }
5056
5057    fn new_implicit(state: &mut ConcurrentState, parent_task: TableId<GuestTask>) -> Result<Self> {
5058        let sync_call_set = state.push(WaitableSet {
5059            is_sync_call_set: true,
5060            ..WaitableSet::default()
5061        })?;
5062        Ok(Self {
5063            context: [0; NUM_COMPONENT_CONTEXT_SLOTS],
5064            parent_task,
5065            wake_on_cancel: None,
5066            state: GuestThreadState::NotStartedImplicit,
5067            instance_rep: None,
5068            sync_call_set,
5069            old_do_not_suspend: None,
5070        })
5071    }
5072
5073    fn new_explicit(
5074        state: &mut ConcurrentState,
5075        parent_task: TableId<GuestTask>,
5076        start_func: Box<
5077            dyn FnOnce(&mut dyn VMStore, QualifiedThreadId) -> Result<()> + Send + Sync,
5078        >,
5079    ) -> Result<Self> {
5080        let sync_call_set = state.push(WaitableSet {
5081            is_sync_call_set: true,
5082            ..WaitableSet::default()
5083        })?;
5084        Ok(Self {
5085            context: [0; NUM_COMPONENT_CONTEXT_SLOTS],
5086            parent_task,
5087            wake_on_cancel: None,
5088            state: GuestThreadState::NotStartedExplicit(start_func),
5089            instance_rep: None,
5090            sync_call_set,
5091            old_do_not_suspend: None,
5092        })
5093    }
5094}
5095
5096impl TableDebug for GuestThread {
5097    fn type_name() -> &'static str {
5098        "GuestThread"
5099    }
5100}
5101
5102enum SyncResult {
5103    NotProduced,
5104    Produced(Option<ValRaw>),
5105    Taken,
5106}
5107
5108impl SyncResult {
5109    fn take(&mut self) -> Result<Option<Option<ValRaw>>> {
5110        Ok(match mem::replace(self, SyncResult::Taken) {
5111            SyncResult::NotProduced => None,
5112            SyncResult::Produced(val) => Some(val),
5113            SyncResult::Taken => {
5114                bail_bug!("attempted to take a synchronous result that was already taken")
5115            }
5116        })
5117    }
5118}
5119
5120#[derive(Debug)]
5121enum HostFutureState {
5122    NotApplicable,
5123    Live,
5124    Dropped,
5125}
5126
5127/// Represents a pending guest task.
5128pub(crate) struct GuestTask {
5129    /// See `WaitableCommon`
5130    common: WaitableCommon,
5131    /// Closure to lower the parameters passed to this task.
5132    lower_params: Option<RawLower>,
5133    /// See `LiftResult`
5134    lift_result: Option<LiftResult>,
5135    /// A place to stash the type-erased lifted result if it can't be delivered
5136    /// immediately.
5137    result: Option<LiftedResult>,
5138    /// Closure to call the callback function for an async-lifted export, if
5139    /// provided.
5140    callback: Option<CallbackFn>,
5141    /// See `Caller`
5142    caller: Caller,
5143    /// Borrow state for this task.
5144    ///
5145    /// Keeps track of `borrow<T>` received to this task to ensure that
5146    /// everything is dropped by the time it exits.
5147    call_context: CallContext,
5148    /// A place to stash the lowered result for a sync-to-async call until it
5149    /// can be returned to the caller.
5150    sync_result: SyncResult,
5151    /// Whether or not the task has been cancelled (i.e. whether the task is
5152    /// permitted to call `task.cancel`).
5153    cancel_sent: bool,
5154    /// Whether or not we've sent a `Status::Starting` event to any current or
5155    /// future waiters for this waitable.
5156    starting_sent: bool,
5157    /// The runtime instance to which the exported function for this guest task
5158    /// belongs.
5159    ///
5160    /// Note that the task may do a sync->sync call via a fused adapter which
5161    /// results in that task executing code in a different instance, and it may
5162    /// call host functions and intrinsics from that other instance.
5163    instance: RuntimeInstance,
5164    /// If present, a pending `Event::None` or `Event::Cancelled` to be
5165    /// delivered to this task.
5166    event: Option<Event>,
5167    /// Whether or not the task has exited.
5168    exited: bool,
5169    /// Threads belonging to this task
5170    threads: HashSet<TableId<GuestThread>>,
5171    /// The state of the host future that represents an async task, which must
5172    /// be dropped before we can delete the task.
5173    host_future_state: HostFutureState,
5174    /// Indicates whether this task was created for a call to an async-typed
5175    /// export.
5176    async_typed: bool,
5177    /// Indicates whether this task was created for a call to an async-lifted
5178    /// export.
5179    async_lifted: bool,
5180
5181    decremented_interesting_task_count: bool,
5182    switch_item: Option<WorkItem>,
5183}
5184
5185impl GuestTask {
5186    fn already_lowered_parameters(&self) -> bool {
5187        // We reset `lower_params` after we lower the parameters
5188        self.lower_params.is_none()
5189    }
5190
5191    fn returned_or_cancelled(&self) -> bool {
5192        // We reset `lift_result` after we return or exit
5193        self.lift_result.is_none()
5194    }
5195
5196    fn ready_to_delete(&self) -> bool {
5197        let threads_completed = self.threads.is_empty();
5198        let has_sync_result = matches!(self.sync_result, SyncResult::Produced(_));
5199        let pending_completion_event = matches!(
5200            self.common.event,
5201            Some(Event::Subtask {
5202                status: Status::Returned | Status::ReturnCancelled
5203            })
5204        );
5205        let ready = threads_completed
5206            && !has_sync_result
5207            && !pending_completion_event
5208            && !matches!(self.host_future_state, HostFutureState::Live);
5209        log::trace!(
5210            "ready to delete? {ready} (threads_completed: {}, has_sync_result: {}, pending_completion_event: {}, host_future_state: {:?})",
5211            threads_completed,
5212            has_sync_result,
5213            pending_completion_event,
5214            self.host_future_state
5215        );
5216        ready
5217    }
5218
5219    fn new(
5220        state: &mut ConcurrentState,
5221        lower_params: RawLower,
5222        lift_result: LiftResult,
5223        caller: Caller,
5224        callback: Option<CallbackFn>,
5225        instance: RuntimeInstance,
5226        async_typed: bool,
5227        async_lifted: bool,
5228    ) -> Result<QualifiedThreadId> {
5229        let host_future_state = match &caller {
5230            Caller::Guest { .. } => HostFutureState::NotApplicable,
5231            Caller::Host {
5232                host_future_present,
5233                ..
5234            } => {
5235                if *host_future_present {
5236                    HostFutureState::Live
5237                } else {
5238                    HostFutureState::NotApplicable
5239                }
5240            }
5241        };
5242        let task = state.push(Self {
5243            common: WaitableCommon::default(),
5244            lower_params: Some(lower_params),
5245            lift_result: Some(lift_result),
5246            result: None,
5247            callback,
5248            caller,
5249            call_context: CallContext::default(),
5250            sync_result: SyncResult::NotProduced,
5251            cancel_sent: false,
5252            starting_sent: false,
5253            instance,
5254            event: None,
5255            exited: false,
5256            threads: HashSet::new(),
5257            host_future_state,
5258            async_typed,
5259            async_lifted,
5260            decremented_interesting_task_count: false,
5261            switch_item: None,
5262        })?;
5263        let new_thread = GuestThread::new_implicit(state, task)?;
5264        let thread = state.push(new_thread)?;
5265        state.get_mut(task)?.threads.insert(thread);
5266        state.interesting_tasks += 1;
5267        let thread = QualifiedThreadId { task, thread };
5268        log::trace!("new implicit thread {thread:?} for instance {instance:?}");
5269        Ok(thread)
5270    }
5271}
5272
5273impl TableDebug for GuestTask {
5274    fn type_name() -> &'static str {
5275        "GuestTask"
5276    }
5277}
5278
5279/// Represents state common to all kinds of waitables.
5280#[derive(Default)]
5281struct WaitableCommon {
5282    /// The currently pending event for this waitable, if any.
5283    event: Option<Event>,
5284    /// The set to which this waitable belongs, if any.
5285    set: Option<TableId<WaitableSet>>,
5286    /// The handle with which the guest refers to this waitable, if any.
5287    handle: Option<u32>,
5288}
5289
5290/// Represents a Component Model Async `waitable`.
5291#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
5292enum Waitable {
5293    /// A host task
5294    Host(TableId<HostTask>),
5295    /// A guest task
5296    Guest(TableId<GuestTask>),
5297    /// The read or write end of a stream or future
5298    Transmit(TableId<TransmitHandle>),
5299}
5300
5301impl Waitable {
5302    /// Retrieve the `Waitable` corresponding to the specified guest-visible
5303    /// handle.
5304    fn from_instance(
5305        state: Pin<&mut ComponentInstance>,
5306        caller_instance: RuntimeComponentInstanceIndex,
5307        waitable: u32,
5308    ) -> Result<Self> {
5309        use crate::runtime::vm::component::Waitable;
5310
5311        let (waitable, kind) = state.instance_states().0[caller_instance]
5312            .handle_table()
5313            .waitable_rep(waitable)?;
5314
5315        Ok(match kind {
5316            Waitable::Subtask { is_host: true } => Self::Host(TableId::new(waitable)),
5317            Waitable::Subtask { is_host: false } => Self::Guest(TableId::new(waitable)),
5318            Waitable::Stream | Waitable::Future => Self::Transmit(TableId::new(waitable)),
5319        })
5320    }
5321
5322    /// Retrieve the host-visible identifier for this `Waitable`.
5323    fn rep(&self) -> u32 {
5324        match self {
5325            Self::Host(id) => id.rep(),
5326            Self::Guest(id) => id.rep(),
5327            Self::Transmit(id) => id.rep(),
5328        }
5329    }
5330
5331    /// Move this `Waitable` to the specified set (when `set` is `Some(_)`) or
5332    /// remove it from any set it may currently belong to (when `set` is
5333    /// `None`).
5334    fn join(&self, state: &mut ConcurrentState, set: Option<TableId<WaitableSet>>) -> Result<()> {
5335        log::trace!("waitable {self:?} join set {set:?}");
5336
5337        let old = mem::replace(&mut self.common(state)?.set, set);
5338
5339        if let Some(old) = old {
5340            match *self {
5341                Waitable::Host(id) => state.remove_child(id, old),
5342                Waitable::Guest(id) => state.remove_child(id, old),
5343                Waitable::Transmit(id) => state.remove_child(id, old),
5344            }?;
5345
5346            state.get_mut(old)?.ready.remove(self);
5347        }
5348
5349        if let Some(set) = set {
5350            match *self {
5351                Waitable::Host(id) => state.add_child(id, set),
5352                Waitable::Guest(id) => state.add_child(id, set),
5353                Waitable::Transmit(id) => state.add_child(id, set),
5354            }?;
5355
5356            if self.common(state)?.event.is_some() {
5357                self.mark_ready(state)?;
5358            }
5359        }
5360
5361        Ok(())
5362    }
5363
5364    /// Retrieve mutable access to the `WaitableCommon` for this `Waitable`.
5365    fn common<'a>(&self, state: &'a mut ConcurrentState) -> Result<&'a mut WaitableCommon> {
5366        Ok(match self {
5367            Self::Host(id) => &mut state.get_mut(*id)?.common,
5368            Self::Guest(id) => &mut state.get_mut(*id)?.common,
5369            Self::Transmit(id) => &mut state.get_mut(*id)?.common,
5370        })
5371    }
5372
5373    /// Trap if this waitable is currently a member of a waitable set.
5374    ///
5375    /// A synchronous stream/future/subtask operation may end up blocking on
5376    /// this waitable, so it is not allowed to run while the waitable is also
5377    /// being watched by a waitable set.
5378    fn trap_if_in_waitable_set(&self, state: &mut ConcurrentState) -> Result<()> {
5379        if self.common(state)?.set.is_some() {
5380            bail!(Trap::WaitableSyncAndAsync);
5381        }
5382        Ok(())
5383    }
5384
5385    /// Set or clear the pending event for this waitable and either deliver it
5386    /// to the first waiter, if any, or mark it as ready to be delivered to the
5387    /// next waiter that arrives.
5388    fn set_event(&self, state: &mut ConcurrentState, event: Option<Event>) -> Result<()> {
5389        log::trace!("set event for {self:?}: {event:?}");
5390        self.common(state)?.event = event;
5391        self.mark_ready(state)
5392    }
5393
5394    /// Take the pending event from this waitable, leaving `None` in its place.
5395    fn take_event(&self, state: &mut ConcurrentState) -> Result<Option<Event>> {
5396        let common = self.common(state)?;
5397        let event = common.event.take();
5398        if let Some(set) = self.common(state)?.set {
5399            state.get_mut(set)?.ready.remove(self);
5400        }
5401
5402        Ok(event)
5403    }
5404
5405    /// Deliver the current event for this waitable to the first waiter, if any,
5406    /// or else mark it as ready to be delivered to the next waiter that
5407    /// arrives.
5408    fn mark_ready(&self, state: &mut ConcurrentState) -> Result<()> {
5409        if let Some(set) = self.common(state)?.set {
5410            let set_state = state.get_mut(set)?;
5411            set_state.ready.insert(*self);
5412
5413            if let Some((thread, mode)) = set_state.waiting.pop_first() {
5414                let wake_on_cancel = state.get_mut(thread.thread)?.wake_on_cancel.take();
5415                assert!(wake_on_cancel.is_none() || wake_on_cancel == Some(set));
5416
5417                let item = match mode {
5418                    WaitMode::Caller { fiber, callee } => {
5419                        // In this case, a caller is waiting for a subtask
5420                        // status update, but we can't necessarily deliver that
5421                        // update immediately because the callee may still be
5422                        // running, nor are we allowed queue delivery in a
5423                        // general-purpose work queues because the CM spec
5424                        // requires deterministic delivery of such updates.
5425                        //
5426                        // Therefore, we'll schedule delivery for when the
5427                        // callee suspends for the first time or exits as
5428                        // required by the spec.
5429
5430                        let item = WorkItem::ResumeFiber {
5431                            instance: state.get_mut(thread.task)?.instance,
5432                            thread,
5433                            fiber,
5434                        };
5435
5436                        if let Some(Event::Subtask {
5437                            status: Status::Starting,
5438                        }) = &self.common(state)?.event
5439                        {
5440                            // `Status::Starting` means we can't invoke the
5441                            // callee yet due to e.g. backpressure, so go ahead
5442                            // and deliver the update now.
5443                            state.set_switch_item(item)?;
5444                        } else {
5445                            if state.get_mut(callee)?.switch_item.is_some() {
5446                                bail_bug!(
5447                                    "`GuestTask::switch_item` is already `Some(_)` when we need \
5448                                     to deliver a subtask status update to the caller"
5449                                );
5450                            }
5451                            state.get_mut(callee)?.switch_item = Some(item);
5452                        }
5453                        None
5454                    }
5455                    WaitMode::Fiber(fiber) => Some(WorkItem::ResumeFiber {
5456                        instance: state.get_mut(thread.task)?.instance,
5457                        thread,
5458                        fiber,
5459                    }),
5460                    WaitMode::Callback(instance) => Some(WorkItem::GuestCall {
5461                        instance: state.get_mut(thread.task)?.instance,
5462                        call: GuestCall {
5463                            thread,
5464                            kind: GuestCallKind::DeliverEvent {
5465                                instance,
5466                                set: Some(set),
5467                            },
5468                        },
5469                    }),
5470                };
5471
5472                if let Some(item) = item {
5473                    state.push_high_priority(item);
5474                }
5475            }
5476        }
5477        Ok(())
5478    }
5479
5480    /// Remove this waitable from the instance's rep table.
5481    fn delete_from(&self, state: &mut ConcurrentState) -> Result<()> {
5482        match self {
5483            Self::Host(task) => {
5484                log::trace!("delete host task {task:?}");
5485                state.delete(*task)?;
5486            }
5487            Self::Guest(task) => {
5488                log::trace!("delete guest task {task:?}");
5489                let task = state.delete(*task)?;
5490
5491                // When a guest task is created it increments the
5492                // `ConcurrentState::interesting_tasks` counter, and that needs
5493                // to be paired with a decrement. There are a few situations in
5494                // which the decrement needs to happen which don't all funnel
5495                // through here, so in lieu of that at least try to catch issues
5496                // where we forgot to do a decrement.
5497                debug_assert!(task.decremented_interesting_task_count);
5498            }
5499            Self::Transmit(task) => {
5500                state.delete(*task)?;
5501            }
5502        }
5503
5504        Ok(())
5505    }
5506}
5507
5508impl fmt::Debug for Waitable {
5509    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
5510        match self {
5511            Self::Host(id) => write!(f, "{id:?}"),
5512            Self::Guest(id) => write!(f, "{id:?}"),
5513            Self::Transmit(id) => write!(f, "{id:?}"),
5514        }
5515    }
5516}
5517
5518/// Represents a Component Model Async `waitable-set`.
5519#[derive(Default)]
5520struct WaitableSet {
5521    /// Which waitables in this set have pending events, if any.
5522    ready: BTreeSet<Waitable>,
5523    /// Which guest threads are currently waiting on this set, if any.
5524    waiting: BTreeMap<QualifiedThreadId, WaitMode>,
5525    /// Whether this set is a synthetic, internal one meant for handling
5526    /// synchronous calls.
5527    is_sync_call_set: bool,
5528}
5529
5530impl TableDebug for WaitableSet {
5531    fn type_name() -> &'static str {
5532        "WaitableSet"
5533    }
5534}
5535
5536/// Type-erased closure to lower the parameters for a guest task.
5537type RawLower =
5538    Box<dyn FnOnce(&mut dyn VMStore, &mut [MaybeUninit<ValRaw>]) -> Result<()> + Send + Sync>;
5539
5540/// Type-erased closure to lift the result for a guest task.
5541type RawLift = Box<
5542    dyn FnOnce(&mut dyn VMStore, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>> + Send + Sync,
5543>;
5544
5545/// Type erased result of a guest task which may be downcast to the expected
5546/// type by a host caller (or simply ignored in the case of a guest caller; see
5547/// `DummyResult`).
5548type LiftedResult = Box<dyn Any + Send + Sync>;
5549
5550/// Used to return a result from a `LiftFn` when the actual result has already
5551/// been lowered to a guest task's stack and linear memory.
5552struct DummyResult;
5553
5554/// Represents the Component Model Async state of a (sub-)component instance.
5555#[derive(Default)]
5556pub struct ConcurrentInstanceState {
5557    /// Whether backpressure is set for this instance (enabled if >0)
5558    backpressure: u16,
5559    /// Whether this instance can be entered
5560    do_not_enter: bool,
5561    /// Whether this instance may suspend (i.e. whether this instance is
5562    /// currently running a sync-typed function).
5563    do_not_suspend: bool,
5564    /// Pending calls for this instance which require `Self::backpressure` to be
5565    /// zero and/or `Self::do_not_enter` to be false before they can proceed.
5566    pending: BTreeMap<QualifiedThreadId, GuestCallKind>,
5567}
5568
5569impl ConcurrentInstanceState {
5570    pub fn pending_is_empty(&self) -> bool {
5571        self.pending.is_empty()
5572    }
5573}
5574
5575#[derive(Debug, Copy, Clone)]
5576pub(crate) enum CurrentThread {
5577    /// The currently running thread is a guest, identified here with its
5578    /// task/thread id combo.
5579    Guest(QualifiedThreadId),
5580    /// The currently running thread is a host task.
5581    Host(TableId<HostTask>),
5582    /// The currently running thread is a host call whose task has not yet been
5583    /// materialized. The contained ID identifies its guest caller.
5584    DeferredHost(QualifiedThreadId),
5585    /// A bit of a kludge to get `StoreOpaque::parent` working with backtraces
5586    /// and this serves as the parent node of a `Host` task. This ideally would
5587    /// get removed in favor of separate backtrace storage.
5588    GuestTask(TableId<GuestTask>),
5589    /// There is no currently running thread because we are in the main event
5590    /// loop or concurrency is disabled.
5591    None,
5592}
5593
5594impl CurrentThread {
5595    fn guest(&self) -> Option<&QualifiedThreadId> {
5596        match self {
5597            Self::Guest(id) => Some(id),
5598            _ => None,
5599        }
5600    }
5601
5602    fn guest_task(&self) -> Option<TableId<GuestTask>> {
5603        match self {
5604            Self::Guest(id) => Some(id.task),
5605            Self::GuestTask(id) => Some(*id),
5606            _ => None,
5607        }
5608    }
5609
5610    fn is_none(&self) -> bool {
5611        matches!(self, Self::None)
5612    }
5613}
5614
5615impl From<QualifiedThreadId> for CurrentThread {
5616    fn from(id: QualifiedThreadId) -> Self {
5617        Self::Guest(id)
5618    }
5619}
5620
5621impl From<TableId<HostTask>> for CurrentThread {
5622    fn from(id: TableId<HostTask>) -> Self {
5623        Self::Host(id)
5624    }
5625}
5626
5627enum Priority {
5628    Switch,
5629    High,
5630    Low,
5631}
5632
5633/// Represents the Component Model Async state of a store.
5634pub struct ConcurrentState {
5635    /// The currently running thread, if any.
5636    ///
5637    /// Note that we lazily materialize threads on-demand and this field is not
5638    /// necessarily up-to-date. The `StoreOpaque::current_thread` method should
5639    /// be preferred over directly accessing this field.
5640    unforced_current_thread: CurrentThread,
5641
5642    /// Borrow state for the deferred host call, if any.
5643    ///
5644    /// This is `Some` if and only if [`Self::unforced_current_thread`] is
5645    /// [`CurrentThread::DeferredHost`]. Materializing the host task moves this
5646    /// context into that task.
5647    deferred_host_call_context: Option<CallContext>,
5648
5649    /// The set of pending host and background tasks, if any.
5650    ///
5651    /// See `ComponentInstance::poll_until` for where we temporarily take this
5652    /// out, poll it, then put it back to avoid any mutable aliasing hazards.
5653    futures: AlwaysMut<Option<FuturesUnordered<HostTaskFuture>>>,
5654    /// The table of waitables, waitable sets, etc.
5655    table: AlwaysMut<ResourceTable>,
5656    /// The next item to switch to if any.
5657    ///
5658    /// This takes precedence over items in the `high_priority` queue below and
5659    /// should be used in cases such as subtask calls and thread resume/promote
5660    /// operations where we must switch to a specific thread at the next turn of
5661    /// the event loop regardless of what happens to be present in the
5662    /// `high_priority` queue.
5663    switch_item: Option<WorkItem>,
5664    /// The "high priority" work queue for this store's event loop.
5665    high_priority: VecDeque<WorkItem>,
5666    /// The "low priority" work queue for this store's event loop.
5667    low_priority: VecDeque<WorkItem>,
5668    /// A place to stash the reason a fiber is suspending so that the code which
5669    /// resumed it will know under what conditions the fiber should be resumed
5670    /// again.
5671    suspend_reason: Option<SuspendReason>,
5672    /// A cached fiber which is waiting for work to do.
5673    ///
5674    /// This helps us avoid creating a new fiber for each `GuestCall` work item.
5675    worker: Option<StoreFiber<'static>>,
5676    /// A place to stash the work item for which we're resuming a worker fiber.
5677    worker_item: Option<WorkerItem>,
5678
5679    /// Reference counts for all component error contexts
5680    ///
5681    /// NOTE: it is possible the global ref count to be *greater* than the sum of
5682    /// (sub)component ref counts as tracked by `error_context_tables`, for
5683    /// example when the host holds one or more references to error contexts.
5684    ///
5685    /// The key of this primary map is often referred to as the "rep" (i.e. host-side
5686    /// component-wide representation) of the index into concurrent state for a given
5687    /// stored `ErrorContext`.
5688    ///
5689    /// Stated another way, `TypeComponentGlobalErrorContextTableIndex` is essentially the same
5690    /// as a `TableId<ErrorContextState>`.
5691    global_error_context_ref_counts:
5692        BTreeMap<TypeComponentGlobalErrorContextTableIndex, GlobalErrorContextRefCount>,
5693
5694    /// The number of "interesting tasks" currently executing in the store.
5695    ///
5696    /// This tracks the concept of a component instance lifetime as defined in
5697    /// https://github.com/WebAssembly/component-model/pull/643. Specifically
5698    /// all tasks currently increment this counter which then gets decremented
5699    /// when they exit. In the future some tasks might not increment this
5700    /// counter, but for now all do.
5701    ///
5702    /// This is used to implement `Accessor::poll_no_interesting_tasks` to
5703    /// inform the embedder when all tasks have completed. This is then
5704    /// used in wasmtime-wasi-http, for example, to know when an instance is
5705    /// idle.
5706    interesting_tasks: usize,
5707
5708    /// Single waker to notify when `interesting_tasks` reaches 0.
5709    ///
5710    /// Used in the implementation of `Accessor::poll_no_interesting_tasks`.
5711    interesting_tasks_empty_waker: Option<Waker>,
5712
5713    /// Single waker to notify when a component instance goes from
5714    /// not-concurrently-callable to concurrently-callable.
5715    ///
5716    /// Used in the implementation of `Accessor::poll_ready_for_concurrent_call`.
5717    ready_for_concurrent_call_waker: Option<Waker>,
5718
5719    /// Whether the `StoreContextMut::poll_until` event loop is running.
5720    event_loop_running: bool,
5721}
5722
5723impl Default for ConcurrentState {
5724    fn default() -> Self {
5725        Self {
5726            unforced_current_thread: CurrentThread::None,
5727            deferred_host_call_context: None,
5728            table: AlwaysMut::new(ResourceTable::new()),
5729            futures: AlwaysMut::new(Some(FuturesUnordered::new())),
5730            switch_item: None,
5731            high_priority: VecDeque::new(),
5732            low_priority: VecDeque::new(),
5733            suspend_reason: None,
5734            worker: None,
5735            worker_item: None,
5736            global_error_context_ref_counts: BTreeMap::new(),
5737            interesting_tasks: 0,
5738            interesting_tasks_empty_waker: None,
5739            ready_for_concurrent_call_waker: None,
5740            event_loop_running: false,
5741        }
5742    }
5743}
5744
5745impl ConcurrentState {
5746    /// Take ownership of any fibers and futures owned by this object.
5747    ///
5748    /// This should be used when disposing of the `Store` containing this object
5749    /// in order to gracefully resolve any and all fibers using
5750    /// `StoreFiber::dispose`.  This is necessary to avoid possible
5751    /// use-after-free bugs due to fibers which may still have access to the
5752    /// `Store`.
5753    ///
5754    /// Additionally, the futures collected with this function should be dropped
5755    /// within a `tls::set` call, which will ensure than any futures closing
5756    /// over an `&Accessor` will have access to the store when dropped, allowing
5757    /// e.g. `WithAccessor[AndValue]` instances to be disposed of without
5758    /// panicking.
5759    ///
5760    /// Note that this will leave the object in an inconsistent and unusable
5761    /// state, so it should only be used just prior to dropping it.
5762    pub(crate) fn take_fibers_and_futures(
5763        &mut self,
5764        fibers: &mut Vec<StoreFiber<'static>>,
5765        futures: &mut Vec<FuturesUnordered<HostTaskFuture>>,
5766    ) {
5767        let mut items = Vec::new();
5768        for entry in self.table.get_mut().iter_mut() {
5769            if let Some(set) = entry.downcast_mut::<WaitableSet>() {
5770                for mode in mem::take(&mut set.waiting).into_values() {
5771                    match mode {
5772                        WaitMode::Fiber(fiber) | WaitMode::Caller { fiber, .. } => {
5773                            fibers.push(fiber);
5774                        }
5775                        WaitMode::Callback(_) => {}
5776                    }
5777                }
5778            } else if let Some(thread) = entry.downcast_mut::<GuestThread>() {
5779                if let GuestThreadState::Suspended(fiber) | GuestThreadState::Ready { fiber, .. } =
5780                    mem::replace(&mut thread.state, GuestThreadState::Completed)
5781                {
5782                    fibers.push(fiber);
5783                }
5784            } else if let Some(task) = entry.downcast_mut::<GuestTask>() {
5785                if let Some(item) = task.switch_item.take() {
5786                    items.push(item);
5787                }
5788            }
5789        }
5790
5791        if let Some(fiber) = self.worker.take() {
5792            fibers.push(fiber);
5793        }
5794
5795        let mut handle_item = |item| match item {
5796            WorkItem::ResumeFiber { fiber, .. } => {
5797                fibers.push(fiber);
5798            }
5799            WorkItem::PushFuture(future) => {
5800                self.futures
5801                    .get_mut()
5802                    .as_mut()
5803                    .unwrap()
5804                    .push(future.into_inner());
5805            }
5806            WorkItem::ResumeThread { .. }
5807            | WorkItem::GuestCall { .. }
5808            | WorkItem::WorkerFunction(_) => {}
5809        };
5810
5811        for item in items {
5812            handle_item(item);
5813        }
5814        if let Some(item) = self.switch_item.take() {
5815            handle_item(item);
5816        }
5817        for item in mem::take(&mut self.high_priority) {
5818            handle_item(item);
5819        }
5820        for item in mem::take(&mut self.low_priority) {
5821            handle_item(item);
5822        }
5823
5824        if let Some(them) = self.futures.get_mut().take() {
5825            futures.push(them);
5826        }
5827    }
5828
5829    #[cfg(feature = "gc")]
5830    pub(crate) fn trace_fiber_roots(
5831        &mut self,
5832        modules: &ModuleRegistry,
5833        unwind: &dyn Unwind,
5834        gc_roots_list: &mut GcRootsList,
5835    ) {
5836        let ConcurrentState {
5837            table,
5838            worker,
5839            switch_item,
5840            high_priority,
5841            low_priority,
5842
5843            // TODO(cm-gc): This field contains `ValRaw`s, but they are never GC
5844            // references because the component model doesn't support GC yet. We
5845            // will need to trace these somehow when it does.
5846            futures: _,
5847
5848            // These fields do not contain GC references.
5849            worker_item: _,
5850            unforced_current_thread: _,
5851            deferred_host_call_context: _,
5852            suspend_reason: _,
5853            global_error_context_ref_counts: _,
5854            interesting_tasks: _,
5855            interesting_tasks_empty_waker: _,
5856            ready_for_concurrent_call_waker: _,
5857            event_loop_running: _,
5858        } = self;
5859
5860        for entry in table.get_mut().iter_mut() {
5861            if let Some(set) = entry.downcast_mut::<WaitableSet>() {
5862                for mode in set.waiting.values_mut() {
5863                    match mode {
5864                        WaitMode::Fiber(fiber) | WaitMode::Caller { fiber, .. } => {
5865                            fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5866                        }
5867                        WaitMode::Callback(_) => {}
5868                    }
5869                }
5870            } else if let Some(thread) = entry.downcast_mut::<GuestThread>() {
5871                if let GuestThreadState::Suspended(fiber) | GuestThreadState::Ready { fiber, .. } =
5872                    &mut thread.state
5873                {
5874                    fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5875                }
5876            }
5877        }
5878
5879        if let Some(fiber) = worker {
5880            fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5881        }
5882
5883        let mut handle_item = |item: &mut WorkItem| match item {
5884            WorkItem::ResumeFiber { fiber, .. } => {
5885                fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5886            }
5887            WorkItem::PushFuture(_future) => {
5888                // TODO(cm-gc): once futures can contain GC roots, we will need
5889                // to trace them.
5890            }
5891            WorkItem::ResumeThread { .. }
5892            | WorkItem::GuestCall { .. }
5893            | WorkItem::WorkerFunction(_) => {}
5894        };
5895
5896        if let Some(item) = switch_item {
5897            handle_item(item);
5898        }
5899        for item in high_priority {
5900            handle_item(item);
5901        }
5902        for item in low_priority {
5903            handle_item(item);
5904        }
5905    }
5906
5907    fn push<V: Send + Sync + 'static>(
5908        &mut self,
5909        value: V,
5910    ) -> Result<TableId<V>, ResourceTableError> {
5911        self.table.get_mut().push(value).map(TableId::from)
5912    }
5913
5914    fn get_mut<V: 'static>(&mut self, id: TableId<V>) -> Result<&mut V, ResourceTableError> {
5915        self.table.get_mut().get_mut(&Resource::from(id))
5916    }
5917
5918    pub fn add_child<T: 'static, U: 'static>(
5919        &mut self,
5920        child: TableId<T>,
5921        parent: TableId<U>,
5922    ) -> Result<(), ResourceTableError> {
5923        self.table
5924            .get_mut()
5925            .add_child(Resource::from(child), Resource::from(parent))
5926    }
5927
5928    pub fn remove_child<T: 'static, U: 'static>(
5929        &mut self,
5930        child: TableId<T>,
5931        parent: TableId<U>,
5932    ) -> Result<(), ResourceTableError> {
5933        self.table
5934            .get_mut()
5935            .remove_child(Resource::from(child), Resource::from(parent))
5936    }
5937
5938    fn delete<V: 'static>(&mut self, id: TableId<V>) -> Result<V, ResourceTableError> {
5939        self.table.get_mut().delete(Resource::from(id))
5940    }
5941
5942    fn push_future(&mut self, future: HostTaskFuture) {
5943        // Note that we can't directly push to `ConcurrentState::futures` here
5944        // since this may be called from a future that's being polled inside
5945        // `Self::poll_until`, which temporarily removes the `FuturesUnordered`
5946        // so it has exclusive access while polling it.  Therefore, we push a
5947        // work item to the "high priority" queue, which will actually push to
5948        // `ConcurrentState::futures` later.
5949        self.push_high_priority(WorkItem::PushFuture(AlwaysMut::new(future)));
5950    }
5951
5952    fn set_switch_item(&mut self, item: WorkItem) -> Result<()> {
5953        log::trace!("set switch item: {item:?}");
5954
5955        if self.switch_item.is_some() {
5956            bail_bug!("switch item already set");
5957        }
5958
5959        self.switch_item = Some(item);
5960
5961        Ok(())
5962    }
5963
5964    fn push_high_priority(&mut self, item: WorkItem) {
5965        log::trace!("push high priority: {item:?}");
5966        self.high_priority.push_front(item);
5967    }
5968
5969    fn push_low_priority(&mut self, item: WorkItem) {
5970        log::trace!("push low priority: {item:?}");
5971        self.low_priority.push_front(item);
5972    }
5973
5974    fn push_work_item(&mut self, item: WorkItem, priority: Priority) -> Result<()> {
5975        match priority {
5976            Priority::Switch => self.set_switch_item(item)?,
5977            Priority::High => self.push_high_priority(item),
5978            Priority::Low => self.push_low_priority(item),
5979        }
5980
5981        Ok(())
5982    }
5983
5984    fn promote_instance_local_thread_work_item(
5985        &mut self,
5986        current_instance: RuntimeInstance,
5987    ) -> Result<bool> {
5988        log::trace!("promote thread work items for {current_instance:?}");
5989
5990        self.promote_work_item_matching(|item: &WorkItem| {
5991            let result = match item {
5992                WorkItem::ResumeThread { instance, .. }
5993                | WorkItem::ResumeFiber { instance, .. }
5994                | WorkItem::GuestCall { instance, .. } => *instance == current_instance,
5995                _ => false,
5996            };
5997
5998            log::trace!("candidate {item:?}: {result}");
5999            result
6000        })
6001    }
6002
6003    fn promote_thread_work_item(&mut self, thread: QualifiedThreadId) -> Result<bool> {
6004        self.promote_work_item_matching(|item: &WorkItem| match item {
6005            WorkItem::ResumeThread {
6006                thread: item_thread,
6007                ..
6008            }
6009            | WorkItem::GuestCall {
6010                call:
6011                    GuestCall {
6012                        thread: item_thread,
6013                        ..
6014                    },
6015                ..
6016            } => *item_thread == thread,
6017            _ => false,
6018        })
6019    }
6020
6021    fn promote_work_item_matching<F>(&mut self, mut predicate: F) -> Result<bool>
6022    where
6023        F: FnMut(&WorkItem) -> bool,
6024    {
6025        // Note the use of `.rev()` below to preserve ordering given that items
6026        // are popped from the back of the `VecDeque`s by `poll_until` and
6027        // pushed to the front by `push_{high,low}_priority`.
6028
6029        for item in mem::take(&mut self.high_priority).into_iter().rev() {
6030            if self.switch_item.is_none() && predicate(&item) {
6031                self.set_switch_item(item)?;
6032            } else {
6033                self.push_high_priority(item);
6034            }
6035        }
6036
6037        if self.switch_item.is_none() {
6038            for item in mem::take(&mut self.low_priority).into_iter().rev() {
6039                if self.switch_item.is_none() && predicate(&item) {
6040                    self.set_switch_item(item)?;
6041                } else {
6042                    self.push_low_priority(item);
6043                }
6044            }
6045        }
6046
6047        Ok(self.switch_item.is_some())
6048    }
6049
6050    /// Used by `ResourceTables` to acquire the current `CallContext` for the
6051    /// specified task.
6052    pub fn call_context(&mut self, task: Scope) -> Result<&mut CallContext> {
6053        match task {
6054            Scope::HostId(task) => {
6055                let task: TableId<HostTask> = TableId::new(task);
6056                Ok(&mut self.get_mut(task)?.call_context)
6057            }
6058            Scope::Id(task) => {
6059                let task: TableId<GuestTask> = TableId::new(task);
6060                Ok(&mut self.get_mut(task)?.call_context)
6061            }
6062        }
6063    }
6064
6065    pub(crate) fn deferred_host_call_context(&mut self) -> Option<&mut CallContext> {
6066        self.deferred_host_call_context.as_mut()
6067    }
6068
6069    fn futures_mut(&mut self) -> Result<&mut FuturesUnordered<HostTaskFuture>> {
6070        match self.futures.get_mut().as_mut() {
6071            Some(f) => Ok(f),
6072            None => bail_bug!("futures field of concurrent state is currently taken"),
6073        }
6074    }
6075
6076    pub(crate) fn table(&mut self) -> &mut ResourceTable {
6077        self.table.get_mut()
6078    }
6079
6080    /// Returns the parent thread, if any, of `cur`.
6081    fn parent(&mut self, cur: CurrentThread) -> Option<CurrentThread> {
6082        let task = match cur {
6083            CurrentThread::GuestTask(task) => task,
6084            CurrentThread::Guest(thread) => thread.task,
6085            CurrentThread::Host(id) => {
6086                return Some(CurrentThread::GuestTask(self.get_mut(id).ok()?.caller));
6087            }
6088            CurrentThread::DeferredHost(caller) => return Some(caller.into()),
6089            CurrentThread::None => return None,
6090        };
6091        let task = self.get_mut(task).ok()?;
6092        Some(match task.caller {
6093            Caller::Host { caller, .. } => caller.map_or(CurrentThread::None, CurrentThread::Host),
6094            Caller::Guest { thread } => thread.into(),
6095        })
6096    }
6097
6098    fn debug_assert_deferred_host_invariant(&self) {
6099        debug_assert_eq!(
6100            self.deferred_host_call_context.is_some(),
6101            matches!(self.unforced_current_thread, CurrentThread::DeferredHost(_)),
6102            "a deferred host thread and call context must exist together",
6103        );
6104    }
6105
6106    fn materialize_host_task(&mut self) -> Result<CurrentThread> {
6107        self.debug_assert_deferred_host_invariant();
6108        let caller = match self.unforced_current_thread {
6109            CurrentThread::DeferredHost(caller) => caller,
6110            thread => return Ok(thread),
6111        };
6112
6113        // Push first so allocation failure leaves the deferred state intact.
6114        let task = self.push(HostTask::new(caller.task, HostTaskState::CalleeStarted))?;
6115        let call_context = self
6116            .deferred_host_call_context
6117            .take()
6118            .expect("deferred host call context should be present");
6119        self.get_mut(task)
6120            .expect("newly inserted host task should be present")
6121            .call_context = call_context;
6122        self.unforced_current_thread = CurrentThread::Host(task);
6123        self.debug_assert_deferred_host_invariant();
6124        log::trace!("new host task materialized {task:?}");
6125        Ok(CurrentThread::Host(task))
6126    }
6127
6128    fn materialize_current_host_task_id(&mut self) -> Result<Option<TableId<HostTask>>> {
6129        match self.materialize_host_task()? {
6130            CurrentThread::Host(id) => Ok(Some(id)),
6131            CurrentThread::None => Ok(None),
6132            CurrentThread::Guest(_) | CurrentThread::GuestTask(_) => {
6133                bail_bug!("tried to materialize a host task id from a guest thread")
6134            }
6135            CurrentThread::DeferredHost(_) => {
6136                bail_bug!(
6137                    "current thread is a deferred host thread which should have been materialized"
6138                )
6139            }
6140        }
6141    }
6142
6143    pub(crate) fn materialize_current_scope(&mut self) -> Result<Scope> {
6144        match self.materialize_host_task()? {
6145            CurrentThread::Host(id) => Ok(Scope::HostId(id.rep())),
6146            _ => bail_bug!("current scope is not a deferred host scope"),
6147        }
6148    }
6149}
6150
6151/// Provide a type hint to compiler about the shape of a parameter lower
6152/// closure.
6153fn for_any_lower<
6154    F: FnOnce(&mut dyn VMStore, &mut [MaybeUninit<ValRaw>]) -> Result<()> + Send + Sync,
6155>(
6156    fun: F,
6157) -> F {
6158    fun
6159}
6160
6161/// Provide a type hint to compiler about the shape of a result lift closure.
6162fn for_any_lift<
6163    F: FnOnce(&mut dyn VMStore, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>> + Send + Sync,
6164>(
6165    fun: F,
6166) -> F {
6167    fun
6168}
6169
6170fn check_ambient_store(id: StoreId) {
6171    let message = "\
6172        `Future`s which depend on asynchronous component tasks, streams, or \
6173        futures to complete may only be polled from the event loop of the \
6174        store to which they belong.  Please use \
6175        `StoreContextMut::{run_concurrent,spawn}` to poll or await them.\
6176    ";
6177    tls::try_get(|store| {
6178        let matched = match store {
6179            tls::TryGet::Some(store) => store.id() == id,
6180            tls::TryGet::Taken | tls::TryGet::None => false,
6181        };
6182
6183        if !matched {
6184            panic!("{message}")
6185        }
6186    });
6187}
6188
6189fn unpack_callback_code(code: u32) -> (u32, u32) {
6190    (code & 0xF, code >> 4)
6191}
6192
6193/// Helper struct for packaging parameters to be passed to
6194/// `ComponentInstance::waitable_check` for calls to `waitable-set.wait` or
6195/// `waitable-set.poll`.
6196struct WaitableCheckParams {
6197    set: TableId<WaitableSet>,
6198    options: OptionsIndex,
6199    payload: u32,
6200}
6201
6202/// Indicates whether `ComponentInstance::waitable_check` is being called for
6203/// `waitable-set.wait` or `waitable-set.poll`.
6204enum WaitableCheck {
6205    Wait,
6206    Poll,
6207}
6208
6209/// An identifier representing a guest task within a component.
6210///
6211/// This can be acquired by calling [`Func::start_call_concurrent`] or
6212/// [`TypedFunc::start_call_concurrent`] and then using the
6213/// [`FuncCallConcurrent::task`] accessor, for example. This can then be
6214/// reflected on with [`StoreContextMut::async_call_stack`].
6215///
6216/// [`TypedFunc::start_call_concurrent`]: crate::component::TypedFunc::start_call_concurrent
6217#[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)]
6218pub struct GuestTaskId(TableId<GuestTask>);
6219
6220/// Represents a guest task called from the host, prepared using `prepare_call`.
6221pub(crate) struct PreparedCall<R> {
6222    /// The guest export to be called
6223    handle: Func,
6224    /// The guest thread created by `prepare_call`
6225    thread: QualifiedThreadId,
6226    /// The number of lowered core Wasm parameters to pass to the call.
6227    param_count: usize,
6228    /// The `oneshot::Receiver` to which the result of the call will be
6229    /// delivered when it is available.
6230    rx: oneshot::Receiver<LiftedResult>,
6231    /// The instance that this call is prepared for.
6232    runtime_instance: RuntimeInstance,
6233    _phantom: PhantomData<R>,
6234}
6235
6236impl<R> PreparedCall<R> {
6237    /// Get a copy of the `TaskId` for this `PreparedCall`.
6238    pub(crate) fn task_id(&self) -> TaskId {
6239        TaskId {
6240            task: self.thread.task,
6241            runtime_instance: self.runtime_instance,
6242        }
6243    }
6244}
6245
6246/// Represents a task created by `prepare_call`.
6247pub(crate) struct TaskId {
6248    task: TableId<GuestTask>,
6249    runtime_instance: RuntimeInstance,
6250}
6251
6252impl TaskId {
6253    /// The host future for an async task was dropped. If the parameters have not been lowered yet,
6254    /// it is no longer valid to do so, as the lowering closure would see a dangling pointer. In this case,
6255    /// we delete the task eagerly. Otherwise, there may be running threads, or ones that are suspended
6256    /// and can be resumed by other tasks for this component, so we mark the future as dropped
6257    /// and delete the task when all threads are done.
6258    pub(crate) fn host_future_dropped(&self, store: &mut StoreOpaque) -> Result<()> {
6259        let task = store.concurrent_state_mut()?.get_mut(self.task)?;
6260        let delete = if !task.already_lowered_parameters() {
6261            store.cancel_guest_subtask_without_lowered_parameters(
6262                self.runtime_instance,
6263                self.task,
6264            )?;
6265            true
6266        } else {
6267            task.host_future_state = HostFutureState::Dropped;
6268            task.ready_to_delete()
6269        };
6270        if delete {
6271            Waitable::Guest(self.task).delete_from(store.concurrent_state_mut()?)?
6272        }
6273        Ok(())
6274    }
6275}
6276
6277/// Prepare a call to the specified exported Wasm function, providing functions
6278/// for lowering the parameters and lifting the result.
6279///
6280/// To enqueue the returned `PreparedCall` in the `ComponentInstance`'s event
6281/// loop, use `stage_call`.
6282pub(crate) fn prepare_call<T, R>(
6283    mut store: StoreContextMut<T>,
6284    handle: Func,
6285    param_count: usize,
6286    host_future_present: bool,
6287    lower_params: impl FnOnce(StoreContextMut<T>, &mut [MaybeUninit<ValRaw>]) -> Result<()>
6288    + Send
6289    + Sync
6290    + 'static,
6291    lift_result: impl FnOnce(&mut StoreOpaque, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>>
6292    + Send
6293    + Sync
6294    + 'static,
6295) -> Result<PreparedCall<R>> {
6296    if !store.0.may_enter() {
6297        bail!(Trap::CannotEnterComponent);
6298    }
6299
6300    let (options, _flags, ty, raw_options) = handle.abi_info(store.0);
6301
6302    let instance = handle.instance().id().get(store.0);
6303    let options = &instance.component().env_component().options[options];
6304    let ty = &instance.component().types()[ty];
6305    let async_typed = ty.async_;
6306    let async_lifted = raw_options.async_;
6307    let task_return_type = ty.results;
6308    let component_instance = raw_options.instance;
6309    let callback = options.callback.map(|i| instance.runtime_callback(i));
6310    let memory = options
6311        .memory()
6312        .map(|i| instance.runtime_memory(i))
6313        .map(SendSyncPtr::new);
6314    let string_encoding = options.string_encoding;
6315    let token = StoreToken::new(store.as_context_mut());
6316    let caller = store.0.materialize_host_task_id()?;
6317    let state = store.0.concurrent_state_mut()?;
6318
6319    let (tx, rx) = oneshot::channel();
6320
6321    let instance = handle.instance().runtime_instance(component_instance);
6322    let thread = GuestTask::new(
6323        state,
6324        Box::new(for_any_lower(move |store, params| {
6325            lower_params(token.as_context_mut(store), params)
6326        })),
6327        LiftResult {
6328            lift: Box::new(for_any_lift(move |store, result| {
6329                lift_result(store, result)
6330            })),
6331            ty: task_return_type,
6332            memory,
6333            string_encoding,
6334        },
6335        Caller::Host {
6336            tx: Some(tx),
6337            host_future_present,
6338            caller,
6339        },
6340        callback.map(|callback| {
6341            let callback = SendSyncPtr::new(callback);
6342            let instance = handle.instance();
6343            Box::new(move |store: &mut dyn VMStore, event, handle| {
6344                let store = token.as_context_mut(store);
6345                // SAFETY: Per the contract of `prepare_call`, the callback
6346                // will remain valid at least as long is this task exists.
6347                unsafe { instance.call_callback(store, callback, event, handle) }
6348            }) as CallbackFn
6349        }),
6350        instance,
6351        async_typed,
6352        async_lifted,
6353    )?;
6354
6355    Ok(PreparedCall {
6356        handle,
6357        thread,
6358        param_count,
6359        runtime_instance: instance,
6360        rx,
6361        _phantom: PhantomData,
6362    })
6363}
6364
6365pub(crate) struct StagedCall<R> {
6366    store: StoreId,
6367    task: TableId<GuestTask>,
6368    rx: oneshot::Receiver<LiftedResult>,
6369    _marker: PhantomData<fn() -> R>,
6370}
6371
6372impl<R> StagedCall<R> {
6373    /// Queue a call previously prepared using `prepare_call` to be run as part of
6374    /// the associated `ComponentInstance`'s event loop.
6375    ///
6376    /// The returned future will resolve to the result once it is available, but
6377    /// must only be polled via the instance's event loop. See
6378    /// `StoreContextMut::run_concurrent` for details.
6379    pub(crate) fn new<T: 'static>(
6380        mut store: StoreContextMut<T>,
6381        prepared: PreparedCall<R>,
6382    ) -> Result<StagedCall<R>> {
6383        let PreparedCall {
6384            handle,
6385            thread,
6386            param_count,
6387            rx,
6388            ..
6389        } = prepared;
6390
6391        stage_call0(store.as_context_mut(), handle, thread, param_count)?;
6392
6393        Ok(StagedCall {
6394            store: store.0.id(),
6395            task: thread.task,
6396            rx,
6397            _marker: PhantomData,
6398        })
6399    }
6400
6401    fn task(&self) -> GuestTaskId {
6402        GuestTaskId(self.task)
6403    }
6404}
6405
6406impl<R> Future for StagedCall<R>
6407where
6408    R: 'static,
6409{
6410    type Output = Result<R>;
6411
6412    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
6413        check_ambient_store(self.store);
6414        Pin::new(&mut self.rx).poll(cx).map(|result| match result {
6415            Ok(r) => match r.downcast() {
6416                Ok(r) => Ok(*r),
6417                Err(_) => bail_bug!("wrong type of value produced"),
6418            },
6419            Err(oneshot::Canceled) => bail_bug!("channel erroneously dropped"),
6420        })
6421    }
6422}
6423
6424/// Queue a call previously prepared using `prepare_call` to be run as part of
6425/// the associated `ComponentInstance`'s event loop.
6426fn stage_call0<T: 'static>(
6427    store: StoreContextMut<T>,
6428    handle: Func,
6429    guest_thread: QualifiedThreadId,
6430    param_count: usize,
6431) -> Result<()> {
6432    let (_options, _, _ty, raw_options) = handle.abi_info(store.0);
6433    let is_concurrent = raw_options.async_;
6434    let callback = raw_options.callback;
6435    let instance = handle.instance();
6436    let callee = handle.lifted_core_func(store.0);
6437    let post_return = raw_options
6438        .post_return
6439        .map(|i| instance.id().get(store.0).runtime_post_return(i));
6440    let callback = callback.map(|i| {
6441        let instance = instance.id().get(store.0);
6442        SendSyncPtr::new(instance.runtime_callback(i))
6443    });
6444
6445    log::trace!("queueing call {guest_thread:?}");
6446
6447    // SAFETY: `callee`, `callback`, and `post_return` are valid pointers
6448    // (with signatures appropriate for this call) and will remain valid as
6449    // long as this instance is valid.
6450    unsafe {
6451        instance.stage_call(
6452            store,
6453            guest_thread,
6454            SendSyncPtr::new(callee),
6455            param_count,
6456            1,
6457            is_concurrent,
6458            callback,
6459            post_return.map(SendSyncPtr::new),
6460            true,
6461        )
6462    }
6463}