Function set_wakeup_fd

crates/vm/src/stdlib/_signal.rs:366–445

(args: SetWakeupFdArgs, vm: &VirtualMachine)

    #[pyfunction]
    fn set_wakeup_fd(args: SetWakeupFdArgs, vm: &VirtualMachine) -> PyResult<i64> {
        // TODO: implement warn_on_full_buffer
        let _ = args.warn_on_full_buffer;
        #[cfg(windows)]
        let fd = args.fd.0;
        #[cfg(not(windows))]
        let fd = args.fd;

        if !vm.is_main_thread() {
            return Err(vm.new_value_error("set_wakeup_fd only works in main thread"));
        }

        #[cfg(windows)]
        let is_socket = if fd != INVALID_WAKEUP {
            use windows_sys::Win32::Networking::WinSock;

            crate::windows::init_winsock();
            let mut res = 0i32;
            let mut res_size = core::mem::size_of::<i32>() as i32;
            let res = unsafe {
                WinSock::getsockopt(
                    fd,
                    WinSock::SOL_SOCKET,
                    WinSock::SO_ERROR,
                    &mut res as *mut i32 as *mut _,
                    &mut res_size,
                )
            };
            // if getsockopt succeeded, fd is for sure a socket
            let is_socket = res == 0;
            if !is_socket {
                let err = std::io::Error::last_os_error();
                // if getsockopt failed for some other reason, throw
                if err.raw_os_error() != Some(WinSock::WSAENOTSOCK) {
                    return Err(err.into_pyexception(vm));
                }
                // Validate that fd is a valid file descriptor using fstat
                // First check if SOCKET can be safely cast to i32 (file descriptor)
                let fd_i32 = i32::try_from(fd).map_err(|_| vm.new_value_error("invalid fd"))?;
                // Verify the fd is valid by trying to fstat it
                let borrowed_fd =
                    unsafe { crate::common::crt_fd::Borrowed::try_borrow_raw(fd_i32) }
                        .map_err(|e| e.into_pyexception(vm))?;
                crate::common::fileutils::fstat(borrowed_fd).map_err(|e| e.into_pyexception(vm))?;
            }
            is_socket
        } else {
            false
        };
        #[cfg(unix)]
        if let Ok(fd) = unsafe { crate::common::crt_fd::Borrowed::try_borrow_raw(fd) } {
            use nix::fcntl;
            let oflags = fcntl::fcntl(fd, fcntl::F_GETFL).map_err(|e| e.into_pyexception(vm))?;
            let nonblock =
                fcntl::OFlag::from_bits_truncate(oflags).contains(fcntl::OFlag::O_NONBLOCK);
            if !nonblock {
                return Err(vm.new_value_error(format!(
                    "the fd {} must be in non-blocking mode",