1use self::error_contexts::GlobalErrorContextRefCount;
54use crate::component::func::{Func, call_post_return};
55use crate::component::{
56 HasData, HasSelf, Instance, Resource, ResourceTable, ResourceTableError, RuntimeInstance,
57};
58use crate::fiber::{self, StoreFiber, StoreFiberYield};
59use crate::hash_set::HashSet;
60#[cfg(feature = "gc")]
61use crate::module::ModuleRegistry;
62use crate::prelude::*;
63use crate::store::{Store, StoreId, StoreInner, StoreOpaque, StoreToken};
64#[cfg(feature = "gc")]
65use crate::vm::GcRootsList;
66use crate::vm::component::{CallContext, ComponentInstance, CurrentScope, InstanceState, Scope};
67use crate::vm::{AlwaysMut, SendSyncPtr, VMFuncRef, VMLazyThread, VMMemoryDefinition, VMStore};
68use crate::{
69 AsContext, AsContextMut, FuncType, Result, StoreContext, StoreContextMut, ValRaw, ValType, bail,
70};
71use crate::{Instance as ModuleInstance, bail_bug};
72use alloc::borrow::ToOwned;
73use alloc::collections::{BTreeMap, BTreeSet, VecDeque};
74use core::any::Any;
75use core::cell::UnsafeCell;
76use core::fmt;
77use core::future;
78use core::future::Future;
79use core::marker::PhantomData;
80use core::mem::{self, ManuallyDrop, MaybeUninit};
81use core::ops::DerefMut;
82use core::pin::{Pin, pin};
83use core::ptr::{self, NonNull};
84use core::task::{Context, Poll, Waker};
85use futures::channel::oneshot;
86use futures::stream::{FuturesUnordered, StreamExt};
87use futures_and_streams::{FlatAbi, ReturnCode, TransmitHandle, TransmitIndex};
88use table::{TableDebug, TableId};
89use wasmtime_environ::component::{
90 CanonicalAbiInfo, CanonicalOptions, CanonicalOptionsDataModel, MAX_FLAT_PARAMS,
91 MAX_FLAT_RESULTS, OptionsIndex, PREPARE_ASYNC_NO_RESULT, PREPARE_ASYNC_WITH_RESULT,
92 RuntimeComponentInstanceIndex, RuntimeTableIndex, StringEncoding,
93 TypeComponentGlobalErrorContextTableIndex, TypeComponentLocalErrorContextTableIndex,
94 TypeFuncIndex, TypeFutureTableIndex, TypeStreamTableIndex, TypeTupleIndex,
95};
96use wasmtime_environ::packed_option::ReservedValue;
97use wasmtime_environ::{NUM_COMPONENT_CONTEXT_SLOTS, Trap};
98#[cfg(feature = "gc")]
99use wasmtime_unwinder::Unwind;
100
101pub use abort::JoinHandle;
102pub use func::{FuncCallConcurrent, TypedFuncCallConcurrent};
103pub use future_stream_any::{FutureAny, StreamAny};
104pub use futures_and_streams::{
105 Destination, DirectDestination, DirectSource, ErrorContext, FutureConsumer, FutureProducer,
106 FutureReader, GuardedFutureReader, GuardedStreamReader, ReadBuffer, Source, StreamConsumer,
107 StreamProducer, StreamReader, StreamResult, VecBuffer, WriteBuffer,
108};
109pub(crate) use futures_and_streams::{ResourcePair, lower_error_context_to_index};
110
111mod abort;
112mod error_contexts;
113mod func;
114mod future_stream_any;
115mod futures_and_streams;
116pub(crate) mod table;
117pub(crate) mod tls;
118
119const BLOCKED: u32 = 0xffff_ffff;
122
123#[derive(Clone, Copy, Eq, PartialEq, Debug)]
125pub enum Status {
126 Starting = 0,
127 Started = 1,
128 Returned = 2,
129 StartCancelled = 3,
130 ReturnCancelled = 4,
131}
132
133impl Status {
134 pub fn pack(self, waitable: Option<u32>) -> u32 {
140 assert!(matches!(self, Status::Returned) == waitable.is_none());
141 let waitable = waitable.unwrap_or(0);
142 assert!(waitable < (1 << 28));
143 (waitable << 4) | (self as u32)
144 }
145}
146
147#[derive(Clone, Copy, Debug)]
150enum Event {
151 None,
152 Subtask {
153 status: Status,
154 },
155 StreamRead {
156 code: ReturnCode,
157 pending: Option<(TypeStreamTableIndex, u32)>,
158 },
159 StreamWrite {
160 code: ReturnCode,
161 pending: Option<(TypeStreamTableIndex, u32)>,
162 },
163 FutureRead {
164 code: ReturnCode,
165 pending: Option<(TypeFutureTableIndex, u32)>,
166 },
167 FutureWrite {
168 code: ReturnCode,
169 pending: Option<(TypeFutureTableIndex, u32)>,
170 },
171 Cancelled,
172}
173
174impl Event {
175 fn parts(self) -> (u32, u32) {
180 const EVENT_NONE: u32 = 0;
181 const EVENT_SUBTASK: u32 = 1;
182 const EVENT_STREAM_READ: u32 = 2;
183 const EVENT_STREAM_WRITE: u32 = 3;
184 const EVENT_FUTURE_READ: u32 = 4;
185 const EVENT_FUTURE_WRITE: u32 = 5;
186 const EVENT_CANCELLED: u32 = 6;
187 match self {
188 Event::None => (EVENT_NONE, 0),
189 Event::Cancelled => (EVENT_CANCELLED, 0),
190 Event::Subtask { status } => (EVENT_SUBTASK, status as u32),
191 Event::StreamRead { code, .. } => (EVENT_STREAM_READ, code.encode()),
192 Event::StreamWrite { code, .. } => (EVENT_STREAM_WRITE, code.encode()),
193 Event::FutureRead { code, .. } => (EVENT_FUTURE_READ, code.encode()),
194 Event::FutureWrite { code, .. } => (EVENT_FUTURE_WRITE, code.encode()),
195 }
196 }
197}
198
199mod callback_code {
201 pub const EXIT: u32 = 0;
202 pub const YIELD: u32 = 1;
203 pub const WAIT: u32 = 2;
204}
205
206const START_FLAG_ASYNC_CALLEE: u32 = wasmtime_environ::component::START_FLAG_ASYNC_CALLEE as u32;
210
211pub struct Access<'a, T: 'static, D: HasData + ?Sized = HasSelf<T>> {
217 store: StoreContextMut<'a, T>,
218 get_data: fn(&mut T) -> D::Data<'_>,
219}
220
221impl<'a, T, D> Access<'a, T, D>
222where
223 D: HasData + ?Sized,
224 T: 'static,
225{
226 pub fn new(store: StoreContextMut<'a, T>, get_data: fn(&mut T) -> D::Data<'_>) -> Self {
228 Self { store, get_data }
229 }
230
231 pub fn data_mut(&mut self) -> &mut T {
233 self.store.data_mut()
234 }
235
236 pub fn get(&mut self) -> D::Data<'_> {
238 (self.get_data)(self.data_mut())
239 }
240
241 pub fn spawn(&mut self, task: impl for<'fut> AccessorTask<'fut, T, D>) -> Result<JoinHandle>
245 where
246 T: 'static,
247 {
248 let accessor = Accessor {
249 get_data: self.get_data,
250 token: StoreToken::new(self.store.as_context_mut()),
251 };
252 self.store
253 .as_context_mut()
254 .spawn_with_accessor(accessor, task)
255 }
256
257 pub fn getter(&self) -> fn(&mut T) -> D::Data<'_> {
260 self.get_data
261 }
262}
263
264impl<'a, T, D> AsContext for Access<'a, T, D>
265where
266 D: HasData + ?Sized,
267 T: 'static,
268{
269 type Data = T;
270
271 fn as_context(&self) -> StoreContext<'_, T> {
272 self.store.as_context()
273 }
274}
275
276impl<'a, T, D> AsContextMut for Access<'a, T, D>
277where
278 D: HasData + ?Sized,
279 T: 'static,
280{
281 fn as_context_mut(&mut self) -> StoreContextMut<'_, T> {
282 self.store.as_context_mut()
283 }
284}
285
286pub struct Accessor<T: 'static, D = HasSelf<T>>
346where
347 D: HasData + ?Sized,
348{
349 token: StoreToken<T>,
350 get_data: fn(&mut T) -> D::Data<'_>,
351}
352
353pub trait AsAccessor {
370 type Data: 'static;
372
373 type AccessorData: HasData + ?Sized;
376
377 fn as_accessor(&self) -> &Accessor<Self::Data, Self::AccessorData>;
379}
380
381impl<T: AsAccessor + ?Sized> AsAccessor for &T {
382 type Data = T::Data;
383 type AccessorData = T::AccessorData;
384
385 fn as_accessor(&self) -> &Accessor<Self::Data, Self::AccessorData> {
386 T::as_accessor(self)
387 }
388}
389
390impl<T, D: HasData + ?Sized> AsAccessor for Accessor<T, D> {
391 type Data = T;
392 type AccessorData = D;
393
394 fn as_accessor(&self) -> &Accessor<T, D> {
395 self
396 }
397}
398
399const _: () = {
422 const fn assert<T: Send + Sync>() {}
423 assert::<Accessor<UnsafeCell<u32>>>();
424};
425
426impl<T> Accessor<T> {
427 pub(crate) fn new(token: StoreToken<T>) -> Self {
436 Self {
437 token,
438 get_data: |x| x,
439 }
440 }
441}
442
443impl<T, D> Accessor<T, D>
444where
445 D: HasData + ?Sized,
446{
447 pub fn with<R>(&self, fun: impl FnOnce(Access<'_, T, D>) -> R) -> R {
465 tls::get(|vmstore| {
466 fun(Access {
467 store: self.token.as_context_mut(vmstore),
468 get_data: self.get_data,
469 })
470 })
471 }
472
473 pub fn getter(&self) -> fn(&mut T) -> D::Data<'_> {
476 self.get_data
477 }
478
479 pub fn with_getter<D2: HasData>(
496 &self,
497 get_data: fn(&mut T) -> D2::Data<'_>,
498 ) -> Accessor<T, D2> {
499 Accessor {
500 token: self.token,
501 get_data,
502 }
503 }
504
505 pub fn spawn(&self, task: impl for<'fut> AccessorTask<'fut, T, D>) -> Result<JoinHandle>
521 where
522 T: 'static,
523 {
524 let accessor = self.clone_for_spawn();
525 self.with(|mut access| access.as_context_mut().spawn_with_accessor(accessor, task))
526 }
527
528 fn clone_for_spawn(&self) -> Self {
529 Self {
530 token: self.token,
531 get_data: self.get_data,
532 }
533 }
534
535 pub fn poll_no_interesting_tasks(&self, cx: &mut Context<'_>) -> Poll<()> {
571 self.with(|mut access| {
572 let store = access.as_context_mut().0;
573 let state = store.concurrent_state_mut_without_forcing_current_thread();
574 if state.interesting_tasks == 0 {
575 Poll::Ready(())
576 } else {
577 state.interesting_tasks_empty_waker = Some(cx.waker().clone());
578 Poll::Pending
579 }
580 })
581 }
582
583 pub fn poll_ready_for_concurrent_call(&self, func: Func, cx: &mut Context<'_>) -> Poll<()> {
600 self.with(|mut access| {
601 let store = access.as_context_mut().0;
602 let (_, _, _, raw_options) = func.abi_info(store);
603 let instance = func.instance().runtime_instance(raw_options.instance);
604 let state = store.instance_state(instance).concurrent_state();
605 if state.backpressure == 0 {
606 Poll::Ready(())
607 } else {
608 store
609 .concurrent_state_mut_without_forcing_current_thread()
610 .ready_for_concurrent_call_waker = Some(cx.waker().clone());
611 Poll::Pending
612 }
613 })
614 }
615}
616
617pub trait AccessorTask<'fut, T, D = HasSelf<T>>:
639 AsyncFnOnce(&Accessor<T, D>) -> Result<()> + Send + 'static
640where
641 D: HasData + ?Sized,
642{
643 fn run(self, accessor: &'fut Accessor<T, D>) -> impl Future<Output = Result<()>> + Send + 'fut;
645}
646
647impl<'fut, F, Fut, T, D> AccessorTask<'fut, T, D> for F
648where
649 T: 'static,
650 F: AsyncFnOnce(&Accessor<T, D>) -> Result<()>,
651 F: FnOnce(&'fut Accessor<T, D>) -> Fut + Send + 'static,
652 Fut: Future<Output = Result<()>> + Send + 'fut,
653 D: HasData,
654{
655 fn run(self, accessor: &'fut Accessor<T, D>) -> impl Future<Output = Result<()>> + Send + 'fut {
656 (self)(accessor)
657 }
658}
659
660enum CallerInfo {
663 Async {
665 params: Vec<ValRaw>,
666 has_result: bool,
667 },
668 Sync {
670 params: Vec<ValRaw>,
671 result_count: u32,
672 },
673}
674
675enum WaitMode {
677 Fiber(StoreFiber<'static>),
679 Callback(Instance),
682 Caller {
683 fiber: StoreFiber<'static>,
684 callee: TableId<GuestTask>,
685 },
686}
687
688#[derive(Debug)]
689enum WaitReason {
690 GuestSubtask(TableId<GuestTask>),
691 Other,
692}
693
694#[derive(Debug)]
696enum SuspendReason {
697 Waiting {
700 set: TableId<WaitableSet>,
701 thread: QualifiedThreadId,
702 },
703 WaitingForGuestSubtask {
704 caller: QualifiedThreadId,
705 callee: TableId<GuestTask>,
706 },
707 NeedWork,
710 Yielding { thread: QualifiedThreadId },
713 ExplicitlySuspending { thread: QualifiedThreadId },
715}
716
717enum GuestCallKind {
719 DeliverEvent {
722 instance: Instance,
724 set: Option<TableId<WaitableSet>>,
729 },
730 StartImplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
736 StartExplicit(Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>),
737}
738
739impl fmt::Debug for GuestCallKind {
740 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
741 match self {
742 Self::DeliverEvent { instance, set } => f
743 .debug_struct("DeliverEvent")
744 .field("instance", instance)
745 .field("set", set)
746 .finish(),
747 Self::StartImplicit(_) => f.debug_tuple("StartImplicit").finish(),
748 Self::StartExplicit(_) => f.debug_tuple("StartExplicit").finish(),
749 }
750 }
751}
752
753#[derive(Copy, Clone, Debug)]
755pub enum SuspensionTarget {
756 Resume(u32),
757 Promote(u32),
758 None,
759}
760
761#[derive(Copy, Clone, Debug)]
763pub enum ResumeThread {
764 Promote,
765 Resume,
766 ResumeLater,
767}
768
769#[derive(Debug)]
771struct GuestCall {
772 thread: QualifiedThreadId,
773 kind: GuestCallKind,
774}
775
776impl GuestCall {
777 fn is_ready(&self, store: &mut StoreOpaque) -> Result<bool> {
787 let task = store.concurrent_state_mut()?.get_mut(self.thread.task)?;
788 let async_typed = task.async_typed;
789 let instance = task.instance;
790 let state = store.instance_state(instance).concurrent_state();
791
792 let ready = match &self.kind {
793 GuestCallKind::DeliverEvent { .. } => !state.do_not_enter,
794 GuestCallKind::StartImplicit(_) => {
795 !async_typed || !(state.do_not_enter || state.backpressure > 0)
796 }
797 GuestCallKind::StartExplicit(_) => true,
798 };
799 log::trace!(
800 "call {self:?} ready? {ready} (do_not_enter: {}; backpressure: {})",
801 state.do_not_enter,
802 state.backpressure
803 );
804 Ok(ready)
805 }
806}
807
808enum WorkerItem {
810 GuestCall(GuestCall),
811 Function(AlwaysMut<Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send>>),
812}
813
814enum WorkItem {
817 PushFuture(AlwaysMut<HostTaskFuture>),
819 ResumeFiber {
821 instance: RuntimeInstance,
822 thread: QualifiedThreadId,
823 fiber: StoreFiber<'static>,
824 },
825 ResumeThread {
827 instance: RuntimeInstance,
828 thread: QualifiedThreadId,
829 },
830 GuestCall {
832 instance: RuntimeInstance,
833 call: GuestCall,
834 },
835 WorkerFunction(AlwaysMut<Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send>>),
837}
838
839impl fmt::Debug for WorkItem {
840 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
841 match self {
842 Self::PushFuture(_) => f.debug_tuple("PushFuture").finish(),
843 Self::ResumeFiber {
844 instance, thread, ..
845 } => f
846 .debug_struct("ResumeFiber")
847 .field("instance", instance)
848 .field("thread", thread)
849 .finish(),
850 Self::ResumeThread { instance, thread } => f
851 .debug_struct("ResumeThread")
852 .field("instance", instance)
853 .field("thread", thread)
854 .finish(),
855 Self::GuestCall { instance, call } => f
856 .debug_struct("GuestCall")
857 .field("instance", instance)
858 .field("call", call)
859 .finish(),
860 Self::WorkerFunction(_) => f.debug_tuple("WorkerFunction").finish(),
861 }
862 }
863}
864
865#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
867pub(crate) enum WaitResult {
868 Cancelled,
869 Completed,
870}
871
872pub(crate) fn poll_and_block<R: Send + Sync + 'static>(
880 store: &mut dyn VMStore,
881 host_task: EnteredHostTask,
882 future: impl Future<Output = Result<R>> + Send + 'static,
883) -> Result<R> {
884 let mut future = Box::pin(future);
891 let poll = tls::set(store, || {
892 future
893 .as_mut()
894 .poll(&mut Context::from_waker(&Waker::noop()))
895 });
896
897 let caller = match host_task {
898 Some(caller) => caller,
899 None => bail_bug!("host task wasn't created but should have been"),
900 };
901
902 let task = match poll {
903 Poll::Ready(result) => return result,
905
906 Poll::Pending => {
911 let Some(task) = store.materialize_host_task_id()? else {
912 bail_bug!("current thread is not a host thread")
913 };
914
915 let future = Box::pin(async move {
918 let result = run_with_host_task_set(task, future).await??;
919 tls::get(move |store| {
920 let state = store.concurrent_state_mut()?;
921 let host_state = &mut state.get_mut(task)?.state;
922 assert!(matches!(host_state, HostTaskState::CalleeStarted));
923 *host_state = HostTaskState::CalleeFinished(Box::new(result));
924
925 Waitable::Host(task).set_event(
926 state,
927 Some(Event::Subtask {
928 status: Status::Returned,
929 }),
930 )?;
931
932 Ok(())
933 })
934 }) as HostTaskFuture;
935
936 let caller_instance = store.concurrent_state_mut()?.get_mut(caller.task)?.instance;
937 store.switch_or_trap_if_may_not_suspend(caller_instance)?;
938
939 let state = store.concurrent_state_mut()?;
940 state.push_future(future);
941
942 let set = state.get_mut(caller.thread)?.sync_call_set;
943 Waitable::Host(task).join(state, Some(set))?;
944
945 store.suspend(SuspendReason::Waiting {
946 set,
947 thread: caller,
948 })?;
949
950 Waitable::Host(task).join(store.concurrent_state_mut()?, None)?;
954 task
955 }
956 };
957
958 let host_state = &mut store.concurrent_state_mut()?.get_mut(task)?.state;
960 match mem::replace(host_state, HostTaskState::CalleeDone { cancelled: false }) {
961 HostTaskState::CalleeFinished(result) => Ok(match result.downcast() {
962 Ok(result) => *result,
963 Err(_) => bail_bug!("host task finished with wrong type of result"),
964 }),
965 _ => bail_bug!("unexpected host task state after completion"),
966 }
967}
968
969fn handle_guest_call(store: &mut dyn VMStore, call: GuestCall) -> Result<()> {
971 match call.kind {
972 GuestCallKind::DeliverEvent { instance, set } => {
973 let (event, waitable) = match instance.get_event(store, call.thread.task, set, true)? {
974 Some(pair) => pair,
975 None => bail_bug!("delivering non-present event"),
976 };
977 let state = store.concurrent_state_mut()?;
978 let task = state.get_mut(call.thread.task)?;
979 let runtime_instance = task.instance;
980 let handle = waitable.map(|(_, v)| v).unwrap_or(0);
981
982 log::trace!(
983 "use callback to deliver event {event:?} to {:?} for {waitable:?}",
984 call.thread,
985 );
986
987 let old_thread = store.set_thread(call.thread)?;
988 log::trace!(
989 "GuestCallKind::DeliverEvent: replaced {old_thread:?} with {:?} as current thread",
990 call.thread
991 );
992
993 store.enter_instance(runtime_instance);
994
995 let Some(callback) = store
996 .concurrent_state_mut()?
997 .get_mut(call.thread.task)?
998 .callback
999 .take()
1000 else {
1001 bail_bug!("guest task callback field not present")
1002 };
1003
1004 let code = callback(store, event, handle)?;
1005
1006 store
1007 .concurrent_state_mut()?
1008 .get_mut(call.thread.task)?
1009 .callback = Some(callback);
1010
1011 store.exit_instance(runtime_instance)?;
1012
1013 store.set_thread(old_thread)?;
1014
1015 instance.handle_callback_code(store, call.thread, runtime_instance.index, code)?;
1016
1017 log::trace!("GuestCallKind::DeliverEvent: restored {old_thread:?} as current thread");
1018 }
1019 GuestCallKind::StartImplicit(fun) => {
1020 fun(store)?;
1021 }
1022 GuestCallKind::StartExplicit(fun) => {
1023 fun(store)?;
1024 }
1025 }
1026
1027 Ok(())
1028}
1029
1030impl<T> Store<T> {
1031 pub async fn run_concurrent<R>(&mut self, fun: impl AsyncFnOnce(&Accessor<T>) -> R) -> Result<R>
1033 where
1034 T: Send + 'static,
1035 {
1036 ensure!(
1037 self.as_context().0.concurrency_support(),
1038 "cannot use `run_concurrent` when Config::concurrency_support disabled",
1039 );
1040 self.as_context_mut().run_concurrent(fun).await
1041 }
1042
1043 #[doc(hidden)]
1044 pub fn assert_concurrent_state_empty(&mut self) {
1045 self.as_context_mut().assert_concurrent_state_empty();
1046 }
1047
1048 #[doc(hidden)]
1049 pub fn concurrent_state_table_size(&mut self) -> usize {
1050 self.as_context_mut().concurrent_state_table_size()
1051 }
1052
1053 pub fn spawn(
1055 &mut self,
1056 task: impl for<'fut> AccessorTask<'fut, T, HasSelf<T>>,
1057 ) -> Result<JoinHandle>
1058 where
1059 T: 'static,
1060 {
1061 self.as_context_mut().spawn(task)
1062 }
1063}
1064
1065impl<T> StoreContextMut<'_, T> {
1066 #[doc(hidden)]
1077 pub fn assert_concurrent_state_empty(self) {
1078 let store = self.0;
1079 store
1080 .store_data_mut()
1081 .components
1082 .assert_instance_states_empty();
1083 let state = store.concurrent_state_mut().unwrap();
1084 assert!(
1085 state.table.get_mut().is_empty(),
1086 "non-empty table: {:?}",
1087 state.table.get_mut()
1088 );
1089 assert!(state.switch_item.is_none());
1090 assert!(state.high_priority.is_empty());
1091 assert!(state.low_priority.is_empty());
1092 assert!(state.unforced_current_thread.is_none());
1093 assert!(state.deferred_host_call_context.is_none());
1094 assert!(state.futures_mut().unwrap().is_empty());
1095 assert!(state.global_error_context_ref_counts.is_empty());
1096 }
1097
1098 #[doc(hidden)]
1103 pub fn concurrent_state_table_size(&mut self) -> usize {
1104 self.0
1105 .concurrent_state_mut()
1106 .unwrap()
1107 .table
1108 .get_mut()
1109 .iter_mut()
1110 .count()
1111 }
1112
1113 pub fn spawn(mut self, task: impl for<'fut> AccessorTask<'fut, T>) -> Result<JoinHandle>
1123 where
1124 T: 'static,
1125 {
1126 let accessor = Accessor::new(StoreToken::new(self.as_context_mut()));
1127 self.spawn_with_accessor(accessor, task)
1128 }
1129
1130 fn spawn_with_accessor<D>(
1133 self,
1134 accessor: Accessor<T, D>,
1135 task: impl for<'fut> AccessorTask<'fut, T, D>,
1136 ) -> Result<JoinHandle>
1137 where
1138 T: 'static,
1139 D: HasData + ?Sized,
1140 {
1141 let (handle, future) = JoinHandle::run(async move { task.run(&accessor).await });
1145 self.0
1146 .concurrent_state_mut()?
1147 .push_future(Box::pin(async move { future.await.unwrap_or(Ok(())) }));
1148 Ok(handle)
1149 }
1150
1151 pub async fn run_concurrent<R>(self, fun: impl AsyncFnOnce(&Accessor<T>) -> R) -> Result<R>
1235 where
1236 T: Send + 'static,
1237 {
1238 ensure!(
1239 self.0.concurrency_support(),
1240 "cannot use `run_concurrent` when Config::concurrency_support disabled",
1241 );
1242 self.do_run_concurrent(fun, false).await
1243 }
1244
1245 pub(super) async fn run_concurrent_trap_on_idle<R>(
1246 self,
1247 fun: impl AsyncFnOnce(&Accessor<T>) -> R,
1248 ) -> Result<R> {
1249 self.do_run_concurrent(fun, true).await
1250 }
1251
1252 async fn do_run_concurrent<R>(
1253 mut self,
1254 fun: impl AsyncFnOnce(&Accessor<T>) -> R,
1255 trap_on_idle: bool,
1256 ) -> Result<R> {
1257 debug_assert!(self.0.concurrency_support());
1258 let already_running = self
1259 .0
1260 .concurrent_state_mut_already_forced_current_thread()
1261 .event_loop_running;
1262 if already_running {
1263 bail!("Recursive `StoreContextMut::run_concurrent` calls not supported")
1264 }
1265 let token = StoreToken::new(self.as_context_mut());
1266
1267 struct Dropper<'a, T: 'static, V> {
1268 store: StoreContextMut<'a, T>,
1269 value: ManuallyDrop<V>,
1270 }
1271
1272 impl<'a, T, V> Drop for Dropper<'a, T, V> {
1273 fn drop(&mut self) {
1274 self.store
1275 .0
1276 .concurrent_state_mut_already_forced_current_thread()
1277 .event_loop_running = false;
1278
1279 tls::set(self.store.0, || {
1280 unsafe { ManuallyDrop::drop(&mut self.value) }
1285 });
1286 }
1287 }
1288
1289 let accessor = &Accessor::new(token);
1290 self.0
1291 .concurrent_state_mut_already_forced_current_thread()
1292 .event_loop_running = true;
1293 let dropper = &mut Dropper {
1294 store: self,
1295 value: ManuallyDrop::new(fun(accessor)),
1296 };
1297 let future = unsafe { Pin::new_unchecked(dropper.value.deref_mut()) };
1299
1300 dropper
1301 .store
1302 .as_context_mut()
1303 .poll_until(future, trap_on_idle)
1304 .await
1305 }
1306
1307 async fn poll_until<R>(
1313 mut self,
1314 mut future: Pin<&mut impl Future<Output = R>>,
1315 trap_on_idle: bool,
1316 ) -> Result<R> {
1317 struct Reset<'a, T: 'static> {
1318 store: StoreContextMut<'a, T>,
1319 futures: Option<FuturesUnordered<HostTaskFuture>>,
1320 }
1321
1322 impl<'a, T> Drop for Reset<'a, T> {
1323 fn drop(&mut self) {
1324 if let Some(futures) = self.futures.take() {
1325 *self
1326 .store
1327 .0
1328 .concurrent_state_mut_already_forced_current_thread()
1329 .futures
1330 .get_mut() = Some(futures);
1331 }
1332 }
1333 }
1334
1335 loop {
1336 let futures = self.0.concurrent_state_mut()?.futures.get_mut().take();
1340 let mut reset = Reset {
1341 store: self.as_context_mut(),
1342 futures,
1343 };
1344 let mut next = match reset.futures.as_mut() {
1345 Some(f) => pin!(f.next()),
1346 None => bail_bug!("concurrent state missing futures field"),
1347 };
1348
1349 enum PollResult<R> {
1350 Complete(R),
1351 ProcessWork {
1352 ready: Option<WorkItem>,
1353 low_priority: bool,
1354 },
1355 }
1356
1357 let result = future::poll_fn(|cx| {
1358 if let Poll::Ready(value) = tls::set(reset.store.0, || future.as_mut().poll(cx)) {
1361 return Poll::Ready(Ok(PollResult::Complete(value)));
1362 }
1363
1364 let next = match tls::set(reset.store.0, || next.as_mut().poll(cx)) {
1368 Poll::Ready(Some(output)) => {
1369 match output {
1370 Err(e) => return Poll::Ready(Err(e)),
1371 Ok(()) => {}
1372 }
1373 Poll::Ready(true)
1374 }
1375 Poll::Ready(None) => Poll::Ready(false),
1376 Poll::Pending => Poll::Pending,
1377 };
1378
1379 let state = reset.store.0.concurrent_state_mut()?;
1394 let mut ready = state.switch_item.take();
1395 let mut low_priority = false;
1396 if ready.is_none() {
1397 ready = state.high_priority.pop_back();
1398 if ready.is_none() {
1399 ready = state.low_priority.pop_back();
1400 low_priority = true;
1401 }
1402 }
1403 if ready.is_some() {
1404 return Poll::Ready(Ok(PollResult::ProcessWork {
1405 ready,
1406 low_priority,
1407 }));
1408 }
1409
1410 return match next {
1414 Poll::Ready(true) => {
1415 Poll::Ready(Ok(PollResult::ProcessWork {
1421 ready: None,
1422 low_priority: false,
1423 }))
1424 }
1425 Poll::Ready(false) => {
1426 if let Poll::Ready(value) =
1430 tls::set(reset.store.0, || future.as_mut().poll(cx))
1431 {
1432 Poll::Ready(Ok(PollResult::Complete(value)))
1433 } else {
1434 if trap_on_idle {
1440 Poll::Ready(Err(if reset.store.0.any_may_not_suspend()? {
1447 Trap::CannotBlockSyncTask.into()
1448 } else {
1449 Trap::AsyncDeadlock.into()
1451 }))
1452 } else {
1453 Poll::Pending
1457 }
1458 }
1459 }
1460 Poll::Pending => Poll::Pending,
1465 };
1466 })
1467 .await;
1468
1469 drop(reset);
1473
1474 match result? {
1475 PollResult::Complete(value) => break Ok(value),
1478 PollResult::ProcessWork {
1481 ready,
1482 low_priority,
1483 } => {
1484 struct Dispose<'a, T: 'static> {
1485 store: StoreContextMut<'a, T>,
1486 ready: Option<WorkItem>,
1487 }
1488
1489 impl<'a, T> Drop for Dispose<'a, T> {
1490 fn drop(&mut self) {
1491 if let Some(item) = self.ready.take() {
1492 match item {
1493 WorkItem::ResumeFiber { mut fiber, .. } => {
1494 fiber.dispose(self.store.0)
1495 }
1496 WorkItem::PushFuture(future) => {
1497 tls::set(self.store.0, move || drop(future))
1498 }
1499 _ => {}
1500 }
1501 }
1502 }
1503 }
1504
1505 let mut dispose = Dispose {
1506 store: self.as_context_mut(),
1507 ready,
1508 };
1509
1510 if low_priority {
1532 dispose.store.0.yield_now().await
1533 }
1534
1535 if let Some(item) = dispose.ready.take() {
1536 dispose
1537 .store
1538 .as_context_mut()
1539 .handle_work_item(item)
1540 .await?;
1541 }
1542 }
1543 }
1544 }
1545 }
1546
1547 async fn handle_work_item(self, item: WorkItem) -> Result<()> {
1549 log::trace!("handle work item {item:?}");
1550 match item {
1551 WorkItem::PushFuture(future) => {
1552 self.0
1553 .concurrent_state_mut()?
1554 .futures_mut()?
1555 .push(future.into_inner());
1556 }
1557 WorkItem::ResumeFiber { fiber, .. } => {
1558 self.0.resume_fiber(fiber).await?;
1559 }
1560 WorkItem::ResumeThread { thread, .. } => {
1561 if let GuestThreadState::Ready { fiber, .. } = mem::replace(
1562 &mut self.0.concurrent_state_mut()?.get_mut(thread.thread)?.state,
1563 GuestThreadState::Running,
1564 ) {
1565 self.0.resume_fiber(fiber).await?;
1566 } else {
1567 bail_bug!("cannot resume non-pending thread {thread:?}");
1568 }
1569 }
1570 WorkItem::GuestCall { call, .. } => {
1571 if call.is_ready(self.0)? {
1572 self.run_on_worker(WorkerItem::GuestCall(call)).await?;
1573 } else {
1574 let state = self.0.concurrent_state_mut()?;
1575 let task = state.get_mut(call.thread.task)?;
1576 if !task.starting_sent {
1577 task.starting_sent = true;
1578 if let GuestCallKind::StartImplicit(_) = &call.kind {
1579 Waitable::Guest(call.thread.task).set_event(
1580 state,
1581 Some(Event::Subtask {
1582 status: Status::Starting,
1583 }),
1584 )?;
1585 }
1586 }
1587
1588 let instance = state.get_mut(call.thread.task)?.instance;
1589 self.0
1590 .instance_state(instance)
1591 .concurrent_state()
1592 .pending
1593 .insert(call.thread, call.kind);
1594 }
1595 }
1596 WorkItem::WorkerFunction(fun) => {
1597 self.run_on_worker(WorkerItem::Function(fun)).await?;
1598 }
1599 }
1600
1601 Ok(())
1602 }
1603
1604 async fn run_on_worker(self, item: WorkerItem) -> Result<()> {
1606 let worker = if let Some(fiber) = self.0.concurrent_state_mut()?.worker.take() {
1607 fiber
1608 } else {
1609 unsafe {
1628 fiber::make_fiber_unchecked(self.0, move |store| {
1629 loop {
1630 let Some(item) = store.concurrent_state_mut()?.worker_item.take() else {
1631 bail_bug!("worker_item not present when resuming fiber")
1632 };
1633 match item {
1634 WorkerItem::GuestCall(call) => handle_guest_call(store, call)?,
1635 WorkerItem::Function(fun) => fun.into_inner()(store)?,
1636 }
1637
1638 store.suspend(SuspendReason::NeedWork)?;
1639 }
1640 })?
1641 }
1642 };
1643
1644 let worker_item = &mut self.0.concurrent_state_mut()?.worker_item;
1645 assert!(worker_item.is_none());
1646 *worker_item = Some(item);
1647
1648 self.0.resume_fiber(worker).await
1649 }
1650
1651 pub(crate) fn wrap_call<F, R>(self, closure: F) -> impl Future<Output = Result<R>> + 'static
1656 where
1657 T: 'static,
1658 F: FnOnce(&Accessor<T>) -> Pin<Box<dyn Future<Output = Result<R>> + Send + '_>>
1659 + Send
1660 + Sync
1661 + 'static,
1662 R: Send + Sync + 'static,
1663 {
1664 let token = StoreToken::new(self);
1665 async move {
1666 let mut accessor = Accessor::new(token);
1667 closure(&mut accessor).await
1668 }
1669 }
1670
1671 pub fn async_call_stack(&mut self) -> Result<impl Iterator<Item = GuestTaskId>> {
1693 let mut cur = Some(self.0.current_thread()?);
1694 let state = self.0.concurrent_state_mut()?;
1695 Ok(core::iter::from_fn(move || {
1696 while let Some(t) = cur {
1697 cur = state.parent(t);
1698 if let Some(task) = t.guest_task() {
1699 return Some(GuestTaskId(task));
1700 }
1701 }
1702
1703 None
1704 }))
1705 }
1706
1707 pub(crate) async fn start_instance(
1708 &mut self,
1709 instance: ModuleInstance,
1710 ) -> Result<ModuleInstance> {
1711 let (tx, rx) = oneshot::channel();
1712 let token = StoreToken::new(self.as_context_mut());
1713 self.0.queue_task(move |store| {
1714 _ = tx.send(
1715 instance
1716 .start_raw(&mut token.as_context_mut(store))
1717 .map(|()| instance),
1718 );
1719 Ok(())
1720 })?;
1721 self.as_context_mut()
1722 .run_concurrent_trap_on_idle(async |_| {
1723 rx.await
1724 .map_err(|_| format_err!("oneshot channel canceled"))
1725 })
1726 .await??
1727 }
1728}
1729
1730pub type EnteredHostTask = Option<QualifiedThreadId>;
1737
1738impl StoreOpaque {
1739 #[inline]
1743 pub(crate) fn current_thread(&mut self) -> Result<CurrentThread> {
1744 if !self.concurrency_support() {
1746 return Ok(CurrentThread::None);
1747 }
1748
1749 if !self
1752 .vm_store_context_mut()
1753 .current_thread_mut()
1754 .is_deferred()
1755 {
1756 return Ok(self
1757 .concurrent_state_mut_already_forced_current_thread()
1758 .unforced_current_thread);
1759 }
1760
1761 self.force_deferred_current_thread()
1762 }
1763
1764 #[cold]
1767 fn force_deferred_current_thread(&mut self) -> Result<CurrentThread> {
1768 let state = self.concurrent_state_mut_without_forcing_current_thread();
1777 let id = match state.unforced_current_thread.guest_task() {
1778 Some(task) => state.get_mut(task)?.instance.instance,
1779 None => bail_bug!("deferred component-model thread with non-guest base"),
1780 };
1781
1782 let mut frames = Vec::new();
1785 let mut cur = *self.vm_store_context_mut().current_thread_mut();
1786 while let Some(ptr) = cur.as_deferred() {
1787 let deferred = unsafe { ptr.as_non_null().as_ref() };
1792 frames.push((
1793 deferred.callee_async != 0,
1794 deferred.callee_instance,
1795 deferred.saved_context,
1796 ));
1797 cur = deferred.parent;
1798 }
1799
1800 *self.vm_store_context_mut().current_thread_mut() = VMLazyThread::forced();
1804
1805 let current_context = *self.vm_store_context_mut().component_context_mut();
1808
1809 for (callee_async, callee_instance, saved_context) in frames.into_iter().rev() {
1813 *self.vm_store_context_mut().component_context_mut() = saved_context;
1817 let callee = RuntimeInstance {
1818 instance: id,
1819 index: RuntimeComponentInstanceIndex::from_u32(callee_instance),
1820 };
1821 self.enter_guest_sync_call(callee_async, callee)?;
1822 }
1823
1824 *self.vm_store_context_mut().component_context_mut() = current_context;
1826
1827 Ok(self
1828 .concurrent_state_mut_without_forcing_current_thread()
1829 .unforced_current_thread)
1830 }
1831
1832 fn current_guest_thread(&mut self) -> Result<QualifiedThreadId> {
1833 match self.current_thread()?.guest() {
1834 Some(id) => Ok(*id),
1835 None => bail_bug!("current thread is not a guest thread"),
1836 }
1837 }
1838
1839 pub(crate) fn current_materialized_host_task(&mut self) -> Result<Option<TableId<HostTask>>> {
1843 match self.current_thread()? {
1844 CurrentThread::Host(id) => Ok(Some(id)),
1845 CurrentThread::DeferredHost(_) | CurrentThread::None => Ok(None),
1846 _ => bail_bug!("current thread is not a host thread"),
1847 }
1848 }
1849
1850 fn materialize_host_task_id(&mut self) -> Result<Option<TableId<HostTask>>> {
1853 Ok(self
1854 .concurrent_state_mut()?
1855 .materialize_current_host_task_id()?)
1856 }
1857
1858 fn enter_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> {
1859 log::trace!("enter sync-typed call {callee:?}");
1860 let state = self.instance_state(callee).concurrent_state();
1861 let old_do_not_suspend = state.do_not_suspend;
1862 state.do_not_suspend = true;
1863
1864 let thread = self.current_guest_thread()?;
1865 let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?;
1866 if thread.old_do_not_suspend.is_some() {
1867 bail_bug!("current thread already has `old_do_not_suspend` value");
1868 }
1869
1870 thread.old_do_not_suspend = Some(old_do_not_suspend);
1871
1872 Ok(())
1873 }
1874
1875 fn exit_sync_call(&mut self, callee: RuntimeInstance) -> Result<()> {
1876 log::trace!("exit sync-typed call {callee:?}");
1877 let thread = self.current_guest_thread()?;
1878 let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?;
1879 let Some(old_do_not_suspend) = thread.old_do_not_suspend.take() else {
1880 bail_bug!("current thread missing `old_do_not_suspend` value");
1881 };
1882 let state = self.instance_state(callee).concurrent_state();
1883 state.do_not_suspend = old_do_not_suspend;
1884 Ok(())
1885 }
1886
1887 pub(crate) fn enter_guest_sync_call(
1899 &mut self,
1900 callee_async_typed: bool,
1901 callee: RuntimeInstance,
1902 ) -> Result<()> {
1903 log::trace!("enter sync-lifted call {callee:?}");
1904 if !self.concurrency_support() {
1905 return self.enter_call_not_concurrent();
1906 }
1907
1908 let thread = self.current_thread()?;
1909 let caller = if let Some(thread) = thread.guest() {
1910 Caller::Guest { thread: *thread }
1911 } else {
1912 Caller::Host {
1913 tx: None,
1914 host_future_present: false,
1915 caller: self.materialize_host_task_id()?,
1916 }
1917 };
1918 let state = self.concurrent_state_mut()?;
1919 let guest_thread = GuestTask::new(
1920 state,
1921 Box::new(move |_, _| bail_bug!("cannot lower params in sync call")),
1922 LiftResult {
1923 lift: Box::new(move |_, _| bail_bug!("cannot lift result in sync call")),
1924 ty: TypeTupleIndex::reserved_value(),
1925 memory: None,
1926 string_encoding: StringEncoding::Utf8,
1927 },
1928 caller,
1929 None,
1930 callee,
1931 callee_async_typed,
1932 true,
1933 )?;
1934
1935 Instance::from_wasmtime(self, callee.instance).add_guest_thread_to_instance_table(
1936 guest_thread.thread,
1937 self,
1938 callee.index,
1939 )?;
1940 self.set_thread(guest_thread)?;
1941
1942 if !callee_async_typed {
1943 self.enter_sync_call(callee)?;
1944 }
1945
1946 Ok(())
1947 }
1948
1949 pub(crate) fn exit_guest_sync_call(&mut self) -> Result<()> {
1957 if !self.concurrency_support() {
1958 return Ok(self.exit_call_not_concurrent());
1959 }
1960
1961 let thread = match self.current_thread()?.guest() {
1962 Some(t) => *t,
1963 None => bail_bug!("expected task when exiting"),
1964 };
1965 let task = self.concurrent_state_mut()?.get_mut(thread.task)?;
1966 let instance = task.instance;
1967
1968 let caller = match &task.caller {
1969 &Caller::Guest { thread } => thread.into(),
1970 &Caller::Host { caller, .. } => caller
1971 .map(CurrentThread::Host)
1972 .unwrap_or(CurrentThread::None),
1973 };
1974 task.lift_result = None;
1975 task.exited = true;
1976 let async_typed = task.async_typed;
1977
1978 if !async_typed {
1979 self.exit_sync_call(instance)?;
1980 }
1981
1982 self.set_thread(caller)?;
1983
1984 log::trace!("exit sync-lifted call {instance:?}");
1985
1986 if async_typed {
1987 self.switch_or_trap_if_may_not_suspend(instance)?;
1992 }
1993
1994 self.cleanup_thread(thread, instance, CleanupTask::Yes)?;
1995
1996 Ok(())
1997 }
1998
1999 pub(crate) fn host_task_create(&mut self) -> Result<EnteredHostTask> {
2006 if !self.concurrency_support() {
2007 self.enter_call_not_concurrent()?;
2008 return Ok(None);
2009 }
2010 let caller = self.current_guest_thread()?;
2011 log::trace!("new deferred host task with caller {caller:?}");
2012 self.set_thread(CurrentThread::DeferredHost(caller))?;
2013 let state = self.concurrent_state_mut()?;
2014 debug_assert!(state.deferred_host_call_context.is_none());
2015 state.deferred_host_call_context = Some(CallContext::default());
2016 state.debug_assert_deferred_host_invariant();
2017 Ok(Some(caller))
2018 }
2019
2020 pub(crate) fn host_task_delete(
2027 &mut self,
2028 original_task: EnteredHostTask,
2029 materialized_task: Option<TableId<HostTask>>,
2030 ) -> Result<()> {
2031 match original_task {
2032 Some(caller) => {
2033 self.set_thread(caller)?;
2034 if materialized_task.is_none() {
2035 let state = self.concurrent_state_mut()?;
2036 let context = state
2037 .deferred_host_call_context
2038 .take()
2039 .expect("deferred host call context should be present");
2040 debug_assert!(context.is_empty());
2041 state.debug_assert_deferred_host_invariant();
2042 }
2043 log::trace!(
2044 "delete host task with caller {original_task:?} and materialized as {materialized_task:?}"
2045 );
2046 if let Some(task) = materialized_task {
2047 self.concurrent_state_mut()?.delete(task)?;
2048 }
2049 }
2050 None => {
2051 debug_assert!(materialized_task.is_none());
2052 self.exit_call_not_concurrent();
2053 }
2054 }
2055 Ok(())
2056 }
2057
2058 fn instance_state(&mut self, instance: RuntimeInstance) -> &mut InstanceState {
2061 self.component_instance_mut(instance.instance)
2062 .instance_state(instance.index)
2063 }
2064
2065 fn set_thread(&mut self, thread: impl Into<CurrentThread>) -> Result<CurrentThread> {
2071 let thread = thread.into();
2072 let state = self.concurrent_state_mut()?;
2073 state.debug_assert_deferred_host_invariant();
2074 let old_thread = mem::replace(&mut state.unforced_current_thread, thread);
2075
2076 if let Some(old_thread) = old_thread.guest() {
2084 let old_context = *self.vm_store_context_mut().component_context_mut();
2085 self.concurrent_state_mut()?
2086 .get_mut(old_thread.thread)?
2087 .context = old_context;
2088 }
2089 if cfg!(debug_assertions) {
2090 *self.vm_store_context_mut().component_context_mut() =
2091 [u32::MAX; NUM_COMPONENT_CONTEXT_SLOTS];
2092 }
2093 if let Some(thread) = thread.guest() {
2094 let thread = self.concurrent_state_mut()?.get_mut(thread.thread)?;
2095 let context = thread.context;
2096 if cfg!(debug_assertions) {
2097 thread.context = [u32::MAX; NUM_COMPONENT_CONTEXT_SLOTS];
2098 }
2099 *self.vm_store_context_mut().component_context_mut() = context;
2100 }
2101
2102 *self.vm_store_context_mut().current_thread_mut() = if thread.is_none() {
2104 VMLazyThread::none()
2105 } else {
2106 VMLazyThread::forced()
2107 };
2108
2109 Ok(old_thread)
2110 }
2111
2112 fn switch_or_trap_if_may_not_suspend(&mut self, instance: RuntimeInstance) -> Result<()> {
2114 if self.switch_if_may_not_suspend(instance)? {
2115 Ok(())
2116 } else {
2117 Err(Trap::CannotBlockSyncTask.into())
2118 }
2119 }
2120
2121 fn switch_if_may_not_suspend(&mut self, instance: RuntimeInstance) -> Result<bool> {
2125 self.concurrent_state_mut()?;
2129
2130 Ok(!self.concurrency_support()
2131 || !self
2132 .instance_state(instance)
2133 .concurrent_state()
2134 .do_not_suspend
2135 || self
2136 .concurrent_state_mut()?
2137 .promote_instance_local_thread_work_item(instance)?)
2138 }
2139
2140 fn enter_instance(&mut self, instance: RuntimeInstance) {
2144 log::trace!("enter {instance:?}");
2145 self.instance_state(instance)
2146 .concurrent_state()
2147 .do_not_enter = true;
2148 }
2149
2150 fn exit_instance(&mut self, instance: RuntimeInstance) -> Result<()> {
2154 log::trace!("exit {instance:?}");
2155 self.instance_state(instance)
2156 .concurrent_state()
2157 .do_not_enter = false;
2158 self.partition_pending(instance)
2159 }
2160
2161 fn partition_pending(&mut self, instance: RuntimeInstance) -> Result<()> {
2169 for (thread, kind) in
2170 mem::take(&mut self.instance_state(instance).concurrent_state().pending).into_iter()
2171 {
2172 let call = GuestCall { thread, kind };
2173 if call.is_ready(self)? {
2174 self.concurrent_state_mut()?
2175 .push_high_priority(WorkItem::GuestCall { instance, call });
2176 } else {
2177 self.instance_state(instance)
2178 .concurrent_state()
2179 .pending
2180 .insert(call.thread, call.kind);
2181 }
2182 }
2183
2184 if let Some(waker) = self
2185 .concurrent_state_mut()?
2186 .ready_for_concurrent_call_waker
2187 .take()
2188 {
2189 waker.wake();
2190 }
2191
2192 Ok(())
2193 }
2194
2195 pub(crate) fn backpressure_modify(
2197 &mut self,
2198 caller_instance: RuntimeInstance,
2199 modify: impl FnOnce(u16) -> Option<u16>,
2200 ) -> Result<()> {
2201 let state = self.instance_state(caller_instance).concurrent_state();
2202 let old = state.backpressure;
2203 let new = modify(old).ok_or_else(|| Trap::BackpressureOverflow)?;
2204 state.backpressure = new;
2205
2206 if old > 0 && new == 0 {
2207 self.partition_pending(caller_instance)?;
2210 }
2211
2212 Ok(())
2213 }
2214
2215 async fn resume_fiber(&mut self, fiber: StoreFiber<'static>) -> Result<()> {
2218 let old_thread = self.current_thread()?;
2219 log::trace!("resume_fiber: save current thread {old_thread:?}");
2220
2221 let fiber = fiber::resolve_or_release(self, fiber).await?;
2222
2223 self.set_thread(old_thread)?;
2224
2225 let state = self.concurrent_state_mut()?;
2226
2227 if let Some(ot) = old_thread.guest() {
2228 state.get_mut(ot.thread)?.state = GuestThreadState::Running;
2229 }
2230 log::trace!("resume_fiber: restore current thread {old_thread:?}");
2231
2232 if let Some(mut fiber) = fiber {
2233 log::trace!("resume_fiber: suspend reason {:?}", &state.suspend_reason);
2234 let reason = match state.suspend_reason.take() {
2236 Some(r) => r,
2237 None => bail_bug!("suspend reason missing when resuming fiber"),
2238 };
2239 match reason {
2240 SuspendReason::NeedWork => {
2241 if state.worker.is_none() {
2242 state.worker = Some(fiber);
2243 } else {
2244 fiber.dispose(self);
2245 }
2246 }
2247 SuspendReason::Yielding { thread } => {
2248 state.get_mut(thread.thread)?.state = GuestThreadState::Ready { fiber };
2249 let instance = state.get_mut(thread.task)?.instance;
2250 state.push_low_priority(WorkItem::ResumeThread { instance, thread });
2251 }
2252 SuspendReason::ExplicitlySuspending { thread } => {
2253 state.get_mut(thread.thread)?.state = GuestThreadState::Suspended(fiber);
2254 }
2255 SuspendReason::Waiting { set, thread } => {
2256 let old = state
2257 .get_mut(set)?
2258 .waiting
2259 .insert(thread, WaitMode::Fiber(fiber));
2260 assert!(old.is_none());
2261 }
2262 SuspendReason::WaitingForGuestSubtask { caller, callee } => {
2263 let set = state.get_mut(caller.thread)?.sync_call_set;
2264 let old = state
2265 .get_mut(set)?
2266 .waiting
2267 .insert(caller, WaitMode::Caller { fiber, callee });
2268 assert!(old.is_none());
2269 }
2270 };
2271 } else {
2272 log::trace!("resume_fiber: fiber has exited");
2273 }
2274
2275 Ok(())
2276 }
2277
2278 fn suspend(&mut self, reason: SuspendReason) -> Result<()> {
2284 log::trace!("suspend fiber: {reason:?}");
2285
2286 let task = match &reason {
2290 SuspendReason::Yielding { thread, .. }
2291 | SuspendReason::Waiting { thread, .. }
2292 | SuspendReason::WaitingForGuestSubtask { caller: thread, .. }
2293 | SuspendReason::ExplicitlySuspending { thread } => Some(thread.task),
2294 SuspendReason::NeedWork => None,
2295 };
2296
2297 let old_guest_thread = if let Some(task) = task {
2298 let state = self.concurrent_state_mut()?;
2304 if state.switch_item.is_none() {
2305 if let Some(item) = state.get_mut(task)?.switch_item.take() {
2306 state.set_switch_item(item)?;
2307 }
2308 }
2309
2310 self.current_thread()?
2311 } else {
2312 CurrentThread::None
2313 };
2314
2315 let suspend_reason = &mut self.concurrent_state_mut()?.suspend_reason;
2316 assert!(suspend_reason.is_none());
2317 *suspend_reason = Some(reason);
2318
2319 if !self.fiber_async_state_mut().can_block() {
2322 return Err(format_err!("future dropped"));
2323 }
2324
2325 self.with_blocking(|_, cx| cx.suspend(StoreFiberYield::ReleaseStore))?;
2326
2327 if task.is_some() {
2328 self.set_thread(old_guest_thread)?;
2329 }
2330
2331 Ok(())
2332 }
2333
2334 fn wait_for_event(
2335 &mut self,
2336 caller_instance: RuntimeInstance,
2337 waitable: Waitable,
2338 reason: WaitReason,
2339 ) -> Result<()> {
2340 let caller = self.current_guest_thread()?;
2341 let state = self.concurrent_state_mut()?;
2342
2343 waitable.trap_if_in_waitable_set(state)?;
2344
2345 let set = state.get_mut(caller.thread)?.sync_call_set;
2346 waitable.join(state, Some(set))?;
2347
2348 self.switch_or_trap_if_may_not_suspend(caller_instance)?;
2349
2350 self.suspend(match reason {
2351 WaitReason::GuestSubtask(callee) => {
2352 SuspendReason::WaitingForGuestSubtask { caller, callee }
2353 }
2354 WaitReason::Other => SuspendReason::Waiting {
2355 set,
2356 thread: caller,
2357 },
2358 })?;
2359 let state = self.concurrent_state_mut()?;
2360 waitable.join(state, None)
2361 }
2362
2363 fn cleanup_thread(
2385 &mut self,
2386 guest_thread: QualifiedThreadId,
2387 runtime_instance: RuntimeInstance,
2388 cleanup_task: CleanupTask,
2389 ) -> Result<()> {
2390 let state = self.concurrent_state_mut()?;
2391 if let Some(item) = state.get_mut(guest_thread.task)?.switch_item.take() {
2394 state.set_switch_item(item)?;
2395 }
2396 let thread_data = state.get_mut(guest_thread.thread)?;
2397 let sync_call_set = thread_data.sync_call_set;
2398 if let Some(guest_id) = thread_data.instance_rep {
2399 self.instance_state(runtime_instance)
2400 .thread_handle_table()
2401 .guest_thread_remove(guest_id)?;
2402 }
2403 let state = self.concurrent_state_mut()?;
2404
2405 for waitable in mem::take(&mut state.get_mut(sync_call_set)?.ready) {
2407 if let Some(Event::Subtask {
2408 status: Status::Returned | Status::ReturnCancelled,
2409 }) = waitable.common(state)?.event
2410 {
2411 waitable.delete_from(state)?;
2412 }
2413 }
2414
2415 state.delete(guest_thread.thread)?;
2416 state.delete(sync_call_set)?;
2417 let task = state.get_mut(guest_thread.task)?;
2418 task.threads.remove(&guest_thread.thread);
2419
2420 if task.threads.is_empty() && !task.returned_or_cancelled() {
2421 bail!(Trap::NoAsyncResult);
2422 }
2423 let ready_to_delete = task.ready_to_delete();
2424
2425 if !task.decremented_interesting_task_count && task.exited && task.returned_or_cancelled() {
2426 task.decremented_interesting_task_count = true;
2427
2428 debug_assert!(state.interesting_tasks > 0);
2429 state.interesting_tasks -= 1;
2430 if state.interesting_tasks == 0
2431 && let Some(waker) = state.interesting_tasks_empty_waker.take()
2432 {
2433 waker.wake();
2434 }
2435 }
2436
2437 match cleanup_task {
2438 CleanupTask::Yes => {
2439 if ready_to_delete {
2440 Waitable::Guest(guest_thread.task).delete_from(state)?;
2441 }
2442 }
2443 CleanupTask::No => {}
2444 }
2445
2446 Ok(())
2447 }
2448
2449 fn cancel_guest_subtask_without_lowered_parameters(
2462 &mut self,
2463 caller_instance: RuntimeInstance,
2464 guest_task: TableId<GuestTask>,
2465 ) -> Result<()> {
2466 let concurrent_state = self.concurrent_state_mut()?;
2467 let task = concurrent_state.get_mut(guest_task)?;
2468 assert!(!task.already_lowered_parameters());
2469 task.lower_params = None;
2473 task.lift_result = None;
2474 task.exited = true;
2475 let instance = task.instance;
2476
2477 assert_eq!(1, task.threads.len());
2480 let thread = *task.threads.iter().next().unwrap();
2481 self.cleanup_thread(
2482 QualifiedThreadId {
2483 task: guest_task,
2484 thread,
2485 },
2486 caller_instance,
2487 CleanupTask::No,
2488 )?;
2489
2490 let pending = &mut self.instance_state(instance).concurrent_state().pending;
2492 let pending_count = pending.len();
2493 pending.retain(|thread, _| thread.task != guest_task);
2494 if pending.len() == pending_count {
2496 bail!(Trap::SubtaskCancelAfterTerminal);
2497 }
2498 Ok(())
2499 }
2500
2501 pub(crate) fn current_scope(&mut self) -> Result<Option<CurrentScope>> {
2504 if !self.concurrency_support() {
2505 return Ok(self
2506 .current_scope_id_not_concurrent()?
2507 .map(|id| CurrentScope::Id(Scope::Id(id))));
2508 }
2509
2510 Ok(match self.current_thread()? {
2511 CurrentThread::Guest(id) => Some(CurrentScope::Id(Scope::Id(id.task.rep()))),
2512 CurrentThread::GuestTask(id) => Some(CurrentScope::Id(Scope::Id(id.rep()))),
2513 CurrentThread::Host(id) => Some(CurrentScope::Id(Scope::HostId(id.rep()))),
2514 CurrentThread::DeferredHost(_) => Some(CurrentScope::DeferredHost),
2515 CurrentThread::None => return Ok(None),
2516 })
2517 }
2518
2519 pub(crate) fn queue_task(
2520 &mut self,
2521 task: impl FnOnce(&mut dyn VMStore) -> Result<()> + Send + 'static,
2522 ) -> Result<()> {
2523 self.concurrent_state_mut()?
2524 .push_high_priority(WorkItem::WorkerFunction(AlwaysMut::new(Box::new(task))));
2525 Ok(())
2526 }
2527
2528 fn any_may_not_suspend(&mut self) -> Result<bool> {
2537 Ok(self
2545 .concurrent_state_mut()?
2546 .table
2547 .get_mut()
2548 .iter_mut()
2549 .filter_map(|entry| {
2550 if let Some(task) = entry.downcast_ref::<GuestTask>() {
2551 Some(task.instance)
2552 } else {
2553 None
2554 }
2555 })
2556 .collect::<Vec<_>>()
2557 .into_iter()
2558 .any(|instance| {
2559 self.instance_state(instance)
2560 .concurrent_state()
2561 .do_not_suspend
2562 }))
2563 }
2564}
2565
2566enum CleanupTask {
2567 Yes,
2568 No,
2569}
2570
2571impl Instance {
2572 fn get_event(
2575 self,
2576 store: &mut StoreOpaque,
2577 guest_task: TableId<GuestTask>,
2578 set: Option<TableId<WaitableSet>>,
2579 cancellable: bool,
2580 ) -> Result<Option<(Event, Option<(Waitable, u32)>)>> {
2581 let state = store.concurrent_state_mut()?;
2582
2583 let task = state.get_mut(guest_task)?;
2584 let event = &mut task.event;
2585 if let Some(ev) = event
2586 && (cancellable || !matches!(ev, Event::Cancelled))
2587 {
2588 log::trace!("deliver event {ev:?} to {guest_task:?}");
2589
2590 if matches!(ev, Event::Cancelled) {
2591 task.cancel_request_delivered = true;
2592 }
2593
2594 let ev = *ev;
2595 *event = None;
2596 return Ok(Some((ev, None)));
2597 }
2598
2599 let set = match set {
2600 Some(set) => set,
2601 None => return Ok(None),
2602 };
2603 let waitable = match state.get_mut(set)?.ready.pop_first() {
2604 Some(v) => v,
2605 None => return Ok(None),
2606 };
2607
2608 let common = waitable.common(state)?;
2609 let handle = match common.handle {
2610 Some(h) => h,
2611 None => bail_bug!("handle not set when delivering event"),
2612 };
2613 let event = match common.event.take() {
2614 Some(e) => e,
2615 None => bail_bug!("event not set when delivering event"),
2616 };
2617
2618 log::trace!(
2619 "deliver event {event:?} to {guest_task:?} for {waitable:?} (handle {handle}); set {set:?}"
2620 );
2621
2622 waitable.on_delivery(store, self, event)?;
2623
2624 Ok(Some((event, Some((waitable, handle)))))
2625 }
2626
2627 fn handle_callback_code(
2633 self,
2634 store: &mut StoreOpaque,
2635 guest_thread: QualifiedThreadId,
2636 runtime_instance: RuntimeComponentInstanceIndex,
2637 code: u32,
2638 ) -> Result<()> {
2639 let (code, set) = unpack_callback_code(code);
2640
2641 log::trace!("received callback code from {guest_thread:?}: {code} (set: {set})");
2642
2643 let state = store.concurrent_state_mut()?;
2644
2645 if let Some(item) = state.get_mut(guest_thread.task)?.switch_item.take() {
2646 state.set_switch_item(item)?;
2647 }
2648
2649 let get_set = |store: &mut StoreOpaque, handle| -> Result<_> {
2650 let set = store
2651 .instance_state(self.runtime_instance(runtime_instance))
2652 .handle_table()
2653 .waitable_set_rep(handle)?;
2654
2655 Ok(TableId::<WaitableSet>::new(set))
2656 };
2657
2658 match code {
2659 callback_code::EXIT => {
2660 log::trace!("implicit thread {guest_thread:?} completed");
2661 let task = store.concurrent_state_mut()?.get_mut(guest_thread.task)?;
2662 task.exited = true;
2663 task.callback = None;
2664
2665 let runtime_instance = self.runtime_instance(runtime_instance);
2666
2667 store.switch_or_trap_if_may_not_suspend(runtime_instance)?;
2672
2673 store.cleanup_thread(guest_thread, runtime_instance, CleanupTask::Yes)?;
2674 }
2675 callback_code::YIELD => {
2676 let task = state.get_mut(guest_thread.task)?;
2677 if let Some(event) = task.event {
2682 assert!(matches!(event, Event::None | Event::Cancelled));
2683 } else {
2684 task.event = Some(Event::None);
2685 }
2686 let call = GuestCall {
2687 thread: guest_thread,
2688 kind: GuestCallKind::DeliverEvent {
2689 instance: self,
2690 set: None,
2691 },
2692 };
2693 state.push_low_priority(WorkItem::GuestCall {
2696 instance: self.runtime_instance(runtime_instance),
2697 call,
2698 });
2699 }
2700 callback_code::WAIT => {
2701 let set = get_set(store, set)?;
2702 let state = store.concurrent_state_mut()?;
2703
2704 if state.get_mut(guest_thread.task)?.event.is_some()
2705 || !state.get_mut(set)?.ready.is_empty()
2706 {
2707 state.push_high_priority(WorkItem::GuestCall {
2709 instance: self.runtime_instance(runtime_instance),
2710 call: GuestCall {
2711 thread: guest_thread,
2712 kind: GuestCallKind::DeliverEvent {
2713 instance: self,
2714 set: Some(set),
2715 },
2716 },
2717 });
2718 } else {
2719 let old = state
2727 .get_mut(guest_thread.thread)?
2728 .wake_on_cancel
2729 .replace(set);
2730 if !old.is_none() {
2731 bail_bug!("thread unexpectedly had wake_on_cancel set");
2732 }
2733 let old = state
2734 .get_mut(set)?
2735 .waiting
2736 .insert(guest_thread, WaitMode::Callback(self));
2737 if !old.is_none() {
2738 bail_bug!("set's waiting set already had this thread registered");
2739 }
2740 }
2741 }
2742 _ => bail!(Trap::UnsupportedCallbackCode),
2743 }
2744
2745 Ok(())
2746 }
2747
2748 unsafe fn stage_call<T: 'static>(
2755 self,
2756 mut store: StoreContextMut<T>,
2757 guest_thread: QualifiedThreadId,
2758 callee: SendSyncPtr<VMFuncRef>,
2759 param_count: usize,
2760 result_count: usize,
2761 async_: bool,
2762 callback: Option<SendSyncPtr<VMFuncRef>>,
2763 post_return: Option<SendSyncPtr<VMFuncRef>>,
2764 host_caller: bool,
2765 ) -> Result<()> {
2766 unsafe fn make_call<T: 'static>(
2781 store: StoreContextMut<T>,
2782 guest_thread: QualifiedThreadId,
2783 callee: SendSyncPtr<VMFuncRef>,
2784 param_count: usize,
2785 result_count: usize,
2786 ) -> impl FnOnce(&mut dyn VMStore) -> Result<[MaybeUninit<ValRaw>; MAX_FLAT_PARAMS]>
2787 + Send
2788 + Sync
2789 + 'static
2790 + use<T> {
2791 let token = StoreToken::new(store);
2792 move |store: &mut dyn VMStore| {
2793 let mut storage = [MaybeUninit::uninit(); MAX_FLAT_PARAMS];
2794
2795 store
2796 .concurrent_state_mut()?
2797 .get_mut(guest_thread.thread)?
2798 .state = GuestThreadState::Running;
2799 let task = store.concurrent_state_mut()?.get_mut(guest_thread.task)?;
2800 let lower = match task.lower_params.take() {
2801 Some(l) => l,
2802 None => bail_bug!("lower_params missing"),
2803 };
2804
2805 lower(store, &mut storage[..param_count])?;
2806
2807 let mut store = token.as_context_mut(store);
2808
2809 unsafe {
2812 crate::Func::call_unchecked_raw(
2813 &mut store,
2814 callee.as_non_null(),
2815 NonNull::new(
2816 &mut storage[..param_count.max(result_count)]
2817 as *mut [MaybeUninit<ValRaw>] as _,
2818 )
2819 .unwrap(),
2820 )?;
2821 }
2822
2823 Ok(storage)
2824 }
2825 }
2826
2827 let call = unsafe {
2831 make_call(
2832 store.as_context_mut(),
2833 guest_thread,
2834 callee,
2835 param_count,
2836 result_count,
2837 )
2838 };
2839
2840 let callee_instance = store
2841 .0
2842 .concurrent_state_mut()?
2843 .get_mut(guest_thread.task)?
2844 .instance;
2845
2846 let fun = if callback.is_some() {
2847 assert!(async_);
2848
2849 Box::new(move |store: &mut dyn VMStore| {
2850 self.add_guest_thread_to_instance_table(
2851 guest_thread.thread,
2852 store,
2853 callee_instance.index,
2854 )?;
2855 let old_thread = store.set_thread(guest_thread)?;
2856 log::trace!(
2857 "stackless call: replaced {old_thread:?} with {guest_thread:?} as current thread"
2858 );
2859
2860 store.enter_instance(callee_instance);
2861
2862 let storage = call(store)?;
2869
2870 store.exit_instance(callee_instance)?;
2871
2872 store.set_thread(old_thread)?;
2873 let state = store.concurrent_state_mut()?;
2874 if let Some(t) = old_thread.guest() {
2875 state.get_mut(t.thread)?.state = GuestThreadState::Running;
2876 }
2877 log::trace!("stackless call: restored {old_thread:?} as current thread");
2878
2879 let code = unsafe { storage[0].assume_init() }.get_i32() as u32;
2882
2883 self.handle_callback_code(store, guest_thread, callee_instance.index, code)
2884 }) as Box<dyn FnOnce(&mut dyn VMStore) -> Result<()> + Send + Sync>
2885 } else {
2886 let token = StoreToken::new(store.as_context_mut());
2887 Box::new(move |store: &mut dyn VMStore| {
2888 self.add_guest_thread_to_instance_table(
2889 guest_thread.thread,
2890 store,
2891 callee_instance.index,
2892 )?;
2893 let old_thread = store.set_thread(guest_thread)?;
2894 log::trace!(
2895 "sync/async-stackful call: replaced {old_thread:?} with {guest_thread:?} as current thread",
2896 );
2897 let flags = self.id().get(store).instance_flags(callee_instance.index);
2898
2899 let callee_async_typed = store
2900 .concurrent_state_mut()?
2901 .get_mut(guest_thread.task)?
2902 .async_typed;
2903
2904 if !async_ && callee_async_typed {
2908 store.enter_instance(callee_instance);
2909 }
2910
2911 if !callee_async_typed {
2912 store.enter_sync_call(callee_instance)?;
2913 }
2914
2915 let storage = call(store)?;
2922
2923 if !callee_async_typed {
2924 store.exit_sync_call(callee_instance)?;
2925 }
2926
2927 if !async_ {
2928 if callee_async_typed {
2934 store.exit_instance(callee_instance)?;
2935 }
2936
2937 let lift = {
2938 let state = store.concurrent_state_mut()?;
2939 if !state.get_mut(guest_thread.task)?.result.is_none() {
2940 bail_bug!("task has already produced a result");
2941 }
2942
2943 match state.get_mut(guest_thread.task)?.lift_result.take() {
2944 Some(lift) => lift,
2945 None => bail_bug!("lift_result field is missing"),
2946 }
2947 };
2948
2949 let result = (lift.lift)(store, unsafe {
2952 mem::transmute::<&[MaybeUninit<ValRaw>], &[ValRaw]>(
2953 &storage[..result_count],
2954 )
2955 })?;
2956
2957 let post_return_arg = match result_count {
2958 0 => ValRaw::i32(0),
2959 1 => unsafe { storage[0].assume_init() },
2962 _ => unreachable!(),
2963 };
2964
2965 unsafe {
2966 call_post_return(
2967 token.as_context_mut(store),
2968 post_return.map(|v| v.as_non_null()),
2969 post_return_arg,
2970 flags,
2971 )?;
2972 }
2973
2974 self.task_complete(store, guest_thread.task, result, Status::Returned)?;
2975 }
2976
2977 store.set_thread(old_thread)?;
2978
2979 store
2980 .concurrent_state_mut()?
2981 .get_mut(guest_thread.task)?
2982 .exited = true;
2983
2984 log::trace!(
2985 "clean up thread; async lifted? {async_} async typed? {callee_async_typed}"
2986 );
2987
2988 if callee_async_typed {
2989 store.switch_or_trap_if_may_not_suspend(callee_instance)?;
2994 }
2995
2996 store.cleanup_thread(guest_thread, callee_instance, CleanupTask::Yes)?;
2998 Ok(())
2999 })
3000 };
3001
3002 store.0.concurrent_state_mut()?.push_work_item(
3003 WorkItem::GuestCall {
3004 instance: callee_instance,
3005 call: GuestCall {
3006 thread: guest_thread,
3007 kind: GuestCallKind::StartImplicit(fun),
3008 },
3009 },
3010 if host_caller {
3011 Priority::High
3012 } else {
3013 Priority::Switch
3014 },
3015 )?;
3016
3017 Ok(())
3018 }
3019
3020 unsafe fn prepare_call<T: 'static>(
3033 self,
3034 mut store: StoreContextMut<T>,
3035 start: NonNull<VMFuncRef>,
3036 return_: NonNull<VMFuncRef>,
3037 caller_instance: RuntimeComponentInstanceIndex,
3038 callee_instance: RuntimeComponentInstanceIndex,
3039 task_return_type: TypeTupleIndex,
3040 callee_async_typed: bool,
3041 memory: *mut VMMemoryDefinition,
3042 string_encoding: StringEncoding,
3043 caller_info: CallerInfo,
3044 ) -> Result<()> {
3045 enum ResultInfo {
3046 Heap { results: u32 },
3047 Stack { result_count: u32 },
3048 }
3049
3050 let result_info = match &caller_info {
3051 CallerInfo::Async {
3052 has_result: true,
3053 params,
3054 } => ResultInfo::Heap {
3055 results: match params.last() {
3056 Some(r) => r.get_u32(),
3057 None => bail_bug!("retptr missing"),
3058 },
3059 },
3060 CallerInfo::Async {
3061 has_result: false, ..
3062 } => ResultInfo::Stack { result_count: 0 },
3063 CallerInfo::Sync {
3064 result_count,
3065 params,
3066 } if *result_count > u32::try_from(MAX_FLAT_RESULTS)? => ResultInfo::Heap {
3067 results: match params.last() {
3068 Some(r) => r.get_u32(),
3069 None => bail_bug!("arg ptr missing"),
3070 },
3071 },
3072 CallerInfo::Sync { result_count, .. } => ResultInfo::Stack {
3073 result_count: *result_count,
3074 },
3075 };
3076
3077 let sync_caller = matches!(caller_info, CallerInfo::Sync { .. });
3078
3079 let start = SendSyncPtr::new(start);
3083 let return_ = SendSyncPtr::new(return_);
3084 let token = StoreToken::new(store.as_context_mut());
3085 let old_thread = store.0.current_guest_thread()?;
3086 let state = store.0.concurrent_state_mut()?;
3087
3088 debug_assert_eq!(
3089 state.get_mut(old_thread.task)?.instance,
3090 self.runtime_instance(caller_instance)
3091 );
3092
3093 let guest_thread = GuestTask::new(
3094 state,
3095 Box::new(move |store, dst| {
3096 let mut store = token.as_context_mut(store);
3097 assert!(dst.len() <= MAX_FLAT_PARAMS);
3098 let mut src = [MaybeUninit::uninit(); MAX_FLAT_PARAMS + 1];
3100 let count = match caller_info {
3101 CallerInfo::Async { params, has_result } => {
3105 let params = ¶ms[..params.len() - usize::from(has_result)];
3106 for (param, src) in params.iter().zip(&mut src) {
3107 src.write(*param);
3108 }
3109 params.len()
3110 }
3111
3112 CallerInfo::Sync { params, .. } => {
3114 for (param, src) in params.iter().zip(&mut src) {
3115 src.write(*param);
3116 }
3117 params.len()
3118 }
3119 };
3120 unsafe {
3127 crate::Func::call_unchecked_raw(
3128 &mut store,
3129 start.as_non_null(),
3130 NonNull::new(
3131 &mut src[..count.max(dst.len())] as *mut [MaybeUninit<ValRaw>] as _,
3132 )
3133 .unwrap(),
3134 )?;
3135 }
3136 dst.copy_from_slice(&src[..dst.len()]);
3137 let task = store.0.current_guest_thread()?.task;
3138 let state = store.0.concurrent_state_mut()?;
3139 Waitable::Guest(task).set_event(
3140 state,
3141 Some(Event::Subtask {
3142 status: Status::Started,
3143 }),
3144 )?;
3145 Ok(())
3146 }),
3147 LiftResult {
3148 lift: Box::new(move |store, src| {
3149 let mut store = token.as_context_mut(store);
3152 let mut my_src = src.to_owned(); if let ResultInfo::Heap { results } = &result_info {
3154 my_src.push(ValRaw::u32(*results));
3155 }
3156
3157 unsafe {
3164 crate::Func::call_unchecked_raw(
3165 &mut store,
3166 return_.as_non_null(),
3167 my_src.as_mut_slice().into(),
3168 )?;
3169 }
3170
3171 let thread = store.0.current_guest_thread()?;
3172 let state = store.0.concurrent_state_mut()?;
3173 if sync_caller {
3174 state.get_mut(thread.task)?.sync_result = SyncResult::Produced(
3175 if let ResultInfo::Stack { result_count } = &result_info {
3176 match result_count {
3177 0 => None,
3178 1 => Some(my_src[0]),
3179 _ => unreachable!(),
3180 }
3181 } else {
3182 None
3183 },
3184 );
3185 }
3186 Ok(Box::new(DummyResult) as Box<dyn Any + Send + Sync>)
3187 }),
3188 ty: task_return_type,
3189 memory: NonNull::new(memory).map(SendSyncPtr::new),
3190 string_encoding,
3191 },
3192 Caller::Guest { thread: old_thread },
3193 None,
3194 self.runtime_instance(callee_instance),
3195 callee_async_typed,
3196 false,
3199 )?;
3200
3201 store.0.set_thread(guest_thread)?;
3204 log::trace!("pushed {guest_thread:?} as current thread; old thread was {old_thread:?}");
3205
3206 Ok(())
3207 }
3208
3209 unsafe fn call_callback<T>(
3214 self,
3215 mut store: StoreContextMut<T>,
3216 function: SendSyncPtr<VMFuncRef>,
3217 event: Event,
3218 handle: u32,
3219 ) -> Result<u32> {
3220 let (ordinal, result) = event.parts();
3221 let params = &mut [
3222 ValRaw::u32(ordinal),
3223 ValRaw::u32(handle),
3224 ValRaw::u32(result),
3225 ];
3226 unsafe {
3231 crate::Func::call_unchecked_raw(
3232 &mut store,
3233 function.as_non_null(),
3234 params.as_mut_slice().into(),
3235 )?;
3236 }
3237 Ok(params[0].get_u32())
3238 }
3239
3240 unsafe fn start_call<T: 'static>(
3253 self,
3254 mut store: StoreContextMut<T>,
3255 callback: *mut VMFuncRef,
3256 post_return: *mut VMFuncRef,
3257 callee: NonNull<VMFuncRef>,
3258 param_count: u32,
3259 result_count: u32,
3260 flags: u32,
3261 storage: Option<&mut [MaybeUninit<ValRaw>]>,
3262 ) -> Result<u32> {
3263 let token = StoreToken::new(store.as_context_mut());
3264 let async_caller = storage.is_none();
3265 let guest_thread = store.0.current_guest_thread()?;
3266 let state = store.0.concurrent_state_mut()?;
3267
3268 if !state.event_loop_running {
3269 bail_bug!("Instance::start_call called without a running event loop");
3270 }
3271
3272 let callee = SendSyncPtr::new(callee);
3273 let param_count = usize::try_from(param_count)?;
3274 assert!(param_count <= MAX_FLAT_PARAMS);
3275 let result_count = usize::try_from(result_count)?;
3276 assert!(result_count <= MAX_FLAT_RESULTS);
3277
3278 let task = state.get_mut(guest_thread.task)?;
3279 let callee_async_typed = task.async_typed;
3280 let callee_instance = task.instance;
3281
3282 task.async_lifted = (flags & START_FLAG_ASYNC_CALLEE) != 0;
3283
3284 if let Some(callback) = NonNull::new(callback) {
3285 let callback = SendSyncPtr::new(callback);
3289 task.callback = Some(Box::new(move |store, event, handle| {
3290 let store = token.as_context_mut(store);
3291 unsafe { self.call_callback::<T>(store, callback, event, handle) }
3292 }));
3293 }
3294
3295 let Caller::Guest { thread: caller } = &task.caller else {
3296 bail_bug!("start_call unexpectedly invoked for host->guest call");
3299 };
3300 let caller = *caller;
3301 let caller_instance = state.get_mut(caller.task)?.instance;
3302
3303 unsafe {
3305 self.stage_call(
3306 store.as_context_mut(),
3307 guest_thread,
3308 callee,
3309 param_count,
3310 result_count,
3311 (flags & START_FLAG_ASYNC_CALLEE) != 0,
3312 NonNull::new(callback).map(SendSyncPtr::new),
3313 NonNull::new(post_return).map(SendSyncPtr::new),
3314 false,
3315 )?;
3316 }
3317
3318 let old_do_not_suspend = if callee_async_typed {
3319 let state = store.0.instance_state(callee_instance).concurrent_state();
3326 let old_do_not_suspend = state.do_not_suspend;
3327 state.do_not_suspend = false;
3328 Some(old_do_not_suspend)
3329 } else {
3330 None
3331 };
3332
3333 let state = store.0.concurrent_state_mut()?;
3334
3335 let guest_waitable = Waitable::Guest(guest_thread.task);
3338 let old_set = guest_waitable.common(state)?.set;
3339 let set = state.get_mut(caller.thread)?.sync_call_set;
3340 guest_waitable.join(state, Some(set))?;
3341
3342 store.0.set_thread(CurrentThread::None)?;
3343
3344 let (status, waitable) = loop {
3360 store.0.suspend(SuspendReason::WaitingForGuestSubtask {
3361 caller,
3362 callee: guest_thread.task,
3363 })?;
3364
3365 if let Some(old_do_not_suspend) = old_do_not_suspend {
3366 store
3367 .0
3368 .instance_state(callee_instance)
3369 .concurrent_state()
3370 .do_not_suspend = old_do_not_suspend;
3371 }
3372
3373 let state = store.0.concurrent_state_mut()?;
3374
3375 log::trace!("taking event for {:?}", guest_thread.task);
3376 let event = guest_waitable.take_event(state)?;
3377 let Some(Event::Subtask { status }) = event else {
3378 bail_bug!("subtasks should only get subtask events, got {event:?}")
3379 };
3380
3381 log::trace!("status {status:?} for {:?}", guest_thread.task);
3382
3383 if status == Status::Returned {
3384 break (status, None);
3386 } else if async_caller {
3387 let handle = store
3391 .0
3392 .instance_state(caller_instance)
3393 .handle_table()
3394 .subtask_insert_guest(guest_thread.task.rep())?;
3395 store
3396 .0
3397 .concurrent_state_mut()?
3398 .get_mut(guest_thread.task)?
3399 .common
3400 .handle = Some(handle);
3401 break (status, Some(handle));
3402 } else {
3403 store.0.switch_or_trap_if_may_not_suspend(caller_instance)?;
3407 }
3408 };
3409
3410 guest_waitable.join(store.0.concurrent_state_mut()?, old_set)?;
3411
3412 store.0.set_thread(caller)?;
3414 store
3415 .0
3416 .concurrent_state_mut()?
3417 .get_mut(caller.thread)?
3418 .state = GuestThreadState::Running;
3419 log::trace!("popped current thread {guest_thread:?}; new thread is {caller:?}");
3420
3421 if let Some(storage) = storage {
3422 let state = store.0.concurrent_state_mut()?;
3426 let task = state.get_mut(guest_thread.task)?;
3427 if let Some(result) = task.sync_result.take()? {
3428 if let Some(result) = result {
3429 storage[0] = MaybeUninit::new(result);
3430 }
3431
3432 if task.exited && task.ready_to_delete() {
3433 Waitable::Guest(guest_thread.task).delete_from(state)?;
3434 }
3435 }
3436 }
3437
3438 Ok(status.pack(waitable))
3439 }
3440
3441 pub(crate) fn first_poll<T: 'static, R: Send + 'static>(
3457 self,
3458 mut store: StoreContextMut<'_, T>,
3459 host_task: EnteredHostTask,
3460 future: impl Future<Output = Result<R>> + Send + 'static,
3461 lower: impl FnOnce(StoreContextMut<T>, Option<R>, bool, Option<TableId<HostTask>>) -> Result<()>
3462 + Send
3463 + 'static,
3464 ) -> Result<u32> {
3465 let token = StoreToken::new(store.as_context_mut());
3466
3467 let (join_handle, future) = JoinHandle::run(future);
3470 let mut future = Box::pin(future);
3471
3472 let poll = tls::set(store.0, || {
3477 future
3478 .as_mut()
3479 .poll(&mut Context::from_waker(&Waker::noop()))
3480 });
3481
3482 match poll {
3483 Poll::Ready(result) => {
3485 let result = result.transpose()?;
3486 let task = store.0.current_materialized_host_task()?;
3489 lower(store.as_context_mut(), result, true, task)?;
3490 return Ok(Status::Returned.pack(None));
3491 }
3492
3493 Poll::Pending => {}
3495 }
3496
3497 let Some(task) = store.0.materialize_host_task_id()? else {
3501 bail_bug!("current thread is not a host thread")
3502 };
3503 {
3504 let state = &mut store.0.concurrent_state_mut()?.get_mut(task)?.state;
3505 assert!(matches!(state, HostTaskState::CalleeStarted));
3506 *state = HostTaskState::CalleeRunning(join_handle);
3507 }
3508
3509 let future = Box::pin(async move {
3517 let result = match run_with_host_task_set(task, future).await? {
3518 Some(result) => Some(result?),
3519 None => None,
3520 };
3521 let on_complete = move |store: &mut dyn VMStore| {
3522 let mut store = token.as_context_mut(store);
3526 let old = store.0.set_thread(task)?;
3527
3528 let status = if result.is_some() {
3529 Status::Returned
3530 } else {
3531 Status::ReturnCancelled
3532 };
3533
3534 lower(store.as_context_mut(), result, false, Some(task))?;
3535 let state = store.0.concurrent_state_mut()?;
3536 match &mut state.get_mut(task)?.state {
3537 HostTaskState::CalleeDone { .. } => {}
3540
3541 other => *other = HostTaskState::CalleeDone { cancelled: false },
3543 }
3544 Waitable::Host(task).set_event(state, Some(Event::Subtask { status }))?;
3545
3546 store.0.set_thread(old)?;
3547 Ok(())
3548 };
3549
3550 tls::get(move |store| {
3555 store
3556 .concurrent_state_mut()?
3557 .push_high_priority(WorkItem::WorkerFunction(AlwaysMut::new(Box::new(
3558 on_complete,
3559 ))));
3560 Ok(())
3561 })
3562 });
3563
3564 let caller = match host_task {
3567 Some(caller) => caller,
3568 None => bail_bug!("host task wasn't created but should have been"),
3569 };
3570 let state = store.0.concurrent_state_mut()?;
3571 state.push_future(future);
3572 let instance = state.get_mut(caller.task)?.instance;
3573 let handle = store
3574 .0
3575 .instance_state(instance)
3576 .handle_table()
3577 .subtask_insert_host(task.rep())?;
3578 store.0.concurrent_state_mut()?.get_mut(task)?.common.handle = Some(handle);
3579 log::trace!("assign {task:?} handle {handle} for {caller:?} instance {instance:?}");
3580
3581 store.0.set_thread(caller)?;
3585 Ok(Status::Started.pack(Some(handle)))
3586 }
3587
3588 pub(crate) fn task_return(
3591 self,
3592 store: &mut dyn VMStore,
3593 ty: TypeTupleIndex,
3594 options: OptionsIndex,
3595 storage: &[ValRaw],
3596 ) -> Result<()> {
3597 let guest_thread = store.current_guest_thread()?;
3598 let state = store.concurrent_state_mut()?;
3599 let lift = state
3600 .get_mut(guest_thread.task)?
3601 .lift_result
3602 .take()
3603 .ok_or_else(|| Trap::TaskCancelOrReturnTwice)?;
3604 if !state.get_mut(guest_thread.task)?.result.is_none() {
3605 bail_bug!("task result unexpectedly already set");
3606 }
3607
3608 let CanonicalOptions {
3609 string_encoding,
3610 data_model,
3611 ..
3612 } = &self.id().get(store).component().env_component().options[options];
3613
3614 let invalid = ty != lift.ty
3615 || string_encoding != &lift.string_encoding
3616 || match data_model {
3617 CanonicalOptionsDataModel::LinearMemory(opts) => match opts.memory {
3618 Some(memory) => {
3619 let expected = lift.memory.map(|v| v.as_ptr()).unwrap_or(ptr::null_mut());
3620 let actual = self.id().get(store).runtime_memory(memory);
3621 expected != actual.as_ptr()
3622 }
3623 None => false,
3626 },
3627 CanonicalOptionsDataModel::Gc { .. } => true,
3629 };
3630
3631 if invalid {
3632 bail!(Trap::TaskReturnInvalid);
3633 }
3634
3635 log::trace!("task.return for {guest_thread:?}");
3636
3637 let result = (lift.lift)(store, storage)?;
3638 self.task_complete(store, guest_thread.task, result, Status::Returned)
3639 }
3640
3641 pub(crate) fn task_cancel(self, store: &mut StoreOpaque) -> Result<()> {
3643 let guest_thread = store.current_guest_thread()?;
3644 let state = store.concurrent_state_mut()?;
3645 let task = state.get_mut(guest_thread.task)?;
3646 if !task.cancel_request_delivered {
3647 bail!(Trap::TaskCancelNotCancelled);
3648 }
3649 _ = task
3650 .lift_result
3651 .take()
3652 .ok_or_else(|| Trap::TaskCancelOrReturnTwice)?;
3653
3654 if !task.result.is_none() {
3655 bail_bug!("task result should not bet set yet");
3656 }
3657
3658 log::trace!("task.cancel for {guest_thread:?}");
3659
3660 self.task_complete(
3661 store,
3662 guest_thread.task,
3663 Box::new(DummyResult),
3664 Status::ReturnCancelled,
3665 )
3666 }
3667
3668 fn task_complete(
3674 self,
3675 store: &mut StoreOpaque,
3676 guest_task: TableId<GuestTask>,
3677 result: Box<dyn Any + Send + Sync>,
3678 status: Status,
3679 ) -> Result<()> {
3680 store
3681 .component_resource_tables(Some(self))?
3682 .validate_scope_exit()?;
3683
3684 let state = store.concurrent_state_mut()?;
3685 let task = state.get_mut(guest_task)?;
3686
3687 if let Caller::Host { tx, .. } = &mut task.caller {
3688 if let Some(tx) = tx.take() {
3689 _ = tx.send(result);
3690 }
3691 } else {
3692 task.result = Some(result);
3693 Waitable::Guest(guest_task).set_event(state, Some(Event::Subtask { status }))?;
3694 }
3695
3696 Ok(())
3697 }
3698
3699 pub(crate) fn waitable_set_new(
3701 self,
3702 store: &mut StoreOpaque,
3703 caller_instance: RuntimeComponentInstanceIndex,
3704 ) -> Result<u32> {
3705 let set = store.concurrent_state_mut()?.push(WaitableSet::default())?;
3706 let handle = store
3707 .instance_state(self.runtime_instance(caller_instance))
3708 .handle_table()
3709 .waitable_set_insert(set.rep())?;
3710 log::trace!("new waitable set {set:?} (handle {handle})");
3711 Ok(handle)
3712 }
3713
3714 pub(crate) fn waitable_set_drop(
3716 self,
3717 store: &mut StoreOpaque,
3718 caller_instance: RuntimeComponentInstanceIndex,
3719 set: u32,
3720 ) -> Result<()> {
3721 let rep = store
3722 .instance_state(self.runtime_instance(caller_instance))
3723 .handle_table()
3724 .waitable_set_remove(set)?;
3725
3726 log::trace!("drop waitable set {rep} (handle {set})");
3727
3728 if !store
3732 .concurrent_state_mut()?
3733 .get_mut(TableId::<WaitableSet>::new(rep))?
3734 .waiting
3735 .is_empty()
3736 {
3737 bail!(Trap::WaitableSetDropHasWaiters);
3738 }
3739
3740 store
3741 .concurrent_state_mut()?
3742 .delete(TableId::<WaitableSet>::new(rep))?;
3743
3744 Ok(())
3745 }
3746
3747 pub(crate) fn waitable_join(
3749 self,
3750 store: &mut StoreOpaque,
3751 caller_instance: RuntimeComponentInstanceIndex,
3752 waitable_handle: u32,
3753 set_handle: u32,
3754 ) -> Result<()> {
3755 let mut instance = self.id().get_mut(store);
3756 let waitable =
3757 Waitable::from_instance(instance.as_mut(), caller_instance, waitable_handle)?;
3758
3759 let set = if set_handle == 0 {
3760 None
3761 } else {
3762 let set = instance.instance_states().0[caller_instance]
3763 .handle_table()
3764 .waitable_set_rep(set_handle)?;
3765
3766 let state = store.concurrent_state_mut()?;
3767 if let Some(old) = waitable.common(state)?.set
3768 && state.get_mut(old)?.is_sync_call_set
3769 {
3770 bail!(Trap::WaitableSyncAndAsync);
3771 }
3772
3773 Some(TableId::<WaitableSet>::new(set))
3774 };
3775
3776 log::trace!(
3777 "waitable {waitable:?} (handle {waitable_handle}) join set {set:?} (handle {set_handle})",
3778 );
3779
3780 waitable.join(store.concurrent_state_mut()?, set)
3781 }
3782
3783 pub(crate) fn subtask_drop(
3785 self,
3786 store: &mut StoreOpaque,
3787 caller_instance: RuntimeComponentInstanceIndex,
3788 task_id: u32,
3789 ) -> Result<()> {
3790 self.waitable_join(store, caller_instance, task_id, 0)?;
3791
3792 let (rep, is_host) = store
3793 .instance_state(self.runtime_instance(caller_instance))
3794 .handle_table()
3795 .subtask_remove(task_id)?;
3796
3797 let concurrent_state = store.concurrent_state_mut()?;
3798 let (waitable, delete) = if is_host {
3799 let id = TableId::<HostTask>::new(rep);
3800 let task = concurrent_state.get_mut(id)?;
3801 match &task.state {
3802 HostTaskState::CalleeRunning(_) => bail!(Trap::SubtaskDropNotResolved),
3803 HostTaskState::CalleeDone { .. } => {}
3804 HostTaskState::CalleeStarted | HostTaskState::CalleeFinished(_) => {
3805 bail_bug!("invalid state for callee in `subtask.drop`")
3806 }
3807 }
3808 (Waitable::Host(id), true)
3809 } else {
3810 let id = TableId::<GuestTask>::new(rep);
3811 let task = concurrent_state.get_mut(id)?;
3812 if task.lift_result.is_some() {
3813 bail!(Trap::SubtaskDropNotResolved);
3814 }
3815 (
3816 Waitable::Guest(id),
3817 concurrent_state.get_mut(id)?.ready_to_delete(),
3818 )
3819 };
3820
3821 waitable.common(concurrent_state)?.handle = None;
3822
3823 if waitable.take_event(concurrent_state)?.is_some() {
3826 bail!(Trap::SubtaskDropNotResolved);
3827 }
3828
3829 if delete {
3830 waitable.delete_from(concurrent_state)?;
3831 }
3832
3833 log::trace!("subtask_drop {waitable:?} (handle {task_id})");
3834 Ok(())
3835 }
3836
3837 pub(crate) fn waitable_set_wait(
3839 self,
3840 store: &mut StoreOpaque,
3841 options: OptionsIndex,
3842 set: u32,
3843 payload: u32,
3844 ) -> Result<u32> {
3845 let &CanonicalOptions {
3846 instance: caller_instance,
3847 ..
3848 } = &self.id().get(store).component().env_component().options[options];
3849 let caller = self.runtime_instance(caller_instance);
3850 let rep = store
3851 .instance_state(self.runtime_instance(caller_instance))
3852 .handle_table()
3853 .waitable_set_rep(set)?;
3854
3855 self.waitable_check(
3856 store,
3857 caller,
3858 WaitableCheck::Wait,
3859 WaitableCheckParams {
3860 set: TableId::new(rep),
3861 options,
3862 payload,
3863 },
3864 )
3865 }
3866
3867 pub(crate) fn waitable_set_poll(
3869 self,
3870 store: &mut StoreOpaque,
3871 options: OptionsIndex,
3872 set: u32,
3873 payload: u32,
3874 ) -> Result<u32> {
3875 let &CanonicalOptions {
3876 instance: caller_instance,
3877 ..
3878 } = &self.id().get(store).component().env_component().options[options];
3879 let caller = self.runtime_instance(caller_instance);
3880 let rep = store
3881 .instance_state(caller)
3882 .handle_table()
3883 .waitable_set_rep(set)?;
3884
3885 self.waitable_check(
3886 store,
3887 caller,
3888 WaitableCheck::Poll,
3889 WaitableCheckParams {
3890 set: TableId::new(rep),
3891 options,
3892 payload,
3893 },
3894 )
3895 }
3896
3897 pub(crate) fn thread_index(&self, store: &mut dyn VMStore) -> Result<u32> {
3899 let thread_id = store.current_guest_thread()?.thread;
3900 match store
3901 .concurrent_state_mut()?
3902 .get_mut(thread_id)?
3903 .instance_rep
3904 {
3905 Some(r) => Ok(r),
3906 None => bail_bug!("thread should have instance_rep by now"),
3907 }
3908 }
3909
3910 pub(crate) fn thread_new_indirect<T: 'static>(
3912 self,
3913 mut store: StoreContextMut<T>,
3914 runtime_instance: RuntimeComponentInstanceIndex,
3915 _func_ty_idx: TypeFuncIndex, start_func_table_idx: RuntimeTableIndex,
3917 start_func_idx: u32,
3918 context: i32,
3919 ) -> Result<u32> {
3920 log::trace!("creating new thread");
3921
3922 let start_func_ty = FuncType::new(store.engine(), [ValType::I32], []);
3923 let (instance, registry) = self.id().get_mut_and_registry(store.0);
3924 let callee = instance
3925 .index_runtime_func_table(registry, start_func_table_idx, start_func_idx as u64)?
3926 .ok_or_else(|| Trap::ThreadNewIndirectUninitialized)?;
3927 if callee.type_index(store.0) != start_func_ty.type_index() {
3928 bail!(Trap::ThreadNewIndirectInvalidType);
3929 }
3930
3931 let token = StoreToken::new(store.as_context_mut());
3932 let start_func = Box::new(
3933 move |store: &mut dyn VMStore, guest_thread: QualifiedThreadId| -> Result<()> {
3934 let old_thread = store.set_thread(guest_thread)?;
3935 log::trace!(
3936 "thread start: replaced {old_thread:?} with {guest_thread:?} as current thread"
3937 );
3938
3939 let mut store = token.as_context_mut(store);
3940 let mut params = [ValRaw::i32(context)];
3941 unsafe { callee.call_unchecked(store.as_context_mut(), &mut params)? };
3944
3945 store.0.set_thread(old_thread)?;
3946
3947 let runtime_instance = self.runtime_instance(runtime_instance);
3948
3949 store
3952 .0
3953 .switch_or_trap_if_may_not_suspend(runtime_instance)?;
3954
3955 store
3956 .0
3957 .cleanup_thread(guest_thread, runtime_instance, CleanupTask::Yes)?;
3958
3959 log::trace!("explicit thread {guest_thread:?} completed");
3960 let state = store.0.concurrent_state_mut()?;
3961 if let Some(t) = old_thread.guest() {
3962 state.get_mut(t.thread)?.state = GuestThreadState::Running;
3963 }
3964 log::trace!("thread start: restored {old_thread:?} as current thread");
3965
3966 Ok(())
3967 },
3968 );
3969
3970 let current_thread = store.0.current_guest_thread()?;
3971 let state = store.0.concurrent_state_mut()?;
3972 let parent_task = current_thread.task;
3973
3974 let new_thread = GuestThread::new_explicit(state, parent_task, start_func)?;
3975 let thread_id = state.push(new_thread)?;
3976 state.get_mut(parent_task)?.threads.insert(thread_id);
3977
3978 log::trace!("new thread with id {thread_id:?} created");
3979
3980 self.add_guest_thread_to_instance_table(thread_id, store.0, runtime_instance)
3981 }
3982
3983 pub(crate) fn resume_thread(
3984 self,
3985 store: &mut StoreOpaque,
3986 runtime_instance: RuntimeComponentInstanceIndex,
3987 thread_idx: u32,
3988 how: ResumeThread,
3989 ) -> Result<bool> {
3990 let thread_id =
3991 GuestThread::from_instance(self.id().get_mut(store), runtime_instance, thread_idx)?;
3992 let state = store.concurrent_state_mut()?;
3993 let guest_thread = QualifiedThreadId::qualify(state, thread_id)?;
3994
3995 if store.current_guest_thread()? == guest_thread {
3996 bail!(Trap::CannotResumeThread);
3997 }
3998
3999 let state = store.concurrent_state_mut()?;
4000 let thread = state.get_mut(guest_thread.thread)?;
4001 let priority = match how {
4002 ResumeThread::Promote | ResumeThread::Resume => Priority::Switch,
4003 ResumeThread::ResumeLater => Priority::Low,
4004 };
4005
4006 match (&how, &thread.state) {
4007 (ResumeThread::Promote, GuestThreadState::Ready { .. }) => {}
4009 (ResumeThread::Promote, _) => return Ok(false),
4010
4011 (
4014 ResumeThread::Resume | ResumeThread::ResumeLater,
4015 GuestThreadState::NotStartedExplicit(_) | GuestThreadState::Suspended(_),
4016 ) => {}
4017 (ResumeThread::Resume | ResumeThread::ResumeLater, _) => {
4018 bail!(Trap::CannotResumeThread)
4019 }
4020 }
4021
4022 match mem::replace(&mut thread.state, GuestThreadState::Running) {
4023 GuestThreadState::NotStartedExplicit(start_func) => {
4024 log::trace!("starting thread {guest_thread:?}");
4025 let guest_call = WorkItem::GuestCall {
4026 instance: self.runtime_instance(runtime_instance),
4027 call: GuestCall {
4028 thread: guest_thread,
4029 kind: GuestCallKind::StartExplicit(Box::new(move |store| {
4030 start_func(store, guest_thread)
4031 })),
4032 },
4033 };
4034 store
4035 .concurrent_state_mut()?
4036 .push_work_item(guest_call, priority)?;
4037 }
4038 GuestThreadState::Suspended(fiber) => {
4039 log::trace!("resuming thread {thread_id:?} that was suspended");
4040 store.concurrent_state_mut()?.push_work_item(
4041 WorkItem::ResumeFiber {
4042 instance: self.runtime_instance(runtime_instance),
4043 thread: guest_thread,
4044 fiber,
4045 },
4046 priority,
4047 )?;
4048 }
4049 GuestThreadState::Ready { fiber } => {
4050 log::trace!("resuming thread {thread_id:?} that was ready");
4051 thread.state = GuestThreadState::Ready { fiber };
4052 store
4053 .concurrent_state_mut()?
4054 .promote_thread_work_item(guest_thread)?;
4055 }
4056 other @ (GuestThreadState::NotStartedImplicit
4057 | GuestThreadState::Running
4058 | GuestThreadState::Completed) => {
4059 thread.state = other;
4060 }
4061 }
4062 Ok(true)
4063 }
4064
4065 fn add_guest_thread_to_instance_table(
4066 self,
4067 thread_id: TableId<GuestThread>,
4068 store: &mut StoreOpaque,
4069 runtime_instance: RuntimeComponentInstanceIndex,
4070 ) -> Result<u32> {
4071 let guest_id = store
4072 .instance_state(self.runtime_instance(runtime_instance))
4073 .thread_handle_table()
4074 .guest_thread_insert(thread_id.rep())?;
4075 store
4076 .concurrent_state_mut()?
4077 .get_mut(thread_id)?
4078 .instance_rep = Some(guest_id);
4079 Ok(guest_id)
4080 }
4081
4082 pub(crate) fn suspension_intrinsic(
4086 self,
4087 store: &mut StoreOpaque,
4088 caller: RuntimeComponentInstanceIndex,
4089 yielding: bool,
4090 to_thread: SuspensionTarget,
4091 ) -> Result<WaitResult> {
4092 let check_suspend = match to_thread {
4093 SuspensionTarget::Promote(thread) => {
4094 !self.resume_thread(store, caller, thread, ResumeThread::Promote)?
4095 }
4096 SuspensionTarget::Resume(thread) => {
4097 if !self.resume_thread(store, caller, thread, ResumeThread::Resume)? {
4098 bail_bug!(
4099 "`resume_thread` should only ever return false \
4100 when `ResumeThread::Promote` is passed to it"
4101 );
4102 }
4103 false
4104 }
4105 SuspensionTarget::None => true,
4106 };
4107
4108 if check_suspend && !store.switch_if_may_not_suspend(self.runtime_instance(caller))? {
4109 return if yielding {
4110 Ok(WaitResult::Completed)
4111 } else {
4112 Err(Trap::CannotBlockSyncTask.into())
4113 };
4114 }
4115
4116 let guest_thread = store.current_guest_thread()?;
4117
4118 let reason = if yielding {
4119 SuspendReason::Yielding {
4120 thread: guest_thread,
4121 }
4122 } else {
4123 SuspendReason::ExplicitlySuspending {
4124 thread: guest_thread,
4125 }
4126 };
4127
4128 store.suspend(reason)?;
4129
4130 Ok(WaitResult::Completed)
4131 }
4132
4133 fn waitable_check(
4135 self,
4136 store: &mut StoreOpaque,
4137 caller: RuntimeInstance,
4138 check: WaitableCheck,
4139 params: WaitableCheckParams,
4140 ) -> Result<u32> {
4141 let guest_thread = store.current_guest_thread()?;
4142
4143 log::trace!("waitable check for {guest_thread:?}; set {:?}", params.set);
4144
4145 let state = store.concurrent_state_mut()?;
4146 let task = state.get_mut(guest_thread.task)?;
4147
4148 match &check {
4151 WaitableCheck::Wait => {
4152 let set = params.set;
4153
4154 if (task.event.is_none() || matches!(task.event, Some(Event::Cancelled)))
4155 && state.get_mut(set)?.ready.is_empty()
4156 {
4157 store.switch_or_trap_if_may_not_suspend(caller)?;
4158
4159 store.suspend(SuspendReason::Waiting {
4160 set,
4161 thread: guest_thread,
4162 })?;
4163 }
4164 }
4165 WaitableCheck::Poll => {}
4166 }
4167
4168 log::trace!(
4169 "waitable check for {guest_thread:?}; set {:?}, part two",
4170 params.set
4171 );
4172
4173 let event = self.get_event(store, guest_thread.task, Some(params.set), false)?;
4175
4176 let (ordinal, handle, result) = match &check {
4177 WaitableCheck::Wait => {
4178 let (event, waitable) = match event {
4179 Some(p) => p,
4180 None => bail_bug!("event expected to be present"),
4181 };
4182 let handle = waitable.map(|(_, v)| v).unwrap_or(0);
4183 let (ordinal, result) = event.parts();
4184 (ordinal, handle, result)
4185 }
4186 WaitableCheck::Poll => {
4187 if let Some((event, waitable)) = event {
4188 let handle = waitable.map(|(_, v)| v).unwrap_or(0);
4189 let (ordinal, result) = event.parts();
4190 (ordinal, handle, result)
4191 } else {
4192 log::trace!(
4193 "no events ready to deliver via waitable-set.poll to {:?}; set {:?}",
4194 guest_thread.task,
4195 params.set
4196 );
4197 let (ordinal, result) = Event::None.parts();
4198 (ordinal, 0, result)
4199 }
4200 }
4201 };
4202 let memory = self.options_memory_mut(store, params.options);
4203 let ptr = crate::component::func::validate_inbounds_dynamic(
4204 &CanonicalAbiInfo::POINTER_PAIR,
4205 memory,
4206 &ValRaw::u32(params.payload),
4207 )?;
4208 memory[ptr + 0..][..4].copy_from_slice(&handle.to_le_bytes());
4209 memory[ptr + 4..][..4].copy_from_slice(&result.to_le_bytes());
4210 Ok(ordinal)
4211 }
4212
4213 pub(crate) fn subtask_cancel(
4215 self,
4216 store: &mut StoreOpaque,
4217 caller_instance: RuntimeComponentInstanceIndex,
4218 async_: bool,
4219 task_id: u32,
4220 ) -> Result<u32> {
4221 let (rep, is_host) = store
4222 .instance_state(self.runtime_instance(caller_instance))
4223 .handle_table()
4224 .subtask_rep(task_id)?;
4225 let waitable = if is_host {
4226 Waitable::Host(TableId::<HostTask>::new(rep))
4227 } else {
4228 Waitable::Guest(TableId::<GuestTask>::new(rep))
4229 };
4230 let concurrent_state = store.concurrent_state_mut()?;
4231
4232 log::trace!("subtask_cancel {waitable:?} (handle {task_id}; async {async_})");
4233
4234 waitable.trap_if_in_waitable_set(concurrent_state)?;
4235
4236 let needs_block;
4237 if let Waitable::Host(host_task) = waitable {
4238 let state = &mut concurrent_state.get_mut(host_task)?.state;
4239 match mem::replace(state, HostTaskState::CalleeDone { cancelled: true }) {
4240 HostTaskState::CalleeRunning(handle) => {
4247 handle.abort();
4248 needs_block = true;
4249 }
4250
4251 HostTaskState::CalleeDone { cancelled } => {
4254 if cancelled {
4255 bail!(Trap::SubtaskCancelAfterTerminal);
4256 } else {
4257 needs_block = false;
4260 }
4261 }
4262
4263 HostTaskState::CalleeStarted | HostTaskState::CalleeFinished(_) => {
4266 bail_bug!("invalid states for host callee")
4267 }
4268 }
4269 } else {
4270 let guest_task = TableId::<GuestTask>::new(rep);
4271 let task = concurrent_state.get_mut(guest_task)?;
4272 if !task.already_lowered_parameters() {
4273 store.cancel_guest_subtask_without_lowered_parameters(
4274 self.runtime_instance(caller_instance),
4275 guest_task,
4276 )?;
4277 return Ok(Status::StartCancelled as u32);
4278 } else if !task.returned_or_cancelled() {
4279 task.event = Some(Event::Cancelled);
4287 let runtime_instance = task.instance;
4288 for thread in task.threads.clone() {
4289 let thread = QualifiedThreadId {
4290 task: guest_task,
4291 thread,
4292 };
4293 let thread_mut = concurrent_state.get_mut(thread.thread)?;
4294
4295 let yield_ = |store: &mut StoreOpaque| {
4296 let state = store.instance_state(runtime_instance).concurrent_state();
4301 let old_do_not_suspend = state.do_not_suspend;
4302 state.do_not_suspend = false;
4303
4304 let caller = store.current_guest_thread()?;
4305
4306 let state = store.concurrent_state_mut()?;
4311 let set = state.get_mut(caller.thread)?.sync_call_set;
4312 waitable.join(state, Some(set))?;
4313
4314 store.suspend(SuspendReason::Yielding { thread: caller })?;
4315
4316 let state = store.concurrent_state_mut()?;
4317 waitable.join(state, None)?;
4318
4319 store
4320 .instance_state(runtime_instance)
4321 .concurrent_state()
4322 .do_not_suspend = old_do_not_suspend;
4323
4324 Ok::<(), crate::Error>(())
4325 };
4326
4327 if let Some(set) = thread_mut.wake_on_cancel.take() {
4328 let item = match concurrent_state.get_mut(set)?.waiting.remove(&thread) {
4330 Some(WaitMode::Fiber(fiber)) => WorkItem::ResumeFiber {
4331 instance: runtime_instance,
4332 thread,
4333 fiber,
4334 },
4335 Some(WaitMode::Callback(instance)) => WorkItem::GuestCall {
4336 instance: runtime_instance,
4337 call: GuestCall {
4338 thread,
4339 kind: GuestCallKind::DeliverEvent {
4340 instance,
4341 set: None,
4342 },
4343 },
4344 },
4345 Some(WaitMode::Caller { .. }) => {
4346 bail_bug!("unexpected `WaitMode::Caller` in wake_on_cancel set")
4347 }
4348 None => bail_bug!("thread not present in wake_on_cancel set"),
4349 };
4350 concurrent_state.set_switch_item(item)?;
4351
4352 yield_(store)?;
4353
4354 break;
4355 }
4356 }
4357
4358 needs_block = !store
4361 .concurrent_state_mut()?
4362 .get_mut(guest_task)?
4363 .returned_or_cancelled()
4364 } else {
4365 needs_block = false;
4366 }
4367 };
4368
4369 if needs_block {
4373 if async_ {
4374 return Ok(BLOCKED);
4375 }
4376
4377 store.wait_for_event(
4380 self.runtime_instance(caller_instance),
4381 waitable,
4382 if is_host {
4383 WaitReason::Other
4384 } else {
4385 WaitReason::GuestSubtask(TableId::<GuestTask>::new(rep))
4386 },
4387 )?;
4388
4389 }
4391
4392 let event = waitable.take_event(store.concurrent_state_mut()?)?;
4393 if let Some(Event::Subtask {
4394 status: status @ (Status::Returned | Status::ReturnCancelled),
4395 }) = event
4396 {
4397 Ok(status as u32)
4398 } else {
4399 bail!(Trap::SubtaskCancelAfterTerminal);
4400 }
4401 }
4402}
4403
4404pub trait VMComponentAsyncStore {
4412 unsafe fn prepare_call(
4418 &mut self,
4419 instance: Instance,
4420 memory: *mut VMMemoryDefinition,
4421 start: NonNull<VMFuncRef>,
4422 return_: NonNull<VMFuncRef>,
4423 caller_instance: RuntimeComponentInstanceIndex,
4424 callee_instance: RuntimeComponentInstanceIndex,
4425 task_return_type: TypeTupleIndex,
4426 callee_async: bool,
4427 string_encoding: StringEncoding,
4428 result_count: u32,
4429 storage: *mut ValRaw,
4430 storage_len: usize,
4431 ) -> Result<()>;
4432
4433 unsafe fn sync_start(
4436 &mut self,
4437 instance: Instance,
4438 callback: *mut VMFuncRef,
4439 callee: NonNull<VMFuncRef>,
4440 param_count: u32,
4441 storage: *mut MaybeUninit<ValRaw>,
4442 storage_len: usize,
4443 ) -> Result<()>;
4444
4445 unsafe fn async_start(
4448 &mut self,
4449 instance: Instance,
4450 callback: *mut VMFuncRef,
4451 post_return: *mut VMFuncRef,
4452 callee: NonNull<VMFuncRef>,
4453 param_count: u32,
4454 result_count: u32,
4455 flags: u32,
4456 ) -> Result<u32>;
4457
4458 fn future_write(
4460 &mut self,
4461 instance: Instance,
4462 caller: RuntimeComponentInstanceIndex,
4463 ty: TypeFutureTableIndex,
4464 options: OptionsIndex,
4465 future: u32,
4466 address: u32,
4467 ) -> Result<u32>;
4468
4469 fn future_read(
4471 &mut self,
4472 instance: Instance,
4473 caller: RuntimeComponentInstanceIndex,
4474 ty: TypeFutureTableIndex,
4475 options: OptionsIndex,
4476 future: u32,
4477 address: u32,
4478 ) -> Result<u32>;
4479
4480 fn future_drop_writable(
4482 &mut self,
4483 instance: Instance,
4484 ty: TypeFutureTableIndex,
4485 writer: u32,
4486 ) -> Result<()>;
4487
4488 fn stream_write(
4490 &mut self,
4491 instance: Instance,
4492 caller: RuntimeComponentInstanceIndex,
4493 ty: TypeStreamTableIndex,
4494 options: OptionsIndex,
4495 stream: u32,
4496 address: u32,
4497 count: u32,
4498 ) -> Result<u32>;
4499
4500 fn stream_read(
4502 &mut self,
4503 instance: Instance,
4504 caller: RuntimeComponentInstanceIndex,
4505 ty: TypeStreamTableIndex,
4506 options: OptionsIndex,
4507 stream: u32,
4508 address: u32,
4509 count: u32,
4510 ) -> Result<u32>;
4511
4512 fn flat_stream_write(
4515 &mut self,
4516 instance: Instance,
4517 caller: RuntimeComponentInstanceIndex,
4518 ty: TypeStreamTableIndex,
4519 options: OptionsIndex,
4520 payload_size: u32,
4521 payload_align: u32,
4522 stream: u32,
4523 address: u32,
4524 count: u32,
4525 ) -> Result<u32>;
4526
4527 fn flat_stream_read(
4530 &mut self,
4531 instance: Instance,
4532 caller: RuntimeComponentInstanceIndex,
4533 ty: TypeStreamTableIndex,
4534 options: OptionsIndex,
4535 payload_size: u32,
4536 payload_align: u32,
4537 stream: u32,
4538 address: u32,
4539 count: u32,
4540 ) -> Result<u32>;
4541
4542 fn stream_drop_writable(
4544 &mut self,
4545 instance: Instance,
4546 ty: TypeStreamTableIndex,
4547 writer: u32,
4548 ) -> Result<()>;
4549
4550 fn error_context_debug_message(
4552 &mut self,
4553 instance: Instance,
4554 ty: TypeComponentLocalErrorContextTableIndex,
4555 options: OptionsIndex,
4556 err_ctx_handle: u32,
4557 debug_msg_address: u32,
4558 ) -> Result<()>;
4559
4560 fn thread_new_indirect(
4562 &mut self,
4563 instance: Instance,
4564 caller: RuntimeComponentInstanceIndex,
4565 func_ty_idx: TypeFuncIndex,
4566 start_func_table_idx: RuntimeTableIndex,
4567 start_func_idx: u32,
4568 context: i32,
4569 ) -> Result<u32>;
4570}
4571
4572impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
4574 unsafe fn prepare_call(
4575 &mut self,
4576 instance: Instance,
4577 memory: *mut VMMemoryDefinition,
4578 start: NonNull<VMFuncRef>,
4579 return_: NonNull<VMFuncRef>,
4580 caller_instance: RuntimeComponentInstanceIndex,
4581 callee_instance: RuntimeComponentInstanceIndex,
4582 task_return_type: TypeTupleIndex,
4583 callee_async: bool,
4584 string_encoding: StringEncoding,
4585 result_count_or_max_if_async: u32,
4586 storage: *mut ValRaw,
4587 storage_len: usize,
4588 ) -> Result<()> {
4589 let params = unsafe { core::slice::from_raw_parts(storage, storage_len) }.to_vec();
4593
4594 unsafe {
4595 instance.prepare_call(
4596 StoreContextMut(self),
4597 start,
4598 return_,
4599 caller_instance,
4600 callee_instance,
4601 task_return_type,
4602 callee_async,
4603 memory,
4604 string_encoding,
4605 match result_count_or_max_if_async {
4606 PREPARE_ASYNC_NO_RESULT => CallerInfo::Async {
4607 params,
4608 has_result: false,
4609 },
4610 PREPARE_ASYNC_WITH_RESULT => CallerInfo::Async {
4611 params,
4612 has_result: true,
4613 },
4614 result_count => CallerInfo::Sync {
4615 params,
4616 result_count,
4617 },
4618 },
4619 )
4620 }
4621 }
4622
4623 unsafe fn sync_start(
4624 &mut self,
4625 instance: Instance,
4626 callback: *mut VMFuncRef,
4627 callee: NonNull<VMFuncRef>,
4628 param_count: u32,
4629 storage: *mut MaybeUninit<ValRaw>,
4630 storage_len: usize,
4631 ) -> Result<()> {
4632 unsafe {
4633 instance
4634 .start_call(
4635 StoreContextMut(self),
4636 callback,
4637 ptr::null_mut(),
4638 callee,
4639 param_count,
4640 1,
4641 START_FLAG_ASYNC_CALLEE,
4642 Some(core::slice::from_raw_parts_mut(storage, storage_len)),
4646 )
4647 .map(drop)
4648 }
4649 }
4650
4651 unsafe fn async_start(
4652 &mut self,
4653 instance: Instance,
4654 callback: *mut VMFuncRef,
4655 post_return: *mut VMFuncRef,
4656 callee: NonNull<VMFuncRef>,
4657 param_count: u32,
4658 result_count: u32,
4659 flags: u32,
4660 ) -> Result<u32> {
4661 unsafe {
4662 instance.start_call(
4663 StoreContextMut(self),
4664 callback,
4665 post_return,
4666 callee,
4667 param_count,
4668 result_count,
4669 flags,
4670 None,
4671 )
4672 }
4673 }
4674
4675 fn future_write(
4676 &mut self,
4677 instance: Instance,
4678 caller: RuntimeComponentInstanceIndex,
4679 ty: TypeFutureTableIndex,
4680 options: OptionsIndex,
4681 future: u32,
4682 address: u32,
4683 ) -> Result<u32> {
4684 instance
4685 .guest_write(
4686 StoreContextMut(self),
4687 caller,
4688 TransmitIndex::Future(ty),
4689 options,
4690 None,
4691 future,
4692 address,
4693 1,
4694 )
4695 .map(|result| result.encode())
4696 }
4697
4698 fn future_read(
4699 &mut self,
4700 instance: Instance,
4701 caller: RuntimeComponentInstanceIndex,
4702 ty: TypeFutureTableIndex,
4703 options: OptionsIndex,
4704 future: u32,
4705 address: u32,
4706 ) -> Result<u32> {
4707 instance
4708 .guest_read(
4709 StoreContextMut(self),
4710 caller,
4711 TransmitIndex::Future(ty),
4712 options,
4713 None,
4714 future,
4715 address,
4716 1,
4717 )
4718 .map(|result| result.encode())
4719 }
4720
4721 fn stream_write(
4722 &mut self,
4723 instance: Instance,
4724 caller: RuntimeComponentInstanceIndex,
4725 ty: TypeStreamTableIndex,
4726 options: OptionsIndex,
4727 stream: u32,
4728 address: u32,
4729 count: u32,
4730 ) -> Result<u32> {
4731 instance
4732 .guest_write(
4733 StoreContextMut(self),
4734 caller,
4735 TransmitIndex::Stream(ty),
4736 options,
4737 None,
4738 stream,
4739 address,
4740 count,
4741 )
4742 .map(|result| result.encode())
4743 }
4744
4745 fn stream_read(
4746 &mut self,
4747 instance: Instance,
4748 caller: RuntimeComponentInstanceIndex,
4749 ty: TypeStreamTableIndex,
4750 options: OptionsIndex,
4751 stream: u32,
4752 address: u32,
4753 count: u32,
4754 ) -> Result<u32> {
4755 instance
4756 .guest_read(
4757 StoreContextMut(self),
4758 caller,
4759 TransmitIndex::Stream(ty),
4760 options,
4761 None,
4762 stream,
4763 address,
4764 count,
4765 )
4766 .map(|result| result.encode())
4767 }
4768
4769 fn future_drop_writable(
4770 &mut self,
4771 instance: Instance,
4772 ty: TypeFutureTableIndex,
4773 writer: u32,
4774 ) -> Result<()> {
4775 instance.guest_drop_writable(self, TransmitIndex::Future(ty), writer)
4776 }
4777
4778 fn flat_stream_write(
4779 &mut self,
4780 instance: Instance,
4781 caller: RuntimeComponentInstanceIndex,
4782 ty: TypeStreamTableIndex,
4783 options: OptionsIndex,
4784 payload_size: u32,
4785 payload_align: u32,
4786 stream: u32,
4787 address: u32,
4788 count: u32,
4789 ) -> Result<u32> {
4790 instance
4791 .guest_write(
4792 StoreContextMut(self),
4793 caller,
4794 TransmitIndex::Stream(ty),
4795 options,
4796 Some(FlatAbi {
4797 size: payload_size,
4798 align: payload_align,
4799 }),
4800 stream,
4801 address,
4802 count,
4803 )
4804 .map(|result| result.encode())
4805 }
4806
4807 fn flat_stream_read(
4808 &mut self,
4809 instance: Instance,
4810 caller: RuntimeComponentInstanceIndex,
4811 ty: TypeStreamTableIndex,
4812 options: OptionsIndex,
4813 payload_size: u32,
4814 payload_align: u32,
4815 stream: u32,
4816 address: u32,
4817 count: u32,
4818 ) -> Result<u32> {
4819 instance
4820 .guest_read(
4821 StoreContextMut(self),
4822 caller,
4823 TransmitIndex::Stream(ty),
4824 options,
4825 Some(FlatAbi {
4826 size: payload_size,
4827 align: payload_align,
4828 }),
4829 stream,
4830 address,
4831 count,
4832 )
4833 .map(|result| result.encode())
4834 }
4835
4836 fn stream_drop_writable(
4837 &mut self,
4838 instance: Instance,
4839 ty: TypeStreamTableIndex,
4840 writer: u32,
4841 ) -> Result<()> {
4842 instance.guest_drop_writable(self, TransmitIndex::Stream(ty), writer)
4843 }
4844
4845 fn error_context_debug_message(
4846 &mut self,
4847 instance: Instance,
4848 ty: TypeComponentLocalErrorContextTableIndex,
4849 options: OptionsIndex,
4850 err_ctx_handle: u32,
4851 debug_msg_address: u32,
4852 ) -> Result<()> {
4853 instance.error_context_debug_message(
4854 StoreContextMut(self),
4855 ty,
4856 options,
4857 err_ctx_handle,
4858 debug_msg_address,
4859 )
4860 }
4861
4862 fn thread_new_indirect(
4863 &mut self,
4864 instance: Instance,
4865 caller: RuntimeComponentInstanceIndex,
4866 func_ty_idx: TypeFuncIndex,
4867 start_func_table_idx: RuntimeTableIndex,
4868 start_func_idx: u32,
4869 context: i32,
4870 ) -> Result<u32> {
4871 instance.thread_new_indirect(
4872 StoreContextMut(self),
4873 caller,
4874 func_ty_idx,
4875 start_func_table_idx,
4876 start_func_idx,
4877 context,
4878 )
4879 }
4880}
4881
4882type HostTaskFuture = Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>;
4883
4884async fn run_with_host_task_set<F>(task: TableId<HostTask>, future: F) -> Result<F::Output>
4887where
4888 F: Future,
4889{
4890 let mut future = pin!(future);
4891 future::poll_fn(|cx| {
4892 let old_thread = match tls::get(|store| store.set_thread(task)) {
4893 Ok(thread) => thread,
4894 Err(error) => return Poll::Ready(Err(error)),
4895 };
4896 let result = future.as_mut().poll(cx);
4897 match tls::get(|store| store.set_thread(old_thread)) {
4898 Ok(_) => result.map(Ok),
4899 Err(error) => Poll::Ready(Err(error)),
4900 }
4901 })
4902 .await
4903}
4904
4905pub(crate) struct HostTask {
4909 common: WaitableCommon,
4910
4911 caller: TableId<GuestTask>,
4918
4919 call_context: CallContext,
4922
4923 state: HostTaskState,
4924}
4925
4926enum HostTaskState {
4927 CalleeStarted,
4932
4933 CalleeRunning(JoinHandle),
4938
4939 CalleeFinished(LiftedResult),
4943
4944 CalleeDone { cancelled: bool },
4947}
4948
4949impl HostTask {
4950 fn new(caller: TableId<GuestTask>, state: HostTaskState) -> Self {
4951 Self {
4952 common: WaitableCommon::default(),
4953 call_context: CallContext::default(),
4954 caller,
4955 state,
4956 }
4957 }
4958}
4959
4960impl TableDebug for HostTask {
4961 fn type_name() -> &'static str {
4962 "HostTask"
4963 }
4964}
4965
4966type CallbackFn = Box<dyn Fn(&mut dyn VMStore, Event, u32) -> Result<u32> + Send + Sync + 'static>;
4967
4968enum Caller {
4970 Host {
4972 tx: Option<oneshot::Sender<LiftedResult>>,
4974 host_future_present: bool,
4977 caller: Option<TableId<HostTask>>,
4981 },
4982 Guest {
4984 thread: QualifiedThreadId,
4986 },
4987}
4988
4989struct LiftResult {
4992 lift: RawLift,
4993 ty: TypeTupleIndex,
4994 memory: Option<SendSyncPtr<VMMemoryDefinition>>,
4995 string_encoding: StringEncoding,
4996}
4997
4998#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
5003pub(crate) struct QualifiedThreadId {
5004 task: TableId<GuestTask>,
5005 thread: TableId<GuestThread>,
5006}
5007
5008impl QualifiedThreadId {
5009 fn qualify(
5010 state: &mut ConcurrentState,
5011 thread: TableId<GuestThread>,
5012 ) -> Result<QualifiedThreadId> {
5013 Ok(QualifiedThreadId {
5014 task: state.get_mut(thread)?.parent_task,
5015 thread,
5016 })
5017 }
5018}
5019
5020impl fmt::Debug for QualifiedThreadId {
5021 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
5022 f.debug_tuple("QualifiedThreadId")
5023 .field(&self.task.rep())
5024 .field(&self.thread.rep())
5025 .finish()
5026 }
5027}
5028
5029enum GuestThreadState {
5030 NotStartedImplicit,
5031 NotStartedExplicit(
5032 Box<dyn FnOnce(&mut dyn VMStore, QualifiedThreadId) -> Result<()> + Send + Sync>,
5033 ),
5034 Running,
5035 Suspended(StoreFiber<'static>),
5036 Ready {
5037 fiber: StoreFiber<'static>,
5038 },
5039 Completed,
5040}
5041
5042impl fmt::Debug for GuestThreadState {
5043 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
5044 match self {
5045 Self::NotStartedImplicit => f.debug_tuple("NotStartedImplicit").finish(),
5046 Self::NotStartedExplicit(_) => f.debug_tuple("NotStartedExplicit").finish(),
5047 Self::Running => f.debug_tuple("Running").finish(),
5048 Self::Suspended(_) => f.debug_tuple("Suspended").finish(),
5049 Self::Ready { .. } => f.debug_struct("Ready").finish(),
5050 Self::Completed => f.debug_tuple("Completed").finish(),
5051 }
5052 }
5053}
5054
5055pub struct GuestThread {
5056 context: [u32; NUM_COMPONENT_CONTEXT_SLOTS],
5059 parent_task: TableId<GuestTask>,
5061 wake_on_cancel: Option<TableId<WaitableSet>>,
5064 state: GuestThreadState,
5066 instance_rep: Option<u32>,
5069 sync_call_set: TableId<WaitableSet>,
5071 old_do_not_suspend: Option<bool>,
5074}
5075
5076impl GuestThread {
5077 fn from_instance(
5080 state: Pin<&mut ComponentInstance>,
5081 caller_instance: RuntimeComponentInstanceIndex,
5082 guest_thread: u32,
5083 ) -> Result<TableId<Self>> {
5084 let rep = state.instance_states().0[caller_instance]
5085 .thread_handle_table()
5086 .guest_thread_rep(guest_thread)?;
5087 Ok(TableId::new(rep))
5088 }
5089
5090 fn new_implicit(state: &mut ConcurrentState, parent_task: TableId<GuestTask>) -> Result<Self> {
5091 let sync_call_set = state.push(WaitableSet {
5092 is_sync_call_set: true,
5093 ..WaitableSet::default()
5094 })?;
5095 Ok(Self {
5096 context: [0; NUM_COMPONENT_CONTEXT_SLOTS],
5097 parent_task,
5098 wake_on_cancel: None,
5099 state: GuestThreadState::NotStartedImplicit,
5100 instance_rep: None,
5101 sync_call_set,
5102 old_do_not_suspend: None,
5103 })
5104 }
5105
5106 fn new_explicit(
5107 state: &mut ConcurrentState,
5108 parent_task: TableId<GuestTask>,
5109 start_func: Box<
5110 dyn FnOnce(&mut dyn VMStore, QualifiedThreadId) -> Result<()> + Send + Sync,
5111 >,
5112 ) -> Result<Self> {
5113 let sync_call_set = state.push(WaitableSet {
5114 is_sync_call_set: true,
5115 ..WaitableSet::default()
5116 })?;
5117 Ok(Self {
5118 context: [0; NUM_COMPONENT_CONTEXT_SLOTS],
5119 parent_task,
5120 wake_on_cancel: None,
5121 state: GuestThreadState::NotStartedExplicit(start_func),
5122 instance_rep: None,
5123 sync_call_set,
5124 old_do_not_suspend: None,
5125 })
5126 }
5127}
5128
5129impl TableDebug for GuestThread {
5130 fn type_name() -> &'static str {
5131 "GuestThread"
5132 }
5133}
5134
5135enum SyncResult {
5136 NotProduced,
5137 Produced(Option<ValRaw>),
5138 Taken,
5139}
5140
5141impl SyncResult {
5142 fn take(&mut self) -> Result<Option<Option<ValRaw>>> {
5143 Ok(match mem::replace(self, SyncResult::Taken) {
5144 SyncResult::NotProduced => None,
5145 SyncResult::Produced(val) => Some(val),
5146 SyncResult::Taken => {
5147 bail_bug!("attempted to take a synchronous result that was already taken")
5148 }
5149 })
5150 }
5151}
5152
5153#[derive(Debug)]
5154enum HostFutureState {
5155 NotApplicable,
5156 Live,
5157 Dropped,
5158}
5159
5160pub(crate) struct GuestTask {
5162 common: WaitableCommon,
5164 lower_params: Option<RawLower>,
5166 lift_result: Option<LiftResult>,
5168 result: Option<LiftedResult>,
5171 callback: Option<CallbackFn>,
5174 caller: Caller,
5176 call_context: CallContext,
5181 sync_result: SyncResult,
5184 cancel_request_delivered: bool,
5188 starting_sent: bool,
5191 instance: RuntimeInstance,
5198 event: Option<Event>,
5201 exited: bool,
5203 threads: HashSet<TableId<GuestThread>>,
5205 host_future_state: HostFutureState,
5208 async_typed: bool,
5211 async_lifted: bool,
5214
5215 decremented_interesting_task_count: bool,
5216 switch_item: Option<WorkItem>,
5217}
5218
5219impl GuestTask {
5220 fn already_lowered_parameters(&self) -> bool {
5221 self.lower_params.is_none()
5223 }
5224
5225 fn returned_or_cancelled(&self) -> bool {
5226 self.lift_result.is_none()
5228 }
5229
5230 fn ready_to_delete(&self) -> bool {
5231 let threads_completed = self.threads.is_empty();
5232 let has_sync_result = matches!(self.sync_result, SyncResult::Produced(_));
5233 let pending_completion_event = matches!(
5234 self.common.event,
5235 Some(Event::Subtask {
5236 status: Status::Returned | Status::ReturnCancelled
5237 })
5238 );
5239 let ready = threads_completed
5240 && !has_sync_result
5241 && !pending_completion_event
5242 && !matches!(self.host_future_state, HostFutureState::Live);
5243 log::trace!(
5244 "ready to delete? {ready} (threads_completed: {}, has_sync_result: {}, pending_completion_event: {}, host_future_state: {:?})",
5245 threads_completed,
5246 has_sync_result,
5247 pending_completion_event,
5248 self.host_future_state
5249 );
5250 ready
5251 }
5252
5253 fn new(
5254 state: &mut ConcurrentState,
5255 lower_params: RawLower,
5256 lift_result: LiftResult,
5257 caller: Caller,
5258 callback: Option<CallbackFn>,
5259 instance: RuntimeInstance,
5260 async_typed: bool,
5261 async_lifted: bool,
5262 ) -> Result<QualifiedThreadId> {
5263 let host_future_state = match &caller {
5264 Caller::Guest { .. } => HostFutureState::NotApplicable,
5265 Caller::Host {
5266 host_future_present,
5267 ..
5268 } => {
5269 if *host_future_present {
5270 HostFutureState::Live
5271 } else {
5272 HostFutureState::NotApplicable
5273 }
5274 }
5275 };
5276 let task = state.push(Self {
5277 common: WaitableCommon::default(),
5278 lower_params: Some(lower_params),
5279 lift_result: Some(lift_result),
5280 result: None,
5281 callback,
5282 caller,
5283 call_context: CallContext::default(),
5284 sync_result: SyncResult::NotProduced,
5285 cancel_request_delivered: false,
5286 starting_sent: false,
5287 instance,
5288 event: None,
5289 exited: false,
5290 threads: HashSet::new(),
5291 host_future_state,
5292 async_typed,
5293 async_lifted,
5294 decremented_interesting_task_count: false,
5295 switch_item: None,
5296 })?;
5297 let new_thread = GuestThread::new_implicit(state, task)?;
5298 let thread = state.push(new_thread)?;
5299 state.get_mut(task)?.threads.insert(thread);
5300 state.interesting_tasks += 1;
5301 let thread = QualifiedThreadId { task, thread };
5302 log::trace!("new implicit thread {thread:?} for instance {instance:?}");
5303 Ok(thread)
5304 }
5305}
5306
5307impl TableDebug for GuestTask {
5308 fn type_name() -> &'static str {
5309 "GuestTask"
5310 }
5311}
5312
5313#[derive(Default)]
5315struct WaitableCommon {
5316 event: Option<Event>,
5318 set: Option<TableId<WaitableSet>>,
5320 handle: Option<u32>,
5322}
5323
5324#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
5326enum Waitable {
5327 Host(TableId<HostTask>),
5329 Guest(TableId<GuestTask>),
5331 Transmit(TableId<TransmitHandle>),
5333}
5334
5335impl Waitable {
5336 fn from_instance(
5339 state: Pin<&mut ComponentInstance>,
5340 caller_instance: RuntimeComponentInstanceIndex,
5341 waitable: u32,
5342 ) -> Result<Self> {
5343 use crate::runtime::vm::component::Waitable;
5344
5345 let (waitable, kind) = state.instance_states().0[caller_instance]
5346 .handle_table()
5347 .waitable_rep(waitable)?;
5348
5349 Ok(match kind {
5350 Waitable::Subtask { is_host: true } => Self::Host(TableId::new(waitable)),
5351 Waitable::Subtask { is_host: false } => Self::Guest(TableId::new(waitable)),
5352 Waitable::Stream | Waitable::Future => Self::Transmit(TableId::new(waitable)),
5353 })
5354 }
5355
5356 fn rep(&self) -> u32 {
5358 match self {
5359 Self::Host(id) => id.rep(),
5360 Self::Guest(id) => id.rep(),
5361 Self::Transmit(id) => id.rep(),
5362 }
5363 }
5364
5365 fn join(&self, state: &mut ConcurrentState, set: Option<TableId<WaitableSet>>) -> Result<()> {
5369 log::trace!("waitable {self:?} join set {set:?}");
5370
5371 let old = mem::replace(&mut self.common(state)?.set, set);
5372
5373 if let Some(old) = old {
5374 match *self {
5375 Waitable::Host(id) => state.remove_child(id, old),
5376 Waitable::Guest(id) => state.remove_child(id, old),
5377 Waitable::Transmit(id) => state.remove_child(id, old),
5378 }?;
5379
5380 state.get_mut(old)?.ready.remove(self);
5381 }
5382
5383 if let Some(set) = set {
5384 match *self {
5385 Waitable::Host(id) => state.add_child(id, set),
5386 Waitable::Guest(id) => state.add_child(id, set),
5387 Waitable::Transmit(id) => state.add_child(id, set),
5388 }?;
5389
5390 if self.common(state)?.event.is_some() {
5391 self.mark_ready(state)?;
5392 }
5393 }
5394
5395 Ok(())
5396 }
5397
5398 fn common<'a>(&self, state: &'a mut ConcurrentState) -> Result<&'a mut WaitableCommon> {
5400 Ok(match self {
5401 Self::Host(id) => &mut state.get_mut(*id)?.common,
5402 Self::Guest(id) => &mut state.get_mut(*id)?.common,
5403 Self::Transmit(id) => &mut state.get_mut(*id)?.common,
5404 })
5405 }
5406
5407 fn trap_if_in_waitable_set(&self, state: &mut ConcurrentState) -> Result<()> {
5413 if self.common(state)?.set.is_some() {
5414 bail!(Trap::WaitableSyncAndAsync);
5415 }
5416 Ok(())
5417 }
5418
5419 fn set_event(&self, state: &mut ConcurrentState, event: Option<Event>) -> Result<()> {
5423 log::trace!("set event for {self:?}: {event:?}");
5424 self.common(state)?.event = event;
5425 self.mark_ready(state)
5426 }
5427
5428 fn take_event(&self, state: &mut ConcurrentState) -> Result<Option<Event>> {
5430 let common = self.common(state)?;
5431 let event = common.event.take();
5432 if let Some(set) = self.common(state)?.set {
5433 state.get_mut(set)?.ready.remove(self);
5434 }
5435
5436 Ok(event)
5437 }
5438
5439 fn mark_ready(&self, state: &mut ConcurrentState) -> Result<()> {
5443 if let Some(set) = self.common(state)?.set {
5444 let set_state = state.get_mut(set)?;
5445 set_state.ready.insert(*self);
5446
5447 if let Some((thread, mode)) = set_state.waiting.pop_first() {
5448 let wake_on_cancel = state.get_mut(thread.thread)?.wake_on_cancel.take();
5449 assert!(wake_on_cancel.is_none() || wake_on_cancel == Some(set));
5450
5451 let item = match mode {
5452 WaitMode::Caller { fiber, callee } => {
5453 let item = WorkItem::ResumeFiber {
5465 instance: state.get_mut(thread.task)?.instance,
5466 thread,
5467 fiber,
5468 };
5469
5470 if let Some(Event::Subtask {
5471 status: Status::Starting,
5472 }) = &self.common(state)?.event
5473 {
5474 state.set_switch_item(item)?;
5478 } else {
5479 if state.get_mut(callee)?.switch_item.is_some() {
5480 bail_bug!(
5481 "`GuestTask::switch_item` is already `Some(_)` when we need \
5482 to deliver a subtask status update to the caller"
5483 );
5484 }
5485 state.get_mut(callee)?.switch_item = Some(item);
5486 }
5487 None
5488 }
5489 WaitMode::Fiber(fiber) => Some(WorkItem::ResumeFiber {
5490 instance: state.get_mut(thread.task)?.instance,
5491 thread,
5492 fiber,
5493 }),
5494 WaitMode::Callback(instance) => Some(WorkItem::GuestCall {
5495 instance: state.get_mut(thread.task)?.instance,
5496 call: GuestCall {
5497 thread,
5498 kind: GuestCallKind::DeliverEvent {
5499 instance,
5500 set: Some(set),
5501 },
5502 },
5503 }),
5504 };
5505
5506 if let Some(item) = item {
5507 state.push_high_priority(item);
5508 }
5509 }
5510 }
5511 Ok(())
5512 }
5513
5514 fn delete_from(&self, state: &mut ConcurrentState) -> Result<()> {
5516 match self {
5517 Self::Host(task) => {
5518 log::trace!("delete host task {task:?}");
5519 state.delete(*task)?;
5520 }
5521 Self::Guest(task) => {
5522 log::trace!("delete guest task {task:?}");
5523 let task = state.delete(*task)?;
5524
5525 debug_assert!(task.decremented_interesting_task_count);
5532 }
5533 Self::Transmit(task) => {
5534 state.delete(*task)?;
5535 }
5536 }
5537
5538 Ok(())
5539 }
5540}
5541
5542impl fmt::Debug for Waitable {
5543 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
5544 match self {
5545 Self::Host(id) => write!(f, "{id:?}"),
5546 Self::Guest(id) => write!(f, "{id:?}"),
5547 Self::Transmit(id) => write!(f, "{id:?}"),
5548 }
5549 }
5550}
5551
5552#[derive(Default)]
5554struct WaitableSet {
5555 ready: BTreeSet<Waitable>,
5557 waiting: BTreeMap<QualifiedThreadId, WaitMode>,
5559 is_sync_call_set: bool,
5562}
5563
5564impl TableDebug for WaitableSet {
5565 fn type_name() -> &'static str {
5566 "WaitableSet"
5567 }
5568}
5569
5570type RawLower =
5572 Box<dyn FnOnce(&mut dyn VMStore, &mut [MaybeUninit<ValRaw>]) -> Result<()> + Send + Sync>;
5573
5574type RawLift = Box<
5576 dyn FnOnce(&mut dyn VMStore, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>> + Send + Sync,
5577>;
5578
5579type LiftedResult = Box<dyn Any + Send + Sync>;
5583
5584struct DummyResult;
5587
5588#[derive(Default)]
5590pub struct ConcurrentInstanceState {
5591 backpressure: u16,
5593 do_not_enter: bool,
5595 do_not_suspend: bool,
5598 pending: BTreeMap<QualifiedThreadId, GuestCallKind>,
5601}
5602
5603impl ConcurrentInstanceState {
5604 pub fn pending_is_empty(&self) -> bool {
5605 self.pending.is_empty()
5606 }
5607}
5608
5609#[derive(Debug, Copy, Clone)]
5610pub(crate) enum CurrentThread {
5611 Guest(QualifiedThreadId),
5614 Host(TableId<HostTask>),
5616 DeferredHost(QualifiedThreadId),
5619 GuestTask(TableId<GuestTask>),
5623 None,
5626}
5627
5628impl CurrentThread {
5629 fn guest(&self) -> Option<&QualifiedThreadId> {
5630 match self {
5631 Self::Guest(id) => Some(id),
5632 _ => None,
5633 }
5634 }
5635
5636 fn guest_task(&self) -> Option<TableId<GuestTask>> {
5637 match self {
5638 Self::Guest(id) => Some(id.task),
5639 Self::GuestTask(id) => Some(*id),
5640 _ => None,
5641 }
5642 }
5643
5644 fn is_none(&self) -> bool {
5645 matches!(self, Self::None)
5646 }
5647}
5648
5649impl From<QualifiedThreadId> for CurrentThread {
5650 fn from(id: QualifiedThreadId) -> Self {
5651 Self::Guest(id)
5652 }
5653}
5654
5655impl From<TableId<HostTask>> for CurrentThread {
5656 fn from(id: TableId<HostTask>) -> Self {
5657 Self::Host(id)
5658 }
5659}
5660
5661enum Priority {
5662 Switch,
5663 High,
5664 Low,
5665}
5666
5667pub struct ConcurrentState {
5669 unforced_current_thread: CurrentThread,
5675
5676 deferred_host_call_context: Option<CallContext>,
5682
5683 futures: AlwaysMut<Option<FuturesUnordered<HostTaskFuture>>>,
5688 table: AlwaysMut<ResourceTable>,
5690 switch_item: Option<WorkItem>,
5698 high_priority: VecDeque<WorkItem>,
5700 low_priority: VecDeque<WorkItem>,
5702 suspend_reason: Option<SuspendReason>,
5706 worker: Option<StoreFiber<'static>>,
5710 worker_item: Option<WorkerItem>,
5712
5713 global_error_context_ref_counts:
5726 BTreeMap<TypeComponentGlobalErrorContextTableIndex, GlobalErrorContextRefCount>,
5727
5728 interesting_tasks: usize,
5741
5742 interesting_tasks_empty_waker: Option<Waker>,
5746
5747 ready_for_concurrent_call_waker: Option<Waker>,
5752
5753 event_loop_running: bool,
5755}
5756
5757impl Default for ConcurrentState {
5758 fn default() -> Self {
5759 Self {
5760 unforced_current_thread: CurrentThread::None,
5761 deferred_host_call_context: None,
5762 table: AlwaysMut::new(ResourceTable::new()),
5763 futures: AlwaysMut::new(Some(FuturesUnordered::new())),
5764 switch_item: None,
5765 high_priority: VecDeque::new(),
5766 low_priority: VecDeque::new(),
5767 suspend_reason: None,
5768 worker: None,
5769 worker_item: None,
5770 global_error_context_ref_counts: BTreeMap::new(),
5771 interesting_tasks: 0,
5772 interesting_tasks_empty_waker: None,
5773 ready_for_concurrent_call_waker: None,
5774 event_loop_running: false,
5775 }
5776 }
5777}
5778
5779impl ConcurrentState {
5780 pub(crate) fn take_fibers_and_futures(
5797 &mut self,
5798 fibers: &mut Vec<StoreFiber<'static>>,
5799 futures: &mut Vec<FuturesUnordered<HostTaskFuture>>,
5800 ) {
5801 let mut items = Vec::new();
5802 for entry in self.table.get_mut().iter_mut() {
5803 if let Some(set) = entry.downcast_mut::<WaitableSet>() {
5804 for mode in mem::take(&mut set.waiting).into_values() {
5805 match mode {
5806 WaitMode::Fiber(fiber) | WaitMode::Caller { fiber, .. } => {
5807 fibers.push(fiber);
5808 }
5809 WaitMode::Callback(_) => {}
5810 }
5811 }
5812 } else if let Some(thread) = entry.downcast_mut::<GuestThread>() {
5813 if let GuestThreadState::Suspended(fiber) | GuestThreadState::Ready { fiber, .. } =
5814 mem::replace(&mut thread.state, GuestThreadState::Completed)
5815 {
5816 fibers.push(fiber);
5817 }
5818 } else if let Some(task) = entry.downcast_mut::<GuestTask>() {
5819 if let Some(item) = task.switch_item.take() {
5820 items.push(item);
5821 }
5822 }
5823 }
5824
5825 if let Some(fiber) = self.worker.take() {
5826 fibers.push(fiber);
5827 }
5828
5829 let mut handle_item = |item| match item {
5830 WorkItem::ResumeFiber { fiber, .. } => {
5831 fibers.push(fiber);
5832 }
5833 WorkItem::PushFuture(future) => {
5834 self.futures
5835 .get_mut()
5836 .as_mut()
5837 .unwrap()
5838 .push(future.into_inner());
5839 }
5840 WorkItem::ResumeThread { .. }
5841 | WorkItem::GuestCall { .. }
5842 | WorkItem::WorkerFunction(_) => {}
5843 };
5844
5845 for item in items {
5846 handle_item(item);
5847 }
5848 if let Some(item) = self.switch_item.take() {
5849 handle_item(item);
5850 }
5851 for item in mem::take(&mut self.high_priority) {
5852 handle_item(item);
5853 }
5854 for item in mem::take(&mut self.low_priority) {
5855 handle_item(item);
5856 }
5857
5858 if let Some(them) = self.futures.get_mut().take() {
5859 futures.push(them);
5860 }
5861 }
5862
5863 #[cfg(feature = "gc")]
5864 pub(crate) fn trace_fiber_roots(
5865 &mut self,
5866 modules: &ModuleRegistry,
5867 unwind: &dyn Unwind,
5868 gc_roots_list: &mut GcRootsList,
5869 ) {
5870 let ConcurrentState {
5871 table,
5872 worker,
5873 switch_item,
5874 high_priority,
5875 low_priority,
5876
5877 futures: _,
5881
5882 worker_item: _,
5884 unforced_current_thread: _,
5885 deferred_host_call_context: _,
5886 suspend_reason: _,
5887 global_error_context_ref_counts: _,
5888 interesting_tasks: _,
5889 interesting_tasks_empty_waker: _,
5890 ready_for_concurrent_call_waker: _,
5891 event_loop_running: _,
5892 } = self;
5893
5894 for entry in table.get_mut().iter_mut() {
5895 if let Some(set) = entry.downcast_mut::<WaitableSet>() {
5896 for mode in set.waiting.values_mut() {
5897 match mode {
5898 WaitMode::Fiber(fiber) | WaitMode::Caller { fiber, .. } => {
5899 fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5900 }
5901 WaitMode::Callback(_) => {}
5902 }
5903 }
5904 } else if let Some(thread) = entry.downcast_mut::<GuestThread>() {
5905 if let GuestThreadState::Suspended(fiber) | GuestThreadState::Ready { fiber, .. } =
5906 &mut thread.state
5907 {
5908 fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5909 }
5910 }
5911 }
5912
5913 if let Some(fiber) = worker {
5914 fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5915 }
5916
5917 let mut handle_item = |item: &mut WorkItem| match item {
5918 WorkItem::ResumeFiber { fiber, .. } => {
5919 fiber.trace_gc_roots(modules, unwind, gc_roots_list);
5920 }
5921 WorkItem::PushFuture(_future) => {
5922 }
5925 WorkItem::ResumeThread { .. }
5926 | WorkItem::GuestCall { .. }
5927 | WorkItem::WorkerFunction(_) => {}
5928 };
5929
5930 if let Some(item) = switch_item {
5931 handle_item(item);
5932 }
5933 for item in high_priority {
5934 handle_item(item);
5935 }
5936 for item in low_priority {
5937 handle_item(item);
5938 }
5939 }
5940
5941 fn push<V: Send + Sync + 'static>(
5942 &mut self,
5943 value: V,
5944 ) -> Result<TableId<V>, ResourceTableError> {
5945 self.table.get_mut().push(value).map(TableId::from)
5946 }
5947
5948 fn get_mut<V: 'static>(&mut self, id: TableId<V>) -> Result<&mut V, ResourceTableError> {
5949 self.table.get_mut().get_mut(&Resource::from(id))
5950 }
5951
5952 pub fn add_child<T: 'static, U: 'static>(
5953 &mut self,
5954 child: TableId<T>,
5955 parent: TableId<U>,
5956 ) -> Result<(), ResourceTableError> {
5957 self.table
5958 .get_mut()
5959 .add_child(Resource::from(child), Resource::from(parent))
5960 }
5961
5962 pub fn remove_child<T: 'static, U: 'static>(
5963 &mut self,
5964 child: TableId<T>,
5965 parent: TableId<U>,
5966 ) -> Result<(), ResourceTableError> {
5967 self.table
5968 .get_mut()
5969 .remove_child(Resource::from(child), Resource::from(parent))
5970 }
5971
5972 fn delete<V: 'static>(&mut self, id: TableId<V>) -> Result<V, ResourceTableError> {
5973 self.table.get_mut().delete(Resource::from(id))
5974 }
5975
5976 fn push_future(&mut self, future: HostTaskFuture) {
5977 self.push_high_priority(WorkItem::PushFuture(AlwaysMut::new(future)));
5984 }
5985
5986 fn set_switch_item(&mut self, item: WorkItem) -> Result<()> {
5987 log::trace!("set switch item: {item:?}");
5988
5989 if self.switch_item.is_some() {
5990 bail_bug!("switch item already set");
5991 }
5992
5993 self.switch_item = Some(item);
5994
5995 Ok(())
5996 }
5997
5998 fn push_high_priority(&mut self, item: WorkItem) {
5999 log::trace!("push high priority: {item:?}");
6000 self.high_priority.push_front(item);
6001 }
6002
6003 fn push_low_priority(&mut self, item: WorkItem) {
6004 log::trace!("push low priority: {item:?}");
6005 self.low_priority.push_front(item);
6006 }
6007
6008 fn push_work_item(&mut self, item: WorkItem, priority: Priority) -> Result<()> {
6009 match priority {
6010 Priority::Switch => self.set_switch_item(item)?,
6011 Priority::High => self.push_high_priority(item),
6012 Priority::Low => self.push_low_priority(item),
6013 }
6014
6015 Ok(())
6016 }
6017
6018 fn promote_instance_local_thread_work_item(
6019 &mut self,
6020 current_instance: RuntimeInstance,
6021 ) -> Result<bool> {
6022 log::trace!("promote thread work items for {current_instance:?}");
6023
6024 self.promote_work_item_matching(|item: &WorkItem| {
6025 let result = match item {
6026 WorkItem::ResumeThread { instance, .. }
6027 | WorkItem::ResumeFiber { instance, .. }
6028 | WorkItem::GuestCall { instance, .. } => *instance == current_instance,
6029 _ => false,
6030 };
6031
6032 log::trace!("candidate {item:?}: {result}");
6033 result
6034 })
6035 }
6036
6037 fn promote_thread_work_item(&mut self, thread: QualifiedThreadId) -> Result<bool> {
6038 self.promote_work_item_matching(|item: &WorkItem| match item {
6039 WorkItem::ResumeThread {
6040 thread: item_thread,
6041 ..
6042 }
6043 | WorkItem::GuestCall {
6044 call:
6045 GuestCall {
6046 thread: item_thread,
6047 ..
6048 },
6049 ..
6050 } => *item_thread == thread,
6051 _ => false,
6052 })
6053 }
6054
6055 fn promote_work_item_matching<F>(&mut self, mut predicate: F) -> Result<bool>
6056 where
6057 F: FnMut(&WorkItem) -> bool,
6058 {
6059 for item in mem::take(&mut self.high_priority).into_iter().rev() {
6064 if self.switch_item.is_none() && predicate(&item) {
6065 self.set_switch_item(item)?;
6066 } else {
6067 self.push_high_priority(item);
6068 }
6069 }
6070
6071 if self.switch_item.is_none() {
6072 for item in mem::take(&mut self.low_priority).into_iter().rev() {
6073 if self.switch_item.is_none() && predicate(&item) {
6074 self.set_switch_item(item)?;
6075 } else {
6076 self.push_low_priority(item);
6077 }
6078 }
6079 }
6080
6081 Ok(self.switch_item.is_some())
6082 }
6083
6084 pub fn call_context(&mut self, task: Scope) -> Result<&mut CallContext> {
6087 match task {
6088 Scope::HostId(task) => {
6089 let task: TableId<HostTask> = TableId::new(task);
6090 Ok(&mut self.get_mut(task)?.call_context)
6091 }
6092 Scope::Id(task) => {
6093 let task: TableId<GuestTask> = TableId::new(task);
6094 Ok(&mut self.get_mut(task)?.call_context)
6095 }
6096 }
6097 }
6098
6099 pub(crate) fn deferred_host_call_context(&mut self) -> Option<&mut CallContext> {
6100 self.deferred_host_call_context.as_mut()
6101 }
6102
6103 fn futures_mut(&mut self) -> Result<&mut FuturesUnordered<HostTaskFuture>> {
6104 match self.futures.get_mut().as_mut() {
6105 Some(f) => Ok(f),
6106 None => bail_bug!("futures field of concurrent state is currently taken"),
6107 }
6108 }
6109
6110 pub(crate) fn table(&mut self) -> &mut ResourceTable {
6111 self.table.get_mut()
6112 }
6113
6114 fn parent(&mut self, cur: CurrentThread) -> Option<CurrentThread> {
6116 let task = match cur {
6117 CurrentThread::GuestTask(task) => task,
6118 CurrentThread::Guest(thread) => thread.task,
6119 CurrentThread::Host(id) => {
6120 return Some(CurrentThread::GuestTask(self.get_mut(id).ok()?.caller));
6121 }
6122 CurrentThread::DeferredHost(caller) => return Some(caller.into()),
6123 CurrentThread::None => return None,
6124 };
6125 let task = self.get_mut(task).ok()?;
6126 Some(match task.caller {
6127 Caller::Host { caller, .. } => caller.map_or(CurrentThread::None, CurrentThread::Host),
6128 Caller::Guest { thread } => thread.into(),
6129 })
6130 }
6131
6132 fn debug_assert_deferred_host_invariant(&self) {
6133 debug_assert_eq!(
6134 self.deferred_host_call_context.is_some(),
6135 matches!(self.unforced_current_thread, CurrentThread::DeferredHost(_)),
6136 "a deferred host thread and call context must exist together",
6137 );
6138 }
6139
6140 fn materialize_host_task(&mut self) -> Result<CurrentThread> {
6141 self.debug_assert_deferred_host_invariant();
6142 let caller = match self.unforced_current_thread {
6143 CurrentThread::DeferredHost(caller) => caller,
6144 thread => return Ok(thread),
6145 };
6146
6147 let task = self.push(HostTask::new(caller.task, HostTaskState::CalleeStarted))?;
6149 let call_context = self
6150 .deferred_host_call_context
6151 .take()
6152 .expect("deferred host call context should be present");
6153 self.get_mut(task)
6154 .expect("newly inserted host task should be present")
6155 .call_context = call_context;
6156 self.unforced_current_thread = CurrentThread::Host(task);
6157 self.debug_assert_deferred_host_invariant();
6158 log::trace!("new host task materialized {task:?}");
6159 Ok(CurrentThread::Host(task))
6160 }
6161
6162 fn materialize_current_host_task_id(&mut self) -> Result<Option<TableId<HostTask>>> {
6163 match self.materialize_host_task()? {
6164 CurrentThread::Host(id) => Ok(Some(id)),
6165 CurrentThread::None => Ok(None),
6166 CurrentThread::Guest(_) | CurrentThread::GuestTask(_) => {
6167 bail_bug!("tried to materialize a host task id from a guest thread")
6168 }
6169 CurrentThread::DeferredHost(_) => {
6170 bail_bug!(
6171 "current thread is a deferred host thread which should have been materialized"
6172 )
6173 }
6174 }
6175 }
6176
6177 pub(crate) fn materialize_current_scope(&mut self) -> Result<Scope> {
6178 match self.materialize_host_task()? {
6179 CurrentThread::Host(id) => Ok(Scope::HostId(id.rep())),
6180 _ => bail_bug!("current scope is not a deferred host scope"),
6181 }
6182 }
6183}
6184
6185fn for_any_lower<
6188 F: FnOnce(&mut dyn VMStore, &mut [MaybeUninit<ValRaw>]) -> Result<()> + Send + Sync,
6189>(
6190 fun: F,
6191) -> F {
6192 fun
6193}
6194
6195fn for_any_lift<
6197 F: FnOnce(&mut dyn VMStore, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>> + Send + Sync,
6198>(
6199 fun: F,
6200) -> F {
6201 fun
6202}
6203
6204fn check_ambient_store(id: StoreId) {
6205 let message = "\
6206 `Future`s which depend on asynchronous component tasks, streams, or \
6207 futures to complete may only be polled from the event loop of the \
6208 store to which they belong. Please use \
6209 `StoreContextMut::{run_concurrent,spawn}` to poll or await them.\
6210 ";
6211 tls::try_get(|store| {
6212 let matched = match store {
6213 tls::TryGet::Some(store) => store.id() == id,
6214 tls::TryGet::Taken | tls::TryGet::None => false,
6215 };
6216
6217 if !matched {
6218 panic!("{message}")
6219 }
6220 });
6221}
6222
6223fn unpack_callback_code(code: u32) -> (u32, u32) {
6224 (code & 0xF, code >> 4)
6225}
6226
6227struct WaitableCheckParams {
6231 set: TableId<WaitableSet>,
6232 options: OptionsIndex,
6233 payload: u32,
6234}
6235
6236enum WaitableCheck {
6239 Wait,
6240 Poll,
6241}
6242
6243#[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)]
6252pub struct GuestTaskId(TableId<GuestTask>);
6253
6254pub(crate) struct PreparedCall<R> {
6256 handle: Func,
6258 thread: QualifiedThreadId,
6260 param_count: usize,
6262 rx: oneshot::Receiver<LiftedResult>,
6265 runtime_instance: RuntimeInstance,
6267 _phantom: PhantomData<R>,
6268}
6269
6270impl<R> PreparedCall<R> {
6271 pub(crate) fn task_id(&self) -> TaskId {
6273 TaskId {
6274 task: self.thread.task,
6275 runtime_instance: self.runtime_instance,
6276 }
6277 }
6278}
6279
6280pub(crate) struct TaskId {
6282 task: TableId<GuestTask>,
6283 runtime_instance: RuntimeInstance,
6284}
6285
6286impl TaskId {
6287 pub(crate) fn host_future_dropped(&self, store: &mut StoreOpaque) -> Result<()> {
6293 let task = store.concurrent_state_mut()?.get_mut(self.task)?;
6294 let delete = if !task.already_lowered_parameters() {
6295 store.cancel_guest_subtask_without_lowered_parameters(
6296 self.runtime_instance,
6297 self.task,
6298 )?;
6299 true
6300 } else {
6301 task.host_future_state = HostFutureState::Dropped;
6302 task.ready_to_delete()
6303 };
6304 if delete {
6305 Waitable::Guest(self.task).delete_from(store.concurrent_state_mut()?)?
6306 }
6307 Ok(())
6308 }
6309}
6310
6311pub(crate) fn prepare_call<T, R>(
6317 mut store: StoreContextMut<T>,
6318 handle: Func,
6319 param_count: usize,
6320 host_future_present: bool,
6321 lower_params: impl FnOnce(StoreContextMut<T>, &mut [MaybeUninit<ValRaw>]) -> Result<()>
6322 + Send
6323 + Sync
6324 + 'static,
6325 lift_result: impl FnOnce(&mut StoreOpaque, &[ValRaw]) -> Result<Box<dyn Any + Send + Sync>>
6326 + Send
6327 + Sync
6328 + 'static,
6329) -> Result<PreparedCall<R>> {
6330 if !store.0.may_enter() {
6331 bail!(Trap::CannotEnterComponent);
6332 }
6333
6334 let (options, _flags, ty, raw_options) = handle.abi_info(store.0);
6335
6336 let instance = handle.instance().id().get(store.0);
6337 let options = &instance.component().env_component().options[options];
6338 let ty = &instance.component().types()[ty];
6339 let async_typed = ty.async_;
6340 let async_lifted = raw_options.async_;
6341 let task_return_type = ty.results;
6342 let component_instance = raw_options.instance;
6343 let callback = options.callback.map(|i| instance.runtime_callback(i));
6344 let memory = options
6345 .memory()
6346 .map(|i| instance.runtime_memory(i))
6347 .map(SendSyncPtr::new);
6348 let string_encoding = options.string_encoding;
6349 let token = StoreToken::new(store.as_context_mut());
6350 let caller = store.0.materialize_host_task_id()?;
6351 let state = store.0.concurrent_state_mut()?;
6352
6353 let (tx, rx) = oneshot::channel();
6354
6355 let instance = handle.instance().runtime_instance(component_instance);
6356 let thread = GuestTask::new(
6357 state,
6358 Box::new(for_any_lower(move |store, params| {
6359 lower_params(token.as_context_mut(store), params)
6360 })),
6361 LiftResult {
6362 lift: Box::new(for_any_lift(move |store, result| {
6363 lift_result(store, result)
6364 })),
6365 ty: task_return_type,
6366 memory,
6367 string_encoding,
6368 },
6369 Caller::Host {
6370 tx: Some(tx),
6371 host_future_present,
6372 caller,
6373 },
6374 callback.map(|callback| {
6375 let callback = SendSyncPtr::new(callback);
6376 let instance = handle.instance();
6377 Box::new(move |store: &mut dyn VMStore, event, handle| {
6378 let store = token.as_context_mut(store);
6379 unsafe { instance.call_callback(store, callback, event, handle) }
6382 }) as CallbackFn
6383 }),
6384 instance,
6385 async_typed,
6386 async_lifted,
6387 )?;
6388
6389 Ok(PreparedCall {
6390 handle,
6391 thread,
6392 param_count,
6393 runtime_instance: instance,
6394 rx,
6395 _phantom: PhantomData,
6396 })
6397}
6398
6399pub(crate) struct StagedCall<R> {
6400 store: StoreId,
6401 task: TableId<GuestTask>,
6402 rx: oneshot::Receiver<LiftedResult>,
6403 _marker: PhantomData<fn() -> R>,
6404}
6405
6406impl<R> StagedCall<R> {
6407 pub(crate) fn new<T: 'static>(
6414 mut store: StoreContextMut<T>,
6415 prepared: PreparedCall<R>,
6416 ) -> Result<StagedCall<R>> {
6417 let PreparedCall {
6418 handle,
6419 thread,
6420 param_count,
6421 rx,
6422 ..
6423 } = prepared;
6424
6425 stage_call0(store.as_context_mut(), handle, thread, param_count)?;
6426
6427 Ok(StagedCall {
6428 store: store.0.id(),
6429 task: thread.task,
6430 rx,
6431 _marker: PhantomData,
6432 })
6433 }
6434
6435 fn task(&self) -> GuestTaskId {
6436 GuestTaskId(self.task)
6437 }
6438}
6439
6440impl<R> Future for StagedCall<R>
6441where
6442 R: 'static,
6443{
6444 type Output = Result<R>;
6445
6446 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
6447 check_ambient_store(self.store);
6448 Pin::new(&mut self.rx).poll(cx).map(|result| match result {
6449 Ok(r) => match r.downcast() {
6450 Ok(r) => Ok(*r),
6451 Err(_) => bail_bug!("wrong type of value produced"),
6452 },
6453 Err(oneshot::Canceled) => bail_bug!("channel erroneously dropped"),
6454 })
6455 }
6456}
6457
6458fn stage_call0<T: 'static>(
6461 store: StoreContextMut<T>,
6462 handle: Func,
6463 guest_thread: QualifiedThreadId,
6464 param_count: usize,
6465) -> Result<()> {
6466 let (_options, _, _ty, raw_options) = handle.abi_info(store.0);
6467 let is_concurrent = raw_options.async_;
6468 let callback = raw_options.callback;
6469 let instance = handle.instance();
6470 let callee = handle.lifted_core_func(store.0);
6471 let post_return = raw_options
6472 .post_return
6473 .map(|i| instance.id().get(store.0).runtime_post_return(i));
6474 let callback = callback.map(|i| {
6475 let instance = instance.id().get(store.0);
6476 SendSyncPtr::new(instance.runtime_callback(i))
6477 });
6478
6479 log::trace!("queueing call {guest_thread:?}");
6480
6481 unsafe {
6485 instance.stage_call(
6486 store,
6487 guest_thread,
6488 SendSyncPtr::new(callee),
6489 param_count,
6490 1,
6491 is_concurrent,
6492 callback,
6493 post_return.map(SendSyncPtr::new),
6494 true,
6495 )
6496 }
6497}