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/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"); }); } 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]