Do not hold a buffer's storage while waiting to hand it over

socket.send()/sendall()/sendto()/sendmsg() and FileIO.write() kept the source buffer's read borrow for the whole call, including the wait for a peer that may never make room. That borrow is a lock every other thread writing to the same object waits on, so threading.Thread(target=lambda: sock.sendall(buf)).start() buf[0] = 1 did not return until the peer read; and a thread parked there is ATTACHED and never reaches a safepoint, so gc.collect() in a third thread waited for the peer too -- the same wedge readinto() had on the receiving side. ArgBytesLike::borrow_buf_unlocked() answers with bytes that survive the borrow being dropped. An immutable object hands out a plain reference and locks nothing, so those are sent where they lie and bytes and memoryviews over them cost nothing; only bytes reached through a lock are copied out first. The export is held throughout either way, so the source still cannot be resized while it is being sent. The regression snippet covers both directions now and is renamed for it. Assisted-by: Claude Assisted-by: Codex:GPT-5
youknowone · Aug 16, 2026 · Aug 14, 2026 · Aug 14, 2026 · Aug 14, 2026 · Aug 14, 2026
commit 125d5aa7cba7b692ee36f3d5112ee8eeeac60d7b
diff --git a/crates/common/src/borrow.rs b/crates/common/src/borrow.rs
@@ -34,6 +34,17 @@ impl_from!('a, T, BorrowedValue<'a, T>,
 );
 
 impl<'a, T: ?Sized> BorrowedValue<'a, T> {
+    /// Whether reaching the value holds a lock that other threads wait on.
+    ///
+    /// An immutable object hands out a plain reference and answers `false`;
+    /// one whose storage can change hands out a guard. A caller about to wait
+    /// for something unrelated -- a peer, a file, a signal -- can use this to
+    /// decide whether it may keep the borrow for the duration.
+    #[must_use]
+    pub const fn is_locked(&self) -> bool {
+        !matches!(self, Self::Ref(_))
+    }
+
     pub fn map<U: ?Sized, F>(s: Self, f: F) -> BorrowedValue<'a, U>
     where
         F: FnOnce(&T) -> &U,

diff --git a/crates/stdlib/src/socket.rs b/crates/stdlib/src/socket.rs
@@ -1694,7 +1694,7 @@ mod _socket {
             vm: &VirtualMachine,
         ) -> Result<usize, IoOrPyException> {
             let flags = flags.unwrap_or(0);
-            let buf = bytes.borrow_buf();
+            let buf = bytes.borrow_buf_unlocked(vm)?;
             let buf = &*buf;
             self.sock_op(vm, SockWaitKind::Write, || {
                 self.sock()?.send_with_flags(buf, flags)
@@ -1714,7 +1714,7 @@ mod _socket {
 
             let deadline = timeout.map(Deadline::new);
 
-            let buf = bytes.borrow_buf();
+            let buf = bytes.borrow_buf_unlocked(vm)?;
             let buf = &*buf;
             let mut buf_offset = 0;
             // now we have like 3 layers of interrupt loop :)
@@ -1751,7 +1751,7 @@ mod _socket {
                 OptionalArg::Missing => (0, arg2),
             };
             let addr = self.extract_address(address, "sendto", vm)?;
-            let buf = bytes.borrow_buf();
+            let buf = bytes.borrow_buf_unlocked(vm)?;
             let buf = &*buf;
             self.sock_op(vm, SockWaitKind::Write, || {
                 self.sock()?.send_to_with_flags(buf, &addr, flags)
@@ -1781,8 +1781,8 @@ mod _socket {
 
             let buffers = buffers
                 .iter()
-                .map(|buf| buf.borrow_buf())
-                .collect::<Vec<_>>();
+                .map(|buf| buf.borrow_buf_unlocked(vm))
+                .collect::<PyResult<Vec<_>>>()?;
             let buffers = buffers
                 .iter()
                 .map(|buf| io::IoSlice::new(buf))

diff --git a/crates/vm/src/function/buffer.rs b/crates/vm/src/function/buffer.rs
@@ -49,6 +49,28 @@ impl ArgBytesLike {
         f(&self.borrow_buf())
     }
 
+    /// The bytes to hand to an operation that may wait, and whatever keeps
+    /// them readable while it does.
+    ///
+    /// `borrow_buf` may answer with a lock that every other thread writing to
+    /// the same object waits on, and a thread waiting on a lock never reaches
+    /// a safepoint, so keeping one across a wait for a peer, a pipe or a
+    /// signal stops the world from being stopped at all. Bytes reached that
+    /// way are copied out first. Bytes that lock nothing -- an immutable
+    /// object's -- are borrowed where they lie, which is all CPython holds in
+    /// either case.
+    pub fn borrow_buf_unlocked(&self, vm: &VirtualMachine) -> PyResult<UnlockedBuf<'_>> {
+        let borrowed = self.borrow_buf();
+        if !borrowed.is_locked() {
+            return Ok(UnlockedBuf::Borrowed(borrowed));
+        }
+        let mut copy = Vec::new();
+        copy.try_reserve_exact(borrowed.len())
+            .map_err(|_| vm.new_memory_error(""))?;
+        copy.extend_from_slice(&borrowed);
+        Ok(UnlockedBuf::Copied(copy))
+    }
+
     #[must_use]
     pub const fn len(&self) -> usize {
         self.0.desc.len
@@ -123,6 +145,24 @@ impl<'a> TryFromBorrowedObject<'a> for ArgContiguousBytesLike {
     }
 }
 
+/// Bytes that stay readable across a wait, from [`ArgBytesLike::borrow_buf_unlocked`].
+#[derive(Debug)]
+pub enum UnlockedBuf<'a> {
+    Borrowed(BorrowedValue<'a, [u8]>),
+    Copied(Vec<u8>),
+}
+
+impl core::ops::Deref for UnlockedBuf<'_> {
+    type Target = [u8];
+
+    fn deref(&self) -> &[u8] {
+        match self {
+            Self::Borrowed(b) => b,
+            Self::Copied(v) => v,
+        }
+    }
+}
+
 /// A memory buffer, read-write access. Like the `w*` format code for `PyArg_Parse` in CPython.
 #[derive(Debug, Traverse)]
 pub struct ArgMemoryBuffer(PyBuffer);

diff --git a/crates/vm/src/stdlib/_io.rs b/crates/vm/src/stdlib/_io.rs
@@ -5899,9 +5899,14 @@ mod fileio {
 
             let handle = zelf.get_fd(vm)?;
 
+            // A pipe, socket or terminal takes the bytes only when the other
+            // end makes room, which may be never; see readinto above for what
+            // holding the source's lock across that wait costs.
+            let buf = obj.borrow_buf_unlocked(vm)?;
+
             // Loop on EINTR (PEP 475)
             let len = loop {
-                match obj.with_ref(|b| vm.allow_threads(|| host_io::write_once(handle, b))) {
+                match vm.allow_threads(|| host_io::write_once(handle, &buf)) {
                     Ok(n) => break n,
                     Err(e) if host_io::is_interrupted_error(&e) => {
                         vm.check_signals()?;

diff --git a/extra_tests/snippets/stdlib_io_blocking_buffer.py b/extra_tests/snippets/stdlib_io_blocking_buffer.py
@@ -0,0 +1,143 @@
+"""Transfers that wait for a peer must not hold the buffer they were given.
+
+A pipe or a socket answers when the other end does, which may be never. The
+buffer is exported for the whole call, so it cannot be resized meanwhile, but
+nothing else about it changes: another thread can still read it, write to it,
+and the interpreter can still stop the world. An implementation that holds the
+buffer's storage for the duration of the wait takes all of that away, and a
+thread parked on that storage never reaches a safepoint, so a collection that
+wants every thread stopped ends up waiting for the peer too.
+"""
+
+import gc
+import os
+import socket
+import threading
+import time
+
+# The peer acts after DELAY; the checks below have to finish well inside it.
+DELAY = 1.0
+SLACK = DELAY / 2
+
+
+def measure(buf, writable):
+    """Time the operations on `buf` that do not need the peer."""
+    start = time.monotonic()
+    assert len(buf) == len(buf), len(buf)
+    assert isinstance(bytes(buf), bytes)
+    if writable:
+        buf[0] = buf[0]
+    gc.collect()
+    return time.monotonic() - start
+
+
+def run(buf, blocking_call, release_peer, writable):
+    started = threading.Event()
+    result = []
+
+    def transfer():
+        started.set()
+        result.append(blocking_call(buf))
+
+    def peer():
+        time.sleep(DELAY)
+        release_peer()
+
+    threads = [threading.Thread(target=transfer), threading.Thread(target=peer)]
+    for t in threads:
+        t.start()
+    started.wait()
+    time.sleep(0.2)  # the transfer is now waiting on its peer
+
+    elapsed = measure(buf, writable)
+    assert elapsed < SLACK, "waited %.2fs on the peer" % elapsed
+
+    # The export is still held either way, so the buffer cannot be resized.
+    try:
+        buf.append(0)
+    except BufferError:
+        pass
+    else:
+        raise AssertionError("append during an export should raise BufferError")
+
+    for t in threads:
+        t.join()
+    return result[0]
+
+
+# --- reading: the buffer is written into, so nothing else may touch it at all
+
+
+read_fd, write_fd = os.pipe()
+pipe = open(read_fd, "rb", buffering=0)
+try:
+    target = bytearray(16)
+    n = run(target, pipe.readinto, lambda: os.write(write_fd, b"pipe"), writable=False)
+    assert n == 4, n
+    assert bytes(target[:4]) == b"pipe", bytes(target)
+finally:
+    pipe.close()
+    os.close(write_fd)
+
+if hasattr(socket, "socketpair"):
+    left, right = socket.socketpair()
+    try:
+        target = bytearray(16)
+        n = run(target, left.recv_into, lambda: right.send(b"socket"), writable=False)
+        assert n == 6, n
+        assert bytes(target[:6]) == b"socket", bytes(target)
+    finally:
+        left.close()
+        right.close()
+
+
+# --- writing: the buffer is only read, so it stays writable meanwhile
+
+
+read_fd, write_fd = os.pipe()
+sink = open(write_fd, "wb", buffering=0)
+try:
+    # More than any pipe will hold, so the write cannot finish on its own.
+    source = bytearray(4 * 1024 * 1024)
+    drained = []
+
+    def drain():
+        with open(read_fd, "rb", buffering=0) as f:
+            while True:
+                chunk = f.read(1 << 16)
+                if not chunk:
+                    break
+                drained.append(len(chunk))
+
+    reader = threading.Thread(target=drain, daemon=True)
+    run(source, sink.write, reader.start, writable=True)
+    sink.close()
+    reader.join()
+    assert sum(drained) == len(source), (sum(drained), len(source))
+finally:
+    if not sink.closed:
+        sink.close()
+
+if hasattr(socket, "socketpair"):
+    left, right = socket.socketpair()
+    try:
+        left.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
+        source = bytearray(4 * 1024 * 1024)
+        received = []
+
+        def receive():
+            while sum(received) < len(source):
+                chunk = right.recv(1 << 16)
+                if not chunk:
+                    break
+                received.append(len(chunk))
+
+        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))
+    finally:
+        left.close()
+        right.close()
+
+print("ok")
diff --git a/extra_tests/snippets/stdlib_io_readinto_blocking.py b/extra_tests/snippets/stdlib_io_readinto_blocking.py