diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py index 212cbaa324f..fc51209a59f 100644 --- a/Lib/test/test_io.py +++ b/Lib/test/test_io.py @@ -4224,10 +4224,6 @@ class PyTextIOWrapperTest(TextIOWrapperTest): io = pyio shutdown_error = "LookupError: unknown encoding: ascii" - @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON; os.set_blocking not available on Windows") - def test_read_non_blocking(self): - return super().test_read_non_blocking() - class IncrementalNewlineDecoderTest(unittest.TestCase): diff --git a/crates/host_env/src/crt_fd.rs b/crates/host_env/src/crt_fd.rs index c9af646122f..5134b1dcf67 100644 --- a/crates/host_env/src/crt_fd.rs +++ b/crates/host_env/src/crt_fd.rs @@ -21,6 +21,8 @@ mod c { unsafe extern "C" { #[link_name = "_chsize_s"] pub(super) fn ftruncate(fd: i32, len: i64) -> i32; + pub(super) fn _get_doserrno(value: *mut c_ulong) -> c_int; + pub(super) fn _set_doserrno(value: c_ulong) -> c_int; } } @@ -427,14 +429,57 @@ pub fn as_handle(fd: Borrowed<'_>) -> io::Result> { fn _write(fd: Raw, buf: &[u8]) -> io::Result { let count = cmp::min(buf.len(), MAX_RW); - let n = cvt(unsafe { suppress_iph!(c::write(fd, buf.as_ptr() as _, count as _)) })?; - Ok(n as usize) + #[cfg(windows)] + { + let mut count = count; + loop { + unsafe { c::_set_doserrno(0) }; + let n = unsafe { suppress_iph!(c::write(fd, buf.as_ptr() as _, count as _)) }; + if n >= 0 { + return Ok(n as usize); + } + let errno = crate::os::get_errno(); + // The CRT reports a successful zero-byte WriteFile as ENOSPC. + // A nonblocking pipe may still accept a smaller prefix. + if errno != libc::ENOSPC || dos_error() != 0 { + return Err(crate::os::io_error_from_errno(errno)); + } + count /= 2; + if count == 0 { + return Err(crate::os::io_error_from_errno(libc::EAGAIN)); + } + } + } + #[cfg(not(windows))] + { + let n = cvt(unsafe { suppress_iph!(c::write(fd, buf.as_ptr() as _, count as _)) })?; + Ok(n as usize) + } } fn _read(fd: Raw, buf: &mut [u8]) -> io::Result { let count = cmp::min(buf.len(), MAX_RW); - let n = cvt(unsafe { suppress_iph!(libc::read(fd, buf.as_mut_ptr() as _, count as _)) })?; - Ok(n as usize) + #[cfg(windows)] + unsafe { + c::_set_doserrno(0); + } + let n = unsafe { suppress_iph!(libc::read(fd, buf.as_mut_ptr() as _, count as _)) }; + #[cfg(windows)] + if n < 0 + && crate::os::get_errno() == libc::EINVAL + && dos_error() == windows_sys::Win32::Foundation::ERROR_NO_DATA + { + return Err(crate::os::io_error_from_errno(libc::EAGAIN)); + } + cvt(n).map(|n| n as usize) +} + +#[cfg(windows)] +fn dos_error() -> u32 { + let mut error = 0; + // SAFETY: the CRT writes to a valid unsigned long on the current thread. + unsafe { c::_get_doserrno(&mut error) }; + error } pub fn write(fd: Borrowed<'_>, buf: &[u8]) -> io::Result { diff --git a/crates/host_env/src/nt.rs b/crates/host_env/src/nt.rs index 9f1c1c23de4..9a083a9826a 100644 --- a/crates/host_env/src/nt.rs +++ b/crates/host_env/src/nt.rs @@ -22,6 +22,7 @@ use crate::{ windows::{CheckWin32Bool, CheckWin32Handle, CheckWin32Sentinel, HandleToOwned}, }; use libc::intptr_t; + use widestring::WideCString; use windows_sys::{ Win32::{ @@ -75,6 +76,25 @@ pub use windows_sys::Win32::Storage::FileSystem::{ FILE_ATTRIBUTE_TEMPORARY, FILE_ATTRIBUTE_VIRTUAL, }; +pub fn get_blocking(fd: crt_fd::Borrowed<'_>) -> io::Result { + let handle = crt_fd::as_handle(fd)?; + let mode = crate::winapi::get_named_pipe_handle_state(handle.as_raw_handle())?; + Ok(mode & windows_sys::Win32::System::Pipes::PIPE_NOWAIT == 0) +} + +pub fn set_blocking(fd: crt_fd::Borrowed<'_>, blocking: bool) -> io::Result<()> { + use windows_sys::Win32::System::Pipes::PIPE_NOWAIT; + + let handle = crt_fd::as_handle(fd)?; + let mode = crate::winapi::get_named_pipe_handle_state(handle.as_raw_handle())?; + let mode = if blocking { + mode & !PIPE_NOWAIT + } else { + mode | PIPE_NOWAIT + }; + crate::winapi::set_named_pipe_handle_state(handle.as_raw_handle(), Some(mode), None, None) +} + pub type Handle = HANDLE; pub const MAX_PATH_USIZE: usize = MAX_PATH as usize; pub const ERROR_INVALID_HANDLE_I32: i32 = ERROR_INVALID_HANDLE as i32; diff --git a/crates/host_env/src/os.rs b/crates/host_env/src/os.rs index f6bd04d21cb..f716d2770c1 100644 --- a/crates/host_env/src/os.rs +++ b/crates/host_env/src/os.rs @@ -698,13 +698,17 @@ impl core::error::Error for CrtErrno {} /// Build an [`io::Error`] that preserves a raw C runtime `errno`. /// -/// The [`io::ErrorKind`] is derived from the closest Win32 mapping so callers -/// matching on `kind()` keep working, while the exact `errno` is preserved for +/// The [`io::ErrorKind`] preserves would-block semantics and otherwise uses the +/// closest Win32 mapping. The exact `errno` is preserved for /// [`ErrorExt::posix_errno`]. #[cfg(windows)] #[must_use] pub fn io_error_from_errno(errno: i32) -> io::Error { - let kind = io::Error::from_raw_os_error(errno_to_winerror(errno)).kind(); + let kind = if errno == libc::EAGAIN { + io::ErrorKind::WouldBlock + } else { + io::Error::from_raw_os_error(errno_to_winerror(errno)).kind() + }; io::Error::new(kind, CrtErrno(errno)) } diff --git a/crates/vm/src/stdlib/nt.rs b/crates/vm/src/stdlib/nt.rs index 812bca83b78..be7fb9de85d 100644 --- a/crates/vm/src/stdlib/nt.rs +++ b/crates/vm/src/stdlib/nt.rs @@ -24,6 +24,18 @@ pub(crate) mod module { use std::os::windows::ffi::{OsStrExt, OsStringExt}; use std::os::windows::io::AsRawHandle; + #[pyfunction] + fn get_blocking(fd: crt_fd::Borrowed<'_>, vm: &VirtualMachine) -> PyResult { + vm.allow_threads(|| host_nt::get_blocking(fd)) + .map_err(|err| err.to_pyexception(vm)) + } + + #[pyfunction] + fn set_blocking(fd: crt_fd::Borrowed<'_>, blocking: bool, vm: &VirtualMachine) -> PyResult<()> { + vm.allow_threads(|| host_nt::set_blocking(fd, blocking)) + .map_err(|err| err.to_pyexception(vm)) + } + #[pyattr] use rustpython_host_env::os::{ O_BINARY, O_NOINHERIT, O_RANDOM, O_SEQUENTIAL, O_TEMPORARY, O_TEXT, diff --git a/extra_tests/snippets/stdlib_os.py b/extra_tests/snippets/stdlib_os.py index 50ce16be191..679c859a8bc 100644 --- a/extra_tests/snippets/stdlib_os.py +++ b/extra_tests/snippets/stdlib_os.py @@ -758,3 +758,34 @@ def __index__(self): assert hasattr(os, "mkfifo") if hasattr(os, "mkfifo"): mkfifo_paths_and_directory_descriptors() + + +import errno + + +def nonblocking_windows_pipe(): + reader, writer = os.pipe() + try: + for fd in (reader, writer): + assert os.get_blocking(fd) + os.set_blocking(fd, False) + assert not os.get_blocking(fd) + os.set_blocking(fd, True) + assert os.get_blocking(fd) + os.set_blocking(fd, False) + + assert os.read(reader, 0) == b"" + assert os.write(writer, b"") == 0 + try: + os.read(reader, 1) + except BlockingIOError as error: + assert error.errno == errno.EAGAIN + else: + raise AssertionError("empty nonblocking pipe did not raise") + finally: + os.close(reader) + os.close(writer) + + +if os.name == "nt": + nonblocking_windows_pipe()