diff --git a/Cargo.lock b/Cargo.lock index 6f8bb87b..c391c775 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3013,9 +3013,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.43" +version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ "once_cell", "ring", @@ -3462,7 +3462,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index bb8be67b..bb427a10 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -107,6 +107,7 @@ gungraun = "0.19.4" # Callgrind instruction/branch/cache benchmarking [features] default = [] +framing-bench = [] zlib-ng = ["flate2/zlib-ng"] # SIMD-optimized compression (requires cmake; ~2-3x faster) tokio-console = ["dep:console-subscriber", "tokio/tracing"] # Runtime/task instrumentation for tokio-console; build with RUSTFLAGS="--cfg tokio_unstable" @@ -140,6 +141,10 @@ harness = false name = "response_parsing" harness = false +[[bench]] +name = "multiline_framing" +harness = false + [[bench]] name = "yenc_decoding" harness = false diff --git a/benches/multiline_framing.rs b/benches/multiline_framing.rs new file mode 100644 index 00000000..cc5ce6fc --- /dev/null +++ b/benches/multiline_framing.rs @@ -0,0 +1,116 @@ +//! Compare the proxy's incremental multiline framing with a stateless control. +//! +//! Run with: +//! `nix develop -c cargo bench --features framing-bench --bench multiline_framing` + +use divan::{Bencher, black_box, counter::BytesCount}; +use nntp_proxy::session::{ + benchmark_incremental_multiline_frame, benchmark_multiline_response, + benchmark_stateless_multiline_frame, +}; +use std::sync::LazyLock; + +fn main() { + divan::main(); +} + +const BODY_64K: usize = 64 * 1024; +const BODY_768K: usize = 768 * 1024; + +static RESPONSE_64K: LazyLock> = LazyLock::new(|| benchmark_multiline_response(BODY_64K)); +static RESPONSE_768K: LazyLock> = LazyLock::new(|| benchmark_multiline_response(BODY_768K)); + +fn bench_incremental(bencher: Bencher, response: &[u8], chunk_size: usize) { + bencher.counter(BytesCount::new(response.len())).bench(|| { + black_box(benchmark_incremental_multiline_frame( + black_box(response), + chunk_size, + )) + }); +} + +fn bench_stateless(bencher: Bencher, response: &[u8], chunk_size: usize) { + bencher.counter(BytesCount::new(response.len())).bench(|| { + black_box(benchmark_stateless_multiline_frame( + black_box(response), + chunk_size, + )) + }); +} + +macro_rules! incremental_benchmarks { + ($module:ident, $response:ident, [$(($name:ident, $chunk:expr)),+ $(,)?]) => { + mod $module { + use super::*; + + $( + #[divan::bench(sample_count = 50, sample_size = 10)] + fn $name(bencher: Bencher) { + bench_incremental(bencher, &$response, $chunk); + } + )+ + } + }; +} + +macro_rules! stateless_benchmarks { + ($module:ident, $response:ident, [$(($name:ident, $chunk:expr, $samples:expr)),+ $(,)?]) => { + mod $module { + use super::*; + + $( + #[divan::bench(sample_count = $samples, sample_size = 5)] + fn $name(bencher: Bencher) { + bench_stateless(bencher, &$response, $chunk); + } + )+ + } + }; +} + +incremental_benchmarks!( + incremental_64k, + RESPONSE_64K, + [ + (one_byte, 1), + (two_bytes, 2), + (four_bytes, 4), + (thirty_one_bytes, 31), + (chunk_256, 256), + (whole_read, usize::MAX), + ] +); + +incremental_benchmarks!( + incremental_768k, + RESPONSE_768K, + [ + (four_bytes, 4), + (thirty_one_bytes, 31), + (chunk_256k, 256 * 1024), + (whole_read, usize::MAX), + ] +); + +stateless_benchmarks!( + stateless_64k, + RESPONSE_64K, + [ + (one_byte, 1, 50), + (two_bytes, 2, 50), + (four_bytes, 4, 50), + (thirty_one_bytes, 31, 50), + (chunk_256, 256, 50), + (whole_read, usize::MAX, 50), + ] +); + +stateless_benchmarks!( + stateless_768k, + RESPONSE_768K, + [ + (thirty_one_bytes, 31, 10), + (chunk_256k, 256 * 1024, 20), + (whole_read, usize::MAX, 20), + ] +); diff --git a/src/session/mod.rs b/src/session/mod.rs index 9ba38dde..ff49c009 100644 --- a/src/session/mod.rs +++ b/src/session/mod.rs @@ -176,6 +176,11 @@ mod handlers; mod metrics_ext; mod mode_state; pub(crate) mod multiline_framing; +#[cfg(feature = "framing-bench")] +pub use multiline_framing::{ + benchmark_incremental_multiline_frame, benchmark_multiline_response, + benchmark_stateless_multiline_frame, +}; mod precheck; pub(crate) mod response_transfer; pub(crate) mod retry; diff --git a/src/session/multiline_framing.rs b/src/session/multiline_framing.rs index 22832abc..8db4f953 100644 --- a/src/session/multiline_framing.rs +++ b/src/session/multiline_framing.rs @@ -11,6 +11,7 @@ use std::borrow::Cow; use std::collections::VecDeque; use std::ops::Range; +use std::sync::LazyLock; use anyhow::Context; use smallvec::SmallVec; @@ -20,6 +21,9 @@ const TERMINATOR: &[u8; 5] = b"\r\n.\r\n"; const TERMINATOR_TAIL_SIZE: usize = 4; const MAX_CAPTURED_MULTILINE_RESPONSE_BYTES: usize = 4 * 1024 * 1024; +static TERMINATOR_FINDER: LazyLock> = + LazyLock::new(|| memchr::memmem::Finder::new(TERMINATOR)); + #[must_use] pub(crate) fn cached_response_completion() -> std::io::IoSlice<'static> { std::io::IoSlice::new(TERMINATOR) @@ -2170,7 +2174,57 @@ fn find_terminator_end(data: &[u8]) -> Option { #[inline] fn terminator_ends(data: &[u8]) -> impl Iterator + '_ { - memchr::memmem::find_iter(data, TERMINATOR).map(|found| found + TERMINATOR.len()) + TERMINATOR_FINDER + .find_iter(data) + .map(|found| found + TERMINATOR.len()) +} + +#[cfg(feature = "framing-bench")] +#[must_use] +pub fn benchmark_multiline_response(body_len: usize) -> Vec { + let mut response = b"220 42 \r\n".to_vec(); + response.extend(std::iter::repeat_n(b'x', body_len)); + response.extend_from_slice(TERMINATOR); + response +} + +/// Measure the production-path baseline while retaining only its rolling tail. +#[cfg(feature = "framing-bench")] +#[must_use] +pub fn benchmark_incremental_multiline_frame(response: &[u8], chunk_size: usize) -> usize { + let mut framer = MultilineFramer::default(); + let chunk_size = chunk_size.max(1); + + for chunk in response.chunks(chunk_size) { + if let Ok(FramedMultilineChunk::Complete(complete)) = + framer.split_chunk(chunk, PackedPendingBytesPolicy::AllowIfStatusPrefix) + { + return complete.response.end; + } + } + + 0 +} + +/// Measure the control path that repeats a full accumulated-response rescan. +#[cfg(feature = "framing-bench")] +#[must_use] +pub fn benchmark_stateless_multiline_frame(response: &[u8], chunk_size: usize) -> usize { + let chunk_size = chunk_size.max(1); + let mut received = 0; + + for chunk in response.chunks(chunk_size) { + received += chunk.len(); + let mut framer = MultilineFramer::default(); + if let Ok(FramedMultilineChunk::Complete(complete)) = framer.split_chunk( + &response[..received], + PackedPendingBytesPolicy::AllowIfStatusPrefix, + ) { + return complete.response.end; + } + } + + 0 } #[inline] @@ -3754,6 +3808,48 @@ mod tests { ); } + #[test] + fn incremental_framer_matches_rescan_for_every_two_push_split() { + let response = b"220 article\r\nbody line\r\n.\r\n"; + let expected = response.len(); + + for split in 0..=response.len() { + let mut framer = MultilineFramer::default(); + let first = framer + .split_chunk( + &response[..split], + PackedPendingBytesPolicy::AllowIfStatusPrefix, + ) + .expect("first push should not reject a valid response"); + let actual = match first { + FramedMultilineChunk::Complete(complete) => Some(complete.response.end), + FramedMultilineChunk::Incomplete(_) => framer + .split_chunk( + &response[split..], + PackedPendingBytesPolicy::AllowIfStatusPrefix, + ) + .ok() + .and_then(|result| match result { + FramedMultilineChunk::Complete(complete) => { + Some(split + complete.response.end) + } + FramedMultilineChunk::Incomplete(_) => None, + }), + }; + + assert_eq!(actual, Some(expected), "split={split}"); + + let mut stateless = MultilineFramer::default(); + let rescanned_end = stateless + .split_chunk(response, PackedPendingBytesPolicy::AllowIfStatusPrefix) + .expect("rescan should accept a valid response"); + let FramedMultilineChunk::Complete(rescanned) = rescanned_end else { + panic!("rescan did not complete for split={split}"); + }; + assert_eq!(actual, Some(rescanned.response.end), "split={split}"); + } + } + #[test] fn backend_reply_tracker_consumes_invalid_status_line() { let mut tracker = BackendReplyTracker::default();