fix: make managed input processing linear

docs/cpp.md

@@ -103,7 +103,7 @@ A struct representing a parser. It has the following fields:
 - `autodetect` (`bool`): If the parser should autodetect requests and responses. Enabled by default.
 - `is_request` (`bool`): The configured or detected message type. Set this when `autodetect` is `false`.
 - `paused` (`bool`): If the parser is paused.
-- `manage_unconsumed` (`bool`): If the parser should automatically copy and prepend unconsumed data.
+- `manage_unconsumed` (`bool`): If the parser should automatically retain and prepend unconsumed data. Retention grows geometrically and incomplete lines resume scanning at the previous endpoint, so byte-fragmented input performs linear aggregate copying and scanning.
 - `suspend_after_headers` (`bool`): If parsing should stop after headers have completed. Disabled by default.
 - `continue_without_data` (`bool`): If the next execution of the parse loop should execute even if there is no more data.
 - `is_connect` (`bool`): If the current request used `CONNECT` method.

docs/js.md

@@ -375,7 +375,7 @@ Returns `true` if the parser is paused.
 
 #### `shouldManageUnconsumed(parser)`
 
-Returns `true` if the parser should automatically copy and prepend unconsumed data.
+Returns `true` if the parser should automatically retain and prepend unconsumed data. Retention grows geometrically and incomplete lines resume scanning at the previous endpoint, so byte-fragmented input performs linear aggregate copying and scanning.
 
 #### `getMaxStartLineLength(parser)`
 

docs/rust.md

@@ -104,7 +104,7 @@ A struct representing a parser. It has the following fields:
 - `autodetect` (`bool`): If the parser should autodetect requests and responses. Enabled by default.
 - `is_request` (`bool`): The configured or detected message type. Set this when `autodetect` is `false`.
 - `paused` (`bool`): If the parser is paused.
-- `manage_unconsumed` (`bool`): If the parser should automatically copy and prepend unconsumed data.
+- `manage_unconsumed` (`bool`): If the parser should automatically retain and prepend unconsumed data. Retention grows geometrically and incomplete lines resume scanning at the previous endpoint, so byte-fragmented input performs linear aggregate copying and scanning.
 - `suspend_after_headers` (`bool`): If parsing should stop after headers have completed. Disabled by default.
 - `continue_without_data` (`bool`): If the next execution of the parse loop should execute even if there is no more data.
 - `is_connect` (`bool`): If the current request used `CONNECT` method.

parser/src/lib.rs

@@ -2,6 +2,7 @@
 
 extern crate alloc;
 
+use alloc::alloc::{Layout, alloc as allocate, dealloc as deallocate, handle_alloc_error, realloc};
 use alloc::ffi::CString;
 use alloc::vec::Vec;
 use alloc::{boxed::Box, format};
@@ -12,6 +13,16 @@ use core::str;
 use core::{mem, ptr};
 use core::{slice, slice::from_raw_parts};
 
+#[repr(C)]
+struct RetainedInput {
+  capacity: usize,
+  scanned: usize,
+  copied: usize,
+  scan_work: usize,
+}
+
+const RETAINED_INPUT_MIN_CAPACITY: usize = 64;
+
 use milo_macros::generate;
 
 #[repr(C)]
@@ -182,15 +193,156 @@ impl Parser {
     }
   }
 
