Visitar URL original
io: decide the readinto path by file type, not by seekability · RustPython/RustPython@d324016 · GitHub
Skip to content

Commit d324016

Browse files
youknowoneyouknowone
committed
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
1 parent 4f782b8 commit d324016

4 files changed

Lines changed: 72 additions & 14 deletions

File tree

‎crates/host_env/src/io.rs‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -199,6 +199,29 @@ pub fn is_seekable(fd: crt_fd::Borrowed<'_>) -> bool {
199199
os::seek_fd(fd, 0, libc::SEEK_CUR).is_ok()
200200
}
201201

202+
/// Whether a read from `fd` answers from data the file already holds, rather
203+
/// than waiting for whoever writes the other end.
204+
///
205+
/// Seeking answers this everywhere but Windows, where a pipe seeks too --
206+
/// `lseek` on one succeeds and reports a position, so a reader that took
207+
/// seekability for an answer would wait on a peer while holding whatever it
208+
/// holds for the length of the call.
209+
#[cfg(not(windows))]
210+
pub fn reads_without_waiting(fd: crt_fd::Borrowed<'_>) -> bool {
211+
is_seekable(fd)
212+
}
213+
214+
#[cfg(windows)]
215+
pub fn reads_without_waiting(fd: crt_fd::Borrowed<'_>) -> bool {
216+
use std::os::windows::io::AsRawHandle;
217+
use windows_sys::Win32::Storage::FileSystem::{FILE_TYPE_DISK, GetFileType};
218+
219+
let Ok(handle) = crt_fd::as_handle(fd) else {
220+
return false;
221+
};
222+
unsafe { GetFileType(handle.as_raw_handle() as _) == FILE_TYPE_DISK }
223+
}
224+
202225
pub fn validate_whence(whence: i32) -> bool {
203226
let standard = (0..=2).contains(&whence);
204227
#[cfg(any(target_os = "dragonfly", target_os = "freebsd", target_os = "linux"))]

‎crates/host_env/src/io_unsupported.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -176,6 +176,10 @@ pub fn is_seekable(_fd: crt_fd::Borrowed<'_>) -> bool {
176176
false
177177
}
178178

179+
pub fn reads_without_waiting(_fd: crt_fd::Borrowed<'_>) -> bool {
180+
false
181+
}
182+
179183
pub fn validate_whence(whence: i32) -> bool {
180184
(0..=2).contains(&whence)
181185
}

‎crates/vm/src/stdlib/_io.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5854,9 +5854,10 @@ mod fileio {
58545854

58555855
let handle = zelf.get_fd(vm)?;
58565856

5857-
if zelf.seekable(vm)? {
5857+
if host_io::reads_without_waiting(handle) {
58585858
// The read answers from the file itself, so it returns without
58595859
// waiting on anyone; write where the caller asked directly.
5860+
// Seekability is not the question -- a pipe on Windows seeks.
58605861
let mut buf = obj.borrow_buf_mut();
58615862
return Self::read_once_into(zelf, handle, &mut buf, vm);
58625863
}

‎extra_tests/snippets/stdlib_io_blocking_buffer.py‎

Lines changed: 43 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -21,14 +21,22 @@
2121

2222

2323
def measure(buf, expected_len, writable):
24-
"""Time the operations on `buf` that do not need the peer."""
25-
start = time.monotonic()
26-
assert len(buf) == expected_len, len(buf)
27-
assert isinstance(bytes(buf), bytes)
24+
"""Time each operation on `buf` that does not need the peer, separately, so
25+
a failure names the one that waited rather than the group."""
26+
elapsed = {}
27+
28+
def timed(name, operation):
29+
start = time.monotonic()
30+
value = operation()
31+
elapsed[name] = time.monotonic() - start
32+
return value
33+
34+
assert timed("len", lambda: len(buf)) == expected_len, len(buf)
35+
assert isinstance(timed("bytes", lambda: bytes(buf)), bytes)
2836
if writable:
29-
buf[0] = buf[0]
30-
gc.collect()
31-
return time.monotonic() - start
37+
timed("setitem", lambda: buf.__setitem__(0, buf[0]))
38+
timed("gc.collect", gc.collect)
39+
return elapsed
3240

3341

3442
def run(buf, blocking_call, release_peer, writable):
@@ -51,9 +59,13 @@ def peer():
5159
time.sleep(0.2) # the transfer is now waiting on its peer
5260

5361
elapsed = measure(buf, expected_len, writable)
54-
assert elapsed < SLACK, "waited %.2fs on the peer" % elapsed
62+
waited = ["%s %.2fs" % item for item in elapsed.items() if item[1] >= SLACK]
63+
assert not waited, "waited on the peer: " + ", ".join(waited)
5564

56-
# The export is still held either way, so the buffer cannot be resized.
65+
# The transfer is still in flight, so its export is still held and the
66+
# buffer cannot be resized. An operating system that took the whole
67+
# transfer without a peer leaves nothing here to observe.
68+
assert not result, "the transfer finished without its peer"
5769
try:
5870
buf.append(0)
5971
except BufferError:
@@ -124,12 +136,26 @@ def drain():
124136
if hasattr(socket, "socketpair"):
125137
left, right = socket.socketpair()
126138
try:
127-
left.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
128-
source = bytearray(4 * 1024 * 1024)
139+
# How much a connection holds before it makes the sender wait is the
140+
# operating system's to decide, and asking for a small send buffer does
141+
# not settle it -- a socketpair is already connected, and on Windows it
142+
# is a loopback pair whose receiver has a window of its own. So fill it
143+
# until it refuses rather than guess a size that outruns it.
144+
left.setblocking(False)
145+
filled = 0
146+
while True:
147+
try:
148+
filled += left.send(bytes(1 << 16))
149+
except (BlockingIOError, InterruptedError):
150+
break
151+
left.setblocking(True)
152+
153+
source = bytearray(1 << 16)
129154
received = []
130155

131156
def receive():
132-
while sum(received) < len(source):
157+
wanted = filled + len(source)
158+
while sum(received) < wanted:
133159
chunk = right.recv(1 << 16)
134160
if not chunk:
135161
break
@@ -138,7 +164,11 @@ def receive():
138164
reader = threading.Thread(target=receive, daemon=True)
139165
run(source, left.sendall, reader.start, writable=True)
140166
reader.join()
141-
assert sum(received) == len(source), (sum(received), len(source))
167+
assert sum(received) == filled + len(source), (
168+
sum(received),
169+
filled,
170+
len(source),
171+
)
142172
finally:
143173
left.close()
144174
right.close()

0 commit comments

Comments
 (0)