[QDP] StreamingProducer: use VecDeque for O(1) buffer advance - #1462
[QDP] StreamingProducer: use VecDeque for O(1) buffer advance#14620lai0 wants to merge 4 commits into
Conversation
viiccwen
left a comment
viiccwen
left a comment
There was a problem hiding this comment.
Thx for the correctness and capacity tests. Since the main change is hot-path performance, could you also provide a small reproducible before/after benchmark?
It would be helpful to compare the current Vec + cursor implementation against this VecDeque implementation.
|
LGTM. |
|
Thanks @viiccwen and @rich7420 Buffer-only (refill chunk 65536), Xeon w3-2435,
E2E via Parquet: ~1.01–1.02x (decode dominates). Win only when batch ≪ chunk; at 1:1+ both do the same work. |
| // 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]; |
There was a problem hiding this 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.
| 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, | ||
| }; |
There was a problem hiding this 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?
| // 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())); |
There was a problem hiding this 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?
3 participants
Related Issues
Closes #1436
Changes
Why
StreamingProducertracked consumed elements with abuffer_cursorand, once the cursor passed the halfway mark (BUFFER_COMPACT_DENOM), reclaimed the prefix withVec::drain(..cursor)— an O(n) memmove of the retained tail, on the streaming hot path.A
VecDequeadvances its head instead, so discarding a consumed prefix never shifts the data that is still live: O(1) amortized buffer advance, no periodic compaction pass, and no change in output.How
buffer: Vec<T>+buffer_cursor: usize→buffer: VecDeque<T>; removed the now-unusedBUFFER_COMPACT_DENOMcompaction heuristicproduce()copies the batch out ofas_slices()(stitching both halves when a batch straddles the ring's wrap boundary), thendrain(..take)s the consumed prefix. Copying via the slices keeps the batch copy onextend_from_slice's bulk path —Drainis notTrustedLen, soextend(drain)would copy element by element. Recycled batch buffers are still reused, so the batch copy itself stays allocation-freeextend(&scratch[..written])rather than.iter().copied():VecDequespecializesExtend<&T> for T: Copyinto a bulkcopy_slicebatch_size * sample_size + initial_cap, the peak live length) so the ring never reallocates mid-run.VecDeque::from(Vec)reuses the existing allocation, so this only tops the capacity up.Checklist