+  #[inline(always)]
+  fn retained_header(&self) -> *mut RetainedInput {
+    unsafe { self.unconsumed.cast_mut().cast::<RetainedInput>().sub(1) }
+  }
+
+  #[inline(always)]
+  fn retained_layout(capacity: usize) -> Layout {
+    Layout::from_size_align(
+      mem::size_of::<RetainedInput>()
+        .checked_add(capacity)
+        .expect("retained input capacity overflow"),
+      mem::align_of::<RetainedInput>(),
+    )
+    .expect("invalid retained input layout")
+  }
+
+  fn allocate_retained(&mut self, data: &[u8], scanned: usize, scan_work: usize) {
+    let capacity = data.len().max(RETAINED_INPUT_MIN_CAPACITY).next_power_of_two();
+    let layout = Self::retained_layout(capacity);
+    let allocation = unsafe { allocate(layout) };
+    if allocation.is_null() {
+      handle_alloc_error(layout);
+    }
+
+    let header = allocation.cast::<RetainedInput>();
+    let retained = unsafe { header.add(1).cast::<c_uchar>() };
+    unsafe {
+      header.write(RetainedInput {
+        capacity,
+        scanned,
+        copied: data.len(),
+        scan_work,
+      });
+      ptr::copy_nonoverlapping(data.as_ptr(), retained, data.len());
+    }
+    self.unconsumed = retained;
+    self.unconsumed_len = data.len();
+  }
+
+  pub(crate) fn append_unconsumed(&mut self, data: &[u8]) {
+    if data.is_empty() {
+      return;
+    }
+    if self.unconsumed_len == 0 {
+      self.allocate_retained(data, 0, 0);
+      return;
+    }
+
+    let required = self
+      .unconsumed_len
+      .checked_add(data.len())
+      .expect("retained input length overflow");
+    let retained_start = self.unconsumed.addr();
+    let retained_end = retained_start
+      .checked_add(self.unconsumed_len)
+      .expect("retained input address overflow");
+    let data_start = data.as_ptr().addr();
+    let data_end = data_start.checked_add(data.len());
+    let aliased_offset = data_end
+      .filter(|end| data_start >= retained_start && *end <= retained_end)
+      .map(|_| data_start - retained_start);
+    let mut header = self.retained_header();
+    let mut capacity = unsafe { (*header).capacity };
+    if required > capacity {
+      while capacity < required {
+        capacity = capacity.checked_mul(2).expect("retained input capacity overflow");
+      }
+      let old_capacity = unsafe { (*header).capacity };
+      let old_layout = Self::retained_layout(old_capacity);
+      let allocation = unsafe { realloc(header.cast::<u8>(), old_layout, Self::retained_layout(capacity).size()) };
+      if allocation.is_null() {
+        handle_alloc_error(Self::retained_layout(capacity));
+      }
+      header = allocation.cast::<RetainedInput>();
+      unsafe {
+        (*header).capacity = capacity;
+        (*header).copied += self.unconsumed_len;
+      }
+      self.unconsumed = unsafe { header.add(1).cast::<c_uchar>() };
+    }
+
+    unsafe {
+      let source = aliased_offset.map_or(data.as_ptr(), |offset| self.unconsumed.add(offset));
+      ptr::copy(source, self.unconsumed.cast_mut().add(self.unconsumed_len), data.len());
+      (*header).copied += data.len();
+    }
+    self.unconsumed_len = required;
+  }
+
+  pub(crate) fn retain_unconsumed_suffix(&mut self, data: &[u8], scanned: usize, scan_work: usize) {
+    if data.is_empty() {
+      self.clear_unconsumed();
+      return;
+    }
+    if self.unconsumed_len == 0 {
+      self.allocate_retained(data, scanned, scan_work);
+      return;
+    }
+
+    let header = self.retained_header();
+    if data.as_ptr() != self.unconsumed {
+      unsafe {
+        ptr::copy(data.as_ptr(), self.unconsumed.cast_mut(), data.len());
+        (*header).copied += data.len();
+      }
+    }
+    self.unconsumed_len = data.len();
+    unsafe {
+      (*header).scanned = scanned.min(data.len());
+      (*header).scan_work = scan_work;
+    }
+  }
+
+  pub(crate) fn unconsumed_scan_state(&self) -> (usize, usize) {
+    if self.unconsumed_len == 0 {
+      (0, 0)
+    } else {
+      let header = self.retained_header();
+      unsafe { ((*header).scanned.min(self.unconsumed_len), (*header).scan_work) }
+    }
+  }
+
+  /// Returns deterministic counters for the managed-input implementation.
+  ///
+  /// This is intended for regression instrumentation. The tuple contains the
+  /// retained allocation capacity, the line-scan resume offset, cumulative
+  /// bytes copied, and cumulative bytes inspected while looking for line ends.
+  #[doc(hidden)]
+  pub fn managed_unconsumed_stats(&self) -> (usize, usize, usize, usize) {
+    if self.unconsumed_len == 0 {
+      (0, 0, 0, 0)
+    } else {
+      let header = self.retained_header();
+      unsafe {
+        (
+          (*header).capacity,
+          (*header).scanned.min(self.unconsumed_len),
+          (*header).copied,
+          (*header).scan_work,
+        )
+      }
+    }
+  }
+
   pub(crate) fn clear_unconsumed(&mut self) {
     if self.unconsumed_len > 0 {
-      // SAFETY: Retained input is allocated as a boxed slice with exactly this
-      // length, and the parser owns it until replacement, reset, or destruction.
+      let header = self.retained_header();
+      let capacity = unsafe { (*header).capacity };
       unsafe {
-        drop(Box::from_raw(ptr::slice_from_raw_parts_mut(
-          self.unconsumed as *mut c_uchar,
-          self.unconsumed_len,
-        )));
+        deallocate(header.cast::<u8>(), Self::retained_layout(capacity));
       }
     }
     self.unconsumed = ptr::null();

parser/src/matchers.rs

@@ -24,12 +24,36 @@ pub fn ensure_valid_line(data: &[u8], cr: usize, available: usize) -> MatchResul
 }
 
 #[inline(always)]
