Function start_new_thread

crates/vm/src/stdlib/_thread.rs:484–566

(mut f_args: FuncArgs, vm: &VirtualMachine)

    #[pyfunction]
    fn start_new_thread(mut f_args: FuncArgs, vm: &VirtualMachine) -> PyResult<u64> {
        if !f_args.kwargs.is_empty() {
            return Err(vm.new_type_error("start_new_thread() takes no keyword arguments"));
        }
        let given = f_args.args.len();
        if given < 2 {
            return Err(vm.new_type_error(format!(
                "start_new_thread expected at least 2 arguments, got {given}"
            )));
        }
        if given > 3 {
            return Err(vm.new_type_error(format!(
                "start_new_thread expected at most 3 arguments, got {given}"
            )));
        }

        let func_obj = f_args.take_positional().unwrap();
        let args_obj = f_args.take_positional().unwrap();
        let kwargs_obj = f_args.take_positional();

        if func_obj.to_callable().is_none() {
            return Err(vm.new_type_error("first arg must be callable"));
        }
        if !args_obj.fast_isinstance(vm.ctx.types.tuple_type) {
            return Err(vm.new_type_error("2nd arg must be a tuple"));
        }
        if kwargs_obj
            .as_ref()
            .is_some_and(|obj| !obj.fast_isinstance(vm.ctx.types.dict_type))
        {
            return Err(vm.new_type_error("optional 3rd arg must be a dictionary"));
        }

        let func: ArgCallable = func_obj.clone().try_into_value(vm)?;
        let args: PyTupleRef = args_obj.clone().try_into_value(vm)?;
        let kwargs: Option<PyDictRef> = kwargs_obj.map(|obj| obj.try_into_value(vm)).transpose()?;

        vm.sys_module.get_attr("audit", vm)?.call(
            (
                "_thread.start_new_thread",
                func_obj,
                args_obj,
                kwargs
                    .as_ref()
                    .map_or_else(|| vm.ctx.none(), |k| k.clone().into()),
            ),
            vm,
        )?;

        if vm
            .state
            .finalizing
            .load(core::sync::atomic::Ordering::Acquire)
        {
            return Err(vm.new_exception_msg(
                vm.ctx.exceptions.python_finalization_error.to_owned(),
                "can't create new thread at interpreter shutdown"
                    .to_owned()