diff --git a/crates/environ/src/component/dfg.rs b/crates/environ/src/component/dfg.rs index fad632be02ef..69e456038884 100644 --- a/crates/environ/src/component/dfg.rs +++ b/crates/environ/src/component/dfg.rs @@ -267,7 +267,6 @@ pub enum CoreDef { InstanceFlags(RuntimeComponentInstanceIndex), Trampoline(TrampolineIndex), UnsafeIntrinsic(ModuleInternedTypeIndex, UnsafeIntrinsic), - TaskMayBlock, /// This is a special variant not present in `info::CoreDef` which /// represents that this definition refers to a fused adapter function. This @@ -913,7 +912,6 @@ impl LinearizeDfg<'_> { } info::CoreDef::UnsafeIntrinsic(*i) } - CoreDef::TaskMayBlock => info::CoreDef::TaskMayBlock, } } diff --git a/crates/environ/src/component/info.rs b/crates/environ/src/component/info.rs index 679564a33a6c..0e96aa047038 100644 --- a/crates/environ/src/component/info.rs +++ b/crates/environ/src/component/info.rs @@ -392,10 +392,6 @@ pub enum CoreDef { Trampoline(TrampolineIndex), /// An intrinsic for compile-time builtins. UnsafeIntrinsic(UnsafeIntrinsic), - /// Reference to a wasm global which represents a runtime-managed boolean - /// indicating whether the currently-running task may perform a blocking - /// operation. - TaskMayBlock, } impl From> for CoreDef diff --git a/crates/environ/src/component/translate.rs b/crates/environ/src/component/translate.rs index 4db699a8d5b4..3cbf3d02643d 100644 --- a/crates/environ/src/component/translate.rs +++ b/crates/environ/src/component/translate.rs @@ -691,9 +691,6 @@ impl<'a, 'data> Translator<'a, 'data> { CoreDef::InstanceFlags(_) => { unreachable!("instance flags are not a function") } - CoreDef::TaskMayBlock => { - unreachable!("task_may_block is not a function") - } // We could in theory inline these trampolines, so it // could potentially make sense to record that we @@ -1976,7 +1973,6 @@ struct Ambiguous { fn component_flags(def: &CoreDef) -> Option { match def { CoreDef::InstanceFlags(instance) => Some(KnownGlobal::ComponentInstanceFlags(*instance)), - CoreDef::TaskMayBlock => Some(KnownGlobal::TaskMayBlock), CoreDef::Export(_) | CoreDef::Trampoline(_) | CoreDef::UnsafeIntrinsic(_) => None, } } @@ -2043,10 +2039,9 @@ fn resolve_core_export( // The chain bottoms out in something that is not an export of // another instance in this component, so there is no defining module // for us to name. - CoreDef::InstanceFlags(_) - | CoreDef::Trampoline(_) - | CoreDef::UnsafeIntrinsic(_) - | CoreDef::TaskMayBlock => return None, + CoreDef::InstanceFlags(_) | CoreDef::Trampoline(_) | CoreDef::UnsafeIntrinsic(_) => { + return None; + } } } } @@ -2111,7 +2106,7 @@ fn ambiguous_entities( } } - CoreDef::InstanceFlags(_) | CoreDef::TaskMayBlock => { + CoreDef::InstanceFlags(_) => { ambiguous.flags.insert(component_flags(def).unwrap()); } diff --git a/crates/environ/src/component/translate/adapt.rs b/crates/environ/src/component/translate/adapt.rs index 61ad82bbe7bf..9555adcb7a6c 100644 --- a/crates/environ/src/component/translate/adapt.rs +++ b/crates/environ/src/component/translate/adapt.rs @@ -164,9 +164,6 @@ pub struct AdapterOptions { /// The Wasmtime-assigned component instance index where the options were /// originally specified. pub instance: RuntimeComponentInstanceIndex, - /// The ancestors (i.e. chain of instantiating instances) of the instance - /// specified in the `instance` field. - pub ancestors: Vec, /// How strings are encoded. pub string_encoding: StringEncoding, /// The async callback function used by these options, if specified. @@ -455,8 +452,7 @@ impl PartitionAdapterModules { // These items can't transitively depend on an adapter dfg::CoreDef::Trampoline(_) | dfg::CoreDef::InstanceFlags(_) - | dfg::CoreDef::UnsafeIntrinsic(..) - | dfg::CoreDef::TaskMayBlock => {} + | dfg::CoreDef::UnsafeIntrinsic(..) => {} } } diff --git a/crates/environ/src/component/translate/inline.rs b/crates/environ/src/component/translate/inline.rs index d5e8c41379e1..4fd2f4a55e9b 100644 --- a/crates/environ/src/component/translate/inline.rs +++ b/crates/environ/src/component/translate/inline.rs @@ -1583,12 +1583,6 @@ impl<'a> Inliner<'a> { let post_return = options.post_return.map(|i| frame.funcs[i].1.clone()); AdapterOptions { instance: frame.instance, - ancestors: frames - .iter() - .rev() - .skip(1) - .map(|(frame, _)| frame.instance) - .collect(), string_encoding: options.string_encoding, callback, post_return, diff --git a/crates/environ/src/fact.rs b/crates/environ/src/fact.rs index 5282d3f0f442..650647decc0a 100644 --- a/crates/environ/src/fact.rs +++ b/crates/environ/src/fact.rs @@ -113,8 +113,6 @@ pub struct Module<'a> { helper_worklist: Vec<(FunctionId, Helper)>, exports: Vec<(u32, String)>, - - task_may_block: Option, } struct AdapterData { @@ -137,9 +135,6 @@ struct AdapterOptions { /// The Wasmtime-assigned component instance index where the options were /// originally specified. instance: RuntimeComponentInstanceIndex, - /// The ancestors (i.e. chain of instantiating instances) of the instance - /// specified in the `instance` field. - ancestors: Vec, /// The ascribed type of this adapter. ty: TypeFuncIndex, /// The global that represents the instance flags for where this adapter @@ -298,7 +293,6 @@ impl<'a> Module<'a> { imported_unsafe_intrinsics: HashMap::new(), imported_traps: HashMap::new(), exports: Vec::new(), - task_may_block: None, } } @@ -352,7 +346,6 @@ impl<'a> Module<'a> { fn import_options(&mut self, ty: TypeFuncIndex, options: &AdapterOptionsDfg) -> AdapterOptions { let AdapterOptionsDfg { instance, - ancestors, string_encoding, post_return: _, // handled above callback, @@ -429,7 +422,6 @@ impl<'a> Module<'a> { AdapterOptions { instance: *instance, - ancestors: ancestors.clone(), ty, flags, post_return: None, @@ -491,25 +483,6 @@ impl<'a> Module<'a> { idx } - fn import_task_may_block(&mut self) -> GlobalIndex { - if let Some(task_may_block) = self.task_may_block { - task_may_block - } else { - let task_may_block = self.import_global( - "instance", - "task_may_block", - GlobalType { - val_type: ValType::I32, - mutable: true, - shared: false, - }, - CoreDef::TaskMayBlock, - ); - self.task_may_block = Some(task_may_block); - task_may_block - } - } - fn import_transcoder(&mut self, transcoder: transcode::Transcoder) -> FuncIndex { *self .imported_transcoders diff --git a/crates/environ/src/fact/trampoline.rs b/crates/environ/src/fact/trampoline.rs index 132f373adb59..f55b4ccf09ed 100644 --- a/crates/environ/src/fact/trampoline.rs +++ b/crates/environ/src/fact/trampoline.rs @@ -113,19 +113,6 @@ pub(super) fn compile(module: &mut Module<'_>, adapter: &AdapterData) { ) } - // If the lift and lower instances are equal, or if one is an ancestor of - // the other, we trap unconditionally. This ensures that recursive - // reentrance via an adapter is impossible. - if adapter.lift.instance == adapter.lower.instance - || adapter.lower.ancestors.contains(&adapter.lift.instance) - || adapter.lift.ancestors.contains(&adapter.lower.instance) - { - let (mut compiler, _, _) = compiler(module, adapter); - compiler.trap(Trap::CannotEnterComponent); - compiler.finish(); - return; - } - // This closure compiles a function to be exported to the host which host to // lift the parameters from the caller and lower them to the callee. // @@ -769,25 +756,7 @@ impl<'a, 'b> Compiler<'a, 'b> { let saved_lower_may_leave = self.trap_if_not_may_leave(adapter.lower.flags, Trap::CannotLeaveComponent); - let old_task_may_block = if self.module.tunables.concurrency_support { - // Save, clear, and later restore the `may_block` field. - let task_may_block = self.module.import_task_may_block(); - let old_task_may_block = if self.types[adapter.lift.ty].async_ { - self.instruction(GlobalGet(task_may_block.as_u32())); - self.instruction(I32Eqz); - self.instruction(If(BlockType::Empty)); - self.trap(Trap::CannotBlockSyncTask); - self.instruction(End); - None - } else { - let task_may_block = self.module.import_task_may_block(); - self.instruction(GlobalGet(task_may_block.as_u32())); - let old_task_may_block = self.local_set_new_tmp(ValType::I32); - self.instruction(I32Const(0)); - self.instruction(GlobalSet(task_may_block.as_u32())); - Some(old_task_may_block) - }; - + if self.module.tunables.concurrency_support { // Push a task onto the current task stack. // // Note that for sync-to-sync calls, we replace this call with @@ -809,8 +778,6 @@ impl<'a, 'b> Compiler<'a, 'b> { )); let enter_sync_call = self.module.import_enter_sync_call(); self.instruction(Call(enter_sync_call.as_u32())); - - old_task_may_block } else if self.emit_resource_call { assert!(!self.types[adapter.lift.ty].async_); self.instruction(I32Const( @@ -822,10 +789,7 @@ impl<'a, 'b> Compiler<'a, 'b> { )); let enter_sync_call = self.module.import_enter_sync_call(); self.instruction(Call(enter_sync_call.as_u32())); - None - } else { - None - }; + } // Perform the translation of arguments. Note that the `may_leave` flag // is cleared around this invocation for the callee as per the @@ -874,7 +838,9 @@ impl<'a, 'b> Compiler<'a, 'b> { // With all the arguments on the stack the actual target function is // now invoked. The core wasm results of the function are then placed // into locals for result translation afterwards. + self.instruction(Call(adapter.callee.as_u32())); + let mut result_locals = Vec::with_capacity(lift_sig.results.len()); let mut temps = Vec::new(); for ty in lift_sig.results.iter().rev() { @@ -944,16 +910,6 @@ impl<'a, 'b> Compiler<'a, 'b> { self.free_temp_local(tmp); } - if self.module.tunables.concurrency_support { - // Restore old `may_block_field` - if let Some(old_task_may_block) = old_task_may_block { - let task_may_block = self.module.import_task_may_block(); - self.instruction(LocalGet(old_task_may_block.idx)); - self.instruction(GlobalSet(task_may_block.as_u32())); - self.free_temp_local(old_task_may_block); - } - } - self.exit_exception_barrier(); self.finish() diff --git a/crates/test-util/src/wast.rs b/crates/test-util/src/wast.rs index 3687e65d63c5..1016c21682ac 100644 --- a/crates/test-util/src/wast.rs +++ b/crates/test-util/src/wast.rs @@ -687,6 +687,14 @@ impl WastTest { return true; } + // This will require a wasm-tools update: + if self + .path + .ends_with("component-model/test/validation/max-value-size.wast") + { + return true; + } + false } } diff --git a/crates/wasmtime/src/runtime/component/concurrent.rs b/crates/wasmtime/src/runtime/component/concurrent.rs index 2bae1c0ccb8d..4031582a3b96 100644 --- a/crates/wasmtime/src/runtime/component/concurrent.rs +++ b/crates/wasmtime/src/runtime/component/concurrent.rs @@ -665,7 +665,6 @@ enum SuspendReason { Waiting { set: TableId, thread: QualifiedThreadId, - skip_may_block_check: bool, }, /// The fiber has finished handling its most recent work item and is waiting /// for another (or to be dropped if it is no longer needed). @@ -675,13 +674,9 @@ enum SuspendReason { Yielding { thread: QualifiedThreadId, cancellable: bool, - skip_may_block_check: bool, }, /// The fiber was explicitly suspended with a call to `thread.suspend` or `thread.switch-to`. - ExplicitlySuspending { - thread: QualifiedThreadId, - skip_may_block_check: bool, - }, + ExplicitlySuspending { thread: QualifiedThreadId }, } /// Represents a pending call into guest code for a given guest task. @@ -702,7 +697,7 @@ enum GuestCallKind { /// /// If the closure returns `Ok(Some(call))`, the `call` should be run /// immediately using `handle_guest_call`. - StartImplicit(Box Result> + Send + Sync>), + StartImplicit(Box Result<()> + Send + Sync>), StartExplicit(Box Result<()> + Send + Sync>), } @@ -723,18 +718,17 @@ impl fmt::Debug for GuestCallKind { /// The target of a suspension intrinsic. #[derive(Copy, Clone, Debug)] pub enum SuspensionTarget { - SomeSuspended(u32), - Some(u32), + Resume(u32), + Promote(u32), None, } -impl SuspensionTarget { - fn is_none(&self) -> bool { - matches!(self, SuspensionTarget::None) - } - fn is_some(&self) -> bool { - !self.is_none() - } +/// Behavior for `resume_thread`. +#[derive(Copy, Clone, Debug)] +pub enum ResumeThread { + Promote, + Resume, + ResumeLater, } /// Represents a pending call into guest code for a given guest thread. @@ -755,15 +749,16 @@ impl GuestCall { /// - the call is for a not-yet started task and the (sub-)component /// instance to be called has backpressure enabled fn is_ready(&self, store: &mut StoreOpaque) -> Result { - let instance = store - .concurrent_state_mut()? - .get_mut(self.thread.task)? - .instance; + let task = store.concurrent_state_mut()?.get_mut(self.thread.task)?; + let async_typed = task.async_typed; + let instance = task.instance; let state = store.instance_state(instance).concurrent_state(); let ready = match &self.kind { GuestCallKind::DeliverEvent { .. } => !state.do_not_enter, - GuestCallKind::StartImplicit(_) => !(state.do_not_enter || state.backpressure > 0), + GuestCallKind::StartImplicit(_) => { + !async_typed || !(state.do_not_enter || state.backpressure > 0) + } GuestCallKind::StartExplicit(_) => true, }; log::trace!( @@ -787,11 +782,21 @@ enum WorkItem { /// A host task to be pushed to `ConcurrentState::futures`. PushFuture(AlwaysMut), /// A fiber to resume. - ResumeFiber(StoreFiber<'static>), + ResumeFiber { + instance: RuntimeInstance, + thread: QualifiedThreadId, + fiber: StoreFiber<'static>, + }, /// A thread to resume. - ResumeThread(RuntimeComponentInstanceIndex, QualifiedThreadId), + ResumeThread { + instance: RuntimeInstance, + thread: QualifiedThreadId, + }, /// A pending call into guest code for a given guest task. - GuestCall(RuntimeComponentInstanceIndex, GuestCall), + GuestCall { + instance: RuntimeInstance, + call: GuestCall, + }, /// A job to run on a worker fiber. WorkerFunction(AlwaysMut Result<()> + Send>>), } @@ -800,16 +805,22 @@ impl fmt::Debug for WorkItem { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match self { Self::PushFuture(_) => f.debug_tuple("PushFuture").finish(), - Self::ResumeFiber(_) => f.debug_tuple("ResumeFiber").finish(), - Self::ResumeThread(instance, thread) => f - .debug_tuple("ResumeThread") - .field(instance) - .field(thread) + Self::ResumeFiber { + instance, thread, .. + } => f + .debug_struct("ResumeFiber") + .field("instance", instance) + .field("thread", thread) .finish(), - Self::GuestCall(instance, call) => f - .debug_tuple("GuestCall") - .field(instance) - .field(call) + Self::ResumeThread { instance, thread } => f + .debug_struct("ResumeThread") + .field("instance", instance) + .field("thread", thread) + .finish(), + Self::GuestCall { instance, call } => f + .debug_struct("GuestCall") + .field("instance", instance) + .field("call", call) .finish(), Self::WorkerFunction(_) => f.debug_tuple("WorkerFunction").finish(), } @@ -882,6 +893,9 @@ pub(crate) fn poll_and_block( // then use `GuestThread::sync_call_set` to wait for the task to // complete, suspending the current fiber until it does so. Poll::Pending => { + let caller_instance = store.concurrent_state_mut()?.get_mut(caller.task)?.instance; + store.switch_or_trap_if_may_not_suspend(caller_instance)?; + let state = store.concurrent_state_mut()?; state.push_future(future); @@ -891,7 +905,6 @@ pub(crate) fn poll_and_block( store.suspend(SuspendReason::Waiting { set, thread: caller, - skip_may_block_check: false, })?; // Remove the `task` from the `sync_call_set` to ensure that when @@ -914,70 +927,59 @@ pub(crate) fn poll_and_block( /// Execute the specified guest call. fn handle_guest_call(store: &mut dyn VMStore, call: GuestCall) -> Result<()> { - let mut next = Some(call); - while let Some(call) = next.take() { - match call.kind { - GuestCallKind::DeliverEvent { instance, set } => { - let (event, waitable) = - match instance.get_event(store, call.thread.task, set, true)? { - Some(pair) => pair, - None => bail_bug!("delivering non-present event"), - }; - let state = store.concurrent_state_mut()?; - let task = state.get_mut(call.thread.task)?; - let runtime_instance = task.instance; - let handle = waitable.map(|(_, v)| v).unwrap_or(0); - - log::trace!( - "use callback to deliver event {event:?} to {:?} for {waitable:?}", - call.thread, - ); - - let old_thread = store.set_thread(call.thread)?; - log::trace!( - "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread", - call.thread - ); - - store.enter_instance(runtime_instance); - - let Some(callback) = store - .concurrent_state_mut()? - .get_mut(call.thread.task)? - .callback - .take() - else { - bail_bug!("guest task callback field not present") - }; + match call.kind { + GuestCallKind::DeliverEvent { instance, set } => { + let (event, waitable) = match instance.get_event(store, call.thread.task, set, true)? { + Some(pair) => pair, + None => bail_bug!("delivering non-present event"), + }; + let state = store.concurrent_state_mut()?; + let task = state.get_mut(call.thread.task)?; + let runtime_instance = task.instance; + let handle = waitable.map(|(_, v)| v).unwrap_or(0); + + log::trace!( + "use callback to deliver event {event:?} to {:?} for {waitable:?}", + call.thread, + ); + + let old_thread = store.set_thread(call.thread)?; + log::trace!( + "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread", + call.thread + ); + + store.enter_instance(runtime_instance); + + let Some(callback) = store + .concurrent_state_mut()? + .get_mut(call.thread.task)? + .callback + .take() + else { + bail_bug!("guest task callback field not present") + }; - let code = callback(store, event, handle)?; + let code = callback(store, event, handle)?; - store - .concurrent_state_mut()? - .get_mut(call.thread.task)? - .callback = Some(callback); + store + .concurrent_state_mut()? + .get_mut(call.thread.task)? + .callback = Some(callback); - store.exit_instance(runtime_instance)?; + store.exit_instance(runtime_instance)?; - store.set_thread(old_thread)?; + store.set_thread(old_thread)?; - next = instance.handle_callback_code( - store, - call.thread, - runtime_instance.index, - code, - )?; + instance.handle_callback_code(store, call.thread, runtime_instance.index, code)?; - log::trace!( - "GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread" - ); - } - GuestCallKind::StartImplicit(fun) => { - next = fun(store)?; - } - GuestCallKind::StartExplicit(fun) => { - fun(store)?; - } + log::trace!("GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread"); + } + GuestCallKind::StartImplicit(fun) => { + fun(store)?; + } + GuestCallKind::StartExplicit(fun) => { + fun(store)?; } } @@ -1197,10 +1199,7 @@ impl StoreContextMut<'_, T> { pub(super) async fn run_concurrent_trap_on_idle( self, fun: impl AsyncFnOnce(&Accessor) -> R, - ) -> Result - where - T: Send + 'static, - { + ) -> Result { self.do_run_concurrent(fun, true).await } @@ -1208,10 +1207,7 @@ impl StoreContextMut<'_, T> { mut self, fun: impl AsyncFnOnce(&Accessor) -> R, trap_on_idle: bool, - ) -> Result - where - T: Send + 'static, - { + ) -> Result { debug_assert!(self.0.concurrency_support()); check_recursive_run(); let token = StoreToken::new(self.as_context_mut()); @@ -1223,6 +1219,11 @@ impl StoreContextMut<'_, T> { impl<'a, T, V> Drop for Dropper<'a, T, V> { fn drop(&mut self) { + self.store + .0 + .concurrent_state_mut_already_forced_current_thread() + .event_loop_running = false; + tls::set(self.store.0, || { // SAFETY: Here we drop the value without moving it for the // first and only time -- per the contract for `Drop::drop`, @@ -1234,6 +1235,9 @@ impl StoreContextMut<'_, T> { } let accessor = &Accessor::new(token); + self.0 + .concurrent_state_mut_already_forced_current_thread() + .event_loop_running = true; let dropper = &mut Dropper { store: self, value: ManuallyDrop::new(fun(accessor)), @@ -1257,10 +1261,7 @@ impl StoreContextMut<'_, T> { mut self, mut future: Pin<&mut impl Future>, trap_on_idle: bool, - ) -> Result - where - T: Send + 'static, - { + ) -> Result { struct Reset<'a, T: 'static> { store: StoreContextMut<'a, T>, futures: Option>, @@ -1327,12 +1328,15 @@ impl StoreContextMut<'_, T> { // any. This will be either all of the high-priority work // items, or if there are none, a single low-priority work item. let state = reset.store.0.concurrent_state_mut()?; - let mut ready = mem::take(&mut state.high_priority); + let mut ready = state.switch_item.take().into_iter().collect::>(); let mut low_priority = false; if ready.is_empty() { - if let Some(item) = state.low_priority.pop_back() { - ready.push(item); - low_priority = true; + ready = mem::take(&mut state.high_priority); + if ready.is_empty() { + if let Some(item) = state.low_priority.pop_back() { + ready.push(item); + low_priority = true; + } } } if !ready.is_empty() { @@ -1374,6 +1378,39 @@ impl StoreContextMut<'_, T> { if trap_on_idle { // `trap_on_idle` is true, so we exit // immediately. + + // If there are any tasks belonging to + // an instance which may not suspend, trap with + // `CannotBlockSyncTask`: + let instances = reset + .store + .0 + .concurrent_state_mut()? + .table + .get_mut() + .iter_mut() + .filter_map(|entry| { + if let Some(task) = entry.downcast_ref::() { + Some(task.instance) + } else { + None + } + }) + .collect::>(); + + for instance in instances { + if reset + .store + .0 + .instance_state(instance) + .concurrent_state() + .do_not_suspend + { + return Poll::Ready(Err(Trap::CannotBlockSyncTask.into())); + } + } + + // Otherwise, trap with `AsyncDeadlock`: Poll::Ready(Err(Trap::AsyncDeadlock.into())) } else { // `trap_on_idle` is false, so we assume @@ -1416,7 +1453,9 @@ impl StoreContextMut<'_, T> { fn drop(&mut self) { while let Some(item) = self.ready.next() { match item { - WorkItem::ResumeFiber(mut fiber) => fiber.dispose(self.store.0), + WorkItem::ResumeFiber { mut fiber, .. } => { + fiber.dispose(self.store.0) + } WorkItem::PushFuture(future) => { tls::set(self.store.0, move || drop(future)) } @@ -1469,10 +1508,7 @@ impl StoreContextMut<'_, T> { } /// Handle the specified work item, possibly resuming a fiber if applicable. - async fn handle_work_item(self, item: WorkItem) -> Result<()> - where - T: Send, - { + async fn handle_work_item(self, item: WorkItem) -> Result<()> { log::trace!("handle work item {item:?}"); match item { WorkItem::PushFuture(future) => { @@ -1481,10 +1517,10 @@ impl StoreContextMut<'_, T> { .futures_mut()? .push(future.into_inner()); } - WorkItem::ResumeFiber(fiber) => { + WorkItem::ResumeFiber { fiber, .. } => { self.0.resume_fiber(fiber).await?; } - WorkItem::ResumeThread(_, thread) => { + WorkItem::ResumeThread { thread, .. } => { if let GuestThreadState::Ready { fiber, .. } = mem::replace( &mut self.0.concurrent_state_mut()?.get_mut(thread.thread)?.state, GuestThreadState::Running, @@ -1494,7 +1530,7 @@ impl StoreContextMut<'_, T> { bail_bug!("cannot resume non-pending thread {thread:?}"); } } - WorkItem::GuestCall(_, call) => { + WorkItem::GuestCall { call, .. } => { if call.is_ready(self.0)? { self.run_on_worker(WorkerItem::GuestCall(call)).await?; } else { @@ -1529,26 +1565,43 @@ impl StoreContextMut<'_, T> { } /// Execute the specified guest call on a worker fiber. - async fn run_on_worker(self, item: WorkerItem) -> Result<()> - where - T: Send, - { + async fn run_on_worker(self, item: WorkerItem) -> Result<()> { let worker = if let Some(fiber) = self.0.concurrent_state_mut()?.worker.take() { fiber } else { - fiber::make_fiber(self.0, move |store| { - loop { - let Some(item) = store.concurrent_state_mut()?.worker_item.take() else { - bail_bug!("worker_item not present when resuming fiber") - }; - match item { - WorkerItem::GuestCall(call) => handle_guest_call(store, call)?, - WorkerItem::Function(fun) => fun.into_inner()(store)?, - } + // SAFETY: the `make_fiber_unchecked` function is unsafe because the + // returned fiber is unconditionally `Send` as opposed to being + // conditionally send depending on the argument (in this case + // `self.0`). This `async` function, however, is conditionally + // `Send` depending on `self`, in this case `StoreContextMut`, + // which is already going to be conditionally `Send` depending on + // `T`. + // + // The returned fiber is possibly stored within the `Store` as + // well. If `T: Send` then that's fine and everything's dandy. If + // `T: !Send`, however, then the store is already not-`Send` meaning + // that putting more actually-not-`Send` things inside of it isn't + // an issue. + // + // The main issue here is that the returned fiber effectively can't + // get transferred outside the context of the store. That's an + // implementation detail we'll have to rely on, but is currently + // true. + unsafe { + fiber::make_fiber_unchecked(self.0, move |store| { + loop { + let Some(item) = store.concurrent_state_mut()?.worker_item.take() else { + bail_bug!("worker_item not present when resuming fiber") + }; + match item { + WorkerItem::GuestCall(call) => handle_guest_call(store, call)?, + WorkerItem::Function(fun) => fun.into_inner()(store)?, + } - store.suspend(SuspendReason::NeedWork)?; - } - })? + store.suspend(SuspendReason::NeedWork)?; + } + })? + } }; let worker_item = &mut self.0.concurrent_state_mut()?.worker_item; @@ -1729,6 +1782,35 @@ impl StoreOpaque { Ok(false) } + fn enter_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> { + log::trace!("enter sync-typed call {callee:?}"); + let state = self.instance_state(callee).concurrent_state(); + let old_do_not_suspend = state.do_not_suspend; + state.do_not_suspend = true; + + let thread = self.current_guest_thread()?; + let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?; + if thread.old_do_not_suspend.is_some() { + bail_bug!("current thread already has `old_do_not_suspend` value"); + } + + thread.old_do_not_suspend = Some(old_do_not_suspend); + + Ok(()) + } + + fn exit_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> { + log::trace!("exit sync-typed call {callee:?}"); + let thread = self.current_guest_thread()?; + let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?; + let Some(old_do_not_suspend) = thread.old_do_not_suspend.take() else { + bail_bug!("current thread missing `old_do_not_suspend` value"); + }; + let state = self.instance_state(callee).concurrent_state(); + state.do_not_suspend = old_do_not_suspend; + Ok(()) + } + /// Push a `GuestTask` onto the task stack for either a sync-to-sync, /// guest-to-guest call or a sync host-to-guest call. /// @@ -1743,10 +1825,10 @@ impl StoreOpaque { pub(crate) fn enter_guest_sync_call( &mut self, guest_caller: Option, - callee_async: bool, + callee_async_typed: bool, callee: RuntimeInstance, ) -> Result<()> { - log::trace!("enter sync call {callee:?}"); + log::trace!("enter sync-lifted call {callee:?}"); if !self.concurrency_support() { return self.enter_call_not_concurrent(); } @@ -1781,7 +1863,8 @@ impl StoreOpaque { }, None, callee, - callee_async, + callee_async_typed, + true, )?; Instance::from_wasmtime(self, callee.instance).add_guest_thread_to_instance_table( @@ -1791,10 +1874,14 @@ impl StoreOpaque { )?; self.set_thread(guest_thread)?; + if !callee_async_typed { + self.enter_sync_call(callee)?; + } + Ok(()) } - /// Pop a `GuestTask` previously pushed using `enter_sync_call`. + /// Pop a `GuestTask` previously pushed using `enter_guest_sync_call`. /// /// NB: for sync-to-sync, guest-to-guest calls we delay task construction in /// fused adapters and then when the call returns we check to see if the @@ -1805,21 +1892,29 @@ impl StoreOpaque { if !self.concurrency_support() { return Ok(self.exit_call_not_concurrent()); } - let thread = match self.set_thread(CurrentThread::None)?.guest() { + + let thread = match self.current_thread()?.guest() { Some(t) => *t, None => bail_bug!("expected task when exiting"), }; let task = self.concurrent_state_mut()?.get_mut(thread.task)?; let instance = task.instance; + let caller = match &task.caller { &Caller::Guest { thread } => thread.into(), &Caller::Host { caller, .. } => caller, }; task.lift_result = None; task.exited = true; + let async_typed = task.async_typed; + + if !async_typed { + self.exit_sync_call(instance)?; + } + self.set_thread(caller)?; - log::trace!("exit sync call {instance:?}"); + log::trace!("exit sync-lifted call {instance:?}"); self.cleanup_thread(thread, instance, CleanupTask::Yes)?; Ok(()) @@ -1865,38 +1960,11 @@ impl StoreOpaque { Ok(()) } - /// Determine whether the specified instance may be entered from the host. + /// Determine whether an instance may be entered from the host. /// - /// We return `true` here only if all of the following hold: - /// - /// - The top-level instance is not already on the current task's call stack. - /// - The instance is not in need of a post-return function call. - /// - `self` has not been poisoned due to a trap. - pub(crate) fn may_enter(&mut self, instance: RuntimeInstance) -> Result { - if self.trapped() { - return Ok(false); - } - if !self.concurrency_support() { - return Ok(true); - } - let mut cur = Some(self.current_thread()?); - let state = self.concurrent_state_mut()?; - while let Some(t) = cur { - if let Some(task) = t.guest_task() { - let task = state.get_mut(task)?; - // Note that we only compare top-level instance IDs here. - // The idea is that the host is not allowed to recursively - // enter a top-level instance even if the specific leaf - // instance is not on the stack. This the behavior defined - // in the spec, and it allows us to elide runtime checks in - // guest-to-guest adapters. - if task.instance.instance == instance.instance { - return Ok(false); - } - } - cur = state.parent(t); - } - Ok(true) + /// We return `false` here only `self` has been poisoned due to a trap. + pub(crate) fn may_enter(&mut self) -> bool { + !self.trapped() } /// Helper function to retrieve the `InstanceState` for the @@ -1942,24 +2010,6 @@ impl StoreOpaque { *self.vm_store_context_mut().component_context_mut() = context; } - // Each time we switch threads, we conservatively set `task_may_block` - // to `false` for the component instance we're switching away from (if - // any), meaning it will be `false` for any new thread created for that - // instance unless explicitly set otherwise. - // - // Additionally if we're switching to a new thread, set its component - // instance's `task_may_block` according to where it left off. - let state = self.concurrent_state_mut()?; - if let Some(old_task) = old_thread.guest_task() { - let instance = state.get_mut(old_task)?.instance.instance; - self.component_instance_mut(instance) - .set_task_may_block(false) - } - - if thread.guest_task().is_some() { - self.set_task_may_block()?; - } - // Keep the JIT-visible current-thread pointer in sync. *self.vm_store_context_mut().current_thread_mut() = if thread.is_none() { VMLazyThread::none() @@ -1970,34 +2020,36 @@ impl StoreOpaque { Ok(old_thread) } - /// Set the global variable representing whether the current task may block - /// prior to entering Wasm code. - fn set_task_may_block(&mut self) -> Result<()> { - let guest_thread = self.current_guest_thread()?; - let state = self.concurrent_state_mut()?; - let instance = state.get_mut(guest_thread.task)?.instance.instance; - let may_block = self.concurrent_state_mut()?.may_block(guest_thread.task)?; - self.component_instance_mut(instance) - .set_task_may_block(may_block); - Ok(()) - } - - pub(crate) fn check_blocking(&mut self) -> Result<()> { - if !self.concurrency_support() { - return Ok(()); - } - let task = self.current_guest_thread()?.task; - let state = self.concurrent_state_mut()?; - let instance = state.get_mut(task)?.instance.instance; - let task_may_block = self.component_instance(instance).get_task_may_block(); - - if task_may_block { + /// Check if the specified instance has a sync-typed call in progress; if so + /// attempt to switch to another ready thread for that instance, and if no + /// such thread exists, trap. + fn switch_or_trap_if_may_not_suspend(&mut self, instance: RuntimeInstance) -> Result<()> { + if self.switch_if_may_not_suspend(instance)? { Ok(()) } else { Err(Trap::CannotBlockSyncTask.into()) } } + /// Check if the specified instance has a sync-typed call in progress; if so + /// attempt to switch to another ready thread for that instance, and if no + /// such thread exists, return false. + fn switch_if_may_not_suspend(&mut self, instance: RuntimeInstance) -> Result { + // Call this for the side effect of forcing any deferred task creation, + // which may influence the value of `ConcurrentState::do_not_suspend` + // below: + self.concurrent_state_mut()?; + + Ok(!self.concurrency_support() + || !self + .instance_state(instance) + .concurrent_state() + .do_not_suspend + || self + .concurrent_state_mut()? + .promote_instance_local_thread_work_item(instance)?) + } + /// Record that we're about to enter a (sub-)component instance which does /// not support more than one concurrent, stackful activation, meaning it /// cannot be entered again until the next call returns. @@ -2033,7 +2085,7 @@ impl StoreOpaque { let call = GuestCall { thread, kind }; if call.is_ready(self)? { self.concurrent_state_mut()? - .push_high_priority(WorkItem::GuestCall(instance.index, call)); + .push_high_priority(WorkItem::GuestCall { instance, call }); } else { self.instance_state(instance) .concurrent_state() @@ -2112,8 +2164,8 @@ impl StoreOpaque { } => { state.get_mut(thread.thread)?.state = GuestThreadState::Ready { fiber, cancellable }; - let instance = state.get_mut(thread.task)?.instance.index; - state.push_low_priority(WorkItem::ResumeThread(instance, thread)); + let instance = state.get_mut(thread.task)?.instance; + state.push_low_priority(WorkItem::ResumeThread { instance, thread }); } SuspendReason::ExplicitlySuspending { thread, .. } => { state.get_mut(thread.thread)?.state = GuestThreadState::Suspended(fiber); @@ -2157,35 +2209,16 @@ impl StoreOpaque { CurrentThread::None }; - // We should not have reached here unless either there's no current - // task, or the current task is permitted to block. In addition, we - // special-case `thread.switch-to` and waiting for a subtask to go from - // `starting` to `started`, both of which we consider non-blocking - // operations despite requiring a suspend. - debug_assert!( - matches!( - reason, - SuspendReason::ExplicitlySuspending { - skip_may_block_check: true, - .. - } | SuspendReason::Waiting { - skip_may_block_check: true, - .. - } | SuspendReason::Yielding { - skip_may_block_check: true, - .. - } - ) || old_guest_thread - .guest_task() - .map(|task| self.concurrent_state_mut()?.may_block(task)) - .transpose()? - .unwrap_or(true) - ); - let suspend_reason = &mut self.concurrent_state_mut()?.suspend_reason; assert!(suspend_reason.is_none()); *suspend_reason = Some(reason); + // We'll panic if we call `Self::with_blocking` when the fiber is being + // disposed, so check for that and exit ASAP if appropriate. + if !self.fiber_async_state_mut().can_block() { + return Err(format_err!("future dropped")); + } + self.with_blocking(|_, cx| cx.suspend(StoreFiberYield::ReleaseStore))?; if task.is_some() { @@ -2195,7 +2228,11 @@ impl StoreOpaque { Ok(()) } - fn wait_for_event(&mut self, waitable: Waitable) -> Result<()> { + fn wait_for_event( + &mut self, + caller_instance: RuntimeInstance, + waitable: Waitable, + ) -> Result<()> { let caller = self.current_guest_thread()?; let state = self.concurrent_state_mut()?; @@ -2203,10 +2240,12 @@ impl StoreOpaque { let set = state.get_mut(caller.thread)?.sync_call_set; waitable.join(state, Some(set))?; + + self.switch_or_trap_if_may_not_suspend(caller_instance)?; + self.suspend(SuspendReason::Waiting { set, thread: caller, - skip_may_block_check: false, })?; let state = self.concurrent_state_mut()?; waitable.join(state, None) @@ -2239,6 +2278,22 @@ impl StoreOpaque { runtime_instance: RuntimeInstance, cleanup_task: CleanupTask, ) -> Result<()> { + let state = self.concurrent_state_mut()?; + let thread_data = state.get_mut(guest_thread.thread)?; + + // If `thread_data.explicit` is `true`, then we're effectively + // suspending this thread forever. In that case we need to make sure we + // don't leave a sync-typed task in a blocked state, so we call + // `switch_or_trap_if_may_not_suspend`. Otherwise, we're dealing with an implicit + // thread which is returning control to a caller, so no need to call + // `switch_or_trap_if_may_not_suspend` yet. + // + // TODO: What if this is an implicit thread but the caller has either + // already exited or is not waiting on this subtask? + if thread_data.explicit { + self.switch_or_trap_if_may_not_suspend(runtime_instance)?; + } + let state = self.concurrent_state_mut()?; let thread_data = state.get_mut(guest_thread.thread)?; let sync_call_set = thread_data.sync_call_set; @@ -2360,6 +2415,15 @@ impl StoreOpaque { assert_eq!((bits << 1) >> 1, bits); Ok(Some((bits << 1) | u32::from(is_host))) } + + pub(crate) fn queue_task( + &mut self, + task: impl FnOnce(&mut dyn VMStore) -> Result<()> + Send + 'static, + ) -> Result<()> { + self.concurrent_state_mut()? + .push_high_priority(WorkItem::WorkerFunction(AlwaysMut::new(Box::new(task)))); + Ok(()) + } } enum CleanupTask { @@ -2428,7 +2492,7 @@ impl Instance { guest_thread: QualifiedThreadId, runtime_instance: RuntimeComponentInstanceIndex, code: u32, - ) -> Result> { + ) -> Result<()> { let (code, set) = unpack_callback_code(code); log::trace!("received callback code from {guest_thread:?}: {code} (set: {set})"); @@ -2444,7 +2508,7 @@ impl Instance { Ok(TableId::::new(set)) }; - Ok(match code { + match code { callback_code::EXIT => { log::trace!("implicit thread {guest_thread:?} completed"); let task = store.concurrent_state_mut()?.get_mut(guest_thread.task)?; @@ -2455,7 +2519,6 @@ impl Instance { self.runtime_instance(runtime_instance), CleanupTask::Yes, )?; - None } callback_code::YIELD => { let task = state.get_mut(guest_thread.task)?; @@ -2475,23 +2538,14 @@ impl Instance { set: None, }, }; - if state.may_block(guest_thread.task)? { - // Push this thread onto the "low priority" queue so it runs - // after any other threads have had a chance to run. - state.push_low_priority(WorkItem::GuestCall(runtime_instance, call)); - None - } else { - // Yielding in a non-blocking context is defined as a no-op - // according to the spec, so we must run this thread - // immediately without allowing any others to run. - Some(call) - } + // Push this thread onto the "low priority" queue so it runs + // after any other threads have had a chance to run. + state.push_low_priority(WorkItem::GuestCall { + instance: self.runtime_instance(runtime_instance), + call, + }); } callback_code::WAIT => { - // The task may only return `WAIT` if it was created for a call - // to an async export). Otherwise, we'll trap. - state.check_blocking_for(guest_thread.task)?; - let set = get_set(store, set)?; let state = store.concurrent_state_mut()?; @@ -2499,16 +2553,16 @@ impl Instance { || !state.get_mut(set)?.ready.is_empty() { // An event is immediately available; deliver it ASAP. - state.push_high_priority(WorkItem::GuestCall( - runtime_instance, - GuestCall { + state.push_high_priority(WorkItem::GuestCall { + instance: self.runtime_instance(runtime_instance), + call: GuestCall { thread: guest_thread, kind: GuestCallKind::DeliverEvent { instance: self, set: Some(set), }, }, - )); + }); } else { // No event is immediately available. // @@ -2532,19 +2586,20 @@ impl Instance { bail_bug!("set's waiting set already had this thread registered"); } } - None } _ => bail!(Trap::UnsupportedCallbackCode), - }) + } + + Ok(()) } - /// Add the specified guest call to the "high priority" work item queue, to - /// be started as soon as backpressure and/or reentrance rules allow. + /// Stage the specified guest call as the work item to run next, to be + /// started as soon as backpressure and/or reentrance rules allow. /// /// SAFETY: The raw pointer arguments must be valid references to guest - /// functions (with the appropriate signatures) when the closures queued by + /// functions (with the appropriate signatures) when the closures staged by /// this function are called. - unsafe fn queue_call( + unsafe fn stage_call( self, mut store: StoreContextMut, guest_thread: QualifiedThreadId, @@ -2654,7 +2709,7 @@ impl Instance { // SAFETY: See the documentation for `make_call` to review the // contract we must uphold for `call` here. // - // Per the contract described in the `queue_call` + // Per the contract described in the `stage_call` // documentation, the `callee` pointer which `call` closes // over must be valid. let storage = call(store)?; @@ -2673,8 +2728,7 @@ impl Instance { let code = unsafe { storage[0].assume_init() }.get_i32() as u32; self.handle_callback_code(store, guest_thread, callee_instance.index, code) - }) - as Box Result> + Send + Sync> + }) as Box Result<()> + Send + Sync> } else { let token = StoreToken::new(store.as_context_mut()); Box::new(move |store: &mut dyn VMStore| { @@ -2689,30 +2743,45 @@ impl Instance { ); let flags = self.id().get(store).instance_flags(callee_instance.index); - // Unless this is a callback-less (i.e. stackful) - // async-lifted export, we need to record that the instance + let callee_async_typed = store + .concurrent_state_mut()? + .get_mut(guest_thread.task)? + .async_typed; + + // Unless this is a callback-less (i.e. stackful) async-lifted + // or sync-typed export, we need to record that the instance // cannot be entered until the call returns. - if !async_ { + if !async_ && callee_async_typed { store.enter_instance(callee_instance); } + if !callee_async_typed { + store.enter_sync_call(callee_instance)?; + } + // SAFETY: See the documentation for `make_call` to review the // contract we must uphold for `call` here. // - // Per the contract described in the `queue_call` + // Per the contract described in the `stage_call` // documentation, the `callee` pointer which `call` closes // over must be valid. let storage = call(store)?; + if !callee_async_typed { + store.exit_sync_call(callee_instance)?; + } + if !async_ { // This is a sync-lifted export, so now is when we lift the // result, optionally call the post-return function, if any, // and finally notify any current or future waiters that the // subtask has returned. - let lift = { + if callee_async_typed { store.exit_instance(callee_instance)?; + } + let lift = { let state = store.concurrent_state_mut()?; if !state.get_mut(guest_thread.task)?.result.is_none() { bail_bug!("task has already produced a result"); @@ -2761,20 +2830,20 @@ impl Instance { // This is a callback-less call, so the implicit thread has now completed store.cleanup_thread(guest_thread, callee_instance, CleanupTask::Yes)?; - Ok(None) + Ok(()) }) }; store .0 .concurrent_state_mut()? - .push_high_priority(WorkItem::GuestCall( - callee_instance.index, - GuestCall { + .set_switch_item(WorkItem::GuestCall { + instance: callee_instance, + call: GuestCall { thread: guest_thread, kind: GuestCallKind::StartImplicit(fun), }, - )); + })?; Ok(()) } @@ -2799,18 +2868,11 @@ impl Instance { caller_instance: RuntimeComponentInstanceIndex, callee_instance: RuntimeComponentInstanceIndex, task_return_type: TypeTupleIndex, - callee_async: bool, + callee_async_typed: bool, memory: *mut VMMemoryDefinition, string_encoding: StringEncoding, caller_info: CallerInfo, ) -> Result<()> { - if let (CallerInfo::Sync { .. }, true) = (&caller_info, callee_async) { - // A task may only call an async-typed function via a sync lower if - // it was created by a call to an async export. Otherwise, we'll - // trap. - store.0.check_blocking()?; - } - enum ResultInfo { Heap { results: u32 }, Stack { result_count: u32 }, @@ -2923,13 +2985,6 @@ impl Instance { my_src.push(ValRaw::u32(*results)); } - // Execute the `return_` hook, generated by Wasmtime's FACT - // compiler, in the context of the old thread. The old - // thread, this thread's caller, may have `realloc` - // callbacks invoked for example and those need the correct - // context set for the current thread. - let prev = store.0.set_thread(old_thread)?; - // SAFETY: `return_` is a valid `*mut VMFuncRef` from // `wasmtime-cranelift`-generated fused adapter code. Based // on how it was constructed (see @@ -2944,10 +2999,6 @@ impl Instance { )?; } - // Restore the previous current thread after the - // lifting/lowering has returned. - store.0.set_thread(prev)?; - let thread = store.0.current_guest_thread()?; let state = store.0.concurrent_state_mut()?; if sync_caller { @@ -2972,7 +3023,10 @@ impl Instance { Caller::Guest { thread: old_thread }, None, self.runtime_instance(callee_instance), - callee_async, + callee_async_typed, + // We don't know whether the callee export was lifted sync or async + // yet, but we'll update this in `start_call`: + false, )?; // Make the new thread the current one so that `Self::start_call` knows @@ -3041,7 +3095,11 @@ impl Instance { let async_caller = storage.is_none(); let guest_thread = store.0.current_guest_thread()?; let state = store.0.concurrent_state_mut()?; - let callee_async = state.get_mut(guest_thread.task)?.async_function; + + if !state.event_loop_running { + bail_bug!("Instance::start_call called without a running event loop"); + } + let callee = SendSyncPtr::new(callee); let param_count = usize::try_from(param_count)?; assert!(param_count <= MAX_FLAT_PARAMS); @@ -3049,6 +3107,11 @@ impl Instance { assert!(result_count <= MAX_FLAT_RESULTS); let task = state.get_mut(guest_thread.task)?; + let callee_async_typed = task.async_typed; + let callee_instance = task.instance; + + task.async_lifted = (flags & START_FLAG_ASYNC_CALLEE) != 0; + if let Some(callback) = NonNull::new(callback) { // We're calling an async-lifted export with a callback, so store // the callback and related context as part of the task so we can @@ -3068,9 +3131,9 @@ impl Instance { let caller = *caller; let caller_instance = state.get_mut(caller.task)?.instance; - // Queue the call as a "high priority" work item. + // Stage the call as the work item to run next, modulo backpressure, etc. unsafe { - self.queue_call( + self.stage_call( store.as_context_mut(), guest_thread, callee, @@ -3082,6 +3145,21 @@ impl Instance { )?; } + let old_do_not_suspend = if callee_async_typed { + // If we're starting an async-typed call, it is permitted to suspend + // since that will just return control back to the caller, which may + // be able to avoid blocking if needed. + // + // We'll restore the old value after the call either returns or + // suspends. + let state = store.0.instance_state(callee_instance).concurrent_state(); + let old_do_not_suspend = state.do_not_suspend; + state.do_not_suspend = false; + Some(old_do_not_suspend) + } else { + None + }; + let state = store.0.concurrent_state_mut()?; // Use the caller's `GuestThread::sync_call_set` to register interest in @@ -3112,16 +3190,16 @@ impl Instance { store.0.suspend(SuspendReason::Waiting { set, thread: caller, - // Normally, `StoreOpaque::suspend` would assert it's being - // called from a context where blocking is allowed. However, if - // `async_caller` is `true`, we'll only "block" long enough for - // the callee to start, i.e. we won't repeat this loop, so we - // tell `suspend` it's okay even if we're not allowed to block. - // Alternatively, if the callee is not an async function, then - // we know it won't block anyway. - skip_may_block_check: async_caller || !callee_async, })?; + if let Some(old_do_not_suspend) = old_do_not_suspend { + store + .0 + .instance_state(callee_instance) + .concurrent_state() + .do_not_suspend = old_do_not_suspend; + } + let state = store.0.concurrent_state_mut()?; log::trace!("taking event for {:?}", guest_thread.task); @@ -3155,6 +3233,7 @@ impl Instance { // The callee hasn't returned yet, and the caller is calling via // a sync-lowered import, so we loop and keep waiting until the // callee returns. + store.0.switch_or_trap_if_may_not_suspend(caller_instance)?; } }; @@ -3581,18 +3660,12 @@ impl Instance { set: u32, payload: u32, ) -> Result { - if !self.options(store, options).async_ { - // The caller may only call `waitable-set.wait` from an async task - // (i.e. a task created via a call to an async export). - // Otherwise, we'll trap. - store.check_blocking()?; - } - let &CanonicalOptions { cancellable, instance: caller_instance, .. } = &self.id().get(store).component().env_component().options[options]; + let caller = self.runtime_instance(caller_instance); let rep = store .instance_state(self.runtime_instance(caller_instance)) .handle_table() @@ -3600,6 +3673,7 @@ impl Instance { self.waitable_check( store, + caller, cancellable, WaitableCheck::Wait, WaitableCheckParams { @@ -3623,13 +3697,15 @@ impl Instance { instance: caller_instance, .. } = &self.id().get(store).component().env_component().options[options]; + let caller = self.runtime_instance(caller_instance); let rep = store - .instance_state(self.runtime_instance(caller_instance)) + .instance_state(caller) .handle_table() .waitable_set_rep(set)?; self.waitable_check( store, + caller, cancellable, WaitableCheck::Poll, WaitableCheckParams { @@ -3724,50 +3800,75 @@ impl Instance { store: &mut StoreOpaque, runtime_instance: RuntimeComponentInstanceIndex, thread_idx: u32, - high_priority: bool, - allow_ready: bool, - ) -> Result<()> { + how: ResumeThread, + ) -> Result { let thread_id = GuestThread::from_instance(self.id().get_mut(store), runtime_instance, thread_idx)?; let state = store.concurrent_state_mut()?; let guest_thread = QualifiedThreadId::qualify(state, thread_id)?; let thread = state.get_mut(guest_thread.thread)?; + let switch = match how { + ResumeThread::Promote | ResumeThread::Resume => true, + ResumeThread::ResumeLater => false, + }; + + match (&how, &thread.state) { + // Promotion is a noop unless the thread is in a ready state. + (ResumeThread::Promote, GuestThreadState::Ready { .. }) => {} + (ResumeThread::Promote, _) => return Ok(false), + + // When resuming a thread it must be in a suspended state otherwise + // this operation is a trap. + ( + ResumeThread::Resume | ResumeThread::ResumeLater, + GuestThreadState::NotStartedExplicit(_) | GuestThreadState::Suspended(_), + ) => {} + (ResumeThread::Resume | ResumeThread::ResumeLater, _) => { + bail!(Trap::CannotResumeThread) + } + } match mem::replace(&mut thread.state, GuestThreadState::Running) { GuestThreadState::NotStartedExplicit(start_func) => { log::trace!("starting thread {guest_thread:?}"); - let guest_call = WorkItem::GuestCall( - runtime_instance, - GuestCall { + let guest_call = WorkItem::GuestCall { + instance: self.runtime_instance(runtime_instance), + call: GuestCall { thread: guest_thread, kind: GuestCallKind::StartExplicit(Box::new(move |store| { start_func(store, guest_thread) })), }, - ); + }; store .concurrent_state_mut()? - .push_work_item(guest_call, high_priority); + .push_work_item(guest_call, switch)?; } GuestThreadState::Suspended(fiber) => { log::trace!("resuming thread {thread_id:?} that was suspended"); - store - .concurrent_state_mut()? - .push_work_item(WorkItem::ResumeFiber(fiber), high_priority); + store.concurrent_state_mut()?.push_work_item( + WorkItem::ResumeFiber { + instance: self.runtime_instance(runtime_instance), + thread: guest_thread, + fiber, + }, + switch, + )?; } - GuestThreadState::Ready { fiber, cancellable } if allow_ready => { + GuestThreadState::Ready { fiber, cancellable } => { log::trace!("resuming thread {thread_id:?} that was ready"); thread.state = GuestThreadState::Ready { fiber, cancellable }; store .concurrent_state_mut()? - .promote_thread_work_item(guest_thread); + .promote_thread_work_item(guest_thread)?; } - other => { + other @ (GuestThreadState::NotStartedImplicit + | GuestThreadState::Running + | GuestThreadState::Completed) => { thread.state = other; - bail!(Trap::CannotResumeThread); } } - Ok(()) + Ok(true) } fn add_guest_thread_to_instance_table( @@ -3787,8 +3888,9 @@ impl Instance { Ok(guest_id) } - /// Helper function for the `thread.yield`, `thread.yield-to-suspended`, `thread.suspend`, - /// `thread.suspend-to`, and `thread.suspend-to-suspended` intrinsics. + /// Helper function for the `thread.yield`, thread.suspend`, + /// `thread.suspend-then-resume`, `thread.suspend-then-promote`, + /// `thread.yield-then-resume`, and `thread.yield-then-promote` intrinsics. pub(crate) fn suspension_intrinsic( self, store: &mut StoreOpaque, @@ -3797,58 +3899,42 @@ impl Instance { yielding: bool, to_thread: SuspensionTarget, ) -> Result { - let guest_thread = store.current_guest_thread()?; - if to_thread.is_none() { - let state = store.concurrent_state_mut()?; - if yielding { - // This is a `thread.yield` call - if !state.may_block(guest_thread.task)? { - // In a non-blocking context, a `thread.yield` may trigger - // other threads in the same component instance to run. - if !state.promote_instance_local_thread_work_item(caller) { - // No other threads are runnable, so just return - return Ok(WaitResult::Completed); - } - } - } else { - // The caller may only call `thread.suspend` from an async task - // (i.e. a task created via a call to an async export). - // Otherwise, we'll trap. - store.check_blocking()?; - } - } - // There could be a pending cancellation from a previous uncancellable wait if cancellable && store.take_pending_cancellation()? { return Ok(WaitResult::Cancelled); } - match to_thread { - SuspensionTarget::SomeSuspended(thread) => { - self.resume_thread(store, caller, thread, true, false)? + let check_suspend = match to_thread { + SuspensionTarget::Promote(thread) => { + !self.resume_thread(store, caller, thread, ResumeThread::Promote)? } - SuspensionTarget::Some(thread) => { - self.resume_thread(store, caller, thread, true, true)? + SuspensionTarget::Resume(thread) => { + if !self.resume_thread(store, caller, thread, ResumeThread::Resume)? { + bail_bug!("resumed thread should have been ready"); + } + false } - SuspensionTarget::None => { /* nothing to do */ } + SuspensionTarget::None => true, + }; + + if check_suspend && !store.switch_if_may_not_suspend(self.runtime_instance(caller))? { + return if yielding { + Ok(WaitResult::Completed) + } else { + Err(Trap::CannotBlockSyncTask.into()) + }; } + let guest_thread = store.current_guest_thread()?; + let reason = if yielding { SuspendReason::Yielding { thread: guest_thread, cancellable, - // Tell `StoreOpaque::suspend` it's okay to suspend here since - // we're handling a `thread.yield-to-suspended` call; otherwise it would - // panic if we called it in a non-blocking context. - skip_may_block_check: to_thread.is_some(), } } else { SuspendReason::ExplicitlySuspending { thread: guest_thread, - // Tell `StoreOpaque::suspend` it's okay to suspend here since - // we're handling a `thread.suspend-to(-suspended)` call; otherwise it would - // panic if we called it in a non-blocking context. - skip_may_block_check: to_thread.is_some(), } }; @@ -3865,6 +3951,7 @@ impl Instance { fn waitable_check( self, store: &mut StoreOpaque, + caller: RuntimeInstance, cancellable: bool, check: WaitableCheck, params: WaitableCheckParams, @@ -3886,8 +3973,11 @@ impl Instance { || (matches!(task.event, Some(Event::Cancelled)) && !cancellable)) && state.get_mut(set)?.ready.is_empty() { + store.switch_or_trap_if_may_not_suspend(caller)?; + if cancellable { - let old = state + let old = store + .concurrent_state_mut()? .get_mut(guest_thread.thread)? .wake_on_cancel .replace(set); @@ -3899,7 +3989,6 @@ impl Instance { store.suspend(SuspendReason::Waiting { set, thread: guest_thread, - skip_may_block_check: false, })?; } } @@ -3959,13 +4048,6 @@ impl Instance { async_: bool, task_id: u32, ) -> Result { - if !async_ { - // The caller may only sync call `subtask.cancel` from an async task - // (i.e. a task created via a call to an async export). Otherwise, - // we'll trap. - store.check_blocking()?; - } - let (rep, is_host) = store .instance_state(self.runtime_instance(caller_instance)) .handle_table() @@ -3977,7 +4059,7 @@ impl Instance { }; let concurrent_state = store.concurrent_state_mut()?; - log::trace!("subtask_cancel {waitable:?} (handle {task_id})"); + log::trace!("subtask_cancel {waitable:?} (handle {task_id}; async {async_})"); if !async_ { waitable.trap_if_in_waitable_set(concurrent_state)?; @@ -4034,7 +4116,7 @@ impl Instance { // `Event::Cancelled` if it was already cancelled), but that's // okay -- this should supersede the previous state. task.event = Some(Event::Cancelled); - let runtime_instance = task.instance.index; + let runtime_instance = task.instance; for thread in task.threads.clone() { let thread = QualifiedThreadId { task: guest_task, @@ -4044,17 +4126,21 @@ impl Instance { if let Some(set) = thread_mut.wake_on_cancel.take() { // The thread is in a cancellable wait, so wake it up: let item = match concurrent_state.get_mut(set)?.waiting.remove(&thread) { - Some(WaitMode::Fiber(fiber)) => WorkItem::ResumeFiber(fiber), - Some(WaitMode::Callback(instance)) => WorkItem::GuestCall( - runtime_instance, - GuestCall { + Some(WaitMode::Fiber(fiber)) => WorkItem::ResumeFiber { + instance: runtime_instance, + thread, + fiber, + }, + Some(WaitMode::Callback(instance)) => WorkItem::GuestCall { + instance: runtime_instance, + call: GuestCall { thread, kind: GuestCallKind::DeliverEvent { instance, set: None, }, }, - ), + }, None => bail_bug!("thread not present in wake_on_cancel set"), }; concurrent_state.push_high_priority(item); @@ -4063,13 +4149,6 @@ impl Instance { store.suspend(SuspendReason::Yielding { thread: caller, cancellable: false, - // We've already checked that for a sync version of - // this intrinsic we're allowed to block (start of - // the function here), and otherwise this is similar - // to `suspension_intrinsic` where we're doing a - // brief yield to deliver the event, so there's no - // need to check may-block again. - skip_may_block_check: true, })?; break; } else if let GuestThreadState::Ready { @@ -4078,13 +4157,11 @@ impl Instance { { // The thread is in a cancellable yield, so yield back // to it. - concurrent_state.promote_thread_work_item(thread); + concurrent_state.promote_thread_work_item(thread)?; let caller = store.current_guest_thread()?; store.suspend(SuspendReason::Yielding { thread: caller, cancellable: false, - // See the comment above for why this is `true` - skip_may_block_check: true, })?; break; } @@ -4111,8 +4188,8 @@ impl Instance { // Wait for this waitable to get signaled with its terminal status // from the completion callback enqueued by `first_poll`. Once - // that's done fall through to the sahred - store.wait_for_event(waitable)?; + // that's done fall through to the shared code. + store.wait_for_event(self.runtime_instance(caller_instance), waitable)?; // .. fall through to determine what event's in store for us. } @@ -4762,6 +4839,11 @@ pub struct GuestThread { instance_rep: Option, /// Scratch waitable set used to watch subtasks during synchronous calls. sync_call_set: TableId, + /// The old value of `do_not_suspend` prior to the sync-typed task for which + /// this thread was created, if relevant. + old_do_not_suspend: Option, + /// Whether this thread was explicitly or implicitly created. + explicit: bool, } impl GuestThread { @@ -4790,6 +4872,8 @@ impl GuestThread { state: GuestThreadState::NotStartedImplicit, instance_rep: None, sync_call_set, + old_do_not_suspend: None, + explicit: false, }) } @@ -4811,6 +4895,8 @@ impl GuestThread { state: GuestThreadState::NotStartedExplicit(start_func), instance_rep: None, sync_call_set, + old_do_not_suspend: None, + explicit: true, }) } } @@ -4893,9 +4979,12 @@ pub(crate) struct GuestTask { /// The state of the host future that represents an async task, which must /// be dropped before we can delete the task. host_future_state: HostFutureState, + /// Indicates whether this task was created for a call to an async-typed + /// export. + async_typed: bool, /// Indicates whether this task was created for a call to an async-lifted /// export. - async_function: bool, + async_lifted: bool, decremented_interesting_task_count: bool, } @@ -4941,7 +5030,8 @@ impl GuestTask { caller: Caller, callback: Option, instance: RuntimeInstance, - async_function: bool, + async_typed: bool, + async_lifted: bool, ) -> Result { let host_future_state = match &caller { Caller::Guest { .. } => HostFutureState::NotApplicable, @@ -4972,14 +5062,17 @@ impl GuestTask { exited: false, threads: HashSet::new(), host_future_state, - async_function, + async_typed, + async_lifted, decremented_interesting_task_count: false, })?; let new_thread = GuestThread::new_implicit(state, task)?; let thread = state.push(new_thread)?; state.get_mut(task)?.threads.insert(thread); state.interesting_tasks += 1; - Ok(QualifiedThreadId { task, thread }) + let thread = QualifiedThreadId { task, thread }; + log::trace!("new implicit thread {thread:?} for instance {instance:?}"); + Ok(thread) } } @@ -5126,17 +5219,21 @@ impl Waitable { assert!(wake_on_cancel.is_none() || wake_on_cancel == Some(set)); let item = match mode { - WaitMode::Fiber(fiber) => WorkItem::ResumeFiber(fiber), - WaitMode::Callback(instance) => WorkItem::GuestCall( - state.get_mut(thread.task)?.instance.index, - GuestCall { + WaitMode::Fiber(fiber) => WorkItem::ResumeFiber { + instance: state.get_mut(thread.task)?.instance, + thread, + fiber, + }, + WaitMode::Callback(instance) => WorkItem::GuestCall { + instance: state.get_mut(thread.task)?.instance, + call: GuestCall { thread, kind: GuestCallKind::DeliverEvent { instance, set: Some(set), }, }, - ), + }, }; state.push_high_priority(item); } @@ -5225,8 +5322,11 @@ pub struct ConcurrentInstanceState { backpressure: u16, /// Whether this instance can be entered do_not_enter: bool, + /// Whether this instance may suspend (i.e. whether this instance is + /// currently running a sync-typed function). + do_not_suspend: bool, /// Pending calls for this instance which require `Self::backpressure` to be - /// `true` and/or `Self::do_not_enter` to be false before they can proceed. + /// zero and/or `Self::do_not_enter` to be false before they can proceed. pending: BTreeMap, } @@ -5307,6 +5407,14 @@ pub struct ConcurrentState { futures: AlwaysMut>>, /// The table of waitables, waitable sets, etc. table: AlwaysMut, + /// The next item to switch to if any. + /// + /// This takes precedence over items in the `high_priority` queue below and + /// should be used in cases such as subtask calls and thread resume/promote + /// operations where we must switch to a specific thread at the next turn of + /// the event loop regardless of what happens to be present in the + /// `high_priority` queue. + switch_item: Option, /// The "high priority" work queue for this store's event loop. high_priority: Vec, /// The "low priority" work queue for this store's event loop. @@ -5361,6 +5469,9 @@ pub struct ConcurrentState { /// /// Used in the implementation of `Accessor::poll_ready_for_concurrent_call`. ready_for_concurrent_call_waker: Option, + + /// Whether the `StoreContextMut::poll_until` event loop is running. + event_loop_running: bool, } impl Default for ConcurrentState { @@ -5369,6 +5480,7 @@ impl Default for ConcurrentState { unforced_current_thread: CurrentThread::None, table: AlwaysMut::new(ResourceTable::new()), futures: AlwaysMut::new(Some(FuturesUnordered::new())), + switch_item: None, high_priority: Vec::new(), low_priority: VecDeque::new(), suspend_reason: None, @@ -5378,6 +5490,7 @@ impl Default for ConcurrentState { interesting_tasks: 0, interesting_tasks_empty_waker: None, ready_for_concurrent_call_waker: None, + event_loop_running: false, } } } @@ -5425,7 +5538,7 @@ impl ConcurrentState { } let mut handle_item = |item| match item { - WorkItem::ResumeFiber(fiber) => { + WorkItem::ResumeFiber { fiber, .. } => { fibers.push(fiber); } WorkItem::PushFuture(future) => { @@ -5435,8 +5548,9 @@ impl ConcurrentState { .unwrap() .push(future.into_inner()); } - WorkItem::ResumeThread(..) | WorkItem::GuestCall(..) | WorkItem::WorkerFunction(..) => { - } + WorkItem::ResumeThread { .. } + | WorkItem::GuestCall { .. } + | WorkItem::WorkerFunction(_) => {} }; for item in mem::take(&mut self.high_priority) { @@ -5463,6 +5577,7 @@ impl ConcurrentState { worker, high_priority, low_priority, + switch_item, // TODO(cm-gc): This field contains `ValRaw`s, but they are never GC // references because the component model doesn't support GC yet. We @@ -5477,6 +5592,7 @@ impl ConcurrentState { interesting_tasks: _, interesting_tasks_empty_waker: _, ready_for_concurrent_call_waker: _, + event_loop_running: _, } = self; for entry in table.get_mut().iter_mut() { @@ -5500,17 +5616,21 @@ impl ConcurrentState { } let mut handle_item = |item: &mut WorkItem| match item { - WorkItem::ResumeFiber(fiber) => { + WorkItem::ResumeFiber { fiber, .. } => { fiber.trace_gc_roots(modules, unwind, gc_roots_list); } WorkItem::PushFuture(_future) => { // TODO(cm-gc): once futures can contain GC roots, we will need // to trace them. } - WorkItem::ResumeThread(..) | WorkItem::GuestCall(..) | WorkItem::WorkerFunction(..) => { - } + WorkItem::ResumeThread { .. } + | WorkItem::GuestCall { .. } + | WorkItem::WorkerFunction(_) => {} }; + if let Some(item) = switch_item { + handle_item(item); + } for item in high_priority { handle_item(item); } @@ -5569,72 +5689,96 @@ impl ConcurrentState { self.high_priority.push(item); } + fn set_switch_item(&mut self, item: WorkItem) -> Result<()> { + log::trace!("set switch item: {item:?}"); + + if self.switch_item.is_some() { + bail_bug!("switch item already set"); + } + + self.switch_item = Some(item); + + Ok(()) + } + fn push_low_priority(&mut self, item: WorkItem) { log::trace!("push low priority: {item:?}"); self.low_priority.push_front(item); } - fn push_work_item(&mut self, item: WorkItem, high_priority: bool) { - if high_priority { - self.push_high_priority(item); + fn push_work_item(&mut self, item: WorkItem, switch: bool) -> Result<()> { + if switch { + self.set_switch_item(item)?; } else { self.push_low_priority(item); } + + Ok(()) } fn promote_instance_local_thread_work_item( &mut self, - current_instance: RuntimeComponentInstanceIndex, - ) -> bool { - self.promote_work_items_matching(|item: &WorkItem| match item { - WorkItem::ResumeThread(instance, _) | WorkItem::GuestCall(instance, _) => { - *instance == current_instance - } - _ => false, + current_instance: RuntimeInstance, + ) -> Result { + log::trace!("promote thread work items for {current_instance:?}"); + + self.promote_work_item_matching(|item: &WorkItem| { + let result = match item { + WorkItem::ResumeThread { instance, .. } + | WorkItem::ResumeFiber { instance, .. } + | WorkItem::GuestCall { instance, .. } => *instance == current_instance, + _ => false, + }; + + log::trace!("candidate {item:?}: {result}"); + result }) } - fn promote_thread_work_item(&mut self, thread: QualifiedThreadId) -> bool { - self.promote_work_items_matching(|item: &WorkItem| match item { - WorkItem::ResumeThread(_, t) | WorkItem::GuestCall(_, GuestCall { thread: t, .. }) => { - *t == thread + fn promote_thread_work_item(&mut self, thread: QualifiedThreadId) -> Result { + self.promote_work_item_matching(|item: &WorkItem| match item { + WorkItem::ResumeThread { + thread: item_thread, + .. } + | WorkItem::GuestCall { + call: + GuestCall { + thread: item_thread, + .. + }, + .. + } => *item_thread == thread, _ => false, }) } - fn promote_work_items_matching(&mut self, mut predicate: F) -> bool + fn promote_work_item_matching(&mut self, mut predicate: F) -> Result where F: FnMut(&WorkItem) -> bool, { - // If there's a high-priority work item to resume the current guest thread, - // we don't need to promote anything, but we return true to indicate that - // work is pending for the current instance. - if self.high_priority.iter().any(&mut predicate) { - true - } - // Otherwise, look for a low-priority work item that matches the current - // instance and promote it to high-priority. - else if let Some(idx) = self.low_priority.iter().position(&mut predicate) { - let item = self.low_priority.remove(idx).unwrap(); - self.push_high_priority(item); - true - } else { - false + for item in mem::take(&mut self.high_priority) { + if self.switch_item.is_none() && predicate(&item) { + self.set_switch_item(item)?; + } else { + self.push_high_priority(item); + } } - } - fn check_blocking_for(&mut self, task: TableId) -> Result<()> { - if self.may_block(task)? { - Ok(()) - } else { - Err(Trap::CannotBlockSyncTask.into()) + if self.switch_item.is_none() { + // Note the use of `.rev()` here to preserve ordering given that + // items are popped from the back of `self.low_priority` by + // `poll_until` and pushed to the front by `push_low_priority`. + for item in mem::take(&mut self.low_priority).into_iter().rev() { + if self.switch_item.is_none() && predicate(&item) { + self.set_switch_item(item)?; + } else { + self.push_low_priority(item); + } + } } - } - fn may_block(&mut self, task: TableId) -> Result { - let task = self.get_mut(task)?; - Ok(task.async_function || task.returned_or_cancelled()) + Ok(self.switch_item.is_some()) } /// Used by `ResourceTables` to acquire the current `CallContext` for the @@ -5822,7 +5966,7 @@ impl TaskId { /// for lowering the parameters and lifting the result. /// /// To enqueue the returned `PreparedCall` in the `ComponentInstance`'s event -/// loop, use `queue_call`. +/// loop, use `stage_call`. pub(crate) fn prepare_call( mut store: StoreContextMut, handle: Func, @@ -5842,7 +5986,8 @@ pub(crate) fn prepare_call( let instance = handle.instance().id().get(store.0); let options = &instance.component().env_component().options[options]; let ty = &instance.component().types()[ty]; - let async_function = ty.async_; + let async_typed = ty.async_; + let async_lifted = raw_options.async_; let task_return_type = ty.results; let component_instance = raw_options.instance; let callback = options.callback.map(|i| instance.runtime_callback(i)); @@ -5887,10 +6032,11 @@ pub(crate) fn prepare_call( }) as CallbackFn }), instance, - async_function, + async_typed, + async_lifted, )?; - if !store.0.may_enter(instance)? { + if !store.0.may_enter() { bail!(Trap::CannotEnterComponent); } @@ -5904,14 +6050,14 @@ pub(crate) fn prepare_call( }) } -pub(crate) struct QueuedCall { +pub(crate) struct StagedCall { store: StoreId, task: TableId, rx: oneshot::Receiver, _marker: PhantomData R>, } -impl QueuedCall { +impl StagedCall { /// Queue a call previously prepared using `prepare_call` to be run as part of /// the associated `ComponentInstance`'s event loop. /// @@ -5921,7 +6067,7 @@ impl QueuedCall { pub(crate) fn new( mut store: StoreContextMut, prepared: PreparedCall, - ) -> Result> { + ) -> Result> { let PreparedCall { handle, thread, @@ -5930,9 +6076,9 @@ impl QueuedCall { .. } = prepared; - queue_call0(store.as_context_mut(), handle, thread, param_count)?; + stage_call0(store.as_context_mut(), handle, thread, param_count)?; - Ok(QueuedCall { + Ok(StagedCall { store: store.0.id(), task: thread.task, rx, @@ -5945,7 +6091,7 @@ impl QueuedCall { } } -impl Future for QueuedCall +impl Future for StagedCall where R: 'static, { @@ -5965,7 +6111,7 @@ where /// Queue a call previously prepared using `prepare_call` to be run as part of /// the associated `ComponentInstance`'s event loop. -fn queue_call0( +fn stage_call0( store: StoreContextMut, handle: Func, guest_thread: QualifiedThreadId, @@ -5990,7 +6136,7 @@ fn queue_call0( // (with signatures appropriate for this call) and will remain valid as // long as this instance is valid. unsafe { - instance.queue_call( + instance.stage_call( store, guest_thread, SendSyncPtr::new(callee), diff --git a/crates/wasmtime/src/runtime/component/concurrent/func.rs b/crates/wasmtime/src/runtime/component/concurrent/func.rs index bedebac33ad5..c1f823cd3d35 100644 --- a/crates/wasmtime/src/runtime/component/concurrent/func.rs +++ b/crates/wasmtime/src/runtime/component/concurrent/func.rs @@ -13,7 +13,7 @@ use wasmtime_environ::component::{InterfaceType, MAX_FLAT_PARAMS, MAX_FLAT_RESUL /// Returned from [`Func::start_call_concurrent`] to represent a /// pending-but-not-yet-resolved call into wasm. pub struct FuncCallConcurrent<'a, T> { - call: concurrent::QueuedCall>, + call: concurrent::StagedCall>, results: &'a mut [Val], _marker: marker::PhantomData, } @@ -136,7 +136,7 @@ impl Func { ) -> Result> { self.check_params_results(store.as_context_mut(), params, results)?; let prepared = self.prepare_call_dynamic(store.as_context_mut(), params.to_vec())?; - let call = concurrent::QueuedCall::new(store.as_context_mut(), prepared)?; + let call = concurrent::StagedCall::new(store.as_context_mut(), prepared)?; Ok(FuncCallConcurrent { call, results, @@ -213,7 +213,7 @@ impl FuncCallConcurrent<'_, T> { /// Returned from [`TypedFunc::start_call_concurrent`] to represent a /// pending-but-not-yet-resolved call into wasm. pub struct TypedFuncCallConcurrent { - call: concurrent::QueuedCall, + call: concurrent::StagedCall, _marker: marker::PhantomData, } @@ -271,7 +271,7 @@ where task: prepared.task_id(), }; - let result = concurrent::QueuedCall::new(wrapper.store.as_context_mut(), prepared)?; + let result = concurrent::StagedCall::new(wrapper.store.as_context_mut(), prepared)?; wrapper .store .as_context_mut() @@ -381,7 +381,7 @@ where let prepared = self.prepare_call(store.as_context_mut(), false, move |cx, ty, dst| { Self::lower_args(cx, ty, dst, ¶ms) })?; - let call = concurrent::QueuedCall::new(store, prepared)?; + let call = concurrent::StagedCall::new(store, prepared)?; Ok(TypedFuncCallConcurrent { call, _marker: marker::PhantomData, diff --git a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs index 4daa4392766a..5c7744a383c1 100644 --- a/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs +++ b/crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs @@ -3502,13 +3502,6 @@ impl Instance { ) -> Result { let count = ItemCount::new(count)?; - if !self.options(store.0, options).async_ { - // The caller may only sync call `{stream,future}.write` from an - // async task (i.e. a task created via a call to an async export). - // Otherwise, we'll trap. - store.0.check_blocking()?; - } - let address = usize::try_from(address)?; self.check_bounds(store.0, options, ty, address, count.as_usize())?; let (rep, state) = self.id().get_mut(store.0).get_mut_by_index(ty, handle)?; @@ -3723,7 +3716,7 @@ impl Instance { }; if result == ReturnCode::Blocked && !self.options(store.0, options).async_ { - result = self.wait_for_write(store.0, transmit_handle)?; + result = self.wait_for_write(store.0, caller, transmit_handle)?; } if result != ReturnCode::Blocked { @@ -3754,13 +3747,6 @@ impl Instance { ) -> Result { let count = ItemCount::new(count)?; - if !self.options(store.0, options).async_ { - // The caller may only sync call `{stream,future}.read` from an - // async task (i.e. a task created via a call to an async export). - // Otherwise, we'll trap. - store.0.check_blocking()?; - } - let address = usize::try_from(address)?; self.check_bounds(store.0, options, ty, address, count.as_usize())?; let (rep, state) = self.id().get_mut(store.0).get_mut_by_index(ty, handle)?; @@ -3955,7 +3941,7 @@ impl Instance { }; if result == ReturnCode::Blocked && !self.options(store.0, options).async_ { - result = self.wait_for_read(store.0, transmit_handle)?; + result = self.wait_for_read(store.0, caller_instance, transmit_handle)?; } if result != ReturnCode::Blocked { @@ -3978,10 +3964,11 @@ impl Instance { fn wait_for_write( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, handle: TableId, ) -> Result { let waitable = Waitable::Transmit(handle); - store.wait_for_event(waitable)?; + store.wait_for_event(self.runtime_instance(caller), waitable)?; let event = waitable.take_event(store.concurrent_state_mut()?)?; if let Some(event @ (Event::StreamWrite { code, .. } | Event::FutureWrite { code, .. })) = event @@ -3997,6 +3984,7 @@ impl Instance { fn cancel_write( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, transmit_id: TableId, async_: bool, ) -> Result { @@ -4043,7 +4031,7 @@ impl Instance { .concurrent_state_mut()? .get_mut(transmit_id)? .write_handle; - self.wait_for_write(store, handle)? + self.wait_for_write(store, caller, handle)? } } else { ReturnCode::Cancelled(ItemCount::ZERO) @@ -4069,10 +4057,11 @@ impl Instance { fn wait_for_read( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, handle: TableId, ) -> Result { let waitable = Waitable::Transmit(handle); - store.wait_for_event(waitable)?; + store.wait_for_event(self.runtime_instance(caller), waitable)?; let event = waitable.take_event(store.concurrent_state_mut()?)?; if let Some(event @ (Event::StreamRead { code, .. } | Event::FutureRead { code, .. })) = event @@ -4088,6 +4077,7 @@ impl Instance { fn cancel_read( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, transmit_id: TableId, async_: bool, ) -> Result { @@ -4135,7 +4125,7 @@ impl Instance { .concurrent_state_mut()? .get_mut(transmit_id)? .read_handle; - self.wait_for_read(store, handle)? + self.wait_for_read(store, caller, handle)? } } else { ReturnCode::Cancelled(ItemCount::ZERO) @@ -4164,17 +4154,11 @@ impl Instance { fn guest_cancel_write( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, ty: TransmitIndex, async_: bool, writer: u32, ) -> Result { - if !async_ { - // The caller may only sync call `{stream,future}.cancel-write` from - // an async task (i.e. a task created via a call to an async - // export). Otherwise, we'll trap. - store.check_blocking()?; - } - let (rep, state) = get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, writer)?; let id = TableId::::new(rep); @@ -4189,7 +4173,7 @@ impl Instance { TransmitLocalState::Busy => {} } let transmit_id = store.concurrent_state_mut()?.get_mut(id)?.state; - let code = self.cancel_write(store, transmit_id, async_)?; + let code = self.cancel_write(store, caller, transmit_id, async_)?; if !matches!(code, ReturnCode::Blocked) { let state = get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, writer)? @@ -4205,17 +4189,11 @@ impl Instance { fn guest_cancel_read( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, ty: TransmitIndex, async_: bool, reader: u32, ) -> Result { - if !async_ { - // The caller may only sync call `{stream,future}.cancel-read` from - // an async task (i.e. a task created via a call to an async - // export). Otherwise, we'll trap. - store.check_blocking()?; - } - let (rep, state) = get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, reader)?; let id = TableId::::new(rep); @@ -4230,7 +4208,7 @@ impl Instance { TransmitLocalState::Busy => {} } let transmit_id = store.concurrent_state_mut()?.get_mut(id)?.state; - let code = self.cancel_read(store, transmit_id, async_)?; + let code = self.cancel_read(store, caller, transmit_id, async_)?; if !matches!(code, ReturnCode::Blocked) { let state = get_mut_by_index_from(self.id().get_mut(store).table_for_transmit(ty), ty, reader)? @@ -4352,11 +4330,12 @@ impl Instance { pub(crate) fn future_cancel_read( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, ty: TypeFutureTableIndex, async_: bool, reader: u32, ) -> Result { - self.guest_cancel_read(store, TransmitIndex::Future(ty), async_, reader) + self.guest_cancel_read(store, caller, TransmitIndex::Future(ty), async_, reader) .map(|v| v.encode()) } @@ -4364,11 +4343,12 @@ impl Instance { pub(crate) fn future_cancel_write( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, ty: TypeFutureTableIndex, async_: bool, writer: u32, ) -> Result { - self.guest_cancel_write(store, TransmitIndex::Future(ty), async_, writer) + self.guest_cancel_write(store, caller, TransmitIndex::Future(ty), async_, writer) .map(|v| v.encode()) } @@ -4376,11 +4356,12 @@ impl Instance { pub(crate) fn stream_cancel_read( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, ty: TypeStreamTableIndex, async_: bool, reader: u32, ) -> Result { - self.guest_cancel_read(store, TransmitIndex::Stream(ty), async_, reader) + self.guest_cancel_read(store, caller, TransmitIndex::Stream(ty), async_, reader) .map(|v| v.encode()) } @@ -4388,11 +4369,12 @@ impl Instance { pub(crate) fn stream_cancel_write( self, store: &mut StoreOpaque, + caller: RuntimeComponentInstanceIndex, ty: TypeStreamTableIndex, async_: bool, writer: u32, ) -> Result { - self.guest_cancel_write(store, TransmitIndex::Stream(ty), async_, writer) + self.guest_cancel_write(store, caller, TransmitIndex::Stream(ty), async_, writer) .map(|v| v.encode()) } diff --git a/crates/wasmtime/src/runtime/component/concurrent_disabled.rs b/crates/wasmtime/src/runtime/component/concurrent_disabled.rs index 2d3af7bf5197..98cc17c21251 100644 --- a/crates/wasmtime/src/runtime/component/concurrent_disabled.rs +++ b/crates/wasmtime/src/runtime/component/concurrent_disabled.rs @@ -170,10 +170,6 @@ impl StoreOpaque { Ok(self.exit_call_not_concurrent()) } - pub(crate) fn check_blocking(&mut self) -> crate::Result<()> { - Ok(()) - } - pub(crate) fn may_enter(&mut self, _instance: RuntimeInstance) -> Result { Ok(!self.trapped()) } diff --git a/crates/wasmtime/src/runtime/component/func.rs b/crates/wasmtime/src/runtime/component/func.rs index 2527a17919c7..1c7d4f8205f6 100644 --- a/crates/wasmtime/src/runtime/component/func.rs +++ b/crates/wasmtime/src/runtime/component/func.rs @@ -461,7 +461,7 @@ impl Func { let instance = self.instance.runtime_instance(raw_options.instance); let async_ = raw_options.async_; - if !store.0.may_enter(instance)? { + if !store.0.may_enter() { bail!(crate::Trap::CannotEnterComponent); } diff --git a/crates/wasmtime/src/runtime/component/func/host.rs b/crates/wasmtime/src/runtime/component/func/host.rs index 3f38679e840a..f5064e576a71 100644 --- a/crates/wasmtime/src/runtime/component/func/host.rs +++ b/crates/wasmtime/src/runtime/component/func/host.rs @@ -283,10 +283,6 @@ where T: 'static, R: Send + Sync + 'static, { - /// Whether or not this is `async` function from the perspective of the - /// component model. - const ASYNC: bool; - /// Performs a type-check to ensure that this host function can be imported /// with the provided signature that a component is using. fn typecheck(ty: TypeFuncIndex, types: &InstanceType<'_>) -> Result<()>; @@ -363,13 +359,6 @@ where let vminstance = instance.id().get(store.0); let async_ = vminstance.component().env_component().options[options].async_; - // If this is a synchronous-lower of a host-async function, then the - // guest is blocking. Test, in the context of the guest task, if that's - // allowed. - if !async_ && Self::ASYNC { - store.0.check_blocking()?; - } - if async_ { #[cfg(feature = "component-model-async")] { @@ -631,8 +620,6 @@ where P: ComponentNamedList + Lift + 'static, R: ComponentNamedList + Lower + 'static, { - const ASYNC: bool = ASYNC; - fn typecheck(ty: TypeFuncIndex, types: &InstanceType<'_>) -> Result<()> { let ty = &types.types[ty]; typecheck_async(ASYNC, ty.async_)?; @@ -709,8 +696,6 @@ where T: 'static, F: Fn(StoreContextMut<'_, T>, ComponentFunc, Vec, usize) -> HostResult>, { - const ASYNC: bool = ASYNC; - /// This function performs dynamic type checks on its parameters and /// results and subsequently does not need to perform up-front type /// checks. However, we _do_ verify async-ness here. diff --git a/crates/wasmtime/src/runtime/component/instance.rs b/crates/wasmtime/src/runtime/component/instance.rs index 95468d3c9d0b..95a6ada5201b 100644 --- a/crates/wasmtime/src/runtime/component/instance.rs +++ b/crates/wasmtime/src/runtime/component/instance.rs @@ -10,13 +10,17 @@ use crate::instance::OwnedImports; use crate::linker::DefinitionType; use crate::prelude::*; use crate::runtime::vm::component::{ComponentInstance, TypedResource, TypedResourceIndex}; -use crate::runtime::vm::{self, VMFuncRef}; +use crate::runtime::vm::{self, VMFuncRef, VMStore}; +#[cfg(feature = "component-model-async")] +use crate::store::StoreToken; use crate::store::{AsStoreOpaque, Asyncness, StoreOpaque}; use crate::{AsContext, AsContextMut, Engine, Module, StoreContextMut}; use alloc::sync::Arc; use core::marker; use core::pin::Pin; use core::ptr::NonNull; +#[cfg(feature = "component-model-async")] +use futures::channel::oneshot; use wasmtime_environ::{EngineOrModuleTypeIndex, component::*}; use wasmtime_environ::{EntityIndex, EntityType, PrimaryMap}; @@ -611,9 +615,6 @@ pub(crate) fn lookup_vmdef( // within that store, so it's safe to create a `Func`. vm::Export::Function(unsafe { crate::Func::from_vm_func_ref(store.id(), funcref) }) } - CoreDef::TaskMayBlock => vm::Export::Global(crate::Global::from_task_may_block( - StoreComponentInstanceId::new(store.id(), id), - )), } } @@ -841,16 +842,74 @@ impl<'a> Instantiator<'a> { // already checked for asyncness and are running on a fiber // if required. - let i = unsafe { - crate::Instance::new_started(store, module, imports.as_ref(), asyncness) + let mut instance = { + let (mut limiter, store) = store.0.resource_limiter_and_store_opaque(); + unsafe { + crate::Instance::new_raw( + store, + limiter.as_mut(), + module, + imports.as_ref(), + ) .await? + } }; + if instance.id.get_mut(store.0).needs_startup() { + if asyncness == Asyncness::No { + instance.start_raw(store)?; + } else { + #[cfg(feature = "async")] + { + #[cfg(feature = "component-model-async")] + { + if store.0.concurrency_support() { + // With concurrency support enabled, we must + // run the start function inside the store's + // event loop in case it calls async + // functions or intrinsics, creates and + // resumes threads, etc. + let (tx, rx) = oneshot::channel(); + let token = StoreToken::new(store.as_context_mut()); + store.0.queue_task(move |store| { + _ = tx.send( + instance + .start_raw(&mut token.as_context_mut(store)) + .map(|()| instance), + ); + Ok(()) + })?; + instance = store + .as_context_mut() + .run_concurrent_trap_on_idle(async |_| { + rx.await.map_err(|_| { + format_err!("oneshot channel canceled") + }) + }) + .await???; + } else { + store.on_fiber(|store| instance.start_raw(store)).await??; + } + } + #[cfg(not(feature = "component-model-async"))] + { + _ = &mut instance; + store.on_fiber(|store| instance.start_raw(store)).await??; + } + } + #[cfg(not(feature = "async"))] + { + _ = &mut instance; + unreachable!(); + } + } + } + if exit { store.0.exit_guest_sync_call()?; } - self.instance_mut(store.0).push_instance_id(i.id())?; + self.instance_mut(store.0).push_instance_id(instance.id())?; } GlobalInitializer::LowerImport { import, index } => { diff --git a/crates/wasmtime/src/runtime/component/resources/any.rs b/crates/wasmtime/src/runtime/component/resources/any.rs index ecb69e4822e8..0cfb1220f1ed 100644 --- a/crates/wasmtime/src/runtime/component/resources/any.rs +++ b/crates/wasmtime/src/runtime/component/resources/any.rs @@ -201,16 +201,8 @@ impl ResourceAny { _ => unreachable!(), }; - // Implement the reentrance check required by the canonical ABI. Note - // that this happens whether or not a destructor is present. - // - // Note that this should be safe because the raw pointer access in - // `flags` is valid due to `store` being the owner of the flags and - // flags are never destroyed within the store. - if let Some(instance) = slot.instance { - if !store.0.may_enter(instance)? { - bail!(Trap::CannotEnterComponent); - } + if slot.instance.is_some() && !store.0.may_enter() { + bail!(Trap::CannotEnterComponent); } let dtor = match slot.dtor { diff --git a/crates/wasmtime/src/runtime/externals/global.rs b/crates/wasmtime/src/runtime/externals/global.rs index ede512da35d3..f3bb40ffd734 100644 --- a/crates/wasmtime/src/runtime/externals/global.rs +++ b/crates/wasmtime/src/runtime/externals/global.rs @@ -349,17 +349,6 @@ impl Global { } } - #[cfg(feature = "component-model")] - pub(crate) fn from_task_may_block( - instance: crate::component::store::StoreComponentInstanceId, - ) -> Global { - Global { - store: instance.store_id(), - instance: instance.instance().as_u32(), - kind: VMGlobalKind::TaskMayBlock, - } - } - pub(crate) fn wasmtime_ty<'a>(&self, store: &'a StoreOpaque) -> &'a wasmtime_environ::Global { self.store.assert_belongs_to(store.id()); match self.kind { @@ -371,7 +360,7 @@ impl Global { } VMGlobalKind::Host(index) => unsafe { &store.host_globals()[index].get().as_ref().ty }, #[cfg(feature = "component-model")] - VMGlobalKind::ComponentFlags(_) | VMGlobalKind::TaskMayBlock => { + VMGlobalKind::ComponentFlags(_) => { const TY: wasmtime_environ::Global = wasmtime_environ::Global { mutability: true, wasm_ty: wasmtime_environ::WasmValType::I32, @@ -389,7 +378,7 @@ impl Global { } VMGlobalKind::Host(_) => None, #[cfg(feature = "component-model")] - VMGlobalKind::ComponentFlags(_) | VMGlobalKind::TaskMayBlock => { + VMGlobalKind::ComponentFlags(_) => { let instance = crate::component::ComponentInstanceId::from_u32(self.instance); Some( VMOpaqueContext::from_vmcomponent(store.component_instance(instance).vmctx()) @@ -421,8 +410,6 @@ impl Global { VMGlobalKind::ComponentFlags(idx) => { u64::from(self.instance) << 32 | u64::from(idx.as_u32()) } - #[cfg(feature = "component-model")] - VMGlobalKind::TaskMayBlock => u64::from(self.instance) << 32 | u64::from(u32::MAX), } } @@ -454,12 +441,6 @@ impl Global { .instance_flags(index) .as_raw() } - #[cfg(feature = "component-model")] - VMGlobalKind::TaskMayBlock => store - .component_instance(crate::component::ComponentInstanceId::from_u32( - self.instance, - )) - .task_may_block(), } } } diff --git a/crates/wasmtime/src/runtime/fiber.rs b/crates/wasmtime/src/runtime/fiber.rs index 50f086f7d4c8..89c5d49b8258 100644 --- a/crates/wasmtime/src/runtime/fiber.rs +++ b/crates/wasmtime/src/runtime/fiber.rs @@ -888,19 +888,6 @@ where }) } -/// Safe wrapper around [`make_fiber_unchecked`] which requires that `S` is -/// `Send`. -#[cfg(feature = "component-model-async")] -pub(crate) fn make_fiber<'a, S>( - store: &mut S, - fun: impl FnOnce(&mut S) -> Result<()> + Send + Sync + 'a, -) -> Result> -where - S: AsStoreOpaque + Send + ?Sized + 'a, -{ - unsafe { make_fiber_unchecked(store, fun) } -} - /// Run the specified function on a newly-created fiber and `.await` its /// completion. pub(crate) async fn on_fiber( diff --git a/crates/wasmtime/src/runtime/instance.rs b/crates/wasmtime/src/runtime/instance.rs index 53fd54faed9b..b46bfa04c68d 100644 --- a/crates/wasmtime/src/runtime/instance.rs +++ b/crates/wasmtime/src/runtime/instance.rs @@ -297,7 +297,7 @@ impl Instance { /// This method is unsafe because it does not type-check the `imports` /// provided. The `imports` provided must be suitable for the module /// provided as well. - async unsafe fn new_raw( + pub(crate) async unsafe fn new_raw( store: &mut StoreOpaque, mut limiter: Option<&mut StoreResourceLimiter<'_>>, module: &Module, @@ -349,7 +349,7 @@ impl Instance { } } - fn start_raw(&self, store: &mut StoreContextMut<'_, T>) -> Result<()> { + pub(crate) fn start_raw(&self, store: &mut StoreContextMut<'_, T>) -> Result<()> { // If a start function is present, invoke it. Make sure we use all the // trap-handling configuration in `store` as well. let store_id = store.0.id(); diff --git a/crates/wasmtime/src/runtime/vm/component.rs b/crates/wasmtime/src/runtime/vm/component.rs index bafaada3d714..e1599ae34f72 100644 --- a/crates/wasmtime/src/runtime/vm/component.rs +++ b/crates/wasmtime/src/runtime/vm/component.rs @@ -961,20 +961,6 @@ impl ComponentInstance { ) } } - - pub(crate) fn task_may_block(&self) -> NonNull { - unsafe { self.vmctx_plus_offset_raw::(self.offsets.task_may_block()) } - } - - #[cfg(feature = "component-model-async")] - pub(crate) fn get_task_may_block(&self) -> bool { - unsafe { *self.task_may_block().as_ref().as_i32() != 0 } - } - - #[cfg(feature = "component-model-async")] - pub(crate) fn set_task_may_block(self: Pin<&mut Self>, val: bool) { - unsafe { *self.task_may_block().as_mut().as_i32_mut() = if val { 1 } else { 0 } } - } } // SAFETY: `layout` should describe this accurately and `OwnedVMContext` is the diff --git a/crates/wasmtime/src/runtime/vm/component/libcalls.rs b/crates/wasmtime/src/runtime/vm/component/libcalls.rs index 11ec01108492..18b881a2e27c 100644 --- a/crates/wasmtime/src/runtime/vm/component/libcalls.rs +++ b/crates/wasmtime/src/runtime/vm/component/libcalls.rs @@ -1,11 +1,13 @@ //! Implementation of string transcoding required by the component model. +#[cfg(feature = "component-model-async")] +use crate::bail_bug; use crate::component::Instance; #[cfg(feature = "component-model-async")] use crate::component::concurrent::WaitResult; use crate::prelude::*; #[cfg(feature = "component-model-async")] -use crate::runtime::component::concurrent::{ResourcePair, SuspensionTarget}; +use crate::runtime::component::concurrent::{ResourcePair, ResumeThread, SuspensionTarget}; use crate::runtime::vm::component::{ComponentInstance, VMComponentContext}; use crate::runtime::vm::{HostResultHasUnwindSentinel, VMStore, VmSafe}; use core::cell::Cell; @@ -1017,13 +1019,14 @@ fn future_read( fn future_cancel_write( store: &mut dyn VMStore, instance: Instance, - _caller_instance: u32, + caller_instance: u32, ty: u32, async_: u8, writer: u32, ) -> Result { instance.future_cancel_write( store, + RuntimeComponentInstanceIndex::from_u32(caller_instance), TypeFutureTableIndex::from_u32(ty), async_ != 0, writer, @@ -1034,13 +1037,14 @@ fn future_cancel_write( fn future_cancel_read( store: &mut dyn VMStore, instance: Instance, - _caller_instance: u32, + caller_instance: u32, ty: u32, async_: u8, reader: u32, ) -> Result { instance.future_cancel_read( store, + RuntimeComponentInstanceIndex::from_u32(caller_instance), TypeFutureTableIndex::from_u32(ty), async_ != 0, reader, @@ -1131,13 +1135,14 @@ fn stream_read( fn stream_cancel_write( store: &mut dyn VMStore, instance: Instance, - _caller_instance: u32, + caller_instance: u32, ty: u32, async_: u8, writer: u32, ) -> Result { instance.stream_cancel_write( store, + RuntimeComponentInstanceIndex::from_u32(caller_instance), TypeStreamTableIndex::from_u32(ty), async_ != 0, writer, @@ -1148,13 +1153,14 @@ fn stream_cancel_write( fn stream_cancel_read( store: &mut dyn VMStore, instance: Instance, - _caller_instance: u32, + caller_instance: u32, ty: u32, async_: u8, reader: u32, ) -> Result { instance.stream_cancel_read( store, + RuntimeComponentInstanceIndex::from_u32(caller_instance), TypeStreamTableIndex::from_u32(ty), async_ != 0, reader, @@ -1324,13 +1330,16 @@ fn thread_resume_later( caller_instance: u32, thread_idx: u32, ) -> Result<()> { - instance.resume_thread( + if !instance.resume_thread( store, RuntimeComponentInstanceIndex::from_u32(caller_instance), thread_idx, - false, - false, - ) + ResumeThread::ResumeLater, + )? { + bail_bug!("resumed thread should have been ready"); + } + + Ok(()) } #[cfg(feature = "component-model-async")] @@ -1383,7 +1392,7 @@ fn thread_suspend_then_resume( RuntimeComponentInstanceIndex::from_u32(caller), cancellable != 0, false, - SuspensionTarget::SomeSuspended(thread_idx), + SuspensionTarget::Resume(thread_idx), ) .map(|r| r == WaitResult::Cancelled) } @@ -1402,7 +1411,7 @@ fn thread_yield_then_resume( RuntimeComponentInstanceIndex::from_u32(caller_instance), cancellable != 0, true, - SuspensionTarget::SomeSuspended(thread_idx), + SuspensionTarget::Resume(thread_idx), ) .map(|r| r == WaitResult::Cancelled) } @@ -1421,7 +1430,7 @@ fn thread_suspend_then_promote( RuntimeComponentInstanceIndex::from_u32(caller), cancellable != 0, false, - SuspensionTarget::Some(thread_idx), + SuspensionTarget::Promote(thread_idx), ) .map(|r| r == WaitResult::Cancelled) } @@ -1440,7 +1449,7 @@ fn thread_yield_then_promote( RuntimeComponentInstanceIndex::from_u32(caller), cancellable != 0, true, - SuspensionTarget::Some(thread_idx), + SuspensionTarget::Promote(thread_idx), ) .map(|r| r == WaitResult::Cancelled) } diff --git a/crates/wasmtime/src/runtime/vm/instance.rs b/crates/wasmtime/src/runtime/vm/instance.rs index 321f140def87..5be8d457ec74 100644 --- a/crates/wasmtime/src/runtime/vm/instance.rs +++ b/crates/wasmtime/src/runtime/vm/instance.rs @@ -744,20 +744,6 @@ impl Instance { index, ) } - #[cfg(feature = "component-model")] - VMGlobalKind::TaskMayBlock => { - // SAFETY: validity of this `&Instance` means validity of its - // imports meaning we can read the id of the vmctx within. - let id = unsafe { - let vmctx = super::component::VMComponentContext::from_opaque( - import.vmctx.unwrap().as_non_null(), - ); - super::component::ComponentInstance::vmctx_instance_id(vmctx) - }; - crate::Global::from_task_may_block( - crate::component::store::StoreComponentInstanceId::new(store, id), - ) - } } } diff --git a/crates/wasmtime/src/runtime/vm/vmcontext.rs b/crates/wasmtime/src/runtime/vm/vmcontext.rs index de5b079425ee..7d02bfb98bbf 100644 --- a/crates/wasmtime/src/runtime/vm/vmcontext.rs +++ b/crates/wasmtime/src/runtime/vm/vmcontext.rs @@ -183,8 +183,6 @@ pub enum VMGlobalKind { /// Flags for a component instance, stored in `VMComponentContext`. #[cfg(feature = "component-model")] ComponentFlags(wasmtime_environ::component::RuntimeComponentInstanceIndex), - #[cfg(feature = "component-model")] - TaskMayBlock, } // SAFETY: the above enum is repr(C) and stores nothing else diff --git a/tests/component-model b/tests/component-model index 73b7ad51d3b5..18e511d15eae 160000 --- a/tests/component-model +++ b/tests/component-model @@ -1 +1 @@ -Subproject commit 73b7ad51d3b5d6f1ef53c923d8c585e28b242bcc +Subproject commit 18e511d15eae87c592fb8915c4f5145429001c90 diff --git a/tests/disas/component-model/sync-adapter-calls-x64.wat b/tests/disas/component-model/sync-adapter-calls-x64.wat index bed2b5db1987..49e9ec05debd 100644 --- a/tests/disas/component-model/sync-adapter-calls-x64.wat +++ b/tests/disas/component-model/sync-adapter-calls-x64.wat @@ -58,39 +58,35 @@ ;; movq 0x18(%r10), %r10 ;; addq $0x20, %r10 ;; cmpq %rsp, %r10 -;; ja 0xe6 +;; ja 0xd2 ;; 39: subq $0x20, %rsp -;; movq 0x48(%rdi), %rdi -;; movq 0xe8(%rdi), %rax +;; movq 0x48(%rdi), %rdx +;; movq 0xe8(%rdx), %rax ;; movl (%rax), %ecx ;; testl %ecx, %ecx -;; je 0xe8 -;; 52: movq 0x100(%rdi), %rdx -;; movl (%rdx), %esi -;; movl $0, (%rdx) -;; movq 8(%rdi), %rdi -;; movq 0x88(%rdi), %r8 -;; leaq (%rsp), %r10 -;; movq %r8, (%rsp) +;; je 0xd4 +;; 52: movq 8(%rdx), %rdx +;; movq 0x88(%rdx), %rsi +;; leaq (%rsp), %r8 +;; movq %rsi, (%rsp) ;; movl $2, 8(%rsp) ;; movl $0, 0xc(%rsp) ;; movl $1, 0x10(%rsp) -;; movl 0x80(%rdi), %r9d -;; movl %r9d, 0x14(%rsp) -;; movl $0, 0x80(%rdi) -;; movl 0x84(%rdi), %r11d -;; movl %r11d, 0x18(%rsp) -;; movl $0, 0x84(%rdi) -;; movq %r10, 0x88(%rdi) -;; movq %r8, 0x88(%rdi) -;; movl %r9d, 0x80(%rdi) -;; movl %r11d, 0x84(%rdi) +;; movl 0x80(%rdx), %edi +;; movl %edi, 0x14(%rsp) +;; movl $0, 0x80(%rdx) +;; movl 0x84(%rdx), %r9d +;; movl %r9d, 0x18(%rsp) +;; movl $0, 0x84(%rdx) +;; movq %r8, 0x88(%rdx) +;; movq %rsi, 0x88(%rdx) +;; movl %edi, 0x80(%rdx) +;; movl %r9d, 0x84(%rdx) ;; movl %ecx, (%rax) -;; movl %esi, (%rdx) ;; movl $0x4fc, %eax ;; addq $0x20, %rsp ;; movq %rbp, %rsp ;; popq %rbp ;; retq -;; e6: ud2 -;; e8: ud2 +;; d2: ud2 +;; d4: ud2 diff --git a/tests/disas/component-model/sync-adapter-calls.wat b/tests/disas/component-model/sync-adapter-calls.wat index 91ee39342691..337851e0f5df 100644 --- a/tests/disas/component-model/sync-adapter-calls.wat +++ b/tests/disas/component-model/sync-adapter-calls.wat @@ -58,18 +58,17 @@ ;; region2 = 1207959576 "VMFunctionImport+0x18" ;; region3 = 1476395008 "VMGlobalImport+0x0" ;; region4 = 738197584 "VMComponentContext+0x50" -;; region5 = 738197536 "VMComponentContext+0x20" -;; region6 = 67109000 "VMStoreContext+0x88" -;; region7 = 1006632960 "VMDeferredThread+0x0" -;; region8 = 1006632968 "VMDeferredThread+0x8" -;; region9 = 1006632972 "VMDeferredThread+0xc" -;; region10 = 1006632976 "VMDeferredThread+0x10" -;; region11 = 67108992 "VMStoreContext+0x80" -;; region12 = 1006632980 "VMDeferredThread+0x14" -;; region13 = 67108996 "VMStoreContext+0x84" -;; region14 = 1006632984 "VMDeferredThread+0x18" -;; region15 = 738197568 "VMComponentContext+0x40" -;; region16 = 1207959560 "VMFunctionImport+0x8" +;; region5 = 67109000 "VMStoreContext+0x88" +;; region6 = 1006632960 "VMDeferredThread+0x0" +;; region7 = 1006632968 "VMDeferredThread+0x8" +;; region8 = 1006632972 "VMDeferredThread+0xc" +;; region9 = 1006632976 "VMDeferredThread+0x10" +;; region10 = 67108992 "VMStoreContext+0x80" +;; region11 = 1006632980 "VMDeferredThread+0x14" +;; region12 = 67108996 "VMStoreContext+0x84" +;; region13 = 1006632984 "VMDeferredThread+0x18" +;; region14 = 738197568 "VMComponentContext+0x40" +;; region15 = 1207959560 "VMFunctionImport+0x8" ;; gv0 = vmctx ;; gv1 = load.i64 notrap aligned readonly can_move region0 gv0+8 ;; gv2 = load.i64 notrap aligned region1 gv1+24 @@ -101,28 +100,25 @@ ;; jump block9 ;; ;; block9: -;; v11 = load.i64 notrap aligned readonly can_move region3 v3+256 -;; v12 = load.i32 notrap aligned region5 v11 +;; v16 = load.i64 notrap aligned readonly can_move region0 v3+8 +;; v17 = load.i64 notrap aligned region5 v16+136 +;; v15 = stack_addr.i64 ss0 +;; store notrap aligned region6 v17, v15 +;; v11 = iconst.i32 2 +;; store notrap aligned region7 v11, v15+8 ; v11 = 2 ;; v8 = iconst.i32 0 -;; store notrap aligned region5 v8, v11 ; v8 = 0 -;; v20 = load.i64 notrap aligned readonly can_move region0 v3+8 -;; v21 = load.i64 notrap aligned region6 v20+136 -;; v19 = stack_addr.i64 ss0 -;; store notrap aligned region7 v21, v19 -;; v15 = iconst.i32 2 -;; store notrap aligned region8 v15, v19+8 ; v15 = 2 -;; store notrap aligned region9 v8, v19+12 ; v8 = 0 -;; v17 = iconst.i32 1 -;; store notrap aligned region10 v17, v19+16 ; v17 = 1 -;; v22 = load.i32 notrap aligned region11 v20+128 -;; store notrap aligned region12 v22, v19+20 -;; store notrap aligned region11 v8, v20+128 ; v8 = 0 -;; v24 = load.i32 notrap aligned region13 v20+132 -;; store notrap aligned region14 v24, v19+24 -;; store notrap aligned region13 v8, v20+132 ; v8 = 0 -;; store notrap aligned region6 v19, v20+136 -;; v26 = load.i64 notrap aligned readonly can_move region3 v3+208 -;; v27 = load.i32 notrap aligned region15 v26 +;; store notrap aligned region8 v8, v15+12 ; v8 = 0 +;; v13 = iconst.i32 1 +;; store notrap aligned region9 v13, v15+16 ; v13 = 1 +;; v18 = load.i32 notrap aligned region10 v16+128 +;; store notrap aligned region11 v18, v15+20 +;; store notrap aligned region10 v8, v16+128 ; v8 = 0 +;; v20 = load.i32 notrap aligned region12 v16+132 +;; store notrap aligned region13 v20, v15+24 +;; store notrap aligned region12 v8, v16+132 ; v8 = 0 +;; store notrap aligned region5 v15, v16+136 +;; v22 = load.i64 notrap aligned readonly can_move region3 v3+208 +;; v23 = load.i32 notrap aligned region14 v22 ;; jump block16 ;; ;; block16: @@ -135,14 +131,13 @@ ;; jump block12 ;; ;; block12: -;; store.i64 notrap aligned region6 v21, v20+136 -;; store.i32 notrap aligned region11 v22, v20+128 -;; store.i32 notrap aligned region13 v24, v20+132 +;; store.i64 notrap aligned region5 v17, v16+136 +;; store.i32 notrap aligned region10 v18, v16+128 +;; store.i32 notrap aligned region12 v20, v16+132 ;; jump block14 ;; ;; block14: ;; store.i32 notrap aligned region4 v10, v9 -;; store.i32 notrap aligned region5 v12, v11 ;; jump block7 ;; ;; block7: @@ -158,6 +153,6 @@ ;; @00f0 jump block1 ;; ;; block1: -;; v50 = iconst.i32 1276 -;; @00f0 return v50 ; v50 = 1276 +;; v45 = iconst.i32 1276 +;; @00f0 return v45 ; v45 = 1276 ;; } diff --git a/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast b/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast index c3648c07e559..528aa22bfbeb 100644 --- a/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast +++ b/tests/misc_testsuite/component-model/async/callback-yield-then-exit.wast @@ -47,5 +47,4 @@ (func (export "run") (alias export $B "run")) ) -(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning") -(assert_trap (invoke "run") "wasm trap: cannot enter component instance") +(assert_return (invoke "run")) diff --git a/tests/misc_testsuite/component-model/async/fused.wast b/tests/misc_testsuite/component-model/async/fused.wast index d7b644f594ab..fad18e8b86a5 100644 --- a/tests/misc_testsuite/component-model/async/fused.wast +++ b/tests/misc_testsuite/component-model/async/fused.wast @@ -145,4 +145,4 @@ (func (export "run") (alias export $lowerer "run")) ) -(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning") +(assert_return (invoke "run")) diff --git a/tests/misc_testsuite/component-model/async/future-read.wast b/tests/misc_testsuite/component-model/async/future-read.wast index 498d2893a513..0f289481a7d3 100644 --- a/tests/misc_testsuite/component-model/async/future-read.wast +++ b/tests/misc_testsuite/component-model/async/future-read.wast @@ -122,7 +122,7 @@ (func (export "run") (alias export $other-child "run")) ) -(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning") +(assert_return (invoke "run")) ;; synchronous future.read; async lift (component diff --git a/tests/misc_testsuite/component-model/async/stackful.wast b/tests/misc_testsuite/component-model/async/stackful.wast index 30cca1d32ced..6c7712f5fdbb 100644 --- a/tests/misc_testsuite/component-model/async/stackful.wast +++ b/tests/misc_testsuite/component-model/async/stackful.wast @@ -104,7 +104,7 @@ (func (export "run") (alias export $lowerer "run")) ) -(assert_trap (invoke "run") "wasm trap: cannot block a synchronous task before returning") +(assert_return (invoke "run")) ;; waitable-set.wait (component diff --git a/tests/wast.rs b/tests/wast.rs index b12d41aa904c..0b78e54a17ce 100644 --- a/tests/wast.rs +++ b/tests/wast.rs @@ -154,7 +154,7 @@ fn run_wast(test: &WastTest, config: WastConfig) -> wasmtime::Result<()> { // panic or segfault as a result. // // Updates to whether a test should pass or fail should be done in the - // `crates/wast-util/src/lib.rs` file. + // `crates/test-util/src/wast.rs` file. let should_fail = test.should_fail(&config); let multi_memory = test_config.multi_memory();