viiccwen commented on code in PR #1462:
URL: https://github.com/apache/mahout/pull/1462#discussion_r3790952946
##########
qdp/qdp-core/src/pipeline_runner.rs:
##########
@@ -1796,6 +1812,117 @@ mod tests {
);
}
+ /// #1436: the `VecDeque` ring buffer must not grow across a long
stream. A deliberately
+ /// small `read_chunk_scratch` forces `produce()` to refill the ring
on nearly every batch,
+ /// so the test exercises the real hot-path pattern — front-drain the
consumed prefix while
+ /// new chunks are appended to the back. A correct ring reuses the
freed front slots for the
+ /// appended tail, so capacity settles to a constant: front-advance is
O(1) amortized with no
+ /// reallocation and no per-batch shift of the retained tail. The
`refills_observed` assert
+ /// guards against a regression where the whole stream is slurped up
front (which would make
+ /// the capacity check vacuous).
+ ///
+ /// The same loop also pins down the wrap-around batch copy: once the
ring wraps, a batch
+ /// can straddle the physical end of the allocation, so `produce()`
must stitch it together
+ /// from both halves of `as_slices()`. Reaching that case needs a read
chunk that is *not*
+ /// a multiple of the batch stride: if every batch drained the ring
empty, `VecDeque` would
+ /// reset the head to 0 on each refill and the ring would never wrap
(`split_observed`
+ /// asserts the straddling case is actually reached).
+ #[test]
+ fn test_streaming_producer_buffer_capacity_constant_over_100_batches()
{
+ const BATCH_SIZE: usize = 4;
+ const SAMPLE_LEN: usize = 4; // 2 qubits, amplitude encoding
+ const BATCHES: usize = 100;
+ // Elements consumed per batch.
+ const STRIDE: usize = BATCH_SIZE * SAMPLE_LEN;
+ // One sample more than a batch: produce() refills on nearly every
batch and always
+ // leaves a remainder behind, so the head walks around the ring
instead of resetting.
+ const READ_CHUNK: usize = STRIDE + SAMPLE_LEN;
+ // Each sample written by write_f32_parquet_n, repeated across the
batch.
+ const SAMPLE: [f32; SAMPLE_LEN] = [0.25, 0.5, 0.75, 1.0];
Review Comment:
Every sample in this fixture has the same values, the test cannot detect
samples being reordered, duplicated, or skipped across the wrap boundary.
Reversing the `front` and `back` slices still passes this test.
##########
qdp/qdp-core/src/pipeline_runner.rs:
##########
@@ -1796,6 +1812,117 @@ mod tests {
);
}
+ /// #1436: the `VecDeque` ring buffer must not grow across a long
stream. A deliberately
+ /// small `read_chunk_scratch` forces `produce()` to refill the ring
on nearly every batch,
+ /// so the test exercises the real hot-path pattern — front-drain the
consumed prefix while
+ /// new chunks are appended to the back. A correct ring reuses the
freed front slots for the
+ /// appended tail, so capacity settles to a constant: front-advance is
O(1) amortized with no
+ /// reallocation and no per-batch shift of the retained tail. The
`refills_observed` assert
+ /// guards against a regression where the whole stream is slurped up
front (which would make
+ /// the capacity check vacuous).
+ ///
+ /// The same loop also pins down the wrap-around batch copy: once the
ring wraps, a batch
+ /// can straddle the physical end of the allocation, so `produce()`
must stitch it together
+ /// from both halves of `as_slices()`. Reaching that case needs a read
chunk that is *not*
+ /// a multiple of the batch stride: if every batch drained the ring
empty, `VecDeque` would
+ /// reset the head to 0 on each refill and the ring would never wrap
(`split_observed`
+ /// asserts the straddling case is actually reached).
+ #[test]
+ fn test_streaming_producer_buffer_capacity_constant_over_100_batches()
{
+ const BATCH_SIZE: usize = 4;
+ const SAMPLE_LEN: usize = 4; // 2 qubits, amplitude encoding
+ const BATCHES: usize = 100;
+ // Elements consumed per batch.
+ const STRIDE: usize = BATCH_SIZE * SAMPLE_LEN;
+ // One sample more than a batch: produce() refills on nearly every
batch and always
+ // leaves a remainder behind, so the head walks around the ring
instead of resetting.
+ const READ_CHUNK: usize = STRIDE + SAMPLE_LEN;
+ // Each sample written by write_f32_parquet_n, repeated across the
batch.
+ const SAMPLE: [f32; SAMPLE_LEN] = [0.25, 0.5, 0.75, 1.0];
+ let path = temp_parquet_path("streaming_cap");
+ // A few extra samples so the 100th produce() is a full batch, not
EOF.
+ write_f32_parquet_n(&path, BATCH_SIZE * (BATCHES + 5));
+
+ let run = (|| -> Result<()> {
+ let reader = ParquetStreamingReader::<f32>::new(
+ &path,
+ Some(DEFAULT_PARQUET_ROW_GROUP_SIZE),
+ NullHandling::FillZero,
+ )?;
+ // Exactly the steady-state peak the production build path
reserves: one batch
+ // plus one refill chunk. Sized this tightly, the head wraps
within the first few
+ // batches, so straddling batch copies are exercised early and
often.
+ let mut producer = StreamingProducer::<f32> {
+ reader,
+ buffer: VecDeque::with_capacity(STRIDE + READ_CHUNK),
+ read_chunk_scratch: vec![0.0_f32; READ_CHUNK],
+ sample_size: SAMPLE_LEN,
+ batch_size: BATCH_SIZE,
+ num_qubits: 2,
+ batches_yielded: 0,
+ batch_limit: usize::MAX,
+ };
Review Comment:
IMO, This test constructs the deque with the expected steady-state capacity
directly, so it does not verify the `reserve` logic in
`build_streaming_producer`, right? The production reservation could be removed
or broken while this test still passes.
Could the test create the producer through the prod initialization path and
then verify that capacity remains stable for 100 batches?
##########
qdp/qdp-core/src/pipeline_runner.rs:
##########
@@ -631,10 +636,17 @@ where
buffer.truncate(written);
let read_chunk_scratch = vec![T::default(); initial_cap];
+ // Pre-reserve the streaming steady state so the ring never reallocates
mid-run. Each batch
+ // drains up to `required` elements from the front while a refill appends
up to one
+ // `initial_cap` chunk to the back, so live length peaks at `required +
initial_cap`.
+ // `VecDeque::from` reuses the Vec's existing `initial_cap` capacity
(O(1), no realloc), so
+ // this only tops it up by `required` — making front-advance realloc-free
from batch 0.
+ let required = config.batch_size * sample_size;
+ let mut buffer = VecDeque::from(buffer);
+ buffer.reserve((required + initial_cap).saturating_sub(buffer.len()));
Review Comment:
This eagerly reserves space for a full batch plus one refill chunk, even
when the file only contains a partial batch. For large samples, this can
substantially increase memory usage; for example, 20-qubit amplitude data with
`batch_size=64` and `f64` grows from roughly 8 MiB to 520 MiB.
Could we avoid reserving the theoretical peak before EOF is known, and use
checked arithmetic for the capacity calculation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
