io: decide the readinto path by file type, not by seekability

FileIO.readinto wrote straight into the caller's buffer, holding its write borrow, when the fd was seekable; otherwise it read aside into scratch and copied. Seekability stood in for "this read answers without waiting on a peer", which a pipe on Windows breaks: lseek on one succeeds, so the pipe took the borrow-holding path and every other thread touching that bytearray waited for the peer. host_io::reads_without_waiting answers it directly -- seekability elsewhere, GetFileType() == FILE_TYPE_DISK on Windows. The regression snippet times each operation separately, so a failure names the one that waited; it asserts the transfer is still in flight before checking the export; and the socket case fills the connection until it refuses rather than assuming a size that outruns it, which SO_SNDBUF on an already-connected pair does not settle. Assisted-by: Claude
youknowone · Aug 16, 2026 · Aug 14, 2026 · Aug 14, 2026 · Aug 14, 2026 · Aug 14, 2026
commit 5a5ba17e147817be69a79c5df43ab283f45f36bb
diff --git a/crates/host_env/src/io.rs b/crates/host_env/src/io.rs
@@ -199,6 +199,29 @@ pub fn is_seekable(fd: crt_fd::Borrowed<'_>) -> bool {
     os::seek_fd(fd, 0, libc::SEEK_CUR).is_ok()
 }
 
+/// Whether a read from `fd` answers from data the file already holds, rather
+/// than waiting for whoever writes the other end.
+///
+/// Seeking answers this everywhere but Windows, where a pipe seeks too --
+/// `lseek` on one succeeds and reports a position, so a reader that took
+/// seekability for an answer would wait on a peer while holding whatever it
+/// holds for the length of the call.
+#[cfg(not(windows))]
+pub fn reads_without_waiting(fd: crt_fd::Borrowed<'_>) -> bool {
+    is_seekable(fd)
+}
+
+#[cfg(windows)]
+pub fn reads_without_waiting(fd: crt_fd::Borrowed<'_>) -> bool {
+    use std::os::windows::io::AsRawHandle;
+    use windows_sys::Win32::Storage::FileSystem::{FILE_TYPE_DISK, GetFileType};
+
+    let Ok(handle) = crt_fd::as_handle(fd) else {
+        return false;
+    };
+    unsafe { GetFileType(handle.as_raw_handle() as _) == FILE_TYPE_DISK }
+}
+
 pub fn validate_whence(whence: i32) -> bool {
     let standard = (0..=2).contains(&whence);
     #[cfg(any(target_os = "dragonfly", target_os = "freebsd", target_os = "linux"))]

diff --git a/crates/host_env/src/io_unsupported.rs b/crates/host_env/src/io_unsupported.rs
@@ -176,6 +176,10 @@ pub fn is_seekable(_fd: crt_fd::Borrowed<'_>) -> bool {
     false
 }
 
+pub fn reads_without_waiting(_fd: crt_fd::Borrowed<'_>) -> bool {
+    false
+}
+
 pub fn validate_whence(whence: i32) -> bool {
     (0..=2).contains(&whence)
 }

diff --git a/crates/vm/src/stdlib/_io.rs b/crates/vm/src/stdlib/_io.rs
@@ -5861,9 +5861,10 @@ mod fileio {
 
             let handle = zelf.get_fd(vm)?;
 
-            if zelf.seekable(vm)? {
+            if host_io::reads_without_waiting(handle) {
                 // The read answers from the file itself, so it returns without
                 // waiting on anyone; write where the caller asked directly.
+                // Seekability is not the question -- a pipe on Windows seeks.
                 let mut buf = obj.borrow_buf_mut();
                 return Self::read_once_into(zelf, handle, &mut buf, vm);
             }

diff --git a/extra_tests/snippets/stdlib_io_blocking_buffer.py b/extra_tests/snippets/stdlib_io_blocking_buffer.py
@@ -21,14 +21,22 @@
 
 
 def measure(buf, expected_len, writable):
-    """Time the operations on `buf` that do not need the peer."""
-    start = time.monotonic()
-    assert len(buf) == expected_len, len(buf)
-    assert isinstance(bytes(buf), bytes)
+    """Time each operation on `buf` that does not need the peer, separately, so
+    a failure names the one that waited rather than the group."""
+    elapsed = {}
+
+    def timed(name, operation):
+        start = time.monotonic()
+        value = operation()
+        elapsed[name] = time.monotonic() - start
+        return value
+
+    assert timed("len", lambda: len(buf)) == expected_len, len(buf)
+    assert isinstance(timed("bytes", lambda: bytes(buf)), bytes)
     if writable:
-        buf[0] = buf[0]
-    gc.collect()
-    return time.monotonic() - start
+        timed("setitem", lambda: buf.__setitem__(0, buf[0]))
+    timed("gc.collect", gc.collect)
+    return elapsed
 
 
 def run(buf, blocking_call, release_peer, writable):
@@ -51,9 +59,13 @@ def peer():
     time.sleep(0.2)  # the transfer is now waiting on its peer
 
     elapsed = measure(buf, expected_len, writable)
-    assert elapsed < SLACK, "waited %.2fs on the peer" % elapsed
+    waited = ["%s %.2fs" % item for item in elapsed.items() if item[1] >= SLACK]
+    assert not waited, "waited on the peer: " + ", ".join(waited)
 
-    # The export is still held either way, so the buffer cannot be resized.
+    # The transfer is still in flight, so its export is still held and the
+    # buffer cannot be resized. An operating system that took the whole
+    # transfer without a peer leaves nothing here to observe.
+    assert not result, "the transfer finished without its peer"
     try:
         buf.append(0)
     except BufferError:
@@ -124,12 +136,26 @@ def drain():
 if hasattr(socket, "socketpair"):
     left, right = socket.socketpair()
     try:
-        left.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
-        source = bytearray(4 * 1024 * 1024)
+        # How much a connection holds before it makes the sender wait is the
+        # operating system's to decide, and asking for a small send buffer does
+        # not settle it -- a socketpair is already connected, and on Windows it
+        # is a loopback pair whose receiver has a window of its own. So fill it
+        # until it refuses rather than guess a size that outruns it.
+        left.setblocking(False)
+        filled = 0
+        while True:
+            try:
+                filled += left.send(bytes(1 << 16))
+            except (BlockingIOError, InterruptedError):
+                break
+        left.setblocking(True)
+
+        source = bytearray(1 << 16)
         received = []
 
         def receive():
-            while sum(received) < len(source):
+            wanted = filled + len(source)
+            while sum(received) < wanted:
                 chunk = right.recv(1 << 16)
                 if not chunk:
                     break
@@ -138,7 +164,11 @@ def receive():
         reader = threading.Thread(target=receive, daemon=True)
         run(source, left.sendall, reader.start, writable=True)
         reader.join()
-        assert sum(received) == len(source), (sum(received), len(source))
+        assert sum(received) == filled + len(source), (
+            sum(received),
+            filled,
+            len(source),
+        )
     finally:
         left.close()
         right.close()
-Original file line number
+Diff line change
@@ Expand Up / @@ -176,6 +176,10 @@ pub fn is_seekable(_fd: crt_fd::Borrowed<'_>) -> bool { @@
         false
     }
+    pub fn reads_without_waiting(_fd: crt_fd::Borrowed<'_>) -> bool {
+        false
+    }
     pub fn validate_whence(whence: i32) -> bool {
         (0..=2).contains(&whence)
     }
@@ Expand Down @@