numinnex commented on code in PR #4025:
URL: https://github.com/apache/iggy/pull/4025#discussion_r3906537316
##########
core/partitions/src/poll_plan.rs:
##########
@@ -1032,6 +1034,54 @@ struct ChunkWalk {
corrupt: bool,
}
+/// A fragment sliced from a disk chunk keeps the WHOLE chunk allocation
+/// alive until the reply frame is written out, and a reply can sit in a
+/// per-connection mailbox for a while. Copy the matched bytes out when they
+/// cover less than this fraction of the chunk, so a sparse match (a
+/// `count=1` poll off a cold partition) cannot pin ~1 MiB per queued reply;
+/// a dense catch-up read keeps the zero-copy path, where retained bytes stay
+/// within this factor of shipped bytes.
+const SPARSE_CHUNK_PIN_DIVISOR: usize = 4;
+
+/// Rewrite the fragments `walk_disk_chunk` pushed (those from index
+/// `walked_from` on) to slices of one compact copy when their combined
+/// length is a sparse fraction of `chunk`. See [`SPARSE_CHUNK_PIN_DIVISOR`].
+fn unpin_sparse_chunk(
+ fragments: &mut PollFragments<4096>,
+ walked_from: usize,
+ chunk: &Frozen<4096>,
+) {
+ let pushed = &mut fragments[walked_from..];
+ // Rewritten-header fragments in the walk own their bytes; only slices of
+ // `chunk` pin it.
+ let borrowed: usize = pushed
+ .iter()
+ .filter(|fragment| fragment.borrows_from(chunk))
+ .map(Fragment::len)
+ .sum();
+ if borrowed == 0 || borrowed >= chunk.len() / SPARSE_CHUNK_PIN_DIVISOR {
+ return;
+ }
+
+ let mut compact = Owned::<4096>::zeroed(borrowed);
+ let mut cursor = 0;
+ for fragment in pushed.iter().filter(|f| f.borrows_from(chunk)) {
+ let bytes = fragment.as_slice();
+ compact.as_mut_slice()[cursor..cursor +
bytes.len()].copy_from_slice(bytes);
+ cursor += bytes.len();
+ }
Review Comment:
`zeroed(borrowed)` memsets every byte the loop below then overwrites — the
borrowing fragments' lengths sum to exactly `borrowed` by construction, so
nothing is left at zero. That doubles the compaction's memory traffic (2x
`borrowed` instead of 1x, up to ~256 KiB of dead memset per 1 MiB chunk),
inside the function added to bound the copy in the first place.
This file already makes the argument, 260 lines up at
`read_chunk_with_retry`:
> `with_capacity` (len == 0, capacity == len) instead of `zeroed`:
`read_exact_at` fills the whole capacity in place and advances the length via
`SetLen`, so the `zeroed` memset of up to 1MiB per chunk was pure waste - every
byte is overwritten by the read.
`Owned::with_capacity` alone will not work here (it gives `len == 0`, so
`as_mut_slice()` is empty). The gather idiom this PR already uses is the fix —
`ResponseBacking::into_contiguous`,
core/server_common/src/consensus_message.rs:122-135:
```rust
let mut compact: AVec<u8, ConstAlign<4096>> = AVec::with_capacity(4096,
borrowed);
for fragment in pushed.iter().filter(|f| f.borrows_from(chunk)) {
compact.extend_from_slice(fragment.as_slice());
}
let compact = Frozen::from(Owned::from(compact));
```
That also drops the `cursor` bookkeeping in the first loop. `From<AVec> for
Owned` is at iobuf.rs:31.
--
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]
