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