Benchmark incremental multiline framing

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

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<Vec<u8>> = LazyLock::new(|| benchmark_multiline_response(BODY_64K));
+static RESPONSE_768K: LazyLock<Vec<u8>> = 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),
+    ]
+);

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;

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<memchr::memmem::Finder<'static>> =
+    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<usize> {
 
 #[inline]
 fn terminator_ends(data: &[u8]) -> impl Iterator<Item = usize> + '_ {
-    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<u8> {
+    let mut response = b"220 42 <benchmark@example.com>\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();