From efaf621d7173df30eb611949da8d150d93534fa7 Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Wed, 14 Oct 2026 13:24:07 +0300 Subject: [PATCH 1/2] Support nonblocking pipes on Windows Assisted-by: Codex:gpt-6 --- Lib/test/test_io.py | 4 --- crates/host_env/src/crt_fd.rs | 53 ++++++++++++++++++++++++--- crates/host_env/src/nt.rs | 20 +++++++++++ crates/host_env/src/os.rs | 10 ++++-- crates/vm/src/stdlib/nt.rs | 12 +++++++ extra_tests/snippets/stdlib_os.py | 59 +++++++++++++++++++++++++++++++ 6 files changed, 147 insertions(+), 11 deletions(-) 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..5098edc03e5 100644 --- a/extra_tests/snippets/stdlib_os.py +++ b/extra_tests/snippets/stdlib_os.py @@ -758,3 +758,62 @@ def __index__(self): assert hasattr(os, "mkfifo") if hasattr(os, "mkfifo"): mkfifo_paths_and_directory_descriptors() + + +import errno +import io + + +def nonblocking_windows_pipe(): + reader, writer = os.pipe() + try: + assert os.get_blocking(reader) and os.get_blocking(writer) + for fd in (reader, writer): + 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"" + try: + os.read(reader, 1) + except BlockingIOError as error: + assert error.errno == errno.EAGAIN + else: + raise AssertionError("empty nonblocking pipe did not raise") + with io.FileIO(reader, "rb", closefd=False) as stream: + assert stream.read(1) is None + assert stream.readinto(bytearray(1)) is None + + payload = b"x" * (1024 * 1024) + first = os.write(writer, payload) + assert 0 < first < len(payload) + total = first + for _ in range(1024): + try: + total += os.write(writer, payload) + except BlockingIOError as error: + assert error.errno == errno.EAGAIN + break + else: + raise AssertionError("pipe did not fill") + with io.FileIO(writer, "wb", closefd=False) as stream: + assert stream.write(b"x") is None + assert os.write(writer, b"") == 0 + received = 0 + while received < total: + data = os.read(reader, total - received) + assert data and data == b"x" * len(data) + received += len(data) + os.close(writer) + writer = None + assert os.read(reader, 1) == b"" + finally: + os.close(reader) + if writer is not None: + os.close(writer) + + +if os.name == "nt": + nonblocking_windows_pipe() From 2a37fafd397de56dc625be96943f6b20770e101a Mon Sep 17 00:00:00 2001 From: 1ndahous3 <9205230+1ndahous3@users.noreply.github.com> Date: Wed, 14 Oct 2026 20:30:28 +0300 Subject: [PATCH 2/2] Remove redundant Windows pipe test coverage Assisted-by: Codex:gpt-6 --- extra_tests/snippets/stdlib_os.py | 34 +++---------------------------- 1 file changed, 3 insertions(+), 31 deletions(-) diff --git a/extra_tests/snippets/stdlib_os.py b/extra_tests/snippets/stdlib_os.py index 5098edc03e5..679c859a8bc 100644 --- a/extra_tests/snippets/stdlib_os.py +++ b/extra_tests/snippets/stdlib_os.py @@ -761,14 +761,13 @@ def __index__(self): import errno -import io def nonblocking_windows_pipe(): reader, writer = os.pipe() try: - assert os.get_blocking(reader) and os.get_blocking(writer) 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) @@ -776,43 +775,16 @@ def nonblocking_windows_pipe(): 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") - with io.FileIO(reader, "rb", closefd=False) as stream: - assert stream.read(1) is None - assert stream.readinto(bytearray(1)) is None - - payload = b"x" * (1024 * 1024) - first = os.write(writer, payload) - assert 0 < first < len(payload) - total = first - for _ in range(1024): - try: - total += os.write(writer, payload) - except BlockingIOError as error: - assert error.errno == errno.EAGAIN - break - else: - raise AssertionError("pipe did not fill") - with io.FileIO(writer, "wb", closefd=False) as stream: - assert stream.write(b"x") is None - assert os.write(writer, b"") == 0 - received = 0 - while received < total: - data = os.read(reader, total - received) - assert data and data == b"x" * len(data) - received += len(data) - os.close(writer) - writer = None - assert os.read(reader, 1) == b"" finally: os.close(reader) - if writer is not None: - os.close(writer) + os.close(writer) if os.name == "nt":