numinnex commented on code in PR #4025:
URL: https://github.com/apache/iggy/pull/4025#discussion_r3906539373
##########
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:
Chunking fix verified: O(n), terminates, FIFO-preserving, and the error path
leaves the caller sane. `into_iter` advances a cursor instead of shifting a
tail, so each element moves exactly once. Boundaries check out too — `len ==
IOV_MAX` correctly takes the fast path (`sendmsg` rejects only `> UIO_MAXIOV`),
and the exact-multiple case terminates on the empty `extend` before issuing a
zero-iovec write.
One residual: the two branches do not treat the caller's allocation the same
way. The fast path round-trips the caller's own Vec, but the chunked path moves
it into `into_iter()` and drops it, handing back a fresh
`Vec::with_capacity(IOV_MAX)`. `writer_loop` then re-grows `iovecs` from 1024
to n on every wide batch — geometric, so amortised O(n) and nowhere near the
quadratic this replaced, but it is ~1.2 MB of memcpy plus a handful of
malloc/free per batch at n=50k. So the doc's "an allocation ready for the next
batch" is true but capped at `IOV_MAX`, not at the high-water mark.
Hoisting `scratch` up into `writer_loop` next to `iovecs` fixes both: each
buffer then persists at its own high-water mark, the per-wide-batch
`Vec::with_capacity(IOV_MAX)` disappears, and the doc claim becomes true for
both branches without rewording.
Also worth a test while this is fresh — the rewritten loop has no boundary
coverage. `run_writes_frames_fragmented_past_iov_max` produces 1793 fragments,
which exercises the chunked branch but misses `len == IOV_MAX`, `IOV_MAX + 1`,
and `2 * IOV_MAX`. The exact multiple is the one worth pinning, since that is
where a wrong loop would emit a zero-iovec `sendmsg`.
--
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]
