From 7fd8238bb593c683ab0b30d80292cae6cc9d7dfc Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 14:53:50 -0500 Subject: [PATCH] fix(io): report what a compio file round sent before it failed On Linux, a compio `send_file` round splices up to a pipe's capacity from the file into a pipe, then drains the pipe into the socket. An error while draining was returned as the round's result even when part of the pipe had already reached the socket, so the caller was told nothing went out. `send_file_all` retries `Interrupted` at the same offset, which would send those bytes a second time. A round that fails after draining part of its pipe now reports the drained bytes, as a write does, drops the pipe with whatever it still holds, and leaves the error to the next call. Refs #159 --- io/src/compio.rs | 34 ++++++++++++-------- io/tests/io.rs | 80 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 13 deletions(-) diff --git a/io/src/compio.rs b/io/src/compio.rs index 814f533..b4ac9bd 100644 --- a/io/src/compio.rs +++ b/io/src/compio.rs @@ -853,7 +853,8 @@ mod splice { } /// One round: up to a pipe's capacity from the file into the pipe, then - /// all of it from the pipe into the socket. + /// all of it from the pipe into the socket, or as much as the socket + /// took before an error, which is then what the round reports. /// /// The second step is not an io_uring splice: io_uring runs every splice /// on an io-wq worker, and with no poll support for the operation a @@ -905,32 +906,39 @@ mod splice { }; let mut drained = 0; - while drained < filled { + let error = loop { + if drained == filled { + stream.pipe.set(Some(pipe)); + return BufResult(Ok(filled), file); + } match crate::sys::pipe::splice( pipe.read.as_fd(), stream.inner.as_fd(), filled - drained, ) { Ok(0) => { - return BufResult( - Err(io::Error::new( - io::ErrorKind::WriteZero, - "stream accepted no bytes", - )), - file, - ); + break io::Error::new(io::ErrorKind::WriteZero, "stream accepted no bytes"); } Ok(n) => drained += n, Err(e) if e.kind() == io::ErrorKind::WouldBlock => { if let Err(e) = ready(stream, Interest::Writable).await { - return BufResult(Err(e), file); + break e; } } - Err(e) => return BufResult(Err(e), file), + Err(e) => break e, } + }; + // The pipe may still hold bytes, so it goes with this round. What was + // drained before the error is in the socket: that is what this call + // sent, as `send_file` reports it, and the error is the next call's + // to report. Reporting the error instead would have a caller that + // retries it (`Interrupted`, in `send_file_all`) send those bytes + // again. + if drained > 0 { + BufResult(Ok(drained), file) + } else { + BufResult(Err(error), file) } - stream.pipe.set(Some(pipe)); - BufResult(Ok(filled), file) } /// Makes the socket `O_NONBLOCK`, once. diff --git a/io/tests/io.rs b/io/tests/io.rs index e7bccc1..d9a3515 100644 --- a/io/tests/io.rs +++ b/io/tests/io.rs @@ -1336,6 +1336,86 @@ mod compio { ); } + /// A round that fails after part of it reached the socket reports the + /// part that did, as any write does, and leaves the error to the next + /// call: reported as the round's result, a retried error (`Interrupted`, + /// which `send_file_all` retries at the same offset) would send those + /// bytes twice. + /// + /// The peer reads nothing and both socket buffers are at their minimum, + /// so the round's pipe drains only partly before the socket is full; + /// the peer then resets the connection, which fails the rest. + #[cfg(any(target_os = "linux", target_os = "android"))] + #[test] + fn send_file_reports_what_it_sent_before_failing() { + use std::{sync::mpsc, time::Duration}; + + use http_kit_io::{BufResult, Listen as _, Listener as _, SendFile as _}; + + // Below the kernel's minimum, which each buffer is raised to + // (`SOCK_MIN_SNDBUF`, `SOCK_MIN_RCVBUF`), far below a pipe's 64 KiB. + const BUFFER: usize = 1; + + let path = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")).join("compio-reset-peer"); + let contents = super::payload(); + std::fs::write(&path, &contents).expect("write the file to send"); + let file = std::rc::Rc::new(std::fs::File::open(&path).expect("open the file to send")); + + let (reset, resets) = mpsc::channel::<()>(); + let (addr_tx, addr_rx) = mpsc::channel(); + let peer = std::thread::spawn(move || { + let addr = addr_rx.recv().expect("listener address"); + let socket = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None) + .expect("peer socket"); + socket + .set_recv_buffer_size(BUFFER) + .expect("peer receive buffer"); + socket + .connect(&socket2::SockAddr::from(addr)) + .expect("connect"); + resets.recv().expect("reset"); + // A zero linger closes with a reset. + socket + .set_linger(Some(Duration::ZERO)) + .expect("zero linger"); + }); + + let (first, next) = run(async { + let io = http_kit_io::compio::Compio::default(); + let listener = io.bind_tcp(super::loopback()).expect("bind"); + addr_tx + .send(listener.local_addr().expect("local address")) + .expect("send address"); + let (mut stream, _) = listener.accept().await.expect("accept"); + socket2::SockRef::from(&stream) + .set_send_buffer_size(BUFFER) + .expect("send buffer"); + let send = stream.send_file(file.clone(), 0, contents.len()); + let watch = async { + // The round fills its pipe and waits on the full socket. + ::compio::time::sleep(Duration::from_millis(300)).await; + reset.send(()).expect("reset the peer"); + }; + let (BufResult(first, _), ()) = futures_lite::future::zip(send, watch).await; + let sent = *first.as_ref().expect("the round reports what it sent"); + let BufResult(next, _) = stream + .send_file(file.clone(), sent as u64, contents.len() - sent) + .await; + (sent, next) + }); + peer.join().expect("peer thread"); + std::fs::remove_file(&path).expect("remove the sent file"); + assert!( + first > 0 && first < contents.len(), + "part of the region was sent: {first} of {} bytes", + contents.len() + ); + assert!( + next.is_err(), + "the next call reports the reset connection: {next:?}" + ); + } + /// A buffer that records when it is dropped. #[cfg(target_os = "linux")] struct Tracked {