hubcio commented on code in PR #4025:
URL: https://github.com/apache/iggy/pull/4025#discussion_r3906321435
##########
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:
fixed in
https://github.com/apache/iggy/commit/7df4ee4b45e64567c713256e61a306560166e4ab.
chunking now drains an `into_iter` cursor through one reused scratch vec, and
the docs say EMSGSIZE for the socket path - EINVAL stays only on the pwritev
side in `messages_writer`.
##########
core/server/src/responses.rs:
##########
@@ -1652,8 +1662,148 @@ where
.map_or(0, VsrConsensus::commit_max)
}
+/// Body head of a `PolledMessages` reply:
+/// `[partition_id:4][current_offset:8][count:4]`, before the batch records.
+const POLLED_HEAD_LEN: usize = 16;
+
+/// Build the `PolledMessages` reply for the wire as a vectored frame: one
+/// buffer holding the reply header and the body head, then the poll
+/// fragments as they are. The record bytes are never copied or gathered;
+/// their reply encoding IS the storage encoding (see
+/// [`build_polled_messages_body`]), so `count` comes from walking the batch
+/// headers in place. At-rest decryption is the one case that must rewrite
+/// records, and it takes the flattening builder instead.
+pub fn build_polled_messages_reply(
+ request_header: &RoutedRequestHeader,
+ commit: u64,
+ partition_id: u32,
+ current_offset: u64,
+ fragments: PollFragments,
+ encryptor: Option<&EncryptorKind>,
+) -> Result<BusMessage, IggyError> {
+ let client_id = request_header.client;
+ let session = request_header.session;
+ if encryptor.is_some() {
+ let body = build_polled_messages_body(partition_id, current_offset,
fragments, encryptor)?;
+ let reply = build_reply_from_bytes(request_header, client_id, session,
commit, &body);
+ return Ok(reply.into_generic().into_frozen().into());
+ }
+
+ let mut frames = ResponseFragments::with_capacity(fragments.len() + 1);
+ frames.extend(fragments.into_iter().map(Fragment::into_frozen));
+ let count = polled_message_count(&frames)?;
+ let records_len: usize = frames.iter().map(Frozen::len).sum();
Review Comment:
good catch, fixed in
https://github.com/apache/iggy/commit/7df4ee4b45e64567c713256e61a306560166e4ab.
matched slices copy out into one compact buffer when they cover under a quarter
of the chunk, so a sparse poll cannot pin the 1 MiB read. dense walks stay
zero-copy.
--
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]
