hubcio commented on code in PR #4025:
URL: https://github.com/apache/iggy/pull/4025#discussion_r3906803110
##########
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;
Review Comment:
fixed in 4a1344c3c. the resident walk now runs the same helper per prepare,
capped at 64 KiB per copy since it sits inline on the pump. doc carries your
bound wording.
##########
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:
fixed in 065b99cc7, `Owned::extend_from_slice` into an exact-capacity
buffer, one pass.
##########
core/message_bus/src/transports/tcp.rs:
##########
@@ -289,6 +294,42 @@ async fn writer_loop(
}
}
+/// `write_vectored_all` over `iovecs`, at most `IOV_MAX` entries per call so
+/// a frame fragmented past the syscall limit (a wide poll reply) cannot fail
+/// with `EMSGSIZE` (the socket path is `io_uring` `sendmsg(2)`). Leaves
+/// `iovecs` empty with an allocation ready for the next batch.
+#[allow(clippy::future_not_send)]
+async fn write_iovecs(
+ write_half: &mut TcpStream,
+ iovecs: &mut Vec<Frozen<MESSAGE_ALIGN>>,
+) -> io::Result<()> {
+ if iovecs.len() <= IOV_MAX {
+ let compio::BufResult(result, mut returned) =
+ write_half.write_vectored_all(mem::take(iovecs)).await;
+ result?;
+ returned.clear();
+ *iovecs = returned;
+ return Ok(());
+ }
+ // Chunk through a cursor and one reused scratch vec. `split_off`-style
+ // chunking memmoves the whole tail per syscall, O(n^2) over a batch of n
+ // iovecs on the shard's single-threaded executor, and n is client-chosen
+ // (an unclamped poll count fans out to 1-2 iovecs per batch).
+ let mut source = mem::take(iovecs).into_iter();
+ let mut scratch = Vec::with_capacity(IOV_MAX);
+ loop {
+ scratch.extend(source.by_ref().take(IOV_MAX));
+ if scratch.is_empty() {
+ *iovecs = scratch;
+ return Ok(());
+ }
+ let compio::BufResult(result, mut returned) =
write_half.write_vectored_all(scratch).await;
+ result?;
+ returned.clear();
+ scratch = returned;
+ }
+}
Review Comment:
fixed in 78097865e. `scratch` lives in `writer_loop` now and the chunked
branch drains the caller's vec in place, so both keep their high-water marks.
boundary test covers `IOV_MAX`, `IOV_MAX + 1` and `2 * IOV_MAX`.
--
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]
