hubcio commented on code in PR #4025:
URL: https://github.com/apache/iggy/pull/4025#discussion_r3906320596
##########
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 95f164b0b. 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.
##########
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 95f164b0b. 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`.
--
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]
