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 {