From c47d27196ed70a1a0d36570b18d099c5ff54b9c4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 17:44:04 +0000 Subject: [PATCH 1/3] Fix memory leaks in streaming parsers and reduce thread pool overhead Streaming parsers (StreamingParser, GeneralStreamingParser, GeneralStreamingParserNewlines) had three memory leak patterns: 1. compact_buffer() used Vec::drain() which preserves peak allocation capacity even after removing most data. Added shrink_excess() to reclaim memory when capacity exceeds 4x length. 2. take_rows() used drain().collect() leaving complete_rows at peak capacity. Now shrinks after draining. 3. finalize() left the internal buffer allocated after extracting the final rows. Now releases buffer memory since parsing is complete. Also: - Reduce rayon thread pool stack from 8 MiB to 2 MiB per thread (saves ~48 MiB virtual memory across 8 persistent threads) - Remove unnecessary field.clone() in parallel encoder's encoding path - Add ExactSizeIterator impls for RowIter, RowFieldIter, FieldIter All 95 tests pass. https://claude.ai/code/session_01QdJE1Gks1uipLWVupAwrbe --- native/rustycsv/Cargo.lock | 2 +- native/rustycsv/src/core/simd_index.rs | 6 ++++ native/rustycsv/src/lib.rs | 22 +++++++-------- native/rustycsv/src/strategy/general.rs | 21 +++++++++++--- native/rustycsv/src/strategy/parallel.rs | 4 +++ native/rustycsv/src/strategy/streaming.rs | 34 ++++++++++++++++++++--- 6 files changed, 68 insertions(+), 21 deletions(-) diff --git a/native/rustycsv/Cargo.lock b/native/rustycsv/Cargo.lock index c6a1eef..d89203e 100644 --- a/native/rustycsv/Cargo.lock +++ b/native/rustycsv/Cargo.lock @@ -180,7 +180,7 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" [[package]] name = "rustycsv" -version = "0.3.7" +version = "0.3.8" dependencies = [ "mimalloc", "rayon", diff --git a/native/rustycsv/src/core/simd_index.rs b/native/rustycsv/src/core/simd_index.rs index 805bf48..0dc1465 100644 --- a/native/rustycsv/src/core/simd_index.rs +++ b/native/rustycsv/src/core/simd_index.rs @@ -144,6 +144,8 @@ impl<'a> Iterator for RowIter<'a> { } } +impl ExactSizeIterator for RowIter<'_> {} + /// A single row from the cursor-based iterator, with its field bounds. pub struct Row<'a> { pub start: u32, @@ -210,6 +212,8 @@ impl<'a> Iterator for RowFieldIter<'a> { } } +impl ExactSizeIterator for RowFieldIter<'_> {} + /// Iterator over fields in a single row. pub struct FieldIter<'a> { seps: &'a [u32], @@ -254,6 +258,8 @@ impl<'a> Iterator for FieldIter<'a> { } } +impl ExactSizeIterator for FieldIter<'_> {} + #[cfg(test)] mod tests { use super::*; diff --git a/native/rustycsv/src/lib.rs b/native/rustycsv/src/lib.rs index 835d9a8..1a88108 100644 --- a/native/rustycsv/src/lib.rs +++ b/native/rustycsv/src/lib.rs @@ -1575,23 +1575,21 @@ fn encode_string_parallel<'a>( continue; } - let utf8_field: Vec = if needs_quoting { + if needs_quoting { let mut buf = Vec::with_capacity(field.len() + 8); write_quoted_field(&mut buf, field, esc); - buf + if needs_encoding { + let encoded = encode_utf8_to_target(&buf, encoding); + out.extend_from_slice(&encoded); + } else { + out.extend_from_slice(&buf); + } } else if needs_encoding { - field.clone() - } else { - // No formula, no quoting, no encoding — direct extend - out.extend_from_slice(field); - continue; - }; - - if needs_encoding { - let encoded = encode_utf8_to_target(&utf8_field, encoding); + // Encode directly — no clone needed + let encoded = encode_utf8_to_target(field, encoding); out.extend_from_slice(&encoded); } else { - out.extend_from_slice(&utf8_field); + out.extend_from_slice(field); } } out.extend_from_slice(&ls_encoded); diff --git a/native/rustycsv/src/strategy/general.rs b/native/rustycsv/src/strategy/general.rs index a93802b..43d01cd 100644 --- a/native/rustycsv/src/strategy/general.rs +++ b/native/rustycsv/src/strategy/general.rs @@ -10,6 +10,7 @@ use std::borrow::Cow; use crate::core::newlines::{match_newline, Newlines}; +use super::streaming::shrink_excess; // ============================================================================ // Helpers @@ -658,12 +659,15 @@ impl GeneralStreamingParser { self.buffer.drain(0..self.partial_row_start); self.scan_pos -= self.partial_row_start; self.partial_row_start = 0; + shrink_excess(&mut self.buffer); } } pub fn take_rows(&mut self, max: usize) -> Vec>> { let take_count = max.min(self.complete_rows.len()); - self.complete_rows.drain(0..take_count).collect() + let rows: Vec<_> = self.complete_rows.drain(0..take_count).collect(); + shrink_excess(&mut self.complete_rows); + rows } pub fn available_rows(&self) -> usize { @@ -684,8 +688,11 @@ impl GeneralStreamingParser { if !row.is_empty() { self.complete_rows.push(row); } - self.partial_row_start = self.buffer.len(); } + // Release the buffer — parsing is done + self.buffer = Vec::new(); + self.partial_row_start = 0; + self.scan_pos = 0; std::mem::take(&mut self.complete_rows) } } @@ -1317,12 +1324,15 @@ impl GeneralStreamingParserNewlines { self.buffer.drain(0..self.partial_row_start); self.scan_pos -= self.partial_row_start; self.partial_row_start = 0; + shrink_excess(&mut self.buffer); } } pub fn take_rows(&mut self, max: usize) -> Vec>> { let take_count = max.min(self.complete_rows.len()); - self.complete_rows.drain(0..take_count).collect() + let rows: Vec<_> = self.complete_rows.drain(0..take_count).collect(); + shrink_excess(&mut self.complete_rows); + rows } pub fn available_rows(&self) -> usize { @@ -1343,8 +1353,11 @@ impl GeneralStreamingParserNewlines { if !row.is_empty() { self.complete_rows.push(row); } - self.partial_row_start = self.buffer.len(); } + // Release the buffer — parsing is done + self.buffer = Vec::new(); + self.partial_row_start = 0; + self.scan_pos = 0; std::mem::take(&mut self.complete_rows) } } diff --git a/native/rustycsv/src/strategy/parallel.rs b/native/rustycsv/src/strategy/parallel.rs index 262774c..3dd6394 100644 --- a/native/rustycsv/src/strategy/parallel.rs +++ b/native/rustycsv/src/strategy/parallel.rs @@ -42,6 +42,10 @@ pub(crate) fn get_pool() -> Option<&'static rayon::ThreadPool> { rayon::ThreadPoolBuilder::new() .num_threads(recommended_threads()) .thread_name(|i| format!("rustycsv-{i}")) + // Reduce per-thread stack from the 8 MiB default to 2 MiB. + // CSV field extraction has shallow call stacks; the default + // wastes ~48 MiB of virtual memory across 8 persistent threads. + .stack_size(2 * 1024 * 1024) .build() .ok() }) diff --git a/native/rustycsv/src/strategy/streaming.rs b/native/rustycsv/src/strategy/streaming.rs index 90b5b2f..d5a82c4 100644 --- a/native/rustycsv/src/strategy/streaming.rs +++ b/native/rustycsv/src/strategy/streaming.rs @@ -13,6 +13,20 @@ use crate::core::{extract_field_owned_with_escape, is_separator}; /// Default maximum buffer size for streaming parsers (256 MB). pub const DEFAULT_MAX_BUFFER: usize = 256 * 1024 * 1024; +/// Shrink a Vec's capacity when it greatly exceeds its length. +/// +/// `Vec::drain` and `Vec::clear` preserve the original allocation. For +/// long-lived streaming parsers this causes memory to grow monotonically +/// to its peak usage and never return to the OS. This helper reclaims +/// that excess when capacity exceeds 4× length (with a 1 KiB floor to +/// avoid thrashing on small buffers). +pub(crate) fn shrink_excess(v: &mut Vec) { + let len = v.len(); + if v.capacity() > len.saturating_mul(4).max(1024) { + v.shrink_to(len.saturating_mul(2)); + } +} + /// Error returned when a streaming `feed()` would exceed the buffer limit. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct BufferOverflow; @@ -220,13 +234,20 @@ impl StreamingParser { // Adjust positions after compaction self.scan_pos -= self.partial_row_start; self.partial_row_start = 0; + // Release excess capacity to prevent unbounded memory growth. + // Vec::drain preserves the original allocation even after removing + // most data, causing the buffer to hold its peak capacity forever. + shrink_excess(&mut self.buffer); } } /// Take up to `max` complete rows from the parser pub fn take_rows(&mut self, max: usize) -> Vec>> { let take_count = max.min(self.complete_rows.len()); - self.complete_rows.drain(0..take_count).collect() + let rows: Vec<_> = self.complete_rows.drain(0..take_count).collect(); + // Prevent complete_rows from retaining peak capacity after draining + shrink_excess(&mut self.complete_rows); + rows } /// Check how many complete rows are available @@ -252,9 +273,13 @@ impl StreamingParser { if !row.is_empty() { self.complete_rows.push(row); } - self.partial_row_start = self.buffer.len(); } + // Release the buffer — parsing is done, no need to hold this memory + self.buffer = Vec::new(); + self.partial_row_start = 0; + self.scan_pos = 0; + // Take all remaining rows std::mem::take(&mut self.complete_rows) } @@ -262,8 +287,9 @@ impl StreamingParser { /// Reset the parser state #[allow(dead_code)] pub fn reset(&mut self) { - self.buffer.clear(); - self.complete_rows.clear(); + // Use = Vec::new() instead of .clear() to actually release memory + self.buffer = Vec::new(); + self.complete_rows = Vec::new(); self.partial_row_start = 0; self.scan_pos = 0; self.in_quotes = false; From 536f777982cbc5aa83eb21e40d791a02e58aef72 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 16 Feb 2026 17:51:28 +0000 Subject: [PATCH 2/3] Apply Rust best practices: thiserror, #[must_use], fmt/clippy clean - Add thiserror for BufferOverflow: implements Display + Error traits as required for idiomatic Rust library error types - Add #[must_use] to key types: StructuralIndex, RowEnd, Newlines, BufferOverflow, StreamingParser, GeneralStreamingParser, GeneralStreamingParserNewlines, GeneralFieldBound, StreamingParserResource - Add #[must_use] to getter methods: available_rows(), has_partial(), buffer_size(), row_count(), max_pattern_len() - Fix import ordering in general.rs to pass cargo fmt - All quality gates pass: cargo fmt, clippy -D warnings, 95 tests https://claude.ai/code/session_01QdJE1Gks1uipLWVupAwrbe --- native/rustycsv/Cargo.lock | 21 +++++++++++++++++++++ native/rustycsv/Cargo.toml | 1 + native/rustycsv/src/core/newlines.rs | 2 ++ native/rustycsv/src/core/simd_index.rs | 3 +++ native/rustycsv/src/resource.rs | 1 + native/rustycsv/src/strategy/general.rs | 11 ++++++++++- native/rustycsv/src/strategy/streaming.rs | 8 +++++++- 7 files changed, 45 insertions(+), 2 deletions(-) diff --git a/native/rustycsv/Cargo.lock b/native/rustycsv/Cargo.lock index d89203e..3e1f633 100644 --- a/native/rustycsv/Cargo.lock +++ b/native/rustycsv/Cargo.lock @@ -185,6 +185,7 @@ dependencies = [ "mimalloc", "rayon", "rustler", + "thiserror", ] [[package]] @@ -204,6 +205,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "unicode-ident" version = "1.0.22" diff --git a/native/rustycsv/Cargo.toml b/native/rustycsv/Cargo.toml index 13fe675..fd6555c 100644 --- a/native/rustycsv/Cargo.toml +++ b/native/rustycsv/Cargo.toml @@ -12,6 +12,7 @@ crate-type = ["cdylib", "lib"] [dependencies] rustler = { git = "https://github.com/rusterlium/rustler.git", branch = "master" } rayon = "1.10" # Data parallelism for parallel parsing +thiserror = "2" # Idiomatic error types for library code mimalloc = { version = "0.1", default-features = false, optional = true } [target.'cfg(target_env = "musl")'.dependencies] diff --git a/native/rustycsv/src/core/newlines.rs b/native/rustycsv/src/core/newlines.rs index 24ae3a6..b3b9b82 100644 --- a/native/rustycsv/src/core/newlines.rs +++ b/native/rustycsv/src/core/newlines.rs @@ -5,6 +5,7 @@ /// checks against the custom patterns. #[derive(Debug, Clone)] +#[must_use] pub struct Newlines { /// Newline patterns sorted longest-first for greedy matching. pub patterns: Vec>, @@ -32,6 +33,7 @@ impl Newlines { } /// Maximum pattern length (used for chunk-boundary safety in streaming). + #[must_use] pub fn max_pattern_len(&self) -> usize { self.patterns.iter().map(|p| p.len()).max().unwrap_or(1) } diff --git a/native/rustycsv/src/core/simd_index.rs b/native/rustycsv/src/core/simd_index.rs index 0dc1465..4edb913 100644 --- a/native/rustycsv/src/core/simd_index.rs +++ b/native/rustycsv/src/core/simd_index.rs @@ -5,6 +5,7 @@ /// A newline terminator position. #[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[must_use] pub struct RowEnd { /// Byte position of terminator start (\n or \r in \r\n). pub pos: u32, @@ -14,6 +15,7 @@ pub struct RowEnd { /// Structural index: positions of all unquoted separators and row endings. #[derive(Debug)] +#[must_use] pub struct StructuralIndex { /// Positions of unquoted field separators (commas, tabs, etc.). pub field_seps: Vec, @@ -58,6 +60,7 @@ impl StructuralIndex { /// Number of rows. #[inline] + #[must_use] pub fn row_count(&self) -> usize { let n = self.row_ends.len(); // If there's content after the last row_end (no trailing newline), there's one more row. diff --git a/native/rustycsv/src/resource.rs b/native/rustycsv/src/resource.rs index 6db53e9..83002a4 100644 --- a/native/rustycsv/src/resource.rs +++ b/native/rustycsv/src/resource.rs @@ -74,6 +74,7 @@ impl StreamingParserEnum { } /// Wrapper for StreamingParser that can be stored in a ResourceArc +#[must_use] pub struct StreamingParserResource { pub inner: Mutex, } diff --git a/native/rustycsv/src/strategy/general.rs b/native/rustycsv/src/strategy/general.rs index 43d01cd..770d6f7 100644 --- a/native/rustycsv/src/strategy/general.rs +++ b/native/rustycsv/src/strategy/general.rs @@ -9,8 +9,8 @@ use std::borrow::Cow; -use crate::core::newlines::{match_newline, Newlines}; use super::streaming::shrink_excess; +use crate::core::newlines::{match_newline, Newlines}; // ============================================================================ // Helpers @@ -188,6 +188,7 @@ fn parse_row_general<'a>( /// Field boundary for general parsing #[derive(Debug, Clone, Copy)] +#[must_use] pub struct GeneralFieldBound { pub start: usize, pub end: usize, @@ -547,6 +548,7 @@ fn parse_row_boundaries_general( // ============================================================================ /// Streaming parser that handles multi-byte separators and escapes +#[must_use] pub struct GeneralStreamingParser { buffer: Vec, complete_rows: Vec>>, @@ -670,14 +672,17 @@ impl GeneralStreamingParser { rows } + #[must_use] pub fn available_rows(&self) -> usize { self.complete_rows.len() } + #[must_use] pub fn has_partial(&self) -> bool { self.partial_row_start < self.buffer.len() } + #[must_use] pub fn buffer_size(&self) -> usize { self.buffer.len() } @@ -1206,6 +1211,7 @@ fn parse_row_boundaries_general_with_newlines( } /// Streaming parser with custom newline support. +#[must_use] pub struct GeneralStreamingParserNewlines { buffer: Vec, complete_rows: Vec>>, @@ -1335,14 +1341,17 @@ impl GeneralStreamingParserNewlines { rows } + #[must_use] pub fn available_rows(&self) -> usize { self.complete_rows.len() } + #[must_use] pub fn has_partial(&self) -> bool { self.partial_row_start < self.buffer.len() } + #[must_use] pub fn buffer_size(&self) -> usize { self.buffer.len() } diff --git a/native/rustycsv/src/strategy/streaming.rs b/native/rustycsv/src/strategy/streaming.rs index d5a82c4..7568ea5 100644 --- a/native/rustycsv/src/strategy/streaming.rs +++ b/native/rustycsv/src/strategy/streaming.rs @@ -28,10 +28,13 @@ pub(crate) fn shrink_excess(v: &mut Vec) { } /// Error returned when a streaming `feed()` would exceed the buffer limit. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +#[error("streaming buffer overflow: feed would exceed maximum buffer size")] +#[must_use] pub struct BufferOverflow; /// State for streaming CSV parser +#[must_use] pub struct StreamingParser { /// Buffer holding unprocessed data buffer: Vec, @@ -251,16 +254,19 @@ impl StreamingParser { } /// Check how many complete rows are available + #[must_use] pub fn available_rows(&self) -> usize { self.complete_rows.len() } /// Check if there's a partial row in the buffer + #[must_use] pub fn has_partial(&self) -> bool { self.partial_row_start < self.buffer.len() } /// Get the size of buffered data (for memory monitoring) + #[must_use] pub fn buffer_size(&self) -> usize { self.buffer.len() } From cc57dcee1c32b66746a1fdef5e296ac35444a748 Mon Sep 17 00:00:00 2001 From: jeffhuen <32542276+jeffhuen@users.noreply.github.com> Date: Mon, 16 Feb 2026 11:19:12 -0800 Subject: [PATCH 3/3] Fix ExactSizeIterator bugs, shrink_excess threshold, add tests Review fixes for PR #2: - Fix RowIter::next(): increment row_idx in trailing-row branch so ExactSizeIterator::len() returns 0 after exhaustion (was returning 1) - Fix RowFieldIter::next(): same trailing-row row_idx fix - Fix FieldIter::size_hint(): check done flag so len() returns 0 after last field is consumed (was returning 1) - Fix shrink_excess(): use byte-based 1 KiB floor via size_of::() instead of element-count 1024 (doc said bytes, code used elements) - Add ExactSizeIterator tests for RowIter, RowFieldIter, FieldIter covering trailing rows and multi-field exhaustion - Add shrink_excess tests: threshold, floor, ratio, large-element types - Add finalize/reset memory release tests --- native/rustycsv/src/core/simd_index.rs | 75 ++++++++++++++++++++ native/rustycsv/src/strategy/streaming.rs | 86 ++++++++++++++++++++++- 2 files changed, 158 insertions(+), 3 deletions(-) diff --git a/native/rustycsv/src/core/simd_index.rs b/native/rustycsv/src/core/simd_index.rs index 4edb913..b2ef8c9 100644 --- a/native/rustycsv/src/core/simd_index.rs +++ b/native/rustycsv/src/core/simd_index.rs @@ -133,6 +133,7 @@ impl<'a> Iterator for RowIter<'a> { let start = self.pos; let end = self.index.input_len; self.pos = end; + self.row_idx += 1; Some((start, end, end)) } else { None @@ -182,6 +183,7 @@ impl<'a> Iterator for RowFieldIter<'a> { let start = self.pos; let end = self.index.input_len; self.pos = end; + self.row_idx += 1; (start, end) } else { return None; @@ -256,6 +258,9 @@ impl<'a> Iterator for FieldIter<'a> { #[inline] fn size_hint(&self) -> (usize, Option) { + if self.done { + return (0, Some(0)); + } let remaining = (self.seps.len() + 1).saturating_sub(self.idx); (remaining, Some(remaining)) } @@ -382,4 +387,74 @@ mod tests { assert_eq!(cursor[0], vec![(0, 1), (2, 3)]); // a,b assert_eq!(cursor[1], vec![(4, 5)]); // c } + + // --- ExactSizeIterator correctness tests --- + + #[test] + fn row_iter_exact_size_with_trailing_row() { + // "a\nb" — 2 rows, second has no trailing newline + let idx = make_index(vec![], vec![RowEnd { pos: 1, len: 1 }], 3); + let mut iter = idx.rows(); + + assert_eq!(iter.len(), 2); + let _ = iter.next(); // consume row 1 + assert_eq!(iter.len(), 1); + let _ = iter.next(); // consume trailing row + assert_eq!(iter.len(), 0); + assert!(iter.next().is_none()); + } + + #[test] + fn row_iter_exact_size_no_trailing_row() { + // "a\n" — 1 row with trailing newline + let idx = make_index(vec![], vec![RowEnd { pos: 1, len: 1 }], 2); + let mut iter = idx.rows(); + + assert_eq!(iter.len(), 1); + let _ = iter.next(); + assert_eq!(iter.len(), 0); + assert!(iter.next().is_none()); + } + + #[test] + fn row_field_iter_exact_size_with_trailing_row() { + // "a\nb" — 2 rows, second has no trailing newline + let idx = make_index(vec![], vec![RowEnd { pos: 1, len: 1 }], 3); + let mut iter = idx.rows_with_fields(); + + assert_eq!(iter.len(), 2); + let _ = iter.next(); + assert_eq!(iter.len(), 1); + let _ = iter.next(); // trailing row + assert_eq!(iter.len(), 0); + assert!(iter.next().is_none()); + } + + #[test] + fn field_iter_exact_size_single_field() { + // Row "abc" — 1 field, no separators + let idx = make_index(vec![], vec![RowEnd { pos: 3, len: 1 }], 4); + let mut fields = idx.fields_in_row(0, 3); + + assert_eq!(fields.len(), 1); + let _ = fields.next(); // consume the only field + assert_eq!(fields.len(), 0); + assert!(fields.next().is_none()); + } + + #[test] + fn field_iter_exact_size_multiple_fields() { + // Row "a,b,c" — 3 fields, seps at 1 and 3 + let idx = make_index(vec![1, 3], vec![RowEnd { pos: 5, len: 1 }], 6); + let mut fields = idx.fields_in_row(0, 5); + + assert_eq!(fields.len(), 3); + let _ = fields.next(); // field "a" + assert_eq!(fields.len(), 2); + let _ = fields.next(); // field "b" + assert_eq!(fields.len(), 1); + let _ = fields.next(); // field "c" (last, sets done=true) + assert_eq!(fields.len(), 0); + assert!(fields.next().is_none()); + } } diff --git a/native/rustycsv/src/strategy/streaming.rs b/native/rustycsv/src/strategy/streaming.rs index 7568ea5..d860ba2 100644 --- a/native/rustycsv/src/strategy/streaming.rs +++ b/native/rustycsv/src/strategy/streaming.rs @@ -18,11 +18,12 @@ pub const DEFAULT_MAX_BUFFER: usize = 256 * 1024 * 1024; /// `Vec::drain` and `Vec::clear` preserve the original allocation. For /// long-lived streaming parsers this causes memory to grow monotonically /// to its peak usage and never return to the OS. This helper reclaims -/// that excess when capacity exceeds 4× length (with a 1 KiB floor to -/// avoid thrashing on small buffers). +/// that excess when capacity exceeds 4× length (with a 1 KiB floor, +/// measured in bytes, to avoid thrashing on small buffers). pub(crate) fn shrink_excess(v: &mut Vec) { let len = v.len(); - if v.capacity() > len.saturating_mul(4).max(1024) { + let floor_elements = 1024 / std::mem::size_of::().max(1); + if v.capacity() > len.saturating_mul(4).max(floor_elements) { v.shrink_to(len.saturating_mul(2)); } } @@ -424,4 +425,83 @@ mod tests { let rows = parser.take_rows(10); assert_eq!(rows[0], vec![b"a\rb".to_vec()]); } + + // --- Memory management tests --- + + #[test] + fn shrink_excess_reclaims_when_capacity_exceeds_threshold() { + let mut v: Vec = Vec::with_capacity(8192); + v.push(1); + // capacity=8192, len=1 → 8192 > 1*4.max(1024) → should shrink + shrink_excess(&mut v); + assert!( + v.capacity() < 8192, + "capacity should have been reduced from 8192, got {}", + v.capacity() + ); + assert_eq!(v.len(), 1); + assert_eq!(v[0], 1); + } + + #[test] + fn shrink_excess_does_not_shrink_below_floor() { + let mut v: Vec = Vec::with_capacity(1024); + // capacity=1024, len=0 → 1024 <= 0*4.max(1024) → should NOT shrink + shrink_excess(&mut v); + assert_eq!( + v.capacity(), + 1024, + "capacity at the floor should not be shrunk" + ); + } + + #[test] + fn shrink_excess_preserves_when_ratio_acceptable() { + let mut v: Vec = Vec::with_capacity(4000); + v.extend_from_slice(&[0u8; 2000]); + // capacity=4000, len=2000 → 4000 <= 2000*4.max(1024) → should NOT shrink + let cap_before = v.capacity(); + shrink_excess(&mut v); + assert_eq!(v.capacity(), cap_before); + } + + #[test] + fn shrink_excess_byte_floor_for_large_elements() { + // For Vec>>, size_of::() = 24 bytes on 64-bit + // floor_elements = 1024 / 24 = 42 + let mut v: Vec>> = Vec::with_capacity(200); + // capacity=200, len=0 → 200 > 0*4.max(42) → should shrink + shrink_excess(&mut v); + assert!( + v.capacity() < 200, + "should shrink large-element Vec past byte-based floor" + ); + } + + #[test] + fn finalize_releases_buffer() { + let mut parser = StreamingParser::new(); + parser.feed(b"a,b,c\n1,2,3").unwrap(); + assert!(parser.buffer_size() > 0); + + let rows = parser.finalize(); + assert_eq!(rows.len(), 2); + assert_eq!( + parser.buffer_size(), + 0, + "buffer should be released after finalize" + ); + } + + #[test] + fn reset_releases_memory() { + let mut parser = StreamingParser::new(); + parser.feed(b"a,b,c\n1,2,3\n").unwrap(); + let _ = parser.take_rows(10); + + parser.reset(); + assert_eq!(parser.buffer_size(), 0); + assert_eq!(parser.available_rows(), 0); + assert!(!parser.has_partial()); + } }