From be4bc7de837bf7c665d3ddad2bab25f47c5932f0 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 16:17:48 -0500 Subject: [PATCH 1/2] build: depend on compio from the lexoliu/compio skyzen branch The branch carries the io_uring changes the receive path needs ahead of a compio release: IORING_CQE_F_BUF_MORE through Extra, a buffer pool that recycles incrementally consumed buffers only once the kernel is done, the second half of 32-byte CQEs and zero-copy notification results. --- h1/Cargo.toml | 2 +- h3/Cargo.toml | 2 +- io/Cargo.toml | 4 ++-- tls/Cargo.toml | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/h1/Cargo.toml b/h1/Cargo.toml index ed06c1e..ce09123 100644 --- a/h1/Cargo.toml +++ b/h1/Cargo.toml @@ -34,7 +34,7 @@ tracing = { version = "0.1.44", default-features = false, features = ["std"] } [dev-dependencies] http-kit-io = { version = "0.6.2", path = "../io", features = ["tokio", "compio"] } -compio = { version = "0.19.2", default-features = false, features = ["runtime"] } +compio = { git = "https://github.com/lexoliu/compio", branch = "skyzen", default-features = false, features = ["runtime"] } tokio = { version = "1.54.1", default-features = false, features = ["rt"] } [lints.rust] diff --git a/h3/Cargo.toml b/h3/Cargo.toml index bb31c5e..f158e7a 100644 --- a/h3/Cargo.toml +++ b/h3/Cargo.toml @@ -49,7 +49,7 @@ tracing = { version = "0.1.44", default-features = false, features = ["std"], op [dev-dependencies] socket2 = { version = "0.6.5", default-features = false } http-kit-io = { version = "0.6.2", path = "../io", features = ["tokio", "compio"] } -compio = { version = "0.19.2", default-features = false, features = ["runtime"] } +compio = { git = "https://github.com/lexoliu/compio", branch = "skyzen", default-features = false, features = ["runtime"] } tokio = { version = "1.54.1", default-features = false, features = ["rt"] } [lints.rust] diff --git a/io/Cargo.toml b/io/Cargo.toml index fdf79bf..3a33500 100644 --- a/io/Cargo.toml +++ b/io/Cargo.toml @@ -21,11 +21,11 @@ tokio = ["dep:tokio", "dep:socket2", "dep:libc", "dep:windows-sys"] async-io = ["dep:async-io", "dep:socket2", "dep:libc", "dep:windows-sys"] [dependencies] -compio-buf = { version = "0.8.3", default-features = false, features = ["bytes"] } +compio-buf = { git = "https://github.com/lexoliu/compio", branch = "skyzen", default-features = false, features = ["bytes"] } futures-core = { version = "0.3.34", default-features = false } futures-io = { version = "0.3.34", default-features = false, features = ["std"] } -compio = { version = "0.19.2", default-features = false, features = ["runtime", "net", "time", "io-uring", "polling"], optional = true } +compio = { git = "https://github.com/lexoliu/compio", branch = "skyzen", default-features = false, features = ["runtime", "net", "time", "io-uring", "polling"], optional = true } tokio = { version = "1.54.1", default-features = false, features = ["net", "time"], optional = true } async-io = { version = "2.6.0", default-features = false, optional = true } socket2 = { version = "0.6.5", default-features = false, optional = true } diff --git a/tls/Cargo.toml b/tls/Cargo.toml index 2a68649..817c929 100644 --- a/tls/Cargo.toml +++ b/tls/Cargo.toml @@ -37,7 +37,7 @@ http-kit-h1 = { version = "0.6.2", path = "../h1", features = ["fs"] } rustls = { version = "0.23.46", default-features = false, features = ["std", "ring", "tls12"] } rcgen = { version = "0.14.10", default-features = false, features = ["crypto", "ring", "pem"] } http = { version = "1.5.0", default-features = false, features = ["std"] } -compio = { version = "0.19.2", default-features = false, features = ["runtime"] } +compio = { git = "https://github.com/lexoliu/compio", branch = "skyzen", default-features = false, features = ["runtime"] } tokio = { version = "1.54.1", default-features = false, features = ["rt"] } [lints.rust] From cf1d3c7f891b73ba42d75a89e4f28fcd50558de2 Mon Sep 17 00:00:00 2001 From: Lexo Liu Date: Wed, 11 Nov 2026 16:17:48 -0500 Subject: [PATCH 2/2] perf(io): consume the receive buffer ring incrementally A multishot receive used a whole ring buffer per completion, however few bytes arrived. On a ring registered with IOU_PBUF_RING_INC the kernel now fills a buffer across completions, so small reads share a buffer instead of draining the ring; compio lends each read its part and recycles the buffer once the kernel is done and every part is released. Kernels before 6.12 reject the flag at registration, and the ring consumes whole buffers. The ring tests run on both consumption modes, and a new test checks that consecutive small reads of a connection continue in one buffer. Closes #163 --- io/src/compio.rs | 35 +++--- io/tests/io.rs | 289 ++++++++++++++++++++++++++++++----------------- 2 files changed, 205 insertions(+), 119 deletions(-) diff --git a/io/src/compio.rs b/io/src/compio.rs index b4ac9bd..175ffd7 100644 --- a/io/src/compio.rs +++ b/io/src/compio.rs @@ -15,20 +15,26 @@ //! On Linux over io_uring a stream receives with a multishot receive into //! the provided buffer ring of the thread's runtime: one submission stays //! armed and completes once per arrival, each into a ring buffer, until the -//! kernel ends it. A reader takes a ring buffer through -//! [`ReadOwned`](crate::ReadOwned), or has it copied into its own through -//! [`Read`](crate::Read); either way the buffer goes back to the ring as soon -//! as its bytes are consumed, so an idle connection holds none. When the -//! ring has no free buffer a read receives into the caller's buffer with a -//! submission of its own, and the next one arms again. A splice from the -//! socket first moves what the ring already received, so it never passes -//! those bytes. +//! kernel ends it. A reader takes the received bytes in the ring through +//! [`ReadOwned`](crate::ReadOwned), or has them copied into its own buffer +//! through [`Read`](crate::Read); either way they are released as soon as +//! they are consumed, so an idle connection holds none of the ring. When the ring has +//! no free buffer a read receives into the caller's buffer with a submission +//! of its own, and the next one arms again. A splice from the socket first +//! moves what the ring already received, so it never passes those bytes. //! //! The ring is compio's buffer pool, one per runtime: size it on the //! runtime's `ProactorBuilder` (`buffer_pool_size`, `buffer_pool_buffer_len`) -//! for the connections one thread serves. Buffers are consumed whole, one per -//! completion: compio reports no `IORING_CQE_F_BUF_MORE`, which incremental -//! consumption (`IOU_PBUF_RING_INC`) needs. +//! for the connections one thread serves, and have the kernel consume its +//! buffers incrementally with `buffer_pool_flag(IOU_PBUF_RING_INC)`. On Linux +//! 6.12 and later a completion then fills only as much of a buffer as it +//! received and marks the buffer `IORING_CQE_F_BUF_MORE` while it keeps it; +//! the next completion, of this connection or another on the thread, +//! continues in it, so small reads share a buffer instead of taking one each. +//! A buffer goes back to the ring once the kernel has finished with it and +//! every read lent out of it is released. An earlier kernel rejects the flag +//! when the ring registers, and compio then registers the ring without it: +//! buffers are consumed whole, one per completion. //! //! On Linux over io_uring a TCP stream's //! [`write_all_owned`](crate::Write::write_all_owned) of at least @@ -1560,9 +1566,10 @@ mod zero_copy { } /// 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 -/// bytes. +/// the reader by [`ReadOwned`](crate::ReadOwned). Dropping it releases them, +/// handing the buffer back to the ring unless the kernel still fills it or +/// other bytes lent out of it are held, so a reader holds it only as long as +/// it needs the bytes. #[cfg(target_os = "linux")] pub struct RecvBuf { buf: ::compio::driver::BufferRef, diff --git a/io/tests/io.rs b/io/tests/io.rs index d9a3515..089ea52 100644 --- a/io/tests/io.rs +++ b/io/tests/io.rs @@ -996,14 +996,29 @@ mod compio { }); } + /// Both ways a ring can consume its buffers: whole, and incrementally + /// where the kernel accepts it (Linux 6.12; whole before). + #[cfg(target_os = "linux")] + const CONSUMPTIONS: [bool; 2] = [false, true]; + /// A runtime whose provided buffer ring has `buffers` buffers of `len` - /// bytes each. + /// bytes each, registered for incremental consumption if `incremental`. #[cfg(target_os = "linux")] - fn run_with_ring(buffers: u16, len: usize, future: F) -> F::Output { + fn run_with_ring( + buffers: u16, + len: usize, + incremental: bool, + future: F, + ) -> F::Output { let mut proactor = ::compio::driver::ProactorBuilder::new(); proactor .buffer_pool_size(std::num::NonZeroU16::new(buffers).expect("a ring has buffers")) .buffer_pool_buffer_len(len); + if incremental { + proactor.buffer_pool_flag( + u16::try_from(io_uring::types::IOU_PBUF_RING_INC).expect("a ring flag"), + ); + } ::compio::runtime::RuntimeBuilder::new() .with_proactor(proactor) .build() @@ -1042,56 +1057,58 @@ mod compio { const MESSAGE: usize = 1024; const MESSAGES: usize = 64; - let data = super::payload()[..MESSAGE * MESSAGES].to_vec(); - let sent = data.clone(); - 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::default() - .sleep(std::time::Duration::from_millis(1)) + for incremental in CONSUMPTIONS { + let data = super::payload()[..MESSAGE * MESSAGES].to_vec(); + let sent = data.clone(); + run_with_ring(BUFFERS, BUFFER_LEN, incremental, 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::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| { + for message in sent.chunks(MESSAGE) { + client.write_all(message).expect("send"); + client.read_exact(&mut [0]).expect("wait for the read"); + } + }) .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| { - for message in sent.chunks(MESSAGE) { - client.write_all(message).expect("send"); - client.read_exact(&mut [0]).expect("wait for the read"); - } - }) - .await; - // The stream takes its slot with a submission of its own; count - // from once it holds it, so the receive is armed on the slot. - until_held(&socket_name(&stream), 1).await; - let before = submitted(); - let mut received = Vec::with_capacity(data.len()); - let mut reads = 0; - while received.len() < data.len() { - let Received::Lent(buf) = stream.read_owned().await.expect("read") else { - panic!("the ring ran dry or the peer closed early"); - }; - reads += 1; - received.extend_from_slice(&buf); - drop(buf); - if received.len() % MESSAGE == 0 { - // The acknowledgement is a send, which is a submission; - // count it out. - let ack = submitted(); - let http_kit_io::BufResult(res, _) = - http_kit_io::Write::write(&mut stream, vec![0]).await; - res.expect("acknowledge"); - assert_eq!(submitted() - ack, 1, "an acknowledgement is one send"); + // The stream takes its slot with a submission of its own; count + // from once it holds it, so the receive is armed on the slot. + until_held(&socket_name(&stream), 1).await; + let before = submitted(); + let mut received = Vec::with_capacity(data.len()); + let mut reads = 0; + while received.len() < data.len() { + let Received::Lent(buf) = stream.read_owned().await.expect("read") else { + panic!("the ring ran dry or the peer closed early"); + }; + reads += 1; + received.extend_from_slice(&buf); + drop(buf); + if received.len() % MESSAGE == 0 { + // The acknowledgement is a send, which is a submission; + // count it out. + let ack = submitted(); + let http_kit_io::BufResult(res, _) = + http_kit_io::Write::write(&mut stream, vec![0]).await; + res.expect("acknowledge"); + assert_eq!(submitted() - ack, 1, "an acknowledgement is one send"); + } } - } - let acks = u32::try_from(MESSAGES).expect("message count"); - assert_eq!( - submitted() - before - acks, - 1, - "receive submissions for {reads} reads" - ); - assert!(reads >= MESSAGES, "{reads} reads"); - super::assert_bytes(&received, &data); - peer.join().expect("peer thread"); - }); + let acks = u32::try_from(MESSAGES).expect("message count"); + assert_eq!( + submitted() - before - acks, + 1, + "receive submissions for {reads} reads" + ); + assert!(reads >= MESSAGES, "{reads} reads"); + super::assert_bytes(&received, &data); + peer.join().expect("peer thread"); + }); + } } /// A connection that carries far more than the ring holds, sent as fast @@ -1103,27 +1120,29 @@ mod compio { fn a_connection_reads_more_than_the_ring_holds() { use std::io::Write as _; - let data = super::payload(); - let sent = data.clone(); - run_with_ring(2, 4096, async move { - let (mut stream, peer) = accept_one(move |mut client| { - client.write_all(&sent).expect("send"); - }) - .await; - let mut received = Vec::with_capacity(data.len()); - let mut buf = Vec::with_capacity(1500); - loop { - let http_kit_io::BufResult(res, filled) = - http_kit_io::Read::read(&mut stream, buf).await; - if res.expect("read") == 0 { - break; + for incremental in CONSUMPTIONS { + let data = super::payload(); + let sent = data.clone(); + run_with_ring(2, 4096, incremental, async move { + let (mut stream, peer) = accept_one(move |mut client| { + client.write_all(&sent).expect("send"); + }) + .await; + let mut received = Vec::with_capacity(data.len()); + let mut buf = Vec::with_capacity(1500); + loop { + let http_kit_io::BufResult(res, filled) = + http_kit_io::Read::read(&mut stream, buf).await; + if res.expect("read") == 0 { + break; + } + received.extend_from_slice(&filled); + buf = filled; } - received.extend_from_slice(&filled); - buf = filled; - } - super::assert_bytes(&received, &data); - peer.join().expect("peer thread"); - }); + super::assert_bytes(&received, &data); + peer.join().expect("peer thread"); + }); + } } /// A splice from a socket that has received ahead of its reader moves @@ -1138,33 +1157,35 @@ mod compio { splice::{Pipe, Spliced}, }; - let data = "世界 せかい 세계 ".repeat(512).into_bytes(); - let sent = data.clone(); - run_with_ring(8, 4096, async move { - let (mut stream, peer) = accept_one(move |mut client| { - client.write_all(&sent).expect("send"); - }) - .await; - // A small read leaves the rest of its ring buffer held. - let http_kit_io::BufResult(res, head) = - http_kit_io::Read::read(&mut stream, Vec::with_capacity(10)).await; - assert_eq!(res.expect("read"), 10); - let mut received = head; - let mut pipe = Pipe::new().expect("pipe"); - loop { - match stream.splice_to_pipe(pipe, usize::MAX).await { - Ok(Spliced::Filled(filled)) => { - let (bytes, empty) = filled.read_out().expect("read the pipe"); - received.extend_from_slice(&bytes); - pipe = empty; + for incremental in CONSUMPTIONS { + let data = "世界 せかい 세계 ".repeat(512).into_bytes(); + let sent = data.clone(); + run_with_ring(8, 4096, incremental, async move { + let (mut stream, peer) = accept_one(move |mut client| { + client.write_all(&sent).expect("send"); + }) + .await; + // A small read leaves the rest of its ring buffer held. + let http_kit_io::BufResult(res, head) = + http_kit_io::Read::read(&mut stream, Vec::with_capacity(10)).await; + assert_eq!(res.expect("read"), 10); + let mut received = head; + let mut pipe = Pipe::new().expect("pipe"); + loop { + match stream.splice_to_pipe(pipe, usize::MAX).await { + Ok(Spliced::Filled(filled)) => { + let (bytes, empty) = filled.read_out().expect("read the pipe"); + received.extend_from_slice(&bytes); + pipe = empty; + } + Ok(Spliced::End(_)) => break, + Err((error, _)) => panic!("splice: {error}"), } - Ok(Spliced::End(_)) => break, - Err((error, _)) => panic!("splice: {error}"), } - } - super::assert_bytes(&received, &data); - peer.join().expect("peer thread"); - }); + super::assert_bytes(&received, &data); + peer.join().expect("peer thread"); + }); + } } /// Stopping the read-ahead hands back what was received and not read, @@ -1176,20 +1197,78 @@ mod compio { use http_kit_io::ReadAhead as _; - let data = "世界 せかい 세계 ".repeat(4096).into_bytes(); + for incremental in CONSUMPTIONS { + let data = "世界 せかい 세계 ".repeat(4096).into_bytes(); + let sent = data.clone(); + run_with_ring(8, 4096, incremental, async move { + let (mut stream, peer) = accept_one(move |mut client| { + client.write_all(&sent).expect("send"); + }) + .await; + let http_kit_io::BufResult(res, head) = + http_kit_io::Read::read(&mut stream, Vec::with_capacity(10)).await; + assert_eq!(res.expect("read"), 10); + let mut received = head; + received.extend_from_slice(&stream.stop_read_ahead().await); + received.extend_from_slice(&super::read_to_end(&mut stream).await); + super::assert_bytes(&received, &data); + peer.join().expect("peer thread"); + }); + } + } + + /// On a ring the kernel consumes incrementally, consecutive small reads + /// of a connection share a ring buffer, each continuing where the one + /// before ended: a ring of two buffers lends eight reads at once, where + /// consuming buffers whole would run it dry after two. + #[cfg(target_os = "linux")] + #[test] + fn small_reads_share_a_ring_buffer() { + use std::io::{Read as _, Write as _}; + + use http_kit_io::{ReadOwned as _, Received}; + + const MESSAGE: usize = 16; + const MESSAGES: usize = 8; + + let data = super::payload()[..MESSAGE * MESSAGES].to_vec(); let sent = data.clone(); - run_with_ring(8, 4096, async move { + run_with_ring(2, 4096, true, async move { + let incremental = ::compio::runtime::Runtime::with_current(|runtime| { + runtime.buffer_pool()?.is_incremental() + }) + .expect("the runtime's ring"); + if !incremental { + eprintln!("the kernel consumes ring buffers whole (before Linux 6.12), skipping"); + return; + } + // Each message waits for the one before it to be read, so each + // read is one message. let (mut stream, peer) = accept_one(move |mut client| { - client.write_all(&sent).expect("send"); + for message in sent.chunks(MESSAGE) { + client.write_all(message).expect("send"); + client.read_exact(&mut [0]).expect("wait for the read"); + } }) .await; - let http_kit_io::BufResult(res, head) = - http_kit_io::Read::read(&mut stream, Vec::with_capacity(10)).await; - assert_eq!(res.expect("read"), 10); - let mut received = head; - received.extend_from_slice(&stream.stop_read_ahead().await); - received.extend_from_slice(&super::read_to_end(&mut stream).await); - super::assert_bytes(&received, &data); + let mut lent = Vec::with_capacity(MESSAGES); + for message in data.chunks(MESSAGE) { + let Received::Lent(buf) = stream.read_owned().await.expect("read") else { + panic!("the ring ran dry after {} reads", lent.len()); + }; + assert_eq!(&*buf, message); + lent.push(buf); + let http_kit_io::BufResult(res, _) = + http_kit_io::Write::write(&mut stream, vec![0]).await; + res.expect("acknowledge"); + } + for pair in lent.windows(2) { + assert_eq!( + pair[1].as_ptr(), + pair[0].as_ptr().wrapping_add(pair[0].len()), + "a read continues in the buffer of the one before" + ); + } peer.join().expect("peer thread"); }); }