-pub fn find_cr(data: &[u8], available: usize) -> Option<usize> {
-  if available == 0 {
+pub fn find_cr(data: &[u8], available: usize, cursor: &mut usize, scan_work: &mut usize) -> Option<usize> {
+  let start = (*cursor).min(available);
+  let found = if start == available {
     None
   } else {
-    find_char(data, 0, available - 1, b'\r')
-  }
+    find_char(data, start, available - 1, b'\r')
+  };
+  let end = found.unwrap_or(available);
+  *scan_work += end - start + usize::from(found.is_some());
+  *cursor = end;
+  found
+}
+
+#[inline(always)]
+pub fn find_header_line_end_from(
+  ptr: *const u8,
+  len: usize,
+  cursor: &mut usize,
+  scan_work: &mut usize,
+) -> HeaderLineScanResult {
+  let start = (*cursor).min(len);
+  let result = unsafe { find_header_line_end(ptr.add(start), len - start) };
+  let (result, end, found) = match result {
+    HeaderLineScanResult::Cr(index) => (HeaderLineScanResult::Cr(start + index), start + index, true),
+    HeaderLineScanResult::Invalid(index) => (HeaderLineScanResult::Invalid(start + index), start + index, true),
+    HeaderLineScanResult::Incomplete => (HeaderLineScanResult::Incomplete, len, false),
+  };
+  *scan_work += end - start + usize::from(found);
+  *cursor = end;
+  result
 }
 
 #[inline(always)]

parser/src/parse.rs

@@ -2,8 +2,7 @@
 
 extern crate alloc;
 
-use alloc::vec::Vec;
-use alloc::{boxed::Box, format};
+use alloc::format;
 use core::cell::{Cell, RefCell};
 use core::ffi::{c_char, c_uchar, c_void};
 use core::fmt::Debug;
@@ -49,22 +48,20 @@ impl Parser {
 
     let input = unsafe { from_raw_parts(input, limit) };
 
-    // Set the data to analyze, prepending unconsumed data from previous iteration
-    // if needed
+    // Set the data to analyze, appending to the amortized retained allocation
+    // when a previous call ended in the middle of a line.
     let mut limit = limit;
-    let aggregate: Vec<c_uchar>;
-    let unconsumed_len = self.unconsumed_len;
-
-    let mut data = if self.manage_unconsumed && unconsumed_len > 0 {
-      unsafe {
-        limit += unconsumed_len;
-        let unconsumed = from_raw_parts(self.unconsumed, unconsumed_len);
+    let had_unconsumed = self.manage_unconsumed && self.unconsumed_len > 0;
+    let (mut line_scan_cursor, mut line_scan_work) = if had_unconsumed {
+      self.unconsumed_scan_state()
+    } else {
+      (0, 0)
+    };
 
-        // Rebuild a contiguous view so state handlers can scan across parse-call
-        // boundaries.
-        aggregate = [unconsumed, input].concat();
-        &aggregate[..]
-      }
+    let mut data = if had_unconsumed {
+      self.append_unconsumed(input);
+      limit = self.unconsumed_len;
+      unsafe { from_raw_parts(self.unconsumed, limit) }
     } else {
       input
     };
@@ -200,7 +197,7 @@ impl Parser {
           }
 
           STATE_REQUEST_LINE => {
-            match find_cr(data, available) {
+            match find_cr(data, available, &mut line_scan_cursor, &mut line_scan_work) {
               // // RFC 9112 section 3
               Some(cr) => {
                 match ensure_valid_line(data, cr, available) {
@@ -348,7 +345,7 @@ impl Parser {
 
           // RFC 9112 section 4
           STATE_STATUS_LINE => {
-            match find_cr(data, available) {
+            match find_cr(data, available, &mut line_scan_cursor, &mut line_scan_work) {
               Some(cr) => {
                 match ensure_valid_line(data, cr, available) {
                   MatchResult::Continue => {}
@@ -457,7 +454,7 @@ impl Parser {
           }
 
           STATE_HEADER => {
-            match find_header_line_end(data.as_ptr(), available) {
+            match find_header_line_end_from(data.as_ptr(), available, &mut line_scan_cursor, &mut line_scan_work) {
               HeaderLineScanResult::Cr(cr) => {
                 match ensure_valid_line(data, cr, available) {
                   MatchResult::Continue => {}
@@ -1009,7 +1006,7 @@ impl Parser {
 
           // RFC 9112 section 7.1
           STATE_CHUNK_HEADER => {
-            match find_cr(data, available) {
+            match find_cr(data, available, &mut line_scan_cursor, &mut line_scan_work) {
               Some(cr) => {
                 match ensure_valid_line(data, cr, available) {
                   MatchResult::Continue => {}
@@ -1091,7 +1088,7 @@ impl Parser {
           }
 
           STATE_CHUNK_EXTENSIONS => {
-            match find_cr(data, available) {
+            match find_cr(data, available, &mut line_scan_cursor, &mut line_scan_work) {
               Some(cr) => {
                 match ensure_valid_line(data, cr, available) {
                   MatchResult::Continue => {}
@@ -1291,7 +1288,7 @@ impl Parser {
 
           // RFC 9112 section 7.1.2
           STATE_TRAILER => {
-            match find_header_line_end(data.as_ptr(), available) {
+            match find_header_line_end_from(data.as_ptr(), available, &mut line_scan_cursor, &mut line_scan_work) {
               HeaderLineScanResult::Cr(cr) => {
                 match ensure_valid_line(data, cr, available) {
                   MatchResult::Continue => {}
@@ -1397,6 +1394,7 @@ impl Parser {
         self.position += advanced;
         data = &data[advanced..];
         available -= advanced;
+        line_scan_cursor = 0;
 
         #[cfg(any(debug_assertions, feature = "debug"))]
         if self.debug {
@@ -1451,14 +1449,10 @@ impl Parser {
     self.parsed += consumed as u64;
 
     if self.manage_unconsumed {
-      self.clear_unconsumed();
-
-      // Use a boxed slice so the length also determines the allocation layout
-      // when releasing retained input; Vec capacity is not stored in the ABI.
       if consumed < limit {
-        let retained = data.to_vec().into_boxed_slice();
-        self.unconsumed_len = retained.len();
-        self.unconsumed = Box::into_raw(retained) as *const c_uchar;
+        self.retain_unconsumed_suffix(data, line_scan_cursor, line_scan_work);
+      } else {
+        self.clear_unconsumed();
       }
     }
 

parser/src/wasm.rs

@@ -187,6 +187,18 @@ pub fn get_state(parser: *const c_void) -> u8 { unsafe { (*(parser as *const Par
 #[unsafe(no_mangle)]
 pub fn get_position(parser: *const c_void) -> usize { unsafe { (*(parser as *const Parser)).position } }
 
+// Get deterministic managed-input copy instrumentation.
+#[unsafe(no_mangle)]
+pub fn get_managed_unconsumed_copy_bytes(parser: *const c_void) -> usize {
+  unsafe { (*(parser as *const Parser)).managed_unconsumed_stats().2 }
+}
+
+// Get deterministic managed-line scan instrumentation.
+#[unsafe(no_mangle)]
+pub fn get_managed_unconsumed_scan_bytes(parser: *const c_void) -> usize {
+  unsafe { (*(parser as *const Parser)).managed_unconsumed_stats().3 }
+}
+
 // Get the parser parsed property.
 #[unsafe(no_mangle)]
 pub fn get_parsed(parser: *const c_void) -> u64 { unsafe { (*(parser as *const Parser)).parsed } }