Support nonblocking pipes on Windows (#9110)

* Support nonblocking pipes on Windows

Assisted-by: Codex:gpt-6

* Remove redundant Windows pipe test coverage

Assisted-by: Codex:gpt-6

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):
 

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<BorrowedHandle<'_>> {
 
 fn _write(fd: Raw, buf: &[u8]) -> io::Result<usize> {
     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<usize> {
     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<usize> {

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<bool> {
+    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;

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))
 }
 

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<bool> {
+        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,

extra_tests/snippets/stdlib_os.py

@@ -780,3 +780,34 @@ if sys.platform.startswith("linux"):
     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()