numinnex commented on code in PR #4025:
URL: https://github.com/apache/iggy/pull/4025#discussion_r3906001386
##########
core/message_bus/src/transports/tcp.rs:
##########
@@ -289,6 +294,33 @@ 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 `EINVAL`. Leaves `iovecs` empty, keeping its allocation 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<()> {
+ let mut chunk = mem::take(iovecs);
+ loop {
+ let rest = if chunk.len() > IOV_MAX {
+ chunk.split_off(IOV_MAX)
+ } else {
+ Vec::new()
+ };
+ let compio::BufResult(result, mut returned) =
write_half.write_vectored_all(chunk).await;
+ result?;
+ if rest.is_empty() {
+ returned.clear();
+ *iovecs = returned;
+ return Ok(());
+ }
+ chunk = rest;
+ }
Review Comment:
`Vec::split_off(at)` allocates a new `Vec` **and** moves the `len - at` tail
elements, so chunking this way costs ~`n^2/2048` element moves for an `n`-iovec
batch rather than `n`.
`n` is client-chosen: `wire.count` reaches `MessageLookup::count` unclamped
(`dispatch.rs:2519`) and `push_selected_batch_fragments` emits 1-2 fragments
per batch. 50k fragments is ~29 MB of memmove; 262k (64 MiB of one-message
batches) is ~805 MB, roughly 80 ms. There is no `.await` in the loop and
`writer_loop` runs on the shard's single-threaded compio executor, so that
blocks every other task on the core, consensus I/O included.
At that shape the new path also moves more bytes than the gather it replaces
(~805 MB vs ~192 MB), which inverts the premise of the change.
Note `drain(..k)` from the front has the same problem (it shifts the tail
too). `into_iter()` advances a cursor instead, which is O(n) overall:
```rust
// `scratch` is reused across chunks and across batches.
let mut it = mem::take(iovecs).into_iter();
loop {
scratch.extend(it.by_ref().take(IOV_MAX));
if scratch.is_empty() {
return Ok(());
}
let compio::BufResult(result, mut returned) =
write_half.write_vectored_all(mem::take(scratch)).await;
returned.clear();
*scratch = returned;
result?;
}
```
Two smaller things in the same function:
- The doc comment says "Leaves `iovecs` empty, keeping its allocation for
the next batch", but the original allocation is dropped at the `chunk = rest`
rebind whenever chunking fires -- what gets restored is the last `split_off`
remainder, i.e. the reuse is lost on exactly the wide-poll path this was
written for.
- "the syscall fails with `EINVAL`" is the wrong errno here.
`write_vectored_all` on a `TcpStream` goes `Socket::send_vectored` ->
`SendVectored` -> io_uring `opcode::SendMsg`
(`compio-driver/src/sys/op/socket/iour.rs:167`), i.e. `sendmsg(2)`, and
`__copy_msghdr` returns **EMSGSIZE** for `msg_iovlen > UIO_MAXIOV`. `EINVAL` is
correct only for the constant's other consumer, `messages_writer.rs:223`, which
goes through `opcode::Writev` / `pwritev(2)`. The same wording is in the new
`iobuf.rs` doc; `config.rs:110`'s existing `EMSGSIZE` was right.
--
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]
