wasmer_wasix/syscalls/wasix/
thread_spawn.rs

1use std::f32::consts::E;
2
3use super::*;
4#[cfg(feature = "journal")]
5use crate::journal::JournalEffector;
6use crate::{
7    WasiThreadHandle,
8    os::task::thread::WasiMemoryLayout,
9    runtime::{
10        TaintReason,
11        task_manager::{TaskWasm, TaskWasmRunProperties},
12    },
13    syscalls::*,
14};
15
16use wasmer::Memory;
17use wasmer_wasix_types::wasi::ThreadStart;
18
19/// ### `thread_spawn()`
20/// Creates a new thread by spawning that shares the same
21/// memory address space, file handles and main event loops.
22///
23/// ## Parameters
24///
25/// * `start_ptr` - Pointer to the structure that describes the thread to be launched
26/// * `ret_tid` - ID of the thread that was launched
27///
28/// ## Return
29///
30/// Returns the thread index of the newly created thread
31/// (indices always start from the same value as `pid` and increments in steps)
32#[instrument(level = "trace", skip_all, ret)]
33pub fn thread_spawn_v2<M: MemorySize>(
34    mut ctx: FunctionEnvMut<'_, WasiEnv>,
35    start_ptr: WasmPtr<ThreadStart<M>, M>,
36    ret_tid: WasmPtr<Tid, M>,
37) -> Result<Errno, WasiError> {
38    WasiEnv::do_pending_operations(&mut ctx)?;
39
40    // Create the thread
41    let tid = wasi_try_ok!(thread_spawn_internal_from_wasi(&mut ctx, start_ptr));
42
43    // Success
44    let memory = unsafe { ctx.data().memory_view(&ctx) };
45    wasi_try_mem_ok!(ret_tid.write(&memory, tid));
46
47    tracing::debug!(
48        tid,
49        from_tid = ctx.data().thread.id().raw(),
50        "spawned new thread"
51    );
52
53    Ok(Errno::Success)
54}
55
56pub fn thread_spawn_internal_from_wasi<M: MemorySize>(
57    ctx: &mut FunctionEnvMut<'_, WasiEnv>,
58    start_ptr: WasmPtr<ThreadStart<M>, M>,
59) -> Result<Tid, Errno> {
60    // Now we use the environment and memory references
61    let env = ctx.data();
62    let memory = unsafe { env.memory_view(&ctx) };
63    let runtime = env.runtime.clone();
64    let tasks = env.tasks().clone();
65    let start_ptr_offset = start_ptr.offset();
66
67    // Read the properties about the stack which we will use for asyncify
68    let layout = {
69        let start: ThreadStart<M> = start_ptr.read(&memory).map_err(mem_error_to_wasi)?;
70        let stack_upper: u64 = start.stack_upper.into();
71        let stack_size: u64 = start.stack_size.into();
72        let guard_size: u64 = start.guard_size.into();
73        let tls_base: u64 = start.tls_base.into();
74        let stack_lower = stack_upper - stack_size;
75
76        WasiMemoryLayout {
77            stack_upper,
78            stack_lower,
79            guard_size,
80            stack_size,
81            tls_base: Some(tls_base),
82        }
83    };
84    tracing::trace!(
85        from_tid = env.thread.id().raw(),
86        "thread_spawn with layout {:?}",
87        layout
88    );
89
90    // Create the handle that represents this thread
91    let thread_start = ThreadStartType::ThreadSpawn {
92        start_ptr: start_ptr_offset.into(),
93    };
94    let mut thread_handle = match env.process.new_thread(layout.clone(), thread_start) {
95        Ok(h) => Arc::new(h),
96        Err(err) => {
97            error!(
98                stack_base = layout.stack_lower,
99                "failed to create thread handle",
100            );
101            // TODO: evaluate the appropriate error code, document it in the spec.
102            return Err(Errno::Access);
103        }
104    };
105    let thread_id: Tid = thread_handle.id().into();
106    Span::current().record("tid", thread_id);
107
108    // Spawn the thread
109    thread_spawn_internal_using_layout::<M>(ctx, thread_handle, layout, start_ptr_offset, None)?;
110
111    // Success
112    Ok(thread_id)
113}
114
115pub fn thread_spawn_internal_using_layout<M: MemorySize>(
116    ctx: &mut FunctionEnvMut<'_, WasiEnv>,
117    thread_handle: Arc<WasiThreadHandle>,
118    layout: WasiMemoryLayout,
119    start_ptr_offset: M::Offset,
120    rewind_state: Option<(RewindState, RewindResultType)>,
121) -> Result<(), Errno> {
122    // We extract the memory which will be passed to the thread
123    let func_env = ctx.as_ref();
124    let mut store = ctx.as_store_mut();
125    let env = func_env.as_ref(&store);
126    let tasks = env.tasks().clone();
127
128    let env_inner = env.inner();
129    let module_handles = env_inner.main_module_instance_handles();
130
131    let thread_memory = module_handles.memory_clone();
132    let linker = env_inner.linker().cloned();
133
134    // We capture some local variables
135    let state = env.state.clone();
136    let mut thread_env = env.clone();
137    thread_env.thread = thread_handle.as_thread();
138    thread_env.layout = layout;
139
140    // TODO: Currently asynchronous threading does not work with multi
141    //       threading in JS but it does work for the main thread. This will
142    //       require more work to find out why.
143    thread_env.enable_deep_sleep = if cfg!(feature = "js") {
144        false
145    } else {
146        unsafe { env.capable_of_deep_sleep() }
147    };
148
149    // This next function gets a context for the local thread and then
150    // calls into the process
151    let mut execute_module = {
152        let thread_handle = thread_handle;
153        move |ctx: WasiFunctionEnv, mut store: Store| {
154            // Call the thread
155            call_module::<M>(ctx, store, start_ptr_offset, thread_handle, rewind_state)
156        }
157    };
158
159    // If the process does not export a thread spawn function then obviously
160    // we can't spawn a background thread
161    if module_handles.thread_spawn.is_none() {
162        warn!("thread failed - the program does not export a `wasi_thread_start` function");
163        return Err(Errno::Notcapable);
164    }
165    let thread_module = module_handles.module_clone();
166    let spawn_type = match linker {
167        Some(linker) => crate::runtime::SpawnType::NewLinkerInstanceGroup(linker, func_env, store),
168        None => crate::runtime::SpawnType::ShareMemory(thread_memory, store.as_store_ref()),
169    };
170
171    // Now spawn a thread
172    trace!("threading: spawning background thread");
173    let run = move |props: TaskWasmRunProperties| {
174        execute_module(props.ctx, props.store);
175    };
176
177    let mut task_wasm = TaskWasm::new(Box::new(run), thread_env, thread_module, false, false)
178        .with_memory(spawn_type);
179
180    tasks.task_wasm(task_wasm).map_err(Into::<Errno>::into)?;
181
182    // Success
183    Ok(())
184}
185
186// This function calls into the module
187fn call_module_internal<M: MemorySize>(
188    env: &WasiFunctionEnv,
189    store: &mut Store,
190    start_ptr_offset: M::Offset,
191) -> Result<(), DeepSleepWork> {
192    // We either call the reactor callback or the thread spawn callback
193    //trace!("threading: invoking thread callback (reactor={})", reactor);
194
195    // Note: we ensure both unwraps can happen before getting to this point
196    let spawn = env
197        .data(&store)
198        .inner()
199        .main_module_instance_handles()
200        .thread_spawn
201        .clone()
202        .unwrap();
203    let tid = env.data(&store).tid();
204    let thread_result = spawn.call(
205        store,
206        tid.raw().try_into().map_err(|_| Errno::Overflow).unwrap(),
207        start_ptr_offset
208            .try_into()
209            .map_err(|_| Errno::Overflow)
210            .unwrap(),
211    );
212    trace!("callback finished (ret={:?})", thread_result);
213
214    let exit_code = handle_thread_result(env, store, thread_result)?;
215
216    // Clean up the environment on exit
217    env.on_exit(store, exit_code);
218    Ok(())
219}
220
221fn handle_thread_result(
222    env: &WasiFunctionEnv,
223    store: &mut Store,
224    err: Result<(), RuntimeError>,
225) -> Result<Option<ExitCode>, DeepSleepWork> {
226    let tid = env.data(&store).tid();
227    let pid = env.data(&store).pid();
228    let Err(err) = err else {
229        trace!("thread exited cleanly without calling thread_exit");
230        return Ok(None);
231    };
232    match err.downcast::<WasiError>() {
233        Ok(WasiError::ThreadExit) => {
234            trace!("thread exited cleanly");
235            Ok(None)
236        }
237        Ok(WasiError::Exit(code)) => {
238            trace!(exit_code = ?code, "thread requested exit");
239            if !code.is_success() {
240                // TODO: Why do we need to taint the runtime on a non-zero exit code? Why not also for zero?
241                env.data(&store)
242                    .runtime
243                    .on_taint(TaintReason::NonZeroExitCode(code));
244            };
245            Ok(Some(code))
246        }
247        Ok(WasiError::DeepSleep(deep)) => {
248            trace!("entered a deep sleep");
249            Err(deep)
250        }
251        Ok(WasiError::UnknownWasiVersion) => {
252            eprintln!(
253                "Thread {tid} of process {pid} failed because it has an unknown wasix version"
254            );
255            env.data(&store)
256                .runtime
257                .on_taint(TaintReason::UnknownWasiVersion);
258            Ok(Some(ExitCode::from(129)))
259        }
260        Ok(WasiError::DlSymbolResolutionFailed(symbol)) => {
261            eprintln!("Thread {tid} of process {pid} failed to find required symbol: {symbol}");
262            env.data(&store)
263                .runtime
264                .on_taint(TaintReason::DlSymbolResolutionFailed(symbol.clone()));
265            Ok(Some(ExitCode::from(129)))
266        }
267        Err(err) => {
268            eprintln!("Thread {tid} of process {pid} failed with runtime error: {err}");
269            env.data(&store)
270                .runtime
271                .on_taint(TaintReason::RuntimeError(err));
272            Ok(Some(ExitCode::from(129)))
273        }
274    }
275}
276
277/// Calls the module
278fn call_module<M: MemorySize>(
279    mut ctx: WasiFunctionEnv,
280    mut store: Store,
281    start_ptr_offset: M::Offset,
282    thread_handle: Arc<WasiThreadHandle>,
283    rewind_state: Option<(RewindState, RewindResultType)>,
284) {
285    let env = ctx.data(&store);
286    let tasks = env.tasks().clone();
287
288    // If we need to rewind then do so
289    if let Some((rewind_state, rewind_result)) = rewind_state {
290        let mut ctx = ctx.env.clone().into_mut(&mut store);
291        let res = rewind_ext::<M>(
292            &mut ctx,
293            Some(rewind_state.memory_stack),
294            rewind_state.rewind_stack,
295            rewind_state.store_data,
296            rewind_result,
297        );
298        if res != Errno::Success {
299            return;
300        }
301    }
302
303    // Now invoke the module
304    let ret = call_module_internal::<M>(&ctx, &mut store, start_ptr_offset);
305
306    // If it went to deep sleep then we need to handle that
307    if let Err(deep) = ret {
308        // Create the callback that will be invoked when the thread respawns after a deep sleep
309        let rewind = deep.rewind;
310        let respawn = {
311            let tasks = tasks.clone();
312            move |ctx, store, trigger_res| {
313                // Call the thread
314                call_module::<M>(
315                    ctx,
316                    store,
317                    start_ptr_offset,
318                    thread_handle,
319                    Some((rewind, RewindResultType::RewindWithResult(trigger_res))),
320                );
321            }
322        };
323
324        /// Spawns the WASM process after a trigger
325        unsafe {
326            tasks.resume_wasm_after_poller(Box::new(respawn), ctx, store, deep.trigger)
327        };
328        return;
329    };
330    // I don't think we need to do this explicitly, but it was done before refactoring so we keep it for now.
331    drop(thread_handle);
332}
wasmer_wasix/syscalls/wasix/thread_spawn.rs

wasmer_wasix/syscalls/wasix/
thread_spawn.rs
