wasmer_wasix/syscalls/wasix/
thread_spawn.rs

1use super::*;
2#[cfg(feature = "journal")]
3use crate::journal::JournalEffector;
4use crate::{
5    WasiThreadHandle,
6    os::task::thread::WasiMemoryLayout,
7    runtime::{
8        TaintReason,
9        task_manager::{TaskWasm, TaskWasmRunProperties},
10    },
11    state::context_switching::ContextSwitchingEnvironment,
12    syscalls::*,
13};
14
15use wasmer::Memory;
16use wasmer_wasix_types::wasi::ThreadStart;
17
18/// ### `thread_spawn()`
19/// Creates a new thread by spawning that shares the same
20/// memory address space, file handles and main event loops.
21///
22/// ## Parameters
23///
24/// * `start_ptr` - Pointer to the structure that describes the thread to be launched
25/// * `ret_tid` - ID of the thread that was launched
26///
27/// ## Return
28///
29/// Returns the thread index of the newly created thread
30/// (indices always start from the same value as `pid` and increments in steps)
31#[instrument(level = "trace", skip_all, ret)]
32pub fn thread_spawn_v2<M: MemorySize>(
33    mut ctx: FunctionEnvMut<'_, WasiEnv>,
34    start_ptr: WasmPtr<ThreadStart<M>, M>,
35    ret_tid: WasmPtr<Tid, M>,
36) -> Result<Errno, WasiError> {
37    WasiEnv::do_pending_operations(&mut ctx)?;
38
39    // Create the thread
40    let tid = wasi_try_ok!(thread_spawn_internal_from_wasi(&mut ctx, start_ptr));
41
42    // Success
43    let memory = unsafe { ctx.data().memory_view(&ctx) };
44    wasi_try_mem_ok!(ret_tid.write(&memory, tid));
45
46    tracing::debug!(
47        tid,
48        from_tid = ctx.data().thread.id().raw(),
49        "spawned new thread"
50    );
51
52    Ok(Errno::Success)
53}
54
55pub fn thread_spawn_internal_from_wasi<M: MemorySize>(
56    ctx: &mut FunctionEnvMut<'_, WasiEnv>,
57    start_ptr: WasmPtr<ThreadStart<M>, M>,
58) -> Result<Tid, Errno> {
59    // Now we use the environment and memory references
60    let env = ctx.data();
61    let memory = unsafe { env.memory_view(&ctx) };
62    let runtime = env.runtime.clone();
63    let tasks = env.tasks().clone();
64    let start_ptr_offset = start_ptr.offset();
65
66    // Read the properties about the stack which we will use for asyncify
67    let layout = {
68        let start: ThreadStart<M> = start_ptr.read(&memory).map_err(mem_error_to_wasi)?;
69        let stack_upper: u64 = start.stack_upper.into();
70        let stack_size: u64 = start.stack_size.into();
71        let guard_size: u64 = start.guard_size.into();
72        let tls_base: u64 = start.tls_base.into();
73        let stack_lower = stack_upper.checked_sub(stack_size).ok_or(Errno::Inval)?;
74
75        WasiMemoryLayout {
76            stack_upper,
77            stack_lower,
78            guard_size,
79            stack_size,
80            tls_base: Some(tls_base),
81        }
82    };
83    tracing::trace!(
84        from_tid = env.thread.id().raw(),
85        "thread_spawn with layout {:?}",
86        layout
87    );
88
89    // Create the handle that represents this thread
90    let thread_start = ThreadStartType::ThreadSpawn {
91        start_ptr: start_ptr_offset.into(),
92    };
93    let mut thread_handle = match env.process.new_thread(layout.clone(), thread_start) {
94        Ok(h) => Arc::new(h),
95        Err(err) => {
96            error!(
97                stack_base = layout.stack_lower,
98                "failed to create thread handle",
99            );
100            // TODO: evaluate the appropriate error code, document it in the spec.
101            return Err(Errno::Access);
102        }
103    };
104    let thread_id: Tid = thread_handle.id().into();
105    Span::current().record("tid", thread_id);
106
107    // Spawn the thread
108    thread_spawn_internal_using_layout::<M>(ctx, thread_handle, layout, start_ptr_offset, None)?;
109
110    // Success
111    Ok(thread_id)
112}
113
114/// Largest thread ID that may be handed to `wasi_thread_start`.
115///
116/// The wasi-threads spec restricts TIDs to `[1, 2^29)`: the sign bit is used
117/// by `thread-spawn` to report errors, and libc implementations reserve the
118/// next bits in their lock words (musl stores the owner TID in the low 30
119/// bits and uses `0x3fffffff` as a sentinel).
120const MAX_THREAD_ID: u32 = (1 << 29) - 1;
121
122/// Arguments for the guest's `wasi_thread_start(tid: i32, start_arg: i32)`.
123///
124/// They are validated when the thread is spawned so that a bad value is
125/// reported to the caller as an errno instead of failing in the new thread.
126#[derive(Debug, Clone, Copy, PartialEq, Eq)]
127struct ThreadStartArgs {
128    tid: i32,
129    start_arg: i32,
130}
131
132impl ThreadStartArgs {
133    fn new(tid: WasiThreadId, start_ptr: u64) -> Result<Self, Errno> {
134        let tid = tid.raw();
135        if !(1..=MAX_THREAD_ID).contains(&tid) {
136            tracing::warn!(
137                tid,
138                "thread ID is outside the range allowed by wasi-threads"
139            );
140            return Err(Errno::Again);
141        }
142        let start_arg = wasm32_ptr_to_i32_arg(start_ptr).inspect_err(|_| {
143            tracing::warn!(
144                start_ptr,
145                "thread start pointer does not fit into the i32 argument of wasi_thread_start"
146            );
147        })?;
148
149        Ok(Self {
150            tid: tid.cast_signed(),
151            start_arg,
152        })
153    }
154
155    fn to_values(self) -> Vec<Value> {
156        vec![Value::I32(self.tid), Value::I32(self.start_arg)]
157    }
158}
159
160pub fn thread_spawn_internal_using_layout<M: MemorySize>(
161    ctx: &mut FunctionEnvMut<'_, WasiEnv>,
162    thread_handle: Arc<WasiThreadHandle>,
163    layout: WasiMemoryLayout,
164    start_ptr_offset: M::Offset,
165    rewind_state: Option<(RewindState, RewindResultType)>,
166) -> Result<(), Errno> {
167    let start_args = ThreadStartArgs::new(thread_handle.id(), start_ptr_offset.into())?;
168
169    // We extract the memory which will be passed to the thread
170    let func_env = ctx.as_ref();
171    let mut store = ctx.as_store_mut();
172    let env = func_env.as_ref(&store);
173    let tasks = env.tasks().clone();
174
175    let env_inner = env.inner();
176    let module_handles = env_inner.main_module_instance_handles();
177
178    let thread_memory = module_handles.memory_clone();
179    let linker = env_inner.linker().cloned();
180
181    // We capture some local variables
182    let state = env.state.clone();
183    let mut thread_env = env.clone();
184    thread_env.thread = thread_handle.as_thread();
185    thread_env.layout = layout;
186
187    // TODO: Currently asynchronous threading does not work with multi
188    //       threading in JS but it does work for the main thread. This will
189    //       require more work to find out why.
190    thread_env.enable_deep_sleep = if cfg!(feature = "js") {
191        false
192    } else {
193        unsafe { env.capable_of_deep_sleep() }
194    };
195
196    // This next function gets a context for the local thread and then
197    // calls into the process
198    let mut execute_module = {
199        let thread_handle = thread_handle;
200        move |ctx: WasiFunctionEnv, mut store: Store| {
201            // Call the thread
202            call_module::<M>(ctx, store, start_args, thread_handle, rewind_state)
203        }
204    };
205
206    // If the process does not export a thread spawn function then obviously
207    // we can't spawn a background thread
208    if module_handles.thread_spawn.is_none() {
209        warn!("thread failed - the program does not export a `wasi_thread_start` function");
210        return Err(Errno::Notcapable);
211    }
212    let thread_module = module_handles.module_clone();
213    let spawn_type = match linker {
214        Some(linker) => {
215            let instance_group_data = linker.prepare_for_instance_group(ctx).map_err(|e| {
216                tracing::warn!("failed to prepare linker for thread spawn: {e}");
217                Errno::Notcapable
218            })?;
219            crate::runtime::SpawnType::NewLinkerInstanceGroup(instance_group_data)
220        }
221        None => crate::runtime::SpawnType::AttachMemory(
222            thread_memory.as_shared(&store).ok_or_else(|| {
223                tracing::warn!("Memory must be shared for thread spawning to work");
224                Errno::Memviolation
225            })?,
226        ),
227    };
228
229    // Now spawn a thread
230    trace!("threading: spawning background thread");
231    let run = move |props: TaskWasmRunProperties| {
232        execute_module(props.ctx, props.store);
233    };
234
235    let mut task_wasm = TaskWasm::new(Box::new(run), thread_env, thread_module, false, false)
236        .with_memory(spawn_type);
237
238    tasks.task_wasm(task_wasm).map_err(Into::<Errno>::into)?;
239
240    // Success
241    Ok(())
242}
243
244// This function calls into the module
245fn call_module_internal(
246    ctx: &WasiFunctionEnv,
247    mut store: Store,
248    start_args: ThreadStartArgs,
249) -> (Store, Result<Option<ExitCode>, DeepSleepWork>) {
250    // The spawning thread checked that the module exports `wasi_thread_start`,
251    // and this thread runs an instance of the same module.
252    let Some(spawn) = ctx
253        .data(&store)
254        .inner()
255        .main_module_instance_handles()
256        .thread_spawn
257        .clone()
258    else {
259        error!("thread failed - the program does not export a `wasi_thread_start` function");
260        return (store, Ok(Some(Errno::Notcapable.into())));
261    };
262
263    let spawn: Function = spawn.into();
264    let (mut store, thread_result) =
265        ContextSwitchingEnvironment::run_main_context(ctx, store, spawn, start_args.to_values());
266    let thread_result = thread_result.map(|_| ());
267
268    trace!("callback finished (ret={:?})", thread_result);
269
270    let exit_code = match handle_thread_result(ctx, &mut store, thread_result) {
271        Ok(code) => code,
272        Err(deep_sleep) => return (store, Err(deep_sleep)),
273    };
274
275    (store, Ok(exit_code))
276}
277
278fn handle_thread_result(
279    env: &WasiFunctionEnv,
280    store: &mut Store,
281    err: Result<(), RuntimeError>,
282) -> Result<Option<ExitCode>, DeepSleepWork> {
283    let tid = env.data(&store).tid();
284    let pid = env.data(&store).pid();
285    let Err(err) = err else {
286        trace!("thread exited cleanly without calling thread_exit");
287        return Ok(None);
288    };
289    match err.downcast::<WasiError>() {
290        Ok(WasiError::ThreadExit) => {
291            trace!("thread exited cleanly");
292            Ok(None)
293        }
294        Ok(WasiError::Exit(code)) => {
295            trace!(exit_code = ?code, "thread requested exit");
296            if !code.is_success() {
297                // TODO: Why do we need to taint the runtime on a non-zero exit code? Why not also for zero?
298                env.data(&store)
299                    .runtime
300                    .on_taint(TaintReason::NonZeroExitCode(code));
301            };
302            Ok(Some(code))
303        }
304        Ok(WasiError::DeepSleep(deep)) => {
305            trace!("entered a deep sleep");
306            Err(deep)
307        }
308        Ok(WasiError::UnknownWasiVersion) => {
309            eprintln!(
310                "Thread {tid} of process {pid} failed because it has an unknown wasix version"
311            );
312            env.data(&store)
313                .runtime
314                .on_taint(TaintReason::UnknownWasiVersion);
315            Ok(Some(ExitCode::from(129)))
316        }
317        Ok(WasiError::DlSymbolResolutionFailed(symbol)) => {
318            eprintln!("Thread {tid} of process {pid} failed to find required symbol: {symbol}");
319            env.data(&store)
320                .runtime
321                .on_taint(TaintReason::DlSymbolResolutionFailed(symbol.clone()));
322            Ok(Some(ExitCode::from(129)))
323        }
324        Err(err) => {
325            if err.clone().to_trap() == Some(wasmer_types::TrapCode::HostInterrupt) {
326                debug!(%tid, %pid, error = %err, "thread interrupted by host");
327            } else {
328                eprintln!("Thread {tid} of process {pid} failed with runtime error: {err}");
329            }
330            env.data(&store)
331                .runtime
332                .on_taint(TaintReason::RuntimeError(err));
333            Ok(Some(ExitCode::from(129)))
334        }
335    }
336}
337
338/// Calls the module
339fn call_module<M: MemorySize>(
340    mut ctx: WasiFunctionEnv,
341    mut store: Store,
342    start_args: ThreadStartArgs,
343    thread_handle: Arc<WasiThreadHandle>,
344    rewind_state: Option<(RewindState, RewindResultType)>,
345) {
346    let env = ctx.data(&store);
347    let tasks = env.tasks().clone();
348
349    // If we need to rewind then do so
350    if let Some((rewind_state, rewind_result)) = rewind_state {
351        let mut ctx = ctx.env.clone().into_mut(&mut store);
352        let res = rewind_ext::<M>(
353            &mut ctx,
354            Some(rewind_state.memory_stack),
355            rewind_state.rewind_stack,
356            rewind_state.store_data,
357            rewind_result,
358        );
359        if res != Errno::Success {
360            return;
361        }
362    }
363
364    // Now invoke the module
365    let (mut store, ret) = call_module_internal(&ctx, store, start_args);
366
367    // If it went to deep sleep then we need to handle that
368    if let Err(deep) = ret {
369        // Create the callback that will be invoked when the thread respawns after a deep sleep
370        let rewind = deep.rewind;
371        let respawn = {
372            let tasks = tasks.clone();
373            move |ctx, store, trigger_res| {
374                // Call the thread
375                call_module::<M>(
376                    ctx,
377                    store,
378                    start_args,
379                    thread_handle,
380                    Some((rewind, RewindResultType::RewindWithResult(trigger_res))),
381                );
382            }
383        };
384
385        /// Spawns the WASM process after a trigger
386        unsafe {
387            tasks.resume_wasm_after_poller(Box::new(respawn), ctx, store, deep.trigger)
388        };
389        return;
390    };
391
392    let exit_code = ret.unwrap_or_else(|_| unreachable!());
393    if let Some(exit_code) = exit_code {
394        ctx.on_exit(&mut store, Some(exit_code));
395        thread_handle.set_status_finished(Ok(exit_code));
396    } else {
397        ctx.on_exit(&mut store, None);
398        thread_handle.set_status_finished(Ok(Errno::Success.into()));
399    }
400
401    drop(thread_handle);
402}
403
404#[cfg(test)]
405mod tests {
406    use super::*;
407
408    const TID: u32 = 2;
409
410    fn start_args(tid: u32, start_ptr: u64) -> Result<ThreadStartArgs, Errno> {
411        ThreadStartArgs::new(WasiThreadId::from(tid), start_ptr)
412    }
413
414    #[test]
415    fn start_args_keep_bits_of_pointers_above_2gib() {
416        let args = start_args(TID, 0x8000_0000).unwrap();
417        assert_eq!(args.start_arg, i32::MIN);
418        assert_eq!(args.start_arg.cast_unsigned(), 0x8000_0000);
419
420        let args = start_args(TID, u32::MAX.into()).unwrap();
421        assert_eq!(args.start_arg, -1);
422
423        let args = start_args(TID, 0x7fff_fff0).unwrap();
424        assert_eq!(args.start_arg, 0x7fff_fff0);
425    }
426
427    #[test]
428    fn start_args_reject_pointers_wider_than_32_bits() {
429        assert_eq!(start_args(TID, 1 << 32), Err(Errno::Overflow));
430        assert_eq!(start_args(TID, u64::MAX), Err(Errno::Overflow));
431    }
432
433    #[test]
434    fn start_args_enforce_wasi_threads_tid_range() {
435        let args = start_args(MAX_THREAD_ID, 0).unwrap();
436        assert_eq!(args.tid, 0x1fff_ffff);
437
438        for tid in [0, MAX_THREAD_ID + 1, i32::MAX.cast_unsigned(), u32::MAX] {
439            assert_eq!(
440                start_args(tid, 0),
441                Err(Errno::Again),
442                "tid {tid:#x} must be rejected"
443            );
444        }
445    }
446}
wasmer_wasix/syscalls/wasix/thread_spawn.rs

wasmer_wasix/syscalls/wasix/
thread_spawn.rs
