From 55606a89543796ff11496905b164e2b51ac01c07 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 07:43:47 -0500 Subject: [PATCH 1/3] feat(io)!: zero-copy send for large writes on io_uring `Write::write_all_owned` writes gathered buffers and hands them back only when the writer is done with them. On Linux over io_uring, a compio TCP stream sends a write of at least `ZeroCopySend::threshold` bytes with `IORING_OP_SENDMSG_ZC` and `MSG_WAITALL`: the call returns at the send's own completion, and a runtime task holds the operation and its buffers until the kernel's notification, which drops them. Unacknowledged zero-copy data is bounded by the socket's send buffer, which counts zero-copy pages. A send the kernel refuses for `RLIMIT_MEMLOCK` (`ENOMEM`, nothing sent) copies instead. Unix-domain streams, other drivers and other platforms copy as before; kernel TLS never reaches the zero-copy path, since its writer only issues copying writes, which TLS's own sendmsg requires (it rejects `MSG_ZEROCOPY` with `EOPNOTSUPP`). `ZeroCopySend` is the runtime-adjustable threshold, shared by every clone across threads; the default, 10 KiB, is where the kernel's MSG_ZEROCOPY documentation finds zero-copy pays off. `Compio` now carries it into every listener and stream it creates, so it is no longer a unit struct, and `Compio::tcp_stream` replaces `From` for adopting a stream. --- h1/src/lib.rs | 9 +- h1/tests/net.rs | 2 +- h1/tests/splice.rs | 2 +- h1/tests/timeouts.rs | 2 +- h1/tests/unix.rs | 2 +- h3/src/lib.rs | 5 +- h3/tests/support/mod.rs | 2 +- io/src/compio.rs | 264 ++++++++++++++++++++++++++++++++++++---- io/src/stream.rs | 137 ++++++++++++++++----- io/tests/io.rs | 197 ++++++++++++++++++++++++++++-- tls/src/lib.rs | 13 +- tls/tests/tls.rs | 2 +- 12 files changed, 554 insertions(+), 83 deletions(-) diff --git a/h1/src/lib.rs b/h1/src/lib.rs index ddf619a..68a473e 100644 --- a/h1/src/lib.rs +++ b/h1/src/lib.rs @@ -72,19 +72,20 @@ //! //! # fn main() -> std::io::Result<()> { //! compio::runtime::Runtime::new()?.block_on(async { -//! let listener = Compio.bind_tcp("127.0.0.1:0".parse().unwrap())?; +//! let io = Compio::default(); +//! let listener = io.bind_tcp("127.0.0.1:0".parse().unwrap())?; //! let addr = listener.local_addr()?; //! let service = |request: Request| async move { //! Response::new(Body::from_bytes(request.uri().path().to_owned())) //! }; //! let server = async { //! let (stream, _) = listener.accept().await?; -//! let connection = ServerConnection::from_stream(stream, Compio, ServerConfig::default()); +//! let connection = ServerConnection::from_stream(stream, io.clone(), ServerConfig::default()); //! connection.serve(&service).await.map_err(std::io::Error::other) //! }; //! let client = async { -//! let stream = Compio.connect(addr).await?; -//! let (send, connection) = handshake_stream(stream, Compio, ClientConfig::default()); +//! let stream = io.connect(addr).await?; +//! let (send, connection) = handshake_stream(stream, io.clone(), ClientConfig::default()); //! let request = async move { //! let mut request = Request::new(Body::empty()); //! *request.uri_mut() = format!("http://{addr}/hello").parse().unwrap(); diff --git a/h1/tests/net.rs b/h1/tests/net.rs index b9bf942..20bdb88 100644 --- a/h1/tests/net.rs +++ b/h1/tests/net.rs @@ -333,7 +333,7 @@ mod compio { .block_on(future) } - suite!(http_kit_io::compio::Compio, run); + suite!(http_kit_io::compio::Compio::default(), run); } mod tokio { diff --git a/h1/tests/splice.rs b/h1/tests/splice.rs index d44c03b..8f9ef1a 100644 --- a/h1/tests/splice.rs +++ b/h1/tests/splice.rs @@ -947,7 +947,7 @@ mod compio { .block_on(future) } - suite!(http_kit_io::compio::Compio, run); + suite!(http_kit_io::compio::Compio::default(), run); } mod tokio { diff --git a/h1/tests/timeouts.rs b/h1/tests/timeouts.rs index 1aba776..7f37369 100644 --- a/h1/tests/timeouts.rs +++ b/h1/tests/timeouts.rs @@ -674,7 +674,7 @@ mod compio { .block_on(future) } - suite!(http_kit_io::compio::Compio, run); + suite!(http_kit_io::compio::Compio::default(), run); } mod tokio { diff --git a/h1/tests/unix.rs b/h1/tests/unix.rs index e14c68c..f69cc3e 100644 --- a/h1/tests/unix.rs +++ b/h1/tests/unix.rs @@ -141,7 +141,7 @@ mod compio { .block_on(future) } - suite!(http_kit_io::compio::Compio, run); + suite!(http_kit_io::compio::Compio::default(), run); } mod tokio { diff --git a/h3/src/lib.rs b/h3/src/lib.rs index a9b6ee2..e99b284 100644 --- a/h3/src/lib.rs +++ b/h3/src/lib.rs @@ -63,8 +63,9 @@ //! let identity = Identity::from_pem(std::fs::read("cert.pem")?, std::fs::read("key.pem")?); //! let config = ServerConfig::new(identity, 64 * 1024); //! compio::runtime::Runtime::new()?.block_on(async { -//! let socket = Compio.bind_udp("[::]:443".parse().unwrap()).await?; -//! let server = Server::new(socket, Compio, &config)?; +//! let io = Compio::default(); +//! let socket = io.bind_udp("[::]:443".parse().unwrap()).await?; +//! let server = Server::new(socket, io, &config)?; //! let service = |request: Request| async move { //! Response::new(Body::from_bytes(request.uri().path().to_owned())) //! }; diff --git a/h3/tests/support/mod.rs b/h3/tests/support/mod.rs index 4f0075c..ac3689e 100644 --- a/h3/tests/support/mod.rs +++ b/h3/tests/support/mod.rs @@ -163,7 +163,7 @@ pub async fn within(io: &I, limit: Duration, test: F) -> F: pub fn on_compio(test: impl FnOnce(http_kit_io::compio::Compio) -> F) -> F::Output { compio::runtime::Runtime::new() .unwrap() - .block_on(test(http_kit_io::compio::Compio)) + .block_on(test(http_kit_io::compio::Compio::default())) } /// Runs a test body on a current-thread tokio runtime. diff --git a/io/src/compio.rs b/io/src/compio.rs index dffeb85..23f9c56 100644 --- a/io/src/compio.rs +++ b/io/src/compio.rs @@ -30,6 +30,18 @@ //! completion: compio reports no `IORING_CQE_F_BUF_MORE`, which incremental //! consumption (`IOU_PBUF_RING_INC`) needs. //! +//! On Linux over io_uring a TCP stream's +//! [`write_all_owned`](crate::Write::write_all_owned) of at least +//! [`ZeroCopySend::threshold`] bytes sends without copying +//! (`IORING_OP_SENDMSG_ZC`): the kernel reads the bytes from the caller's +//! buffers until the peer acknowledges them. The write returns as soon as +//! the kernel has taken every byte; the stream keeps the buffers until the +//! kernel's notification says it no longer reads them, and drops them then. +//! What the kernel holds this way is bounded like any other unacknowledged +//! data, by the socket's send buffer, which counts zero-copy pages too. +//! Smaller writes, the other write methods, Unix-domain streams (which the +//! kernel does not send zero-copy) and every other platform copy. +//! //! The poll-style capability is a [`CopyingPoll`]: a completion-based driver //! cannot read into a buffer borrowed for one poll. @@ -60,10 +72,89 @@ use crate::{ #[cfg(unix)] use crate::{DialUnix, FromStdUnix, ListenUnix}; -/// The compio implementation. It holds nothing: compio keeps its runtime per -/// thread, and every operation goes to the current thread's. -#[derive(Debug, Clone, Copy, Default)] -pub struct Compio; +/// The compio implementation. compio keeps its runtime per thread, and every +/// operation goes to the current thread's; this holds only the +/// [`ZeroCopySend`] setting, which every stream it creates or adopts shares. +#[derive(Debug, Clone, Default)] +pub struct Compio { + zero_copy: ZeroCopySend, +} + +impl Compio { + /// An implementation whose streams send zero-copy as `zero_copy` says. + #[must_use] + pub const fn with_zero_copy(zero_copy: ZeroCopySend) -> Self { + Self { zero_copy } + } + + /// The zero-copy setting its streams share. + #[must_use] + pub const fn zero_copy(&self) -> &ZeroCopySend { + &self.zero_copy + } + + /// Adopts a connected compio stream: one a runtime of this thread + /// registered, such as a socket accepted elsewhere and handed over as a + /// std stream (on Windows a socket joins one completion port only, so a + /// thread-per-core server accepts where no port is involved and each + /// thread adopts its share). + #[must_use] + pub fn tcp_stream(&self, stream: net::TcpStream) -> TcpStream { + TcpStream::new(stream, self.zero_copy.clone()) + } +} + +/// When a write goes out through io_uring's zero-copy send: from +/// [`threshold`](Self::threshold) bytes up, on Linux. +/// +/// It is the operator's runtime-adjustable handle: every clone shares one +/// value, across threads, and a change applies to the next write of every +/// stream holding a clone. A threshold of `0` sends every such write +/// zero-copy; `usize::MAX` none. +/// +/// A zero-copy send pins the pages it sends until the peer acknowledges +/// them, and io_uring charges them to the user's `RLIMIT_MEMLOCK` unless the +/// process has `CAP_IPC_LOCK` (`io_notif_account_mem`). When that limit is +/// reached the kernel refuses the send before sending anything, and the write +/// copies instead. +#[derive(Debug, Clone)] +pub struct ZeroCopySend { + threshold: std::sync::Arc, +} + +impl ZeroCopySend { + /// The default threshold. The kernel's `MSG_ZEROCOPY` documentation + /// (`Documentation/networking/msg_zerocopy.rst`) finds pinning pages and + /// waiting for the notification "generally only effective at writes over + /// around 10 KB". + pub const DEFAULT_THRESHOLD: usize = 10 * 1024; + + /// A setting with `threshold`. + #[must_use] + pub fn new(threshold: usize) -> Self { + Self { + threshold: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(threshold)), + } + } + + /// The fewest bytes a write sends zero-copy. + #[must_use] + pub fn threshold(&self) -> usize { + self.threshold.load(std::sync::atomic::Ordering::Relaxed) + } + + /// Sets the threshold for every clone of this setting. + pub fn set_threshold(&self, threshold: usize) { + self.threshold + .store(threshold, std::sync::atomic::Ordering::Relaxed); + } +} + +impl Default for ZeroCopySend { + fn default() -> Self { + Self::new(Self::DEFAULT_THRESHOLD) + } +} impl Timer for Compio { fn sleep_until(&self, deadline: Instant) -> impl Future { @@ -75,7 +166,8 @@ impl Dial for Compio { type Connection = TcpStream; async fn connect(&self, addr: SocketAddr) -> io::Result { - net::TcpStream::connect(addr).await.map(TcpStream::new) + let stream = net::TcpStream::connect(addr).await?; + Ok(self.tcp_stream(stream)) } } @@ -84,7 +176,8 @@ impl DialUnix for Compio { type UnixConnection = UnixStream; async fn connect_unix(&self, path: &std::path::Path) -> io::Result { - net::UnixStream::connect(path).await.map(UnixStream::new) + let stream = net::UnixStream::connect(path).await?; + Ok(UnixStream::new(stream, self.zero_copy.clone())) } } @@ -119,13 +212,15 @@ impl FromStdUnix for Compio { &self, listener: std::os::unix::net::UnixListener, ) -> io::Result { - net::UnixListener::from_std(listener).map(UnixListener::new) + let listener = net::UnixListener::from_std(listener)?; + Ok(UnixListener::new(listener, self.zero_copy.clone())) } } impl FromStd for Compio { fn tcp_listener(&self, listener: std::net::TcpListener) -> io::Result { - net::TcpListener::from_std(listener).map(TcpListener::new) + let listener = net::TcpListener::from_std(listener)?; + Ok(TcpListener::new(listener, self.zero_copy.clone())) } fn udp_socket(&self, socket: std::net::UdpSocket) -> io::Result { @@ -161,6 +256,11 @@ impl Fd for T {} /// A stream socket of compio's, which the operations below drive through a /// shared reference. trait Socket: Clone + Fd + 'static { + /// Whether the kernel sends from this socket zero-copy: it sets + /// `SOCK_SUPPORT_ZC`, which io_uring's zero-copy send requires, on TCP + /// sockets only (`tcp_init_sock`). + #[cfg(target_os = "linux")] + const ZERO_COPY: bool; /// The descriptor, shared with an operation that outlives a call. #[cfg(target_os = "linux")] fn shared_fd(&self) -> ::compio::driver::SharedFd; @@ -187,8 +287,11 @@ trait Socket: Clone + Fd + 'static { } macro_rules! impl_socket { - ($ty:ty) => { + ($ty:ty, zero_copy: $zero_copy:literal) => { impl Socket for $ty { + #[cfg(target_os = "linux")] + const ZERO_COPY: bool = $zero_copy; + #[cfg(target_os = "linux")] fn shared_fd(&self) -> ::compio::driver::SharedFd { ::compio::driver::ToSharedFd::to_shared_fd(self) @@ -251,9 +354,9 @@ macro_rules! impl_socket { }; } -impl_socket!(net::TcpStream); +impl_socket!(net::TcpStream, zero_copy: true); #[cfg(unix)] -impl_socket!(net::UnixStream); +impl_socket!(net::UnixStream, zero_copy: false); /// A connected stream socket. pub struct Stream { @@ -271,10 +374,16 @@ pub struct Stream { /// The multishot receive and what it received ahead of the reader. #[cfg(target_os = "linux")] recv: recv::Recv, + /// From what size a write sends zero-copy. + #[cfg(target_os = "linux")] + zero_copy: ZeroCopySend, } impl Stream { - fn new(inner: S) -> Self { + fn new(inner: S, zero_copy: ZeroCopySend) -> Self { + // Only Linux sends zero-copy. + #[cfg(not(target_os = "linux"))] + drop(zero_copy); Self { inner, #[cfg(any(target_os = "linux", target_os = "android"))] @@ -283,6 +392,8 @@ impl Stream { nonblocking: std::cell::Cell::new(false), #[cfg(target_os = "linux")] recv: recv::Recv::default(), + #[cfg(target_os = "linux")] + zero_copy, } } @@ -293,16 +404,6 @@ impl Stream { } } -/// Adopts a connected compio stream: one a runtime of this thread registered, -/// such as a socket accepted elsewhere and handed over as a std stream (on -/// Windows a socket joins one completion port only, so a thread-per-core -/// server accepts where no port is involved and each thread adopts its share). -impl From for TcpStream { - fn from(stream: net::TcpStream) -> Self { - Self::new(stream) - } -} - #[cfg(unix)] impl Stream { /// The local address. @@ -414,6 +515,16 @@ impl SharedStream for Stream { self.inner.write_vectored(buf).await } + #[cfg(target_os = "linux")] + async fn write_all_owned(&self, buf: B) -> BufResult<(), Option> { + if S::ZERO_COPY && buf.total_len() >= self.zero_copy.threshold() { + return zero_copy::write_all(&self.inner, buf).await; + } + let BufResult(res, buf) = + crate::stream::write_all_vectored(&mut crate::stream::ByRef(self), buf).await; + BufResult(res, Some(buf)) + } + async fn shutdown(&self) -> io::Result<()> { self.inner.shutdown().await } @@ -707,6 +818,99 @@ mod splice { } } +/// Zero-copy sends on io_uring. +#[cfg(target_os = "linux")] +mod zero_copy { + use std::{io, rc::Rc}; + + use ::compio::driver::{ + OpCode, + op::{SendFlags, SendVectoredZc}, + }; + use compio_buf::{BufResult, IoVectoredBuf}; + use futures_core::stream::FusedStream as _; + use futures_lite::StreamExt as _; + + use super::Socket; + + /// The caller's buffers, shared by the sends of one write: a short send + /// is continued by another while the first still holds what it sent. + struct Shared(Rc); + + impl IoVectoredBuf for Shared { + fn iter_slice(&self) -> impl Iterator { + self.0.iter_slice() + } + } + + /// [`Write::write_all_owned`](crate::Write::write_all_owned) through + /// zero-copy sends. + /// + /// With `MSG_WAITALL` io_uring finishes a short send itself, waiting for + /// room in the socket with a poll (`io_net_retry`), so a write is one + /// send and one notification unless an error cuts it short. + pub(super) async fn write_all( + socket: &S, + buf: B, + ) -> BufResult<(), Option> { + let flags = SendFlags::from_bits_retain(libc::MSG_WAITALL.cast_unsigned()); + let len = buf.total_len(); + let buf = Rc::new(buf); + let mut written = 0; + let mut zero_copy = true; + let res = loop { + if written == len { + break Ok(()); + } + let rest = Shared(Rc::clone(&buf)).slice(written); + let res = if zero_copy { + send(SendVectoredZc::new(socket.shared_fd(), rest, flags)).await + } else { + socket.write_vectored(rest).await.0 + }; + match res { + Ok(0) => { + break Err(io::Error::new( + io::ErrorKind::WriteZero, + "stream accepted no bytes", + )); + } + Ok(n) => written += n, + Err(e) if e.kind() == io::ErrorKind::Interrupted => {} + // The pages would exceed `RLIMIT_MEMLOCK`: the kernel refused + // the send before sending anything, so the rest copies. + Err(e) if zero_copy && e.raw_os_error() == Some(libc::ENOMEM) => { + zero_copy = false; + } + Err(e) => break Err(e), + } + }; + BufResult(res, Rc::into_inner(buf)) + } + + /// Submits a zero-copy send and returns what the send itself completed + /// with, as soon as it has. The operation, with the buffers it holds, + /// goes to a task of the runtime until the kernel's notification that it + /// no longer reads them, which drops them. + /// + /// It takes any zero-copy operation, whatever buffers it sends from. + pub(super) async fn send(op: T) -> io::Result { + let mut op = ::compio::runtime::submit_multi(op); + let BufResult(res, _) = op + .next() + .await + .expect("a submitted operation completes at least once"); + // The send's own completion carries `IORING_CQE_F_MORE` when a + // notification follows. Without one (an error the kernel posts + // alone, or a kernel without the operation, where compio sends by + // copying) the first completion is the last. + if !op.is_terminated() { + ::compio::runtime::spawn(async move { while op.next().await.is_some() {} }).detach(); + } + res + } +} + /// Bytes a multishot receive put in a buffer of the runtime's ring, lent to /// the reader by [`ReadOwned`](crate::ReadOwned). Dropping it hands the /// buffer back to the ring, so a reader holds it only as long as it needs the @@ -1145,14 +1349,17 @@ pub struct TcpListener { inner: net::TcpListener, #[cfg(target_os = "linux")] accepts: multishot::Accepts, + /// The setting each accepted stream shares. + zero_copy: ZeroCopySend, } impl TcpListener { - fn new(inner: net::TcpListener) -> Self { + fn new(inner: net::TcpListener, zero_copy: ZeroCopySend) -> Self { Self { #[cfg(target_os = "linux")] accepts: multishot::Accepts::new(&inner), inner, + zero_copy, } } @@ -1181,10 +1388,10 @@ impl Listener for TcpListener { // completions, so it reports no peer. let stream = net::TcpStream::from_std(accepted?.into())?; let peer = stream.peer_addr()?; - return Ok((TcpStream::new(stream), peer)); + return Ok((TcpStream::new(stream, self.zero_copy.clone()), peer)); } let (stream, peer) = self.inner.accept().await?; - Ok((TcpStream::new(stream), peer)) + Ok((TcpStream::new(stream, self.zero_copy.clone()), peer)) } fn local_addr(&self) -> io::Result { @@ -1199,15 +1406,18 @@ pub struct UnixListener { inner: net::UnixListener, #[cfg(target_os = "linux")] accepts: multishot::Accepts, + /// The setting each accepted stream shares. + zero_copy: ZeroCopySend, } #[cfg(unix)] impl UnixListener { - fn new(inner: net::UnixListener) -> Self { + fn new(inner: net::UnixListener, zero_copy: ZeroCopySend) -> Self { Self { #[cfg(target_os = "linux")] accepts: multishot::Accepts::new(&inner), inner, + zero_copy, } } @@ -1238,7 +1448,7 @@ impl Listener for UnixListener { }; #[cfg(not(target_os = "linux"))] let (stream, _) = self.inner.accept().await?; - let stream = UnixStream::new(stream); + let stream = UnixStream::new(stream, self.zero_copy.clone()); let peer = stream.peer_addr()?; Ok((stream, peer)) } diff --git a/io/src/stream.rs b/io/src/stream.rs index 65e2771..cac0d76 100644 --- a/io/src/stream.rs +++ b/io/src/stream.rs @@ -94,6 +94,29 @@ pub trait Write { /// Shuts down the write direction (a TCP FIN) after flushing. fn shutdown(&mut self) -> impl Future>; + /// Writes all the bytes of `buf`, several buffers gathered as + /// [`write_vectored`](Self::write_vectored) takes them, and hands `buf` + /// back unless the writer still needs it. + /// + /// A writer that sends without copying (the compio implementation's + /// zero-copy send on Linux) returns once the kernel has taken every byte, + /// but the kernel goes on reading them from `buf` until the peer + /// acknowledges them: such a writer keeps `buf`, drops it when the kernel + /// is done with it, and answers `None`. A write that copies answers + /// `Some(buf)`, so the caller can reuse the allocation. + /// + /// A zero-length write before the end is reported as + /// [`io::ErrorKind::WriteZero`]. + fn write_all_owned( + &mut self, + buf: B, + ) -> impl Future>> { + async move { + let BufResult(res, buf) = write_all_vectored(self, buf).await; + BufResult(res, Some(buf)) + } + } + /// Writes all of `buf`'s initialized bytes. /// /// A zero-length write before the end is reported as @@ -126,6 +149,35 @@ pub trait Write { } } +/// Writes all of `buf` through [`Write::write_vectored`], copying. +pub(crate) async fn write_all_vectored( + writer: &mut W, + buf: B, +) -> BufResult<(), B> { + let len = buf.total_len(); + let mut buf = buf; + let mut written = 0; + while written < len { + let BufResult(res, slice) = writer.write_vectored(buf.slice(written)).await; + buf = slice.into_inner(); + match res { + Ok(0) => { + return BufResult( + Err(io::Error::new( + io::ErrorKind::WriteZero, + "stream accepted no bytes", + )), + buf, + ); + } + Ok(n) => written += n, + Err(e) if e.kind() == io::ErrorKind::Interrupted => {} + Err(e) => return BufResult(Err(e), buf), + } + } + BufResult(Ok(()), buf) +} + /// Splits a stream into halves that read and write concurrently, for /// full-duplex protocols (an upgraded connection, a proxied tunnel). pub trait Split { @@ -270,6 +322,21 @@ pub(crate) mod sealed { &self, buf: B, ) -> impl Future>; + /// [`Write::write_all_owned`](crate::Write::write_all_owned); a + /// socket that sends without copying overrides it. + fn write_all_owned( + &self, + buf: B, + ) -> impl Future>> + where + Self: Sized, + { + async move { + let BufResult(res, buf) = + super::write_all_vectored(&mut super::ByRef(self), buf).await; + BufResult(res, Some(buf)) + } + } fn shutdown(&self) -> impl Future>; fn send_file + 'static>( &self, @@ -299,6 +366,31 @@ pub(crate) mod sealed { use sealed::SharedStream; +/// [`Write`] over a shared reference to a socket, for the copying paths +/// that reuse the trait's provided methods. +pub(crate) struct ByRef<'a, S>(pub(crate) &'a S); + +impl Write for ByRef<'_, S> { + fn write(&mut self, buf: B) -> impl Future> { + self.0.write(buf) + } + + fn write_vectored( + &mut self, + buf: B, + ) -> impl Future> { + self.0.write_vectored(buf) + } + + async fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + + fn shutdown(&mut self) -> impl Future> { + self.0.shutdown() + } +} + /// Implements the public stream traits for a whole socket type over its /// [`SharedStream`] operations, and [`Split`] into [`ReadHalf`]/[`WriteHalf`]. #[cfg(any(feature = "compio", feature = "tokio", feature = "async-io"))] @@ -348,6 +440,13 @@ macro_rules! impl_stream { $crate::stream::sealed::SharedStream::write_vectored(&*self, buf) } + fn write_all_owned( + &mut self, + buf: B, + ) -> impl ::std::future::Future>> { + $crate::stream::sealed::SharedStream::write_all_owned(&*self, buf) + } + async fn flush(&mut self) -> ::std::io::Result<()> { Ok(()) } @@ -515,6 +614,13 @@ impl Write for WriteHalf { self.0.write_vectored(buf) } + fn write_all_owned( + &mut self, + buf: B, + ) -> impl Future>> { + self.0.write_all_owned(buf) + } + async fn flush(&mut self) -> io::Result<()> { Ok(()) } @@ -584,11 +690,11 @@ impl WriteControl for WriteHalf { not(any(target_os = "linux", target_os = "android", target_vendor = "apple")) ))] pub(crate) mod copy { - use std::{borrow::Borrow, fs::File, future::Future, io}; + use std::{borrow::Borrow, fs::File}; - use compio_buf::{BufResult, IoBuf, IoVectoredBuf}; + use compio_buf::BufResult; - use super::{SharedStream, Write}; + use super::{ByRef, SharedStream, Write}; /// One positional read into a buffer, then a write of all of it. The read is a /// synchronous call, which is what `sendfile` costs on a page-cache miss too. @@ -622,29 +728,4 @@ pub(crate) mod copy { let BufResult(res, _) = ByRef(stream).write_all(chunk).await; BufResult(res.map(|()| n), file) } - - /// [`Write`] over a shared reference, so the fallback reuses - /// [`Write::write_all`]. - struct ByRef<'a, S>(&'a S); - - impl Write for ByRef<'_, S> { - fn write(&mut self, buf: B) -> impl Future> { - self.0.write(buf) - } - - fn write_vectored( - &mut self, - buf: B, - ) -> impl Future> { - self.0.write_vectored(buf) - } - - async fn flush(&mut self) -> io::Result<()> { - Ok(()) - } - - fn shutdown(&mut self) -> impl Future> { - self.0.shutdown() - } - } } diff --git a/io/tests/io.rs b/io/tests/io.rs index 2cb7e87..4577653 100644 --- a/io/tests/io.rs +++ b/io/tests/io.rs @@ -872,13 +872,13 @@ mod compio { .block_on(future) } - suite!(http_kit_io::compio::Compio, run); + suite!(http_kit_io::compio::Compio::default(), run); /// A connection accepted by a plain std listener on one thread is adopted /// by a compio runtime on another and serves both directions there. #[test] fn a_std_stream_accepted_elsewhere_is_adopted() { - use http_kit_io::{Read as _, Write as _, compio::TcpStream}; + use http_kit_io::{Read as _, Write as _, compio::Compio}; use std::io::{Read as _, Write as _}; let listener = std::net::TcpListener::bind(super::loopback()).expect("bind"); @@ -888,8 +888,8 @@ mod compio { let accepted = accepted.join().expect("accept thread"); let server = std::thread::spawn(move || { run(async move { - let mut stream = - TcpStream::from(::compio::net::TcpStream::from_std(accepted).expect("adopt")); + let mut stream = Compio::default() + .tcp_stream(::compio::net::TcpStream::from_std(accepted).expect("adopt")); let http_kit_io::BufResult(read, buf) = stream.read(Vec::with_capacity(64)).await; read.expect("read"); let http_kit_io::BufResult(written, _) = stream.write(buf).await; @@ -943,9 +943,10 @@ mod compio { run(async { // The driver arms a poll on its own wake-up eventfd the first time // it parks; park it once so that submission is not counted. - Compio.sleep(std::time::Duration::from_millis(1)).await; + let io = Compio::default(); + io.sleep(std::time::Duration::from_millis(1)).await; - let listener = Compio.bind_tcp(super::loopback()).expect("bind"); + let listener = io.bind_tcp(super::loopback()).expect("bind"); let addr = listener.local_addr().expect("local address"); let before = submitted(); // Submits the accept, then drops the call that submitted it. @@ -1012,7 +1013,7 @@ mod compio { ) -> (http_kit_io::compio::TcpStream, std::thread::JoinHandle<()>) { use http_kit_io::{Listen as _, Listener as _, compio::Compio}; - let listener = Compio.bind_tcp(super::loopback()).expect("bind"); + let listener = Compio::default().bind_tcp(super::loopback()).expect("bind"); let addr = listener.local_addr().expect("local address"); let peer = std::thread::spawn(move || peer(std::net::TcpStream::connect(addr).expect("connect"))); @@ -1041,7 +1042,9 @@ mod compio { run_with_ring(BUFFERS, BUFFER_LEN, async move { // The driver arms a poll on its own wake-up eventfd the first time // it parks; park it once so that submission is not counted. - Compio.sleep(std::time::Duration::from_millis(1)).await; + Compio::default() + .sleep(std::time::Duration::from_millis(1)) + .await; // Each message waits for the one before it to be read, so the // receive always finds a free buffer. let (mut stream, peer) = accept_one(move |mut client| { @@ -1225,7 +1228,7 @@ mod compio { }); let (busy, bytes) = run(async { - let io = http_kit_io::compio::Compio; + 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")) @@ -1292,7 +1295,7 @@ mod compio { }); let busy = run(async { - let io = http_kit_io::compio::Compio; + 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")) @@ -1324,6 +1327,180 @@ mod compio { "kernel workers held during the stall: {busy:?}" ); } + + /// A buffer that records when it is dropped. + #[cfg(target_os = "linux")] + struct Tracked { + bytes: Vec, + dropped: std::rc::Rc>, + } + + #[cfg(target_os = "linux")] + impl http_kit_io::IoBuf for Tracked { + fn as_init(&self) -> &[u8] { + &self.bytes + } + } + + #[cfg(target_os = "linux")] + impl Drop for Tracked { + fn drop(&mut self) { + self.dropped.set(true); + } + } + + /// Runs `future`, failing the test if it takes longer than `limit`. + #[cfg(target_os = "linux")] + async fn within(limit: std::time::Duration, what: &str, future: F) -> F::Output { + let expired = async { + ::compio::time::sleep(limit).await; + panic!("{what} took longer than {limit:?}"); + }; + futures_lite::future::or(future, expired).await + } + + /// A zero-copy write returns while the kernel still reads the caller's + /// buffer: a peer that reads nothing leaves most of it unsent. The stream + /// keeps the buffer, and the kernel's notification, which comes once the + /// peer has read every byte, drops it. + #[cfg(target_os = "linux")] + #[test] + fn the_notification_releases_a_zero_copy_buffer() { + use std::{cell::Cell, io::Read as _, rc::Rc, sync::mpsc, time::Duration}; + + use http_kit_io::{BufResult, Listen as _, Listener as _, Write as _, compio::Compio}; + + /// What the peer's receive buffer is asked to hold: far less than + /// the write, so the rest waits in the sender's queue. + const PEER_BUFFER: usize = 16 * 1024; + /// Above the default threshold, and within the sender's send buffer + /// (which zero-copy pages count against) once raised. + const LEN: usize = 256 * 1024; + + let data = super::payload()[..LEN].to_vec(); + let (release, released) = 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(PEER_BUFFER) + .expect("peer receive buffer"); + socket + .connect(&socket2::SockAddr::from(addr)) + .expect("connect"); + let mut stream = std::net::TcpStream::from(socket); + released.recv().expect("release"); + let mut received = Vec::new(); + stream.read_to_end(&mut received).expect("read"); + received + }); + + let sent = data.clone(); + run(async move { + let io = Compio::default(); + assert!(LEN >= io.zero_copy().threshold()); + 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(4 * LEN) + .expect("send buffer"); + + let dropped = Rc::new(Cell::new(false)); + let buf = Tracked { + bytes: sent, + dropped: Rc::clone(&dropped), + }; + let BufResult(res, kept) = within( + Duration::from_secs(10), + "the zero-copy write", + stream.write_all_owned([buf]), + ) + .await; + res.expect("zero-copy write"); + assert!( + kept.is_none(), + "the stream handed back a buffer the kernel may still read" + ); + assert!( + !dropped.get(), + "the buffer was dropped while the peer had read nothing" + ); + + release.send(()).expect("release the peer"); + within(Duration::from_secs(10), "the notification", async { + while !dropped.get() { + ::compio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await; + http_kit_io::Write::shutdown(&mut stream) + .await + .expect("shutdown"); + }); + super::assert_bytes(&peer.join().expect("peer thread"), &data); + } + + /// A write below the threshold copies and hands its buffer back, and so + /// does any write once the threshold is raised on a clone of the setting + /// the stream already holds. + #[cfg(target_os = "linux")] + #[test] + fn the_zero_copy_threshold_is_a_runtime_setting() { + use std::io::Read as _; + + use http_kit_io::{ + BufResult, Listen as _, Listener as _, Write as _, + compio::{Compio, ZeroCopySend}, + }; + + let data = super::payload()[..64 * 1024].to_vec(); + let (addr_tx, addr_rx) = std::sync::mpsc::channel(); + let peer = std::thread::spawn(move || { + let addr = addr_rx.recv().expect("listener address"); + let mut stream = std::net::TcpStream::connect(addr).expect("connect"); + let mut received = Vec::new(); + stream.read_to_end(&mut received).expect("read"); + received + }); + + let sent = data.clone(); + run(async move { + let setting = ZeroCopySend::new(sent.len() + 1); + let io = Compio::with_zero_copy(setting.clone()); + 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"); + + let BufResult(res, kept) = stream.write_all_owned([sent]).await; + res.expect("write below the threshold"); + let [sent] = kept.expect("a copying write hands the buffer back"); + + setting.set_threshold(0); + let BufResult(res, kept) = stream.write_all_owned([sent]).await; + res.expect("zero-copy write"); + assert!(kept.is_none(), "the lowered threshold did not apply"); + + setting.set_threshold(usize::MAX); + let BufResult(res, kept) = stream.write_all_owned([vec![0; 64 * 1024]]).await; + res.expect("write with zero-copy off"); + assert!(kept.is_some(), "the raised threshold did not apply"); + http_kit_io::Write::shutdown(&mut stream) + .await + .expect("shutdown"); + }); + let received = peer.join().expect("peer thread"); + let mut expected = data.clone(); + expected.extend_from_slice(&data); + expected.extend_from_slice(&[0; 64 * 1024]); + super::assert_bytes(&received, &expected); + } } #[cfg(feature = "tokio")] diff --git a/tls/src/lib.rs b/tls/src/lib.rs index 655c56a..8eaea9f 100644 --- a/tls/src/lib.rs +++ b/tls/src/lib.rs @@ -95,23 +95,24 @@ //! let connector = TlsConnector::new(Arc::new(client)); //! //! compio::runtime::Runtime::new()?.block_on(async { -//! let listener = Compio.bind_tcp("127.0.0.1:0".parse().unwrap())?; +//! let io = Compio::default(); +//! let listener = io.bind_tcp("127.0.0.1:0".parse().unwrap())?; //! let addr = listener.local_addr()?; //! let service = |request: Request| async move { //! Response::new(Body::from_bytes(request.uri().path().to_owned())) //! }; //! let server = async { //! let (stream, _) = listener.accept().await?; -//! let stream = acceptor.accept(stream, &Compio).await.map_err(std::io::Error::other)?; +//! let stream = acceptor.accept(stream, &io).await.map_err(std::io::Error::other)?; //! assert_eq!(stream.alpn_protocol(), Some(&b"http/1.1"[..])); -//! let connection = ServerConnection::from_stream(stream, Compio, ServerConfig::default()); +//! let connection = ServerConnection::from_stream(stream, io.clone(), ServerConfig::default()); //! connection.serve(&service).await.map_err(std::io::Error::other) //! }; //! let client = async { -//! let stream = Compio.connect(addr).await?; +//! let stream = io.connect(addr).await?; //! let name = "localhost".try_into().unwrap(); -//! let stream = connector.connect(name, stream, &Compio).await.map_err(std::io::Error::other)?; -//! let (send, connection) = handshake_stream(stream, Compio, ClientConfig::default()); +//! let stream = connector.connect(name, stream, &io).await.map_err(std::io::Error::other)?; +//! let (send, connection) = handshake_stream(stream, io.clone(), ClientConfig::default()); //! let request = async move { //! let mut request = Request::new(Body::empty()); //! *request.uri_mut() = format!("https://localhost:{}/hello", addr.port()).parse().unwrap(); diff --git a/tls/tests/tls.rs b/tls/tests/tls.rs index df49f69..15d40f8 100644 --- a/tls/tests/tls.rs +++ b/tls/tests/tls.rs @@ -1352,7 +1352,7 @@ mod compio { .block_on(future) } - suite!(http_kit_io::compio::Compio, run); + suite!(http_kit_io::compio::Compio::default(), run); } mod tokio { From 6a1a0af698f72dc5d5ea029f11520e7c6da0f089 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 07:43:52 -0500 Subject: [PATCH 2/3] feat(h1): send queued buffers through write_all_owned A message's head and data frames go to the writer as owned buffers, so on io_uring a large flush is sent zero-copy; the buffer list's allocation is reused only when the writer hands it back. --- h1/src/conn.rs | 48 ++++++++++++------------------------------------ 1 file changed, 12 insertions(+), 36 deletions(-) diff --git a/h1/src/conn.rs b/h1/src/conn.rs index 3e332da..94b0103 100644 --- a/h1/src/conn.rs +++ b/h1/src/conn.rs @@ -144,44 +144,20 @@ impl Outbox { self.bufs.len() >= MAX_WRITE_BUFFERS } - /// Writes everything queued. + /// Writes everything queued. The buffers go to the writer, which may + /// keep them until the kernel has sent them (a zero-copy send); a writer + /// that hands them back has their allocation reused for the next message. pub(crate) async fn flush(&mut self, writer: &mut W) -> Result<(), Error> { - let mut bufs = mem::take(&mut self.bufs); - while !bufs.is_empty() { - let BufResult(result, returned) = writer.write_vectored(bufs).await; - bufs = returned; - let mut written = match result { - Ok(0) => { - return Err(io::Error::new( - io::ErrorKind::WriteZero, - "stream accepted no bytes", - ) - .into()); - } - Ok(n) => n, - Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, - Err(error) => return Err(error.into()), - }; - let done = bufs - .iter() - .take_while(|buf| { - if written >= buf.len() { - written -= buf.len(); - true - } else { - false - } - }) - .count(); - bufs.drain(..done); - if written > 0 { - let first = &mut bufs[0]; - *first = Bytes::slice(first, written..); - } + if self.bufs.is_empty() { + return Ok(()); + } + let bufs = mem::take(&mut self.bufs); + let BufResult(result, returned) = writer.write_all_owned(bufs).await; + if let Some(mut bufs) = returned { + bufs.clear(); + self.bufs = bufs; } - // Hand the allocation back for the next message. - self.bufs = bufs; - Ok(()) + result.map_err(Error::from) } } From 0a9d895484fae5b06979315559772c323a44b089 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 07:50:36 -0500 Subject: [PATCH 3/3] test(h3): clone the compio driver into the curl server --- h3/tests/curl.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/h3/tests/curl.rs b/h3/tests/curl.rs index 489a483..da1a3b6 100644 --- a/h3/tests/curl.rs +++ b/h3/tests/curl.rs @@ -54,7 +54,7 @@ fn answers_curl_http3() { on_compio(|io| async move { let socket = io.bind_udp(loopback()).await.unwrap(); let port = socket.local_addr().unwrap().port(); - let server = Server::new(socket, io, &server_config(&pki)).unwrap(); + let server = Server::new(socket, io.clone(), &server_config(&pki)).unwrap(); let (stop, stopped) = async_channel::bounded::<()>(1); let served = server.serve_until(&echo, async move { stopped.recv().await.ok();