You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/01 18:33:08 UTC
(iggy) branch master updated: perf(server): serve poll replies without copying record bytes (#4025)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 831b2a652 perf(server): serve poll replies without copying record bytes (#4025)
831b2a652 is described below
commit 831b2a652ba71af89497aca9a08e2b8742d61e03
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Sep 1 20:33:03 2026 +0200
perf(server): serve poll replies without copying record bytes (#4025)
---
core/consensus/src/impls.rs | 9 +-
core/consensus/src/metadata_helpers.rs | 8 +-
core/consensus/src/plane_helpers.rs | 6 +-
core/message_bus/src/lib.rs | 26 +-
.../src/lifecycle/connection_registry.rs | 53 ++-
core/message_bus/src/transports/mod.rs | 18 +-
core/message_bus/src/transports/quic.rs | 39 +-
core/message_bus/src/transports/tcp.rs | 143 +++++++-
core/message_bus/src/transports/tcp_tls.rs | 18 +-
core/message_bus/src/transports/ws.rs | 14 +-
core/message_bus/src/transports/wss.rs | 20 +-
core/message_bus/tests/tcp_tls_client_listener.rs | 7 +-
core/message_bus/tests/tcp_tls_client_roundtrip.rs | 7 +-
core/message_bus/tests/wss_client_listener.rs | 7 +-
core/message_bus/tests/wss_client_roundtrip.rs | 7 +-
core/metadata/src/impls/metadata.rs | 8 +-
core/partitions/src/iggy_partition.rs | 8 +-
core/partitions/src/journal.rs | 243 +++++++++++-
core/partitions/src/messages_writer.rs | 6 +-
core/partitions/src/poll_plan.rs | 7 +-
core/partitions/src/types.rs | 22 ++
core/server/src/dispatch.rs | 72 ++--
core/server/src/http/reply.rs | 13 +-
core/server/src/http/submit.rs | 10 +-
core/server/src/partition_reconciler.rs | 2 +-
core/server/src/responses.rs | 408 ++++++++++++++++++++-
core/server_common/src/consensus_message.rs | 75 +++-
core/server_common/src/iobuf.rs | 27 ++
core/server_common/src/lib.rs | 1 +
core/shard/src/lib.rs | 9 +-
core/simulator/src/bus.rs | 9 +-
31 files changed, 1128 insertions(+), 174 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index edc60c086..3f732bb48 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -4377,6 +4377,7 @@ mod timestamp_clamp_tests {
use super::*;
use crate::LocalPipeline;
+ use message_bus::BusMessage;
use server_common::MESSAGE_ALIGN;
use server_common::iobuf::Frozen;
@@ -4398,7 +4399,7 @@ mod timestamp_clamp_tests {
async fn send_to_client(
&self,
_client_id: u128,
- _data: Frozen<MESSAGE_ALIGN>,
+ _data: impl Into<BusMessage>,
) -> Result<(), message_bus::SendError> {
Ok(())
}
@@ -4711,6 +4712,7 @@ mod timestamp_clamp_tests {
#[cfg(test)]
mod vsr_consensus_tests {
use super::*;
+ use message_bus::BusMessage;
#[test]
fn stage_transitions_follow_the_machine() {
@@ -4753,7 +4755,7 @@ mod vsr_consensus_tests {
async fn send_to_client(
&self,
_client_id: u128,
- _data: server_common::iobuf::Frozen<{ server_common::MESSAGE_ALIGN }>,
+ _data: impl Into<BusMessage>,
) -> Result<(), message_bus::SendError> {
Ok(())
}
@@ -5284,6 +5286,7 @@ mod quorum_tests {
use super::*;
use crate::LocalPipeline;
+ use message_bus::BusMessage;
use server_common::MESSAGE_ALIGN;
use server_common::iobuf::Frozen;
@@ -5293,7 +5296,7 @@ mod quorum_tests {
async fn send_to_client(
&self,
_client_id: u128,
- _data: Frozen<MESSAGE_ALIGN>,
+ _data: impl Into<BusMessage>,
) -> Result<(), message_bus::SendError> {
Ok(())
}
diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs
index 43af22a87..188375f4b 100644
--- a/core/consensus/src/metadata_helpers.rs
+++ b/core/consensus/src/metadata_helpers.rs
@@ -471,7 +471,7 @@ mod tests {
use crate::client_table::{REGISTER_REQUEST_ID, REPLY_RING_RETENTION_BYTES};
use crate::{CLIENTS_TABLE_MAX, LocalPipeline};
use iggy_binary_protocol::{Command, Operation, ReplyHeader};
- use message_bus::SendError;
+ use message_bus::{BusMessage, SendError};
/// Acting user for register fixtures; these tests exercise preflight /
/// replay, not user resolution, so the exact value is immaterial.
@@ -502,9 +502,11 @@ mod tests {
async fn send_to_client(
&self,
client_id: u128,
- data: Frozen<MESSAGE_ALIGN>,
+ data: impl Into<BusMessage>,
) -> Result<(), SendError> {
- self.client_sends.borrow_mut().push((client_id, data));
+ self.client_sends
+ .borrow_mut()
+ .push((client_id, data.into().into_contiguous()));
Ok(())
}
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index ff7df19cc..195120af8 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -877,7 +877,7 @@ mod tests {
use aligned_vec::{AVec, ConstAlign};
use iggy_binary_protocol::{ConsensusHeader, Operation, StartViewChangeHeader};
use iggy_common::calculate_checksum;
- use message_bus::SendError;
+ use message_bus::{BusMessage, SendError};
use server_common::{MESSAGE_ALIGN, iobuf::Frozen};
use std::collections::BTreeMap;
@@ -896,7 +896,7 @@ mod tests {
async fn send_to_client(
&self,
_client_id: u128,
- _data: Frozen<MESSAGE_ALIGN>,
+ _data: impl Into<BusMessage>,
) -> Result<(), SendError> {
Ok(())
}
@@ -1942,7 +1942,7 @@ mod tests {
async fn send_to_client(
&self,
_client_id: u128,
- _data: Frozen<MESSAGE_ALIGN>,
+ _data: impl Into<BusMessage>,
) -> Result<(), SendError> {
Ok(())
}
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index f41f50e30..74836bb8e 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -53,8 +53,9 @@
//! `Ready` on first poll. Consensus code relies on this for
//! reentrancy reasoning; any `.await` in the body breaks it.
//! - The TCP transport's writer task coalesces up to
-//! `MessageBusConfig::max_batch` (default 256) `Frozen<MESSAGE_ALIGN>`
-//! into one `write_vectored_all`. Don't introduce per-message
+//! `MessageBusConfig::max_batch` (default 256) [`BusMessage`] frames
+//! (their `Frozen<MESSAGE_ALIGN>` fragments flattened, chunked at
+//! `IOV_MAX`) into `write_vectored_all`. Don't introduce per-message
//! syscalls or per-message encryption on the plaintext TCP plane.
//! - fd-delegation ([`fd_transfer`]) is TCP-only. TLS / QUIC
//! connections have no dupable plaintext fd, so shard 0 terminates
@@ -279,7 +280,7 @@ pub type ReplicaForwardFn = Box<dyn Fn(u8, u16, Frozen<MESSAGE_ALIGN>) -> Result
/// shift. `client_id` is carried so the receiver can rebuild the
/// `ShardFrame::Lifecycle`'s `ForwardClientSend { client_id, msg }`
/// payload.
-pub type ClientForwardFn = Box<dyn Fn(u128, u16, Frozen<MESSAGE_ALIGN>) -> Result<(), SendError>>;
+pub type ClientForwardFn = Box<dyn Fn(u128, u16, BusMessage) -> Result<(), SendError>>;
/// Callback invoked when a client connection metadata entry is removed.
pub type ClientConnectionLostFn = std::rc::Rc<dyn Fn(u128)>;
@@ -496,10 +497,13 @@ pub type ConnectionLostFn = std::rc::Rc<dyn Fn(u8)>;
///
/// A bus impl must preserve this divergence - see each method.
pub trait MessageBus {
+ /// Queue one frame for `client_id`. Takes anything that converts into a
+ /// [`BusMessage`]: a single `Frozen<MESSAGE_ALIGN>` buffer or an
+ /// already fragmented frame.
fn send_to_client(
&self,
client_id: u128,
- data: Frozen<MESSAGE_ALIGN>,
+ data: impl Into<BusMessage>,
) -> impl Future<Output = Result<(), SendError>>;
fn send_to_replica(
@@ -1261,7 +1265,7 @@ impl<T: MessageBus + ?Sized> MessageBus for std::rc::Rc<T> {
fn send_to_client(
&self,
client_id: u128,
- data: Frozen<MESSAGE_ALIGN>,
+ data: impl Into<BusMessage>,
) -> impl Future<Output = Result<(), SendError>> {
(**self).send_to_client(client_id, data)
}
@@ -1313,11 +1317,12 @@ impl MessageBus for IggyMessageBus {
async fn send_to_client(
&self,
client_id: u128,
- message: Frozen<MESSAGE_ALIGN>,
+ message: impl Into<BusMessage>,
) -> Result<(), SendError> {
if self.is_shutting_down() {
return Err(SendError::BusShuttingDown);
}
+ let message = message.into();
// Owning shard is encoded in the top 16 bits of client_id.
let owning_shard = client_id_owning_shard(client_id);
if owning_shard == self.shard_id {
@@ -1357,9 +1362,9 @@ impl MessageBus for IggyMessageBus {
// Fast path: this shard owns a connection to the replica. On no-slot
// the registry returns the message unchanged so the slow path can
// forward it via the inter-shard channel without a wasted clone.
- let message = match self.replicas.try_send_or_return(replica, message) {
+ let message = match self.replicas.try_send_or_return(replica, message.into()) {
Ok(send_result) => return send_result.map_err(map_try_send_err),
- Err(message) => message,
+ Err(message) => message.into_contiguous(),
};
// Slow path: route via the inter-shard channel to the owning shard.
// The shard-shared `owner_table` is the authoritative routing
@@ -1438,7 +1443,7 @@ pub const fn is_auto_commit_client(client_id: u128) -> bool {
/// Shape-matches `Result::map_err` (takes the error by value) so it can be
/// used directly as a function reference rather than a closure.
#[allow(clippy::needless_pass_by_value)] // signature required by map_err
-fn map_try_send_err(e: async_channel::TrySendError<Frozen<MESSAGE_ALIGN>>) -> SendError {
+fn map_try_send_err(e: async_channel::TrySendError<BusMessage>) -> SendError {
match e {
async_channel::TrySendError::Full(_) => SendError::Backpressure,
async_channel::TrySendError::Closed(_) => SendError::ConnectionClosed,
@@ -1448,9 +1453,10 @@ fn map_try_send_err(e: async_channel::TrySendError<Frozen<MESSAGE_ALIGN>>) -> Se
/// Peek `ReplyHeader.request` (the originating request id) from a reply
/// buffer at its fixed header offset. Only the in-process reply path calls
/// this; the socket path never decodes.
-fn reply_request_id(reply: &Frozen<MESSAGE_ALIGN>) -> u64 {
+fn reply_request_id(reply: &BusMessage) -> u64 {
const OFFSET: usize = std::mem::offset_of!(ReplyHeader, request);
reply
+ .first()
.as_slice()
.get(OFFSET..OFFSET + std::mem::size_of::<u64>())
.and_then(|bytes| bytes.try_into().ok())
diff --git a/core/message_bus/src/lifecycle/connection_registry.rs b/core/message_bus/src/lifecycle/connection_registry.rs
index 0fa89a13b..24c709742 100644
--- a/core/message_bus/src/lifecycle/connection_registry.rs
+++ b/core/message_bus/src/lifecycle/connection_registry.rs
@@ -34,7 +34,7 @@
use compio::runtime::JoinHandle;
use futures::channel::oneshot;
-use server_common::{MESSAGE_ALIGN, iobuf::Frozen};
+use server_common::ResponseBacking;
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::collections::hash_map::Entry as HmEntry;
@@ -70,12 +70,14 @@ pub const READER_DRAIN_FLOOR: Duration = Duration::from_millis(250);
/// Payload type carried over every per-peer queue.
///
-/// Consensus messages are `Frozen<MESSAGE_ALIGN>` by the time they hit
-/// this queue: the dispatch layer freezes once and fan-out becomes a
-/// refcount bump per target. The writer task reads `Frozen` out of the
-/// queue and passes it straight to `write_vectored_all`, so no
-/// conversion happens on the hot path.
-pub type BusMessage = Frozen<MESSAGE_ALIGN>;
+/// One frame as a list of `Frozen<MESSAGE_ALIGN>` fragments written back
+/// to back. Consensus messages and most replies are a single fragment: the
+/// dispatch layer freezes once and fan-out becomes a refcount bump per
+/// target. A poll reply is a header fragment followed by the stored batch
+/// buffers as they are, so record bytes are never copied. The writer task
+/// hands the fragments straight to the socket write, so no conversion
+/// happens on the hot path.
+pub type BusMessage = ResponseBacking;
/// Producer side of a per-peer queue. Cloned out of the registry by
/// `send_to_*` and used with `try_send`.
@@ -960,6 +962,7 @@ mod tests {
h.size = HEADER_SIZE as u32;
})
.into_frozen()
+ .into()
}
fn spawn_dummy_writer(rx: BusReceiver) -> JoinHandle<()> {
@@ -1289,7 +1292,7 @@ mod tests {
// Receiver drains the message; closing the writer's rx end via
// close_peer is not necessary for this assertion.
let received = rx.recv().await.expect("queue had message");
- assert_eq!(received.len(), HEADER_SIZE);
+ assert_eq!(received.total_len(), HEADER_SIZE);
}
/// No slot for `key`: the message is handed back to the caller so the
@@ -1298,13 +1301,13 @@ mod tests {
async fn try_send_or_return_returns_msg_when_slot_missing() {
let reg: ConnectionRegistry<u8> = ConnectionRegistry::new();
let msg = make_bus_msg();
- let original_len = msg.len();
+ let original_len = msg.total_len();
let outcome = reg.try_send_or_return(99u8, msg);
let ReplyRoute::NoSlot(returned) = outcome else {
panic!("missing slot should return msg");
};
- assert_eq!(returned.len(), original_len);
+ assert_eq!(returned.total_len(), original_len);
}
/// Slot present but queue full: inner `Err(TrySendError::Full(msg))` so
@@ -1345,11 +1348,11 @@ mod tests {
async fn replica_try_send_or_return_returns_msg_when_slot_missing() {
let reg = ReplicaRegistry::new();
let msg = make_bus_msg();
- let original_len = msg.len();
+ let original_len = msg.total_len();
let outcome = reg.try_send_or_return(7u8, msg);
let returned = outcome.expect_err("missing slot should return msg");
- assert_eq!(returned.len(), original_len);
+ assert_eq!(returned.total_len(), original_len);
}
/// End-to-end registry seam: install entry + slot, route, fire, await.
@@ -1366,7 +1369,7 @@ mod tests {
reg.fire_in_process(1u8, 42, msg).expect("waiter present");
let received = reply_rx.await.expect("reply delivered");
- assert_eq!(received.len(), HEADER_SIZE);
+ assert_eq!(received.total_len(), HEADER_SIZE);
drop(guard);
assert!(reg.contains(1u8), "guard drop must not remove the entry");
}
@@ -1418,7 +1421,10 @@ mod tests {
reg.fire_in_process(1u8, 42, make_bus_msg())
.expect("first waiter still registered");
- assert_eq!(reply_rx.await.expect("reply delivered").len(), HEADER_SIZE);
+ assert_eq!(
+ reply_rx.await.expect("reply delivered").total_len(),
+ HEADER_SIZE
+ );
}
/// Reply for a request id nobody waits on: handed back, no panic, and
@@ -1430,15 +1436,18 @@ mod tests {
let (_guard, reply_rx) = reg.install_reply_slot(1u8, 42).expect("slot installs");
let msg = make_bus_msg();
- let original_len = msg.len();
+ let original_len = msg.total_len();
let returned = reg
.fire_in_process(1u8, 7, msg)
.expect_err("no waiter for request 7");
- assert_eq!(returned.len(), original_len);
+ assert_eq!(returned.total_len(), original_len);
reg.fire_in_process(1u8, 42, make_bus_msg())
.expect("slot 42 untouched");
- assert_eq!(reply_rx.await.expect("reply delivered").len(), HEADER_SIZE);
+ assert_eq!(
+ reply_rx.await.expect("reply delivered").total_len(),
+ HEADER_SIZE
+ );
}
/// Timeout / cancellation path: dropping the guard removes the slot,
@@ -1473,7 +1482,10 @@ mod tests {
.expect("waiter present");
drop(guard);
- assert_eq!(reply_rx.await.expect("reply delivered").len(), HEADER_SIZE);
+ assert_eq!(
+ reply_rx.await.expect("reply delivered").total_len(),
+ HEADER_SIZE
+ );
let (_guard, _reply_rx) = reg
.install_reply_slot(1u8, 42)
.expect("request id reusable after fire + drop");
@@ -1497,6 +1509,9 @@ mod tests {
drop(stale_guard);
reg.fire_in_process(1u8, 42, make_bus_msg())
.expect("stale guard must not evict the new slot");
- assert_eq!(reply_rx.await.expect("reply delivered").len(), HEADER_SIZE);
+ assert_eq!(
+ reply_rx.await.expect("reply delivered").total_len(),
+ HEADER_SIZE
+ );
}
}
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index 121afaa8b..7a703884e 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -39,20 +39,22 @@
//! - **Batch ordering**: each transport drains the per-peer
//! `BusReceiver` in FIFO order, but the dispatch shape differs:
//! * **TCP (vectored batch)** assembles up to `max_batch` frames
-//! into one `write_vectored_all` call. The kernel may short-write
-//! the iovec set (so `writev` is not atomic), but FIFO order
-//! across the batch is preserved and any short or failed write
-//! tears the connection down rather than retrying on a
+//! (every fragment of each [`BusMessage`](crate::BusMessage), chunked at `IOV_MAX`
+//! iovecs) into `write_vectored_all` calls. The kernel may
+//! short-write the iovec set (so `writev` is not atomic), but FIFO
+//! order across the batch is preserved and any short or failed
+//! write tears the connection down rather than retrying on a
//! half-written batch.
//! * **TCP-TLS, WS, WSS (drain-and-flush)** drain the same
//! `max_batch` window into a `Vec<BusMessage>` and then write
//! each frame through the per-record API (`AsyncWriteExt::write_all`
-//! for TLS, `WebSocketStream::send` for WS / WSS) followed by ONE
-//! trailing `flush()` per batch. This avoids per-frame TCP/TLS
+//! per fragment for TLS, `WebSocketStream::send` for WS / WSS, which
+//! joins a multi-fragment frame into one payload first) followed by
+//! ONE trailing `flush()` per batch. This avoids per-frame TCP/TLS
//! record overhead while staying inside the per-record API
//! constraints of those transports.
//! * **QUIC (per-frame)** uses one bidirectional stream per peer
-//! and writes each `BusMessage` with a separate
+//! and writes each `BusMessage` fragment with a separate
//! `SendStream::write_all` call; quinn coalesces at the datagram
//! layer.
//!
@@ -127,7 +129,7 @@ pub struct ActorContext {
/// message to the bus's per-plane handler (`MessageHandler` /
/// `RequestHandler`).
pub in_tx: Sender<Message<GenericHeader>>,
- /// Outbound channel: the bus's `send_to_*` path pushes `Frozen`
+ /// Outbound channel: the bus's `send_to_*` path pushes [`BusMessage`](crate::BusMessage)
/// frames into the matching `Sender`; the transport drains here.
pub rx: BusReceiver,
/// Cooperative cancellation. Fires on bus shutdown OR on
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
index 8968a18e4..da9296cb6 100644
--- a/core/message_bus/src/transports/quic.rs
+++ b/core/message_bus/src/transports/quic.rs
@@ -319,7 +319,7 @@ enum ReplyRoute {
/// Decide how to route an outbound frame. Only `Command::Reply` carries a
/// `request` id; everything else must pass through unfiltered.
fn reply_route(frame: &BusMessage) -> ReplyRoute {
- let bytes = frame.as_slice();
+ let bytes = frame.first().as_slice();
let is_reply = bytes
.get(COMMAND_OFFSET)
.is_some_and(|&command| command == Command::Reply as u8);
@@ -503,11 +503,19 @@ impl TransportConn for QuicTransportConn {
continue;
};
- // Write the reply, then `finish()` so quinn-proto flushes pending
- // data + a FIN, signalling the SDK that the reply is complete.
- let BufResult(result, _frozen) = send.write_all(first).await;
- if let Err(e) = result {
- debug!(%label, %peer, error = ?e, "quic: write failed");
+ // Write the reply fragment by fragment, then `finish()` so
+ // quinn-proto flushes pending data + a FIN, signalling the SDK
+ // that the reply is complete.
+ let mut write_failed = false;
+ for fragment in first.into_fragments() {
+ let BufResult(result, _frozen) = send.write_all(fragment).await;
+ if let Err(e) = result {
+ debug!(%label, %peer, error = ?e, "quic: write failed");
+ write_failed = true;
+ break;
+ }
+ }
+ if write_failed {
continue;
}
if let Err(e) = send.finish() {
@@ -611,12 +619,12 @@ mod tests {
fn drive(
conn: QuicTransportConn,
) -> (
- async_channel::Sender<Frozen<MESSAGE_ALIGN>>,
+ async_channel::Sender<BusMessage>,
async_channel::Receiver<Message<GenericHeader>>,
Shutdown,
compio::runtime::JoinHandle<()>,
) {
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(16);
+ let (out_tx, out_rx) = bounded::<BusMessage>(16);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(16);
let (shutdown, token) = Shutdown::new();
let ctx = ActorContext {
@@ -654,11 +662,20 @@ mod tests {
// and then awaits exactly one reply on `rx` before
// accepting the next bidi. Feed three replies in step.
let a = in_rx.recv().await.unwrap();
- out_tx.send(header_only(Command::Reply)).await.unwrap();
+ out_tx
+ .send(header_only(Command::Reply).into())
+ .await
+ .unwrap();
let b = in_rx.recv().await.unwrap();
- out_tx.send(header_only(Command::Reply)).await.unwrap();
+ out_tx
+ .send(header_only(Command::Reply).into())
+ .await
+ .unwrap();
let c = in_rx.recv().await.unwrap();
- out_tx.send(header_only(Command::Reply)).await.unwrap();
+ out_tx
+ .send(header_only(Command::Reply).into())
+ .await
+ .unwrap();
shutdown.trigger();
let _ = handle.await;
(a.header().command, b.header().command, c.header().command)
diff --git a/core/message_bus/src/transports/tcp.rs b/core/message_bus/src/transports/tcp.rs
index 87a310380..53845b41e 100644
--- a/core/message_bus/src/transports/tcp.rs
+++ b/core/message_bus/src/transports/tcp.rs
@@ -35,6 +35,8 @@ use compio::io::AsyncWriteExt;
use compio::net::{TcpListener, TcpStream};
use compio::runtime::fd::PollFd;
use futures::FutureExt;
+use server_common::MESSAGE_ALIGN;
+use server_common::iobuf::{Frozen, IOV_MAX};
use std::io;
use std::mem;
use std::net::SocketAddr;
@@ -242,6 +244,9 @@ async fn writer_loop(
) {
let _wake_reader = scopeguard::guard(conn_shutdown, |s| s.trigger());
let mut batch: Vec<BusMessage> = Vec::with_capacity(max_batch);
+ let mut iovecs: Vec<Frozen<MESSAGE_ALIGN>> = Vec::with_capacity(max_batch);
+ // Sized by the first wide batch; most connections never need it.
+ let mut scratch: Vec<Frozen<MESSAGE_ALIGN>> = Vec::new();
let mut shutdown_fut = Box::pin(shutdown.wait().fuse());
loop {
@@ -271,10 +276,12 @@ async fn writer_loop(
let drained = batch.len();
trace!(%label, %peer, batch = drained, "writev batch");
- let owned = mem::take(&mut batch);
- let compio::BufResult(result, mut returned) = write_half.write_vectored_all(owned).await;
- returned.clear();
- batch = returned;
+ // `drain(..)` keeps the batch allocation for the next round.
+ #[allow(clippy::iter_with_drain)]
+ for frame in batch.drain(..) {
+ iovecs.extend(frame.into_fragments());
+ }
+ let result = write_iovecs(&mut write_half, &mut iovecs, &mut scratch).await;
if let Err(e) = result {
error!(
@@ -289,6 +296,43 @@ 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)`). `scratch`
+/// carries each chunk in flight. Leaves both vecs empty with their
+/// allocations intact for the next batch.
+#[allow(clippy::future_not_send)]
+async fn write_iovecs(
+ write_half: &mut TcpStream,
+ iovecs: &mut Vec<Frozen<MESSAGE_ALIGN>>,
+ scratch: &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(());
+ }
+ // A full-range drain shifts nothing, so the batch moves into `scratch` in
+ // O(n). Draining `IOV_MAX` at a time from the front would memmove the tail
+ // per syscall, O(n^2) over a client-chosen n (an unclamped poll count fans
+ // out to 1-2 iovecs per batch) on the shard's single-threaded executor.
+ let mut source = iovecs.drain(..);
+ loop {
+ scratch.extend(source.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;
+ result?;
+ returned.clear();
+ *scratch = returned;
+ }
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -325,12 +369,12 @@ mod tests {
fn drive(
conn: TcpTransportConn,
) -> (
- async_channel::Sender<Frozen<MESSAGE_ALIGN>>,
+ async_channel::Sender<BusMessage>,
async_channel::Receiver<Message<GenericHeader>>,
Shutdown,
compio::runtime::JoinHandle<()>,
) {
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(16);
+ let (out_tx, out_rx) = bounded::<BusMessage>(16);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(16);
let (shutdown, token) = Shutdown::new();
let ctx = ActorContext {
@@ -370,7 +414,7 @@ mod tests {
drive(TcpTransportConn::new(server));
for cmd in [Command::Ping, Command::Prepare, Command::Request] {
- client_out.send(header_only(cmd)).await.unwrap();
+ client_out.send(header_only(cmd).into()).await.unwrap();
}
let recv_with_timeout = |rx: &async_channel::Receiver<Message<GenericHeader>>| {
@@ -395,6 +439,91 @@ mod tests {
let _ = server_handle.await;
}
+ /// A frame fragmented to `IOV_MAX` or past it must reach the peer whole
+ /// and in order: `sendmsg` rejects more than `IOV_MAX` iovecs with
+ /// `EMSGSIZE`, so the writer has to chunk the flattened batch instead of
+ /// failing it. Each frame goes out alone in its batch (the follow-up frame
+ /// is sent only once the peer has it), so the iovec count is exactly the
+ /// fragment count and the chunk boundaries land where the cases say.
+ #[compio::test]
+ #[allow(clippy::future_not_send, clippy::cast_possible_truncation)]
+ async fn run_writes_frames_fragmented_at_and_past_iov_max() {
+ let (client, server) = local_pair().await;
+ let (client_out, _client_in, client_shutdown, client_handle) =
+ drive(TcpTransportConn::new(client));
+ let (_server_out, server_in, server_shutdown, server_handle) =
+ drive(TcpTransportConn::new(server));
+ let recv_with_timeout = || async {
+ compio::time::timeout(Duration::from_secs(2), server_in.recv())
+ .await
+ .expect("recv within 2s")
+ .expect("ok")
+ };
+
+ // One connection for all cases so a chunked batch is followed by both
+ // a single-iovec one and a wider one. A full single chunk, one iovec
+ // over it, the original `2 * IOV_MAX`-byte repro, and an exact multiple
+ // of the chunk size.
+ for fragment_count in [
+ IOV_MAX,
+ IOV_MAX + 1,
+ 2 * IOV_MAX - HEADER_SIZE + 1,
+ 2 * IOV_MAX,
+ ] {
+ // Header whole in the first fragment (the frame contract), body as
+ // one byte per fragment.
+ let frame_len = HEADER_SIZE + fragment_count - 1;
+ let whole = Message::<GenericHeader>::new(frame_len)
+ .transmute_header(|_, h: &mut GenericHeader| {
+ h.command = Command::Ping;
+ h.size = frame_len as u32;
+ })
+ .into_frozen();
+ let mut fragments = server_common::ResponseFragments::with_capacity(fragment_count);
+ fragments.push(whole.slice(..HEADER_SIZE));
+ fragments.extend((HEADER_SIZE..frame_len).map(|at| whole.slice(at..=at)));
+ assert_eq!(fragments.len(), fragment_count);
+ let fragmented =
+ Message::<GenericHeader, server_common::ResponseBacking>::try_from(fragments)
+ .expect("fragments form a valid frame")
+ .into_inner();
+ client_out.send(fragmented).await.unwrap();
+
+ let ping = recv_with_timeout().await;
+ assert_eq!(
+ ping.header().command,
+ Command::Ping,
+ "{fragment_count} fragments"
+ );
+ assert_eq!(
+ ping.header().size as usize,
+ frame_len,
+ "{fragment_count} fragments"
+ );
+ assert_eq!(
+ ping.as_slice(),
+ whole.as_slice(),
+ "{fragment_count} fragments"
+ );
+
+ client_out
+ .send(header_only(Command::Request).into())
+ .await
+ .unwrap();
+ let request = recv_with_timeout().await;
+ assert_eq!(
+ request.header().command,
+ Command::Request,
+ "{fragment_count} fragments"
+ );
+ }
+
+ client_shutdown.trigger();
+ server_shutdown.trigger();
+ let _ = client_handle.await;
+ let _ = server_handle.await;
+ }
+
#[compio::test]
#[allow(clippy::future_not_send)]
async fn run_exits_on_shutdown_signal() {
diff --git a/core/message_bus/src/transports/tcp_tls.rs b/core/message_bus/src/transports/tcp_tls.rs
index a248fb85c..c3fa5ac21 100644
--- a/core/message_bus/src/transports/tcp_tls.rs
+++ b/core/message_bus/src/transports/tcp_tls.rs
@@ -573,9 +573,9 @@ async fn run_pump(tls: &mut TlsStream<TcpStream>, ctx: ActorContext) {
// preserving the Vec's allocation for the next
// iteration; `into_iter()` would move the buffer out.
#[allow(clippy::iter_with_drain)]
- for msg in batch.drain(..) {
- let len = msg.buf_len();
- let compio::BufResult(result, _) = tls.write_all(msg).await;
+ for fragment in batch.drain(..).flat_map(BusMessage::into_fragments) {
+ let len = fragment.buf_len();
+ let compio::BufResult(result, _) = tls.write_all(fragment).await;
if let Err(e) = result {
warn!(
%label,
@@ -742,12 +742,12 @@ mod tests {
fn drive(
conn: TcpTlsTransportConn,
) -> (
- Sender<Frozen<MESSAGE_ALIGN>>,
+ Sender<BusMessage>,
Receiver<Message<GenericHeader>>,
Shutdown,
compio::runtime::JoinHandle<()>,
) {
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(16);
+ let (out_tx, out_rx) = bounded::<BusMessage>(16);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(16);
let (shutdown, token) = Shutdown::new();
let ctx = ActorContext {
@@ -780,7 +780,7 @@ mod tests {
let (client_out, client_in, client_shutdown, client_handle) = drive(client_conn);
client_out
- .send(header_only(Command::Request))
+ .send(header_only(Command::Request).into())
.await
.expect("client send");
let received = compio::time::timeout(Duration::from_secs(5), server_in.recv())
@@ -790,7 +790,7 @@ mod tests {
assert_eq!(received.header().command, Command::Request);
server_out
- .send(header_only(Command::Reply))
+ .send(header_only(Command::Reply).into())
.await
.expect("server send");
let reply = compio::time::timeout(Duration::from_secs(5), client_in.recv())
@@ -854,7 +854,7 @@ mod tests {
let (client_out, _client_in, client_shutdown, client_handle) = drive(client_conn);
client_out
- .send(padded(Command::Request, total))
+ .send(padded(Command::Request, total).into())
.await
.expect("client send 1 MiB");
let received = compio::time::timeout(Duration::from_secs(10), server_in.recv())
@@ -924,7 +924,7 @@ mod tests {
let (client_out, _client_in, client_shutdown, client_handle) = drive(client_conn);
client_out
- .send(header_only(Command::Request))
+ .send(header_only(Command::Request).into())
.await
.expect("client send");
let received = compio::time::timeout(Duration::from_secs(5), server_in.recv())
diff --git a/core/message_bus/src/transports/ws.rs b/core/message_bus/src/transports/ws.rs
index f985191bb..15b5a963a 100644
--- a/core/message_bus/src/transports/ws.rs
+++ b/core/message_bus/src/transports/ws.rs
@@ -234,8 +234,12 @@ async fn run_pump(ws: &mut WebSocketStream<TcpStream>, ctx: ActorContext) {
}
let drained = batch.len();
#[allow(clippy::iter_with_drain)]
+ // One WS frame per message: tungstenite takes a single payload
+ // buffer, so a multi-fragment frame is joined here (the only
+ // record copy left on this plane; single-fragment frames move).
for msg in batch.drain(..) {
- if let Err(e) = ws.send(WsMessage::Binary(Bytes::from_owner(msg))).await {
+ let payload = Bytes::from_owner(msg.into_contiguous());
+ if let Err(e) = ws.send(WsMessage::Binary(payload)).await {
warn!(%label, %peer, error = ?e, batch_len = drained, "ws writer: send failed");
return;
}
@@ -364,7 +368,7 @@ mod tests {
fn drive(
conn: WsTransportConn,
) -> (
- Sender<Frozen<MESSAGE_ALIGN>>,
+ Sender<BusMessage>,
Receiver<Message<GenericHeader>>,
Shutdown,
compio::runtime::JoinHandle<()>,
@@ -377,12 +381,12 @@ mod tests {
conn: WsTransportConn,
max_message_size: usize,
) -> (
- Sender<Frozen<MESSAGE_ALIGN>>,
+ Sender<BusMessage>,
Receiver<Message<GenericHeader>>,
Shutdown,
compio::runtime::JoinHandle<()>,
) {
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(16);
+ let (out_tx, out_rx) = bounded::<BusMessage>(16);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(16);
let (shutdown, token) = Shutdown::new();
let ctx = ActorContext {
@@ -442,7 +446,7 @@ mod tests {
// Server replies via its outbound mailbox; the serial pump writes
// the reply on the same bidi the request arrived on, client reads.
server_out
- .send(header_only(Command::Reply))
+ .send(header_only(Command::Reply).into())
.await
.expect("server send");
let reply = compio::time::timeout(Duration::from_secs(5), raw_recv(&mut client_ws))
diff --git a/core/message_bus/src/transports/wss.rs b/core/message_bus/src/transports/wss.rs
index e30b6a21f..20168d008 100644
--- a/core/message_bus/src/transports/wss.rs
+++ b/core/message_bus/src/transports/wss.rs
@@ -381,8 +381,12 @@ async fn run_pump(ws: &mut WebSocketStream<TcpStream>, ctx: ActorContext) {
// preserving the Vec's allocation for the next
// iteration; `into_iter()` would move the buffer out.
#[allow(clippy::iter_with_drain)]
+ // One WS frame per message: tungstenite takes a single payload
+ // buffer, so a multi-fragment frame is joined here (the only
+ // record copy left on this plane; single-fragment frames move).
for msg in batch.drain(..) {
- if let Err(e) = ws.send(WsMessage::Binary(Bytes::from_owner(msg))).await {
+ let payload = Bytes::from_owner(msg.into_contiguous());
+ if let Err(e) = ws.send(WsMessage::Binary(payload)).await {
warn!(%label, %peer, error = ?e, batch_len = drained, "wss writer: send failed");
return;
}
@@ -554,7 +558,7 @@ mod tests {
fn drive(
conn: WssTransportConn,
) -> (
- Sender<Frozen<MESSAGE_ALIGN>>,
+ Sender<BusMessage>,
Receiver<Message<GenericHeader>>,
Shutdown,
compio::runtime::JoinHandle<()>,
@@ -567,12 +571,12 @@ mod tests {
conn: WssTransportConn,
max_message_size: usize,
) -> (
- Sender<Frozen<MESSAGE_ALIGN>>,
+ Sender<BusMessage>,
Receiver<Message<GenericHeader>>,
Shutdown,
compio::runtime::JoinHandle<()>,
) {
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(16);
+ let (out_tx, out_rx) = bounded::<BusMessage>(16);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(16);
let (shutdown, token) = Shutdown::new();
let ctx = ActorContext {
@@ -604,7 +608,7 @@ mod tests {
let (client_out, client_in, client_shutdown, client_handle) = drive(client_conn);
client_out
- .send(header_only(Command::Request))
+ .send(header_only(Command::Request).into())
.await
.expect("client send");
let received = compio::time::timeout(Duration::from_secs(5), server_in.recv())
@@ -614,7 +618,7 @@ mod tests {
assert_eq!(received.header().command, Command::Request);
server_out
- .send(header_only(Command::Reply))
+ .send(header_only(Command::Reply).into())
.await
.expect("server send");
let reply = compio::time::timeout(Duration::from_secs(5), client_in.recv())
@@ -647,7 +651,7 @@ mod tests {
let (client_out, _client_in, client_shutdown, client_handle) = drive(client_conn);
client_out
- .send(padded(Command::Request, total))
+ .send(padded(Command::Request, total).into())
.await
.expect("client send 1 MiB");
let received = compio::time::timeout(Duration::from_secs(15), server_in.recv())
@@ -683,7 +687,7 @@ mod tests {
drive_with_cap(client_conn, framing::MAX_MESSAGE_SIZE);
client_out
- .send(padded(Command::Request, OVER_CAP))
+ .send(padded(Command::Request, OVER_CAP).into())
.await
.expect("client send oversize");
diff --git a/core/message_bus/tests/tcp_tls_client_listener.rs b/core/message_bus/tests/tcp_tls_client_listener.rs
index 0ce3ed030..d477d1cd4 100644
--- a/core/message_bus/tests/tcp_tls_client_listener.rs
+++ b/core/message_bus/tests/tcp_tls_client_listener.rs
@@ -22,6 +22,7 @@ use common::{header_only, install_tls_clients_locally, loopback};
use compio::net::TcpStream;
use iggy_binary_protocol::Command;
use iggy_binary_protocol::GenericHeader;
+use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
use message_bus::client_listener::tcp_tls::{bind, run};
use message_bus::transports::tcp_tls::TcpTlsTransportConn;
@@ -30,7 +31,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, MessageBusConfig, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
-use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
+use server_common::Message;
use std::rc::Rc;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -83,7 +84,7 @@ async fn tcp_tls_client_listener_accepts_and_round_trips() {
let client_tcp = TcpStream::connect(server_addr).await.expect("client dial");
let conn = TcpTlsTransportConn::new_client(client_tcp, client_cfg, server_name);
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(8);
+ let (out_tx, out_rx) = bounded::<BusMessage>(8);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(8);
let (client_shutdown, client_token) = Shutdown::new();
let ctx = ActorContext {
@@ -99,7 +100,7 @@ async fn tcp_tls_client_listener_accepts_and_round_trips() {
let client_handle = compio::runtime::spawn(async move { conn.run(ctx).await });
let request = header_only(Command::Request, 42, 0).into_frozen();
- out_tx.send(request).await.expect("client send");
+ out_tx.send(request.into()).await.expect("client send");
let reply = compio::time::timeout(Duration::from_secs(5), in_rx.recv())
.await
diff --git a/core/message_bus/tests/tcp_tls_client_roundtrip.rs b/core/message_bus/tests/tcp_tls_client_roundtrip.rs
index f4d55f5c5..b3154f99c 100644
--- a/core/message_bus/tests/tcp_tls_client_roundtrip.rs
+++ b/core/message_bus/tests/tcp_tls_client_roundtrip.rs
@@ -25,6 +25,7 @@ use common::{
use compio::net::TcpStream;
use iggy_binary_protocol::Command;
use iggy_binary_protocol::GenericHeader;
+use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
use message_bus::replica::io::start_on_shard_zero;
@@ -35,7 +36,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
-use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
+use server_common::Message;
use std::rc::Rc;
use std::sync::Arc;
use std::time::Duration;
@@ -116,7 +117,7 @@ async fn start_on_shard_zero_tcp_tls_round_trip() {
let client_tcp = TcpStream::connect(server_addr).await.expect("client dial");
let conn = TcpTlsTransportConn::new_client(client_tcp, client_cfg, server_name);
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(8);
+ let (out_tx, out_rx) = bounded::<BusMessage>(8);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(8);
let (client_shutdown, client_token) = Shutdown::new();
let ctx = ActorContext {
@@ -132,7 +133,7 @@ async fn start_on_shard_zero_tcp_tls_round_trip() {
let client_handle = compio::runtime::spawn(async move { conn.run(ctx).await });
let request = header_only(Command::Request, CLUSTER, 0).into_frozen();
- out_tx.send(request).await.expect("client send");
+ out_tx.send(request.into()).await.expect("client send");
let reply = compio::time::timeout(Duration::from_secs(5), in_rx.recv())
.await
diff --git a/core/message_bus/tests/wss_client_listener.rs b/core/message_bus/tests/wss_client_listener.rs
index 56566a6c3..f280d8ebf 100644
--- a/core/message_bus/tests/wss_client_listener.rs
+++ b/core/message_bus/tests/wss_client_listener.rs
@@ -22,6 +22,7 @@ use common::{header_only, install_wss_clients_locally, loopback};
use compio::net::TcpStream;
use iggy_binary_protocol::Command;
use iggy_binary_protocol::GenericHeader;
+use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
use message_bus::client_listener::wss::{bind, run};
use message_bus::transports::tls::{install_default_crypto_provider, self_signed_for_loopback};
@@ -30,7 +31,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, MessageBusConfig, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
-use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
+use server_common::Message;
use std::rc::Rc;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -78,7 +79,7 @@ async fn wss_client_listener_accepts_and_round_trips() {
let client_tcp = TcpStream::connect(server_addr).await.expect("client dial");
let conn = WssTransportConn::new_client(client_tcp, client_cfg, server_name);
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(8);
+ let (out_tx, out_rx) = bounded::<BusMessage>(8);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(8);
let (client_shutdown, client_token) = Shutdown::new();
let ctx = ActorContext {
@@ -94,7 +95,7 @@ async fn wss_client_listener_accepts_and_round_trips() {
let client_handle = compio::runtime::spawn(async move { conn.run(ctx).await });
let request = header_only(Command::Request, 7, 0).into_frozen();
- out_tx.send(request).await.expect("client send");
+ out_tx.send(request.into()).await.expect("client send");
let reply = compio::time::timeout(Duration::from_secs(5), in_rx.recv())
.await
diff --git a/core/message_bus/tests/wss_client_roundtrip.rs b/core/message_bus/tests/wss_client_roundtrip.rs
index 24cb27273..a1e2950fd 100644
--- a/core/message_bus/tests/wss_client_roundtrip.rs
+++ b/core/message_bus/tests/wss_client_roundtrip.rs
@@ -25,6 +25,7 @@ use common::{
use compio::net::TcpStream;
use iggy_binary_protocol::Command;
use iggy_binary_protocol::GenericHeader;
+use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
use message_bus::replica::io::start_on_shard_zero;
@@ -35,7 +36,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
-use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
+use server_common::Message;
use std::rc::Rc;
use std::sync::Arc;
use std::time::Duration;
@@ -117,7 +118,7 @@ async fn start_on_shard_zero_wss_round_trip() {
let client_tcp = TcpStream::connect(server_addr).await.expect("client dial");
let conn = WssTransportConn::new_client(client_tcp, client_cfg, server_name);
- let (out_tx, out_rx) = bounded::<Frozen<MESSAGE_ALIGN>>(8);
+ let (out_tx, out_rx) = bounded::<BusMessage>(8);
let (in_tx, in_rx) = bounded::<Message<GenericHeader>>(8);
let (client_shutdown, client_token) = Shutdown::new();
let ctx = ActorContext {
@@ -133,7 +134,7 @@ async fn start_on_shard_zero_wss_round_trip() {
let client_handle = compio::runtime::spawn(async move { conn.run(ctx).await });
let request = header_only(Command::Request, CLUSTER, 0).into_frozen();
- out_tx.send(request).await.expect("client send");
+ out_tx.send(request.into()).await.expect("client send");
let reply = compio::time::timeout(Duration::from_secs(5), in_rx.recv())
.await
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 6ff355e47..31ce620cd 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -4056,7 +4056,9 @@ mod tests {
use iggy_binary_protocol::requests::topics::CreateTopicRequest;
use iggy_common::variadic;
use journal::prepare_journal::PrepareJournal;
- use message_bus::{ClientForwardFn, ConnectionLostFn, JoinHandle, ReplicaForwardFn, SendError};
+ use message_bus::{
+ BusMessage, ClientForwardFn, ConnectionLostFn, JoinHandle, ReplicaForwardFn, SendError,
+ };
use server_common::MESSAGE_ALIGN;
use server_common::iobuf::Frozen;
use std::cell::RefCell;
@@ -4389,7 +4391,7 @@ mod tests {
async fn send_to_client(
&self,
_client_id: u128,
- _data: Frozen<MESSAGE_ALIGN>,
+ _data: impl Into<BusMessage>,
) -> Result<(), SendError> {
Ok(())
}
@@ -4545,7 +4547,7 @@ mod tests {
async fn send_to_client(
&self,
_client_id: u128,
- _data: Frozen<MESSAGE_ALIGN>,
+ _data: impl Into<BusMessage>,
) -> Result<(), SendError> {
if self.stall.get() {
self.stall_hits.set(self.stall_hits.get() + 1);
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 7b538e68d..b08a4e2db 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -5183,7 +5183,7 @@ mod tests {
use compio::io::AsyncWriteAtExt;
use consensus::LocalPipeline;
use iggy_binary_protocol::{Command, ReplyHeader, WireConsumer, WireEncode};
- use message_bus::SendError;
+ use message_bus::{BusMessage, SendError};
use server_common::MESSAGE_ALIGN;
use server_common::send_messages::{
COMMAND_HEADER_SIZE, IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned,
@@ -5570,9 +5570,11 @@ mod tests {
async fn send_to_client(
&self,
client_id: u128,
- data: Frozen<MESSAGE_ALIGN>,
+ data: impl Into<BusMessage>,
) -> Result<(), SendError> {
- self.sent_to_clients.borrow_mut().push((client_id, data));
+ self.sent_to_clients
+ .borrow_mut()
+ .push((client_id, data.into().into_contiguous()));
Ok(())
}
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index fcb3a9af7..45f3b6d1d 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -1153,6 +1153,68 @@ pub fn push_selected_batch_fragments(
*matched_messages += selection.matched_messages;
}
+/// A fragment sliced from a storage buffer keeps the WHOLE 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 source, so a sparse match (a
+/// `count=1` poll off a cold partition, a short poll into a large resident
+/// batch) cannot pin a ~1 MiB chunk or a whole prepare per queued reply; a
+/// dense match keeps the zero-copy path.
+///
+/// On the disk tier, the chunk allocations a poll reply keeps alive are
+/// bounded by `SPARSE_CHUNK_PIN_DIVISOR` times the record bytes it serves
+/// from disk, plus one page per compacted chunk. This is a ratio, not an
+/// absolute cap: a grown chunk can still retain tens of MiB, and it does not
+/// cover the reply's absolute size. The resident tier applies the same ratio
+/// per prepare entry but only copies up to [`RESIDENT_SPARSE_COPY_MAX_BYTES`];
+/// a sparse selection past that stays a slice and pins its prepare.
+const SPARSE_CHUNK_PIN_DIVISOR: usize = 4;
+
+/// Resident copies run inline on the shard pump (the disk walk is detached),
+/// and a poll selects at most two partial batches (its first and last), so
+/// this caps the pump's per-poll memcpy at about twice this many bytes.
+const RESIDENT_SPARSE_COPY_MAX_BYTES: usize = 64 * 1024;
+
+/// Rewrite the fragments pushed from index `pushed_from` on that slice
+/// `source` to slices of one compact copy when their combined length is a
+/// sparse fraction of `source` and at most `copy_max_bytes`. Fragments that
+/// own their bytes (rewritten batch headers) are left alone. See
+/// [`SPARSE_CHUNK_PIN_DIVISOR`].
+pub fn unpin_sparse_source(
+ fragments: &mut PollFragments<4096>,
+ pushed_from: usize,
+ source: &Frozen<4096>,
+ copy_max_bytes: usize,
+) {
+ let pushed = &mut fragments[pushed_from..];
+ let borrowed: usize = pushed
+ .iter()
+ .filter(|fragment| fragment.borrows_from(source))
+ .map(Fragment::len)
+ .sum();
+ if borrowed == 0
+ || borrowed >= source.len() / SPARSE_CHUNK_PIN_DIVISOR
+ || borrowed > copy_max_bytes
+ {
+ return;
+ }
+
+ let mut compact = Owned::<4096>::with_capacity(borrowed);
+ for fragment in pushed.iter().filter(|f| f.borrows_from(source)) {
+ compact.extend_from_slice(fragment.as_slice());
+ }
+ let compact = Frozen::from(compact);
+ let mut cursor = 0;
+ for fragment in pushed.iter_mut() {
+ if !fragment.borrows_from(source) {
+ continue;
+ }
+ let len = fragment.len();
+ *fragment = Fragment::slice(compact.clone(), cursor, cursor + len);
+ cursor += len;
+ }
+}
+
/// Decode one resident `Frozen` entry and push its matching fragments. Shared by
/// the live storage walk and the owned-snapshot walk so the corrupt-header skip
/// and `SendMessages` filter live in one place. Skips (never panics) on a short
@@ -1188,6 +1250,7 @@ fn try_push_resident_entry(
};
// The batch's 256B header sits right after the prepare header in a resident
// entry (see `decode_prepare_slice`), so the batch base is `PREPARE_HEADER_SIZE`.
+ let pushed_from = fragments.len();
push_selected_batch_fragments(
fragments,
last_matching_offset,
@@ -1197,6 +1260,16 @@ fn try_push_resident_entry(
&batch,
selection,
);
+ // `evict_prefix` drains the storage on the routine commit flush, so a
+ // queued reply that still slices this prepare becomes its sole owner.
+ // Accounted per entry here; the disk walk accounts per chunk, where many
+ // batches share one allocation.
+ unpin_sparse_source(
+ fragments,
+ pushed_from,
+ prepare,
+ RESIDENT_SPARSE_COPY_MAX_BYTES,
+ );
}
/// Poll an owned, point-in-time snapshot of the resident journal tail.
@@ -1251,7 +1324,8 @@ mod tests {
use journal::Journal;
use server_common::Message;
use server_common::send_messages::{
- IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned, decode_batch_slice,
+ BatchHeader, IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned,
+ decode_batch_slice,
};
use server_common::sharding::IggyNamespace;
@@ -1516,14 +1590,58 @@ mod tests {
let mut owned = SendMessagesOwned::from_messages(IggyNamespace::new(1, 1, 0), &messages)
.expect("build send_messages batch");
owned.header.base_timestamp = base_timestamp;
- owned.header.batch_checksum = owned.header.checksum_for_blob(&owned.blob);
+ stamped_batch_record(owned)
+ }
+ /// Stamp `owned`'s checksum and lay it out as the `[256B batch header][blob]`
+ /// record a batch occupies in storage.
+ fn stamped_batch_record(mut owned: SendMessagesOwned) -> Vec<u8> {
+ owned.header.batch_checksum = owned.header.checksum_for_blob(&owned.blob);
let mut record = vec![0u8; COMMAND_HEADER_SIZE + owned.blob.len()];
owned.header.encode_into(&mut record[..COMMAND_HEADER_SIZE]);
record[COMMAND_HEADER_SIZE..].copy_from_slice(&owned.blob);
record
}
+ /// A resident `SendMessages` prepare entry holding one batch of
+ /// `message_count` records with distinct `payload_len`-byte payloads, in
+ /// the `[PrepareHeader][256B batch header][blob]` layout
+ /// `try_push_resident_entry` decodes.
+ fn build_resident_prepare(message_count: usize, payload_len: usize) -> Frozen<4096> {
+ let mut messages = IggyMessages::with_capacity(message_count);
+ for index in 0..message_count {
+ let fill = u8::try_from(index % usize::from(u8::MAX)).expect("bounded by u8::MAX");
+ messages.push(IggyMessage {
+ header: IggyMessageHeader {
+ payload_length: u32::try_from(payload_len).expect("payload_len fits u32"),
+ ..Default::default()
+ },
+ payload: Bytes::from(vec![fill; payload_len]),
+ user_headers: None,
+ });
+ }
+ let owned = SendMessagesOwned::from_messages(IggyNamespace::new(1, 1, 0), &messages)
+ .expect("build send_messages batch");
+ let record = stamped_batch_record(owned);
+
+ let mut prepare = build_prepare(1, PREPARE_HEADER_SIZE + record.len()).transmute_header(
+ |header: PrepareHeader, send_messages: &mut PrepareHeader| {
+ *send_messages = header;
+ send_messages.operation = Operation::SendMessages;
+ },
+ );
+ prepare.as_mut_slice()[PREPARE_HEADER_SIZE..].copy_from_slice(&record);
+ prepare.into_frozen()
+ }
+
+ fn offset_lookup(offset: u64, count: u32) -> MessageLookup {
+ MessageLookup::Offset {
+ offset,
+ count,
+ ceiling: u64::MAX,
+ }
+ }
+
#[test]
fn timestamp_poll_at_exact_broker_timestamp_includes_the_batch() {
// A client polls with a timestamp read from a previous reply, which is
@@ -1562,4 +1680,125 @@ mod tests {
"poll past the broker timestamp must match nothing"
);
}
+
+ /// A short poll into a large resident batch ships a rewritten header plus
+ /// a body slice. Left as a slice of the prepare, that body would keep the
+ /// whole entry alive after `evict_prefix` drained it from the journal.
+ #[test]
+ fn resident_partial_selection_copies_out_of_a_large_prepare() {
+ let prepare = build_resident_prepare(1_000, 300);
+ let query = offset_lookup(500, 1);
+ let batch = decode_prepare_slice_trusted(prepare.as_slice()).expect("prepare decodes");
+ let selection = select_batch_slice(&batch, query, 0).expect("record 500 is selected");
+ let expected_body = &batch.blob()[selection.start..selection.end];
+ assert!(
+ expected_body.len() < prepare.len() / SPARSE_CHUNK_PIN_DIVISOR,
+ "fixture must select a sparse fraction of the prepare"
+ );
+
+ let (fragments, last_matching_offset) =
+ select_resident(std::slice::from_ref(&prepare), query).expect("one record matches");
+ assert_eq!(last_matching_offset, Some(500));
+ assert_eq!(fragments.len(), 2, "rewritten header plus body slice");
+ let (header, body) = (&fragments[0], &fragments[1]);
+ assert!(
+ !body.borrows_from(&prepare),
+ "sparse body must be copied out of the prepare"
+ );
+ assert_eq!(body.as_slice(), expected_body);
+ assert_eq!(header.len(), COMMAND_HEADER_SIZE);
+ assert!(
+ !header.borrows_from(&body.clone().into_frozen()),
+ "the owned header must stay out of the compact copy"
+ );
+ let rewritten = BatchHeader::decode(header.as_slice()).expect("rewritten header decodes");
+ assert_eq!(rewritten.message_count, 1);
+ }
+
+ #[test]
+ fn resident_whole_batch_selection_keeps_the_zero_copy_slice() {
+ let prepare = build_resident_prepare(1_000, 300);
+
+ let (fragments, last_matching_offset) =
+ select_resident(std::slice::from_ref(&prepare), offset_lookup(0, 1_000))
+ .expect("whole batch matches");
+ assert_eq!(last_matching_offset, Some(999));
+ assert_eq!(fragments.len(), 1, "a whole batch ships its original bytes");
+ assert!(
+ fragments[0].borrows_from(&prepare),
+ "dense selection keeps the zero-copy path"
+ );
+ assert_eq!(fragments[0].len(), prepare.len() - PREPARE_HEADER_SIZE);
+ }
+
+ /// The resident copy runs inline on the shard pump, so past the byte cap
+ /// a sparse selection stays a zero-copy slice even though it pins the
+ /// prepare.
+ #[test]
+ fn resident_sparse_copy_stops_at_the_byte_cap() {
+ let prepare = build_resident_prepare(2_000, 300);
+ let query = offset_lookup(0, 200);
+ let batch = decode_prepare_slice_trusted(prepare.as_slice()).expect("prepare decodes");
+ let selection = select_batch_slice(&batch, query, 0).expect("200 records are selected");
+ let selected = selection.end - selection.start;
+ assert!(
+ selected > RESIDENT_SPARSE_COPY_MAX_BYTES
+ && selected < prepare.len() / SPARSE_CHUNK_PIN_DIVISOR,
+ "fixture must select a sparse fraction that is over the copy cap"
+ );
+
+ let (fragments, _) =
+ select_resident(std::slice::from_ref(&prepare), query).expect("records match");
+ assert_eq!(fragments.len(), 2, "rewritten header plus body slice");
+ assert!(
+ fragments[1].borrows_from(&prepare),
+ "over the cap the body must stay a slice of the prepare"
+ );
+ }
+
+ /// A sparse match must not pin the whole disk chunk: the matched bytes
+ /// are copied out byte-for-byte and the fragments stop borrowing the
+ /// chunk allocation. A dense match keeps the zero-copy slices, and
+ /// fragments that already own their bytes (rewritten batch headers) are
+ /// never touched.
+ #[test]
+ fn unpin_sparse_source_bounds_chunk_retention() {
+ let chunk_len = 1 << 20;
+ let mut backing = Owned::<4096>::zeroed(chunk_len);
+ for (position, byte) in backing.as_mut_slice().iter_mut().enumerate() {
+ *byte = u8::try_from(position % 251).unwrap();
+ }
+ let chunk = Frozen::from(backing);
+
+ let mut fragments = PollFragments::<4096>::new();
+ fragments.push(Fragment::whole(Owned::<4096>::zeroed(256).into()));
+ fragments.push(Fragment::slice(chunk.clone(), 512, 512 + 600));
+ fragments.push(Fragment::slice(chunk.clone(), 4096, 4096 + 300));
+ let first = fragments[1].as_slice().to_vec();
+ let second = fragments[2].as_slice().to_vec();
+ unpin_sparse_source(&mut fragments, 0, &chunk, usize::MAX);
+ assert!(
+ !fragments[1].borrows_from(&chunk) && !fragments[2].borrows_from(&chunk),
+ "sparse slices must be copied out of the chunk"
+ );
+ assert_eq!(fragments[1].as_slice(), &first[..]);
+ assert_eq!(fragments[2].as_slice(), &second[..]);
+ assert!(
+ fragments[1].borrows_from(&fragments[2].clone().into_frozen()),
+ "copies pack into one compact allocation"
+ );
+ assert_eq!(fragments[0].len(), 256);
+
+ let mut fragments = PollFragments::<4096>::new();
+ fragments.push(Fragment::slice(
+ chunk.clone(),
+ 0,
+ chunk_len / SPARSE_CHUNK_PIN_DIVISOR,
+ ));
+ unpin_sparse_source(&mut fragments, 0, &chunk, usize::MAX);
+ assert!(
+ fragments[0].borrows_from(&chunk),
+ "dense slice keeps the zero-copy path"
+ );
+ }
}
diff --git a/core/partitions/src/messages_writer.rs b/core/partitions/src/messages_writer.rs
index d68912719..63ad93c11 100644
--- a/core/partitions/src/messages_writer.rs
+++ b/core/partitions/src/messages_writer.rs
@@ -20,7 +20,7 @@ use compio::{
io::AsyncWriteAtExt,
};
use iggy_common::{IggyByteSize, IggyError};
-use server_common::iobuf::Frozen;
+use server_common::iobuf::{Frozen, IOV_MAX};
use std::{
rc::Rc,
sync::atomic::{AtomicU64, Ordering},
@@ -30,8 +30,6 @@ use tracing::{error, warn};
#[cfg(target_os = "linux")]
use nix::fcntl::{FallocateFlags, fallocate};
-const MAX_IOV_COUNT: usize = 1024;
-
#[derive(Debug)]
pub struct MessagesWriter {
file_path: String,
@@ -217,7 +215,7 @@ async fn write_frozen_chunked<const ALIGN: usize>(
mut position: u64,
buffers: &[Frozen<ALIGN>],
) -> Result<(), IggyError> {
- for chunk in buffers.chunks(MAX_IOV_COUNT) {
+ for chunk in buffers.chunks(IOV_MAX) {
let chunk_size: usize = chunk.iter().map(Frozen::len).sum();
let chunk_vec: Vec<_> = chunk.to_vec();
diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs
index d982fa1bc..7d3fae869 100644
--- a/core/partitions/src/poll_plan.rs
+++ b/core/partitions/src/poll_plan.rs
@@ -33,7 +33,9 @@
use crate::PollFragments;
use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndexCache};
use crate::iggy_index_reader::IggyIndexReader;
-use crate::journal::{MessageLookup, push_selected_batch_fragments, select_batch_slice};
+use crate::journal::{
+ MessageLookup, push_selected_batch_fragments, select_batch_slice, unpin_sparse_source,
+};
use compio::io::AsyncReadAtExt;
use iggy_common::{
ConsumerGroupId, ConsumerGroupOffsets, ConsumerKind, ConsumerOffset, ConsumerOffsets, IggyError,
@@ -572,6 +574,7 @@ impl DiskReadPlan {
faulted = true;
break 'walk;
};
+ let fragments_before_chunk = fragments.len();
let ChunkWalk { consumed, corrupt } = walk_disk_chunk(
&chunk,
query,
@@ -586,6 +589,8 @@ impl DiskReadPlan {
},
self.namespace_raw,
);
+ // Detached from the pump, so the ratio alone bounds the copy.
+ unpin_sparse_source(&mut fragments, fragments_before_chunk, &chunk, usize::MAX);
if corrupt {
// A batch that does not match its own checksum. Fail closed like
// an IO fault: serving it hands a consumer data provably not what
diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs
index 6c956f7e3..4762b772a 100644
--- a/core/partitions/src/types.rs
+++ b/core/partitions/src/types.rs
@@ -75,6 +75,28 @@ impl<const ALIGN: usize> Fragment<ALIGN> {
self.source.slice(self.start..self.end)
}
}
+
+ #[must_use]
+ pub fn as_slice(&self) -> &[u8] {
+ &self.source[self.start..self.end]
+ }
+
+ #[must_use]
+ pub const fn len(&self) -> usize {
+ self.end - self.start
+ }
+
+ #[must_use]
+ pub const fn is_empty(&self) -> bool {
+ self.start == self.end
+ }
+
+ /// Whether this fragment refcounts `source`'s allocation (and thus keeps
+ /// all of it alive, not just the sliced window).
+ #[must_use]
+ pub fn borrows_from(&self, source: &Frozen<ALIGN>) -> bool {
+ self.source.shares_allocation(source)
+ }
}
/// Arguments for polling messages from a partition.
diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs
index 784b86bdf..3bf5af28a 100644
--- a/core/server/src/dispatch.rs
+++ b/core/server/src/dispatch.rs
@@ -41,7 +41,7 @@ use crate::pat::maybe_rewrite_pat_request;
use crate::responses::{
NonReplicatedResponse, build_consumer_offset_body, build_deny_reply, build_empty_reply,
build_get_me_response, build_get_personal_access_tokens_response,
- build_non_replicated_response, build_polled_messages_body, build_raw_pat_reply,
+ build_non_replicated_response, build_polled_messages_reply, build_raw_pat_reply,
connected_client_to_response, current_metadata_commit, resolve_partition_namespace,
resolve_partition_request_namespace,
};
@@ -103,10 +103,10 @@ use iggy_common::{
};
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
-use message_bus::AUTO_COMMIT_CLIENT_ID;
use message_bus::client_listener::RequestHandler;
use message_bus::framing::MAX_MESSAGE_SIZE;
use message_bus::replica::listener::MessageHandler;
+use message_bus::{AUTO_COMMIT_CLIENT_ID, BusMessage};
use metadata::impls::metadata::{
BoundSession, MetadataSubmitError, StreamsFrontend, build_truncate_partition_client_message,
build_truncate_partition_client_message_with_identifiers,
@@ -1941,11 +1941,30 @@ async fn send_non_replicated_bytes<B, MJ, S, SB>(
request.header().session,
commit,
);
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
+ send_reply_frame(
+ shard,
+ transport_client_id,
+ reply.into_generic().into_frozen(),
+ label,
+ )
+ .await;
+}
+
+/// Hand a built reply frame to the bus for `transport_client_id`.
+#[allow(clippy::future_not_send)]
+async fn send_reply_frame<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ frame: impl Into<BusMessage>,
+ label: &'static str,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ if let Err(error) = shard.bus.send_to_client(transport_client_id, frame).await {
warn!(transport_client_id, label, error = %error, "failed to send non-replicated reply");
}
}
@@ -2169,20 +2188,27 @@ async fn handle_poll_messages<B, MJ, S, SB>(
Some(PartitionReadReply::Poll {
fragments,
current_offset,
- }) => build_polled_messages_body(
+ }) => match build_polled_messages_reply(
+ request.header(),
+ current_metadata_commit(shard),
partition_id,
current_offset,
fragments,
shard.plane.partitions().config().encryptor.as_deref(),
- )
- .unwrap_or_else(|error| {
- warn!(
- transport_client_id,
- error = %error,
- "failed to re-encode polled batches; replying empty poll"
- );
- empty_polled_messages_body(partition_id)
- }),
+ ) {
+ Ok(reply) => {
+ send_reply_frame(shard, transport_client_id, reply, "poll_messages").await;
+ return;
+ }
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to re-encode polled batches; replying empty poll"
+ );
+ empty_polled_messages_body(partition_id)
+ }
+ },
other => {
warn!(
transport_client_id,
@@ -3823,6 +3849,7 @@ mod tests {
use iggy_common::defaults::DEFAULT_ROOT_USER_ID;
use iggy_common::variadic;
use journal::prepare_journal::PrepareJournal;
+ use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
use message_bus::fd_transfer::DupedFd;
use message_bus::installer::ConnectionInstaller;
@@ -3895,11 +3922,11 @@ mod tests {
async fn send_to_client(
&self,
client_id: u128,
- data: Frozen<MESSAGE_ALIGN>,
+ data: impl Into<BusMessage>,
) -> Result<(), SendError> {
self.client_replies
.borrow_mut()
- .push((client_id, data.as_slice().to_vec()));
+ .push((client_id, data.into().into_contiguous().as_slice().to_vec()));
Ok(())
}
async fn send_to_replica(
@@ -4496,7 +4523,10 @@ mod tests {
fn reply_lane_forward(client_id: u128) -> ShardFrame {
ShardFrame::lifecycle(LifecycleFrame::ForwardClientSend {
client_id,
- msg: server_common::iobuf::Owned::<MESSAGE_ALIGN>::zeroed(64).into(),
+ msg: server_common::iobuf::Frozen::from(
+ server_common::iobuf::Owned::<MESSAGE_ALIGN>::zeroed(64),
+ )
+ .into(),
})
}
@@ -4646,7 +4676,7 @@ mod tests {
if let ShardFrame::Lifecycle(LifecycleFrame::ForwardClientSend { client_id, msg }) =
frame
{
- denies.push((client_id, msg.as_slice().to_vec()));
+ denies.push((client_id, msg.into_contiguous().as_slice().to_vec()));
}
}
assert_eq!(
diff --git a/core/server/src/http/reply.rs b/core/server/src/http/reply.rs
index c17f5ce78..0ce0b14f4 100644
--- a/core/server/src/http/reply.rs
+++ b/core/server/src/http/reply.rs
@@ -33,8 +33,7 @@ use iggy_common::{
ConsumerGroupDetails, IggyError, StreamDetails, TopicDetails, UserInfoDetails,
eviction_reason_to_error,
};
-use message_bus::BusMessage;
-use server_common::Message;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use tracing::warn;
use crate::http::error::{PartitionWriteError, WriteError};
@@ -60,7 +59,7 @@ use crate::login_register::LoginRegisterError;
/// keeps that a single parse whose failure mode is already decided here, rather
/// than a second one whose fallback would have to invent a body.
pub(in crate::http) fn classify_partition_reply(
- reply: &BusMessage,
+ reply: &Frozen<MESSAGE_ALIGN>,
) -> Result<ReplyHeader, PartitionWriteError> {
let header = reply
.as_slice()
@@ -93,7 +92,7 @@ pub(in crate::http) fn classify_partition_reply(
/// `header` is the one [`classify_partition_reply`] graded, so this reads the
/// body without re-deriving its extent.
pub(in crate::http) fn send_confirmations(
- reply: &BusMessage,
+ reply: &Frozen<MESSAGE_ALIGN>,
header: &ReplyHeader,
) -> Option<SendMessagesResponse> {
let body = partition_reply_body(reply, header);
@@ -115,7 +114,7 @@ pub(in crate::http) fn send_confirmations(
/// A partition reply's body past the header, bounded by the header's `size`
/// rather than by the buffer length: `size` is the frame's authoritative
/// extent, and the typed decoders reject trailing bytes.
-fn partition_reply_body<'a>(reply: &'a BusMessage, header: &ReplyHeader) -> &'a [u8] {
+fn partition_reply_body<'a>(reply: &'a Frozen<MESSAGE_ALIGN>, header: &ReplyHeader) -> &'a [u8] {
reply
.as_slice()
.get(HEADER_SIZE..header.size as usize)
@@ -362,7 +361,7 @@ mod tests {
);
}
- fn frozen(reply: Message<iggy_binary_protocol::ReplyHeader>) -> BusMessage {
+ fn frozen(reply: Message<iggy_binary_protocol::ReplyHeader>) -> Frozen<MESSAGE_ALIGN> {
reply.into_generic().into_frozen()
}
@@ -486,7 +485,7 @@ mod tests {
));
}
- fn send_reply(body: &Bytes) -> BusMessage {
+ fn send_reply(body: &Bytes) -> Frozen<MESSAGE_ALIGN> {
let prepare = PrepareHeader {
command: Command::Prepare,
operation: Operation::SendMessages,
diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs
index eeaf9dc1a..e7dd75efd 100644
--- a/core/server/src/http/submit.rs
+++ b/core/server/src/http/submit.rs
@@ -28,9 +28,8 @@ use futures::channel::oneshot;
use iggy_binary_protocol::consensus::Command;
use iggy_binary_protocol::{GenericHeader, Operation, ReplyHeader, RoutedRequestHeader};
use iggy_common::IggyError;
-use message_bus::BusMessage;
use metadata::impls::metadata::StreamsFrontend;
-use server_common::Message;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use tracing::warn;
use crate::dispatch::{
@@ -370,7 +369,7 @@ pub(in crate::http) async fn partition_write_replicated(
session: &HttpSession,
operation: Operation,
body: &[u8],
-) -> Result<(BusMessage, ReplyHeader), PartitionWriteError> {
+) -> Result<(Frozen<MESSAGE_ALIGN>, ReplyHeader), PartitionWriteError> {
// Admission sits here rather than before body decode: axum's extractors
// already buffered and deserialized the body (bounded by the router-wide
// `DefaultBodyLimit`) before the handler ran, so the caps gate what is
@@ -414,7 +413,10 @@ pub(in crate::http) async fn partition_write_replicated(
// reply after a timeout sheds at the bus instead of leaking a waiter.
drop(guard);
match outcome {
- Ok(Ok(reply)) => classify_partition_reply(&reply).map(|header| (reply, header)),
+ Ok(Ok(reply)) => {
+ let reply = reply.into_contiguous();
+ classify_partition_reply(&reply).map(|header| (reply, header))
+ }
// Cancelled (reply target torn down by session eviction mid-wait) or
// elapsed: same caller contract either way - outcome unknown, 504.
Ok(Err(_)) | Err(_) => Err(PartitionWriteError::Timeout(operation)),
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index 5de84bcc1..d0216aea7 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -4470,7 +4470,7 @@ mod tests {
fn reply_status(reply: &message_bus::BusMessage) -> u32 {
bytemuck::checked::try_from_bytes::<ReplyHeader>(
- &reply.as_ref()[..size_of::<ReplyHeader>()],
+ &reply.first().as_slice()[..size_of::<ReplyHeader>()],
)
.expect("deny reply carries a valid ReplyHeader")
.status
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index 6575456ae..4c48c0514 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -88,11 +88,13 @@ use iggy_common::{
};
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
+use message_bus::BusMessage;
use metadata::impls::metadata::StreamsFrontend;
-use partitions::PollFragments;
-use server_common::Message;
+use partitions::{Fragment, PollFragments};
+use server_common::iobuf::{Frozen, Owned};
use server_common::send_messages;
use server_common::sharding::IggyNamespace;
+use server_common::{MESSAGE_ALIGN, Message, ResponseBacking, ResponseFragments};
use shard::ConnectedClientInfo;
use std::cell::RefCell;
use std::net::IpAddr;
@@ -1610,15 +1612,25 @@ pub fn build_reply_with_body(
) -> Message<ReplyHeader> {
let header_len = std::mem::size_of::<ReplyHeader>();
let total_size = header_len + body_len;
+ let size = u32::try_from(total_size).expect("reply size must fit into u32");
let mut reply = Message::<ReplyHeader>::new(total_size);
- let header_size = u32::try_from(total_size).expect("reply size must fit into u32");
- let header = bytemuck::checked::try_from_bytes_mut::<ReplyHeader>(
- &mut reply.as_mut_slice()[..header_len],
- )
- .expect("zeroed bytes are valid");
- *header = ReplyHeader {
+ let header = reply_header(request_header, client_id, session, commit, size);
+ reply.as_mut_slice()[..header_len].copy_from_slice(bytemuck::bytes_of(&header));
+ write_body(&mut reply.as_mut_slice()[header_len..total_size]);
+ reply
+}
+
+/// The header of a `size`-byte reply frame answering `request_header`.
+fn reply_header(
+ request_header: &RoutedRequestHeader,
+ client_id: u128,
+ session: u64,
+ commit: u64,
+ size: u32,
+) -> ReplyHeader {
+ ReplyHeader {
cluster: request_header.cluster,
- size: header_size,
+ size,
view: request_header.view,
release: request_header.release,
command: Command::Reply,
@@ -1631,9 +1643,7 @@ pub fn build_reply_with_body(
request: request_header.request,
operation: request_header.operation,
..Default::default()
- };
- write_body(&mut reply.as_mut_slice()[header_len..total_size]);
- reply
+ }
}
pub fn current_metadata_commit<B, MJ, S, SB>(shard: &Rc<ShellShard<B, MJ, S, SB>>) -> u64
@@ -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();
+
+ let header_len = std::mem::size_of::<ReplyHeader>();
+ let size = u32::try_from(header_len + POLLED_HEAD_LEN + records_len)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let header = reply_header(request_header, client_id, session, commit, size);
+ let mut head = Owned::<MESSAGE_ALIGN>::zeroed(header_len + POLLED_HEAD_LEN);
+ let (header_bytes, body_head) = head.as_mut_slice().split_at_mut(header_len);
+ header_bytes.copy_from_slice(bytemuck::bytes_of(&header));
+ body_head[..4].copy_from_slice(&partition_id.to_le_bytes());
+ body_head[4..12].copy_from_slice(¤t_offset.to_le_bytes());
+ body_head[12..].copy_from_slice(&count.to_le_bytes());
+ frames.insert(0, head.into());
+
+ // Re-checks the header and that the fragments cover `size`.
+ Message::<ReplyHeader, ResponseBacking>::try_from(frames)
+ .map(Message::into_inner)
+ .map_err(|_| IggyError::InvalidCommand)
+}
+
+/// Sum of `message_count` over the batch records spanning `fragments`, read
+/// from each batch header in place. Rejects a stream that is not a whole
+/// number of batches, as [`build_polled_messages_body`] does.
+fn polled_message_count(fragments: &[Frozen<MESSAGE_ALIGN>]) -> Result<u32, IggyError> {
+ let mut cursor = FragmentCursor::new(fragments);
+ let mut count = 0u32;
+ let mut header = [0u8; send_messages::COMMAND_HEADER_SIZE];
+ while !cursor.is_exhausted() {
+ cursor.read_exact(&mut header)?;
+ let batch =
+ send_messages::BatchHeader::decode(&header).map_err(|_| IggyError::InvalidCommand)?;
+ cursor.skip(batch.blob_len().map_err(|_| IggyError::InvalidCommand)?)?;
+ count = count
+ .checked_add(batch.message_count)
+ .ok_or(IggyError::InvalidCommand)?;
+ }
+ Ok(count)
+}
+
+/// Byte cursor over the virtual concatenation of `fragments`. Rests on an
+/// unread byte or at the end of the stream, never inside an exhausted
+/// fragment, so a batch header split across fragments reads the same as one
+/// stored whole.
+struct FragmentCursor<'a> {
+ fragments: &'a [Frozen<MESSAGE_ALIGN>],
+ index: usize,
+ offset: usize,
+}
+
+impl<'a> FragmentCursor<'a> {
+ fn new(fragments: &'a [Frozen<MESSAGE_ALIGN>]) -> Self {
+ let mut cursor = Self {
+ fragments,
+ index: 0,
+ offset: 0,
+ };
+ cursor.settle();
+ cursor
+ }
+
+ const fn is_exhausted(&self) -> bool {
+ self.index == self.fragments.len()
+ }
+
+ fn read_exact(&mut self, out: &mut [u8]) -> Result<(), IggyError> {
+ let mut filled = 0;
+ while filled < out.len() {
+ let available = self.available()?;
+ let take = available.len().min(out.len() - filled);
+ out[filled..filled + take].copy_from_slice(&available[..take]);
+ filled += take;
+ self.advance(take);
+ }
+ Ok(())
+ }
+
+ fn skip(&mut self, mut len: usize) -> Result<(), IggyError> {
+ while len > 0 {
+ let take = self.available()?.len().min(len);
+ len -= take;
+ self.advance(take);
+ }
+ Ok(())
+ }
+
+ /// Unread bytes of the current fragment; `Err` past the end of the stream.
+ fn available(&self) -> Result<&'a [u8], IggyError> {
+ self.fragments
+ .get(self.index)
+ .map(|fragment| &fragment.as_slice()[self.offset..])
+ .ok_or(IggyError::InvalidCommand)
+ }
+
+ fn advance(&mut self, len: usize) {
+ self.offset += len;
+ self.settle();
+ }
+
+ /// Step past the current fragment once it is used up, and past empty ones.
+ fn settle(&mut self) {
+ while let Some(fragment) = self.fragments.get(self.index) {
+ if self.offset < fragment.len() {
+ break;
+ }
+ self.offset = 0;
+ self.index += 1;
+ }
+ }
+}
+
/// Build the `PolledMessages` reply body from the owning shard's poll
-/// fragments.
+/// fragments, gathered into one buffer.
///
/// Fragments carry the stored batch records (a 256-byte batch header plus
/// `[48B header][payload][user_headers]` frames, deltas resolved against the
@@ -1662,6 +1812,10 @@ where
/// decryption: stored sections are ciphertext, and this reply is the single
/// decrypt point, so encrypted records are rebuilt over the plaintext.
///
+/// The binary transports reply through [`build_polled_messages_reply`], which
+/// ships the fragments without gathering them; this builder serves the
+/// decrypt path and the HTTP handler, which decodes the body into JSON.
+///
/// Body layout: `[partition_id:4][current_offset:8][count:4][batch records...]`.
pub fn build_polled_messages_body(
partition_id: u32,
@@ -2005,4 +2159,232 @@ mod tests {
assert_eq!(unlimited.max_topic_size, u64::MAX);
assert_eq!(unlimited.message_expiry, u64::MAX);
}
+
+ // Vectored `PolledMessages` replies against the flattening builder as the
+ // byte-for-byte oracle.
+
+ use iggy_common::Aes256GcmEncryptor;
+ use server_common::send_messages::{
+ BatchHeader, COMMAND_HEADER_SIZE, IggyMessage, IggyMessageHeader, IggyMessages,
+ PREPARE_SPLIT_POINT, SendMessagesOwned, encrypt_batch_request, frozen_batch_header,
+ };
+ use server_common::sharding::IggyNamespace;
+
+ const POLL_PARTITION_ID: u32 = 9;
+ const POLL_CURRENT_OFFSET: u64 = 1_234;
+ const POLL_COMMIT: u64 = 17;
+
+ fn poll_request_header() -> RoutedRequestHeader {
+ pat_request_header()
+ }
+
+ /// A stored batch record over an opaque blob. Both builders decode only
+ /// the 256-byte batch header, so the blob needs no message framing.
+ fn batch_record(base_offset: u64, message_count: u32, blob: &[u8]) -> Frozen<MESSAGE_ALIGN> {
+ let batch_length = u64::try_from(COMMAND_HEADER_SIZE + blob.len()).expect("fits u64");
+ let mut header =
+ BatchHeader::new(u64::from(POLL_PARTITION_ID), 5, batch_length, message_count);
+ header.base_offset = base_offset;
+ let mut bytes = vec![0u8; COMMAND_HEADER_SIZE + blob.len()];
+ header.encode_into(&mut bytes[..COMMAND_HEADER_SIZE]);
+ bytes[COMMAND_HEADER_SIZE..].copy_from_slice(blob);
+ Owned::<MESSAGE_ALIGN>::copy_from_slice(&bytes).into()
+ }
+
+ /// The wire bytes the flattening builder ships for `fragments`.
+ fn flattened_reply(fragments: PollFragments, encryptor: Option<&EncryptorKind>) -> Vec<u8> {
+ let header = poll_request_header();
+ let body = build_polled_messages_body(
+ POLL_PARTITION_ID,
+ POLL_CURRENT_OFFSET,
+ fragments,
+ encryptor,
+ )
+ .expect("flattening builder accepts the fragments");
+ build_reply_from_bytes(&header, header.client, header.session, POLL_COMMIT, &body)
+ .into_generic()
+ .into_frozen()
+ .as_slice()
+ .to_vec()
+ }
+
+ fn vectored_reply(
+ fragments: PollFragments,
+ encryptor: Option<&EncryptorKind>,
+ ) -> Result<BusMessage, IggyError> {
+ build_polled_messages_reply(
+ &poll_request_header(),
+ POLL_COMMIT,
+ POLL_PARTITION_ID,
+ POLL_CURRENT_OFFSET,
+ fragments,
+ encryptor,
+ )
+ }
+
+ /// The vectored reply must be byte-identical to the flattened one and
+ /// ship exactly `fragment_count` buffers.
+ fn assert_vectored_matches_flattened(fragments: PollFragments, fragment_count: usize) {
+ let expected = flattened_reply(fragments.clone(), None);
+ let reply =
+ vectored_reply(fragments, None).expect("vectored builder accepts the fragments");
+ assert_eq!(reply.fragments().len(), fragment_count);
+ assert_eq!(reply.total_len(), expected.len());
+ assert_eq!(reply.into_contiguous().as_slice(), expected.as_slice());
+ }
+
+ fn polled_count(reply: &[u8]) -> u32 {
+ let count_at = std::mem::size_of::<ReplyHeader>() + 12;
+ u32::from_le_bytes(reply[count_at..count_at + 4].try_into().expect("4 bytes"))
+ }
+
+ #[test]
+ fn polled_reply_single_fragment_matches_flattened_builder() {
+ let record = batch_record(0, 3, &[0xAB; 100]);
+ let fragments = PollFragments::from_iter([Fragment::whole(record)]);
+ assert_vectored_matches_flattened(fragments.clone(), 2);
+
+ let reply = vectored_reply(fragments, None)
+ .expect("reply")
+ .into_contiguous();
+ let header = bytemuck::checked::try_from_bytes::<ReplyHeader>(
+ &reply.as_slice()[..std::mem::size_of::<ReplyHeader>()],
+ )
+ .expect("reply header decodes");
+ assert_eq!(header.size as usize, reply.len());
+ assert_eq!(header.client, 42);
+ assert_eq!(header.op, 7);
+ assert_eq!(header.commit, POLL_COMMIT);
+ assert_eq!(polled_count(reply.as_slice()), 3);
+ }
+
+ #[test]
+ fn polled_reply_split_batch_matches_flattened_builder() {
+ // The journal slices a partially selected batch into a rewritten header
+ // plus a blob slice, exactly how `push_selected_batch_fragments` does.
+ let source = batch_record(10, 4, &[0x11; 400]);
+ let (start, end) = (100, 300);
+ let batch_length = u64::try_from(COMMAND_HEADER_SIZE + (end - start)).expect("fits u64");
+ let mut rewritten = BatchHeader::new(u64::from(POLL_PARTITION_ID), 5, batch_length, 2);
+ rewritten.base_offset = 10;
+ let fragments = PollFragments::from_iter([
+ Fragment::whole(frozen_batch_header(&rewritten)),
+ Fragment::slice(
+ source,
+ COMMAND_HEADER_SIZE + start,
+ COMMAND_HEADER_SIZE + end,
+ ),
+ ]);
+ assert_vectored_matches_flattened(fragments.clone(), 3);
+ let reply = vectored_reply(fragments, None)
+ .expect("reply")
+ .into_contiguous();
+ assert_eq!(polled_count(reply.as_slice()), 2);
+ }
+
+ #[test]
+ fn polled_reply_multiple_batches_counts_every_header() {
+ let first = batch_record(0, 1, &[0x01; 50]);
+ let second = batch_record(1, 4, &[0x02; 700]);
+ let third = batch_record(5, 7, &[0x03; 20]);
+ // `second` arrives cut mid-header so the count walk has to read a batch
+ // header spanning two fragments.
+ let fragments = PollFragments::from_iter([
+ Fragment::whole(first),
+ Fragment::slice(second.clone(), 0, 100),
+ Fragment::slice(second.clone(), 100, second.len()),
+ Fragment::whole(third),
+ ]);
+ assert_vectored_matches_flattened(fragments.clone(), 5);
+ let reply = vectored_reply(fragments, None)
+ .expect("reply")
+ .into_contiguous();
+ assert_eq!(polled_count(reply.as_slice()), 12);
+ }
+
+ #[test]
+ fn polled_reply_empty_poll_is_the_head_alone() {
+ assert_vectored_matches_flattened(PollFragments::new(), 1);
+ let reply = vectored_reply(PollFragments::new(), None)
+ .expect("reply")
+ .into_contiguous();
+ assert_eq!(
+ reply.len(),
+ std::mem::size_of::<ReplyHeader>() + POLLED_HEAD_LEN
+ );
+ assert_eq!(polled_count(reply.as_slice()), 0);
+ }
+
+ #[test]
+ fn polled_reply_rejects_a_truncated_record() {
+ let record = batch_record(0, 3, &[0xAB; 100]);
+ let truncated =
+ PollFragments::from_iter([Fragment::slice(record, 0, COMMAND_HEADER_SIZE + 99)]);
+ assert!(matches!(
+ build_polled_messages_body(
+ POLL_PARTITION_ID,
+ POLL_CURRENT_OFFSET,
+ truncated.clone(),
+ None
+ ),
+ Err(IggyError::InvalidCommand)
+ ));
+ assert!(matches!(
+ vectored_reply(truncated, None),
+ Err(IggyError::InvalidCommand)
+ ));
+ }
+
+ /// A stored record encrypted the way the primary encrypts at ingestion.
+ fn encrypted_record(encryptor: &EncryptorKind) -> Frozen<MESSAGE_ALIGN> {
+ let namespace = IggyNamespace::new(1, 1, 3);
+ let mut messages = IggyMessages::with_capacity(2);
+ for (id, payload) in [(7u128, &b"first-payload"[..]), (8, &b"second-payload"[..])] {
+ messages.push(IggyMessage {
+ header: IggyMessageHeader {
+ id,
+ origin_timestamp: 1_000,
+ ..Default::default()
+ },
+ payload: Bytes::copy_from_slice(payload),
+ user_headers: None,
+ });
+ }
+ let owned = SendMessagesOwned::from_messages(namespace, &messages).expect("build batch");
+ let header_size = std::mem::size_of::<RoutedRequestHeader>();
+ let total = header_size + owned.header.total_size();
+ let mut buffer = Owned::<MESSAGE_ALIGN>::zeroed(total);
+ {
+ let header: &mut RoutedRequestHeader =
+ bytemuck::checked::try_from_bytes_mut(&mut buffer.as_mut_slice()[..header_size])
+ .expect("zeroed bytes form a valid RoutedRequestHeader");
+ header.command = Command::Request;
+ header.operation = Operation::SendMessages;
+ header.client = 1;
+ header.session = 1;
+ header.request = 1;
+ header.size = u32::try_from(total).expect("size fits u32");
+ }
+ let bytes = buffer.as_mut_slice();
+ owned
+ .header
+ .encode_into(&mut bytes[header_size..header_size + COMMAND_HEADER_SIZE]);
+ bytes[PREPARE_SPLIT_POINT..].copy_from_slice(&owned.blob);
+ let canonical = Message::try_from(buffer).expect("request message is valid");
+ let encrypted = encrypt_batch_request(canonical, encryptor).expect("encrypt batch");
+ let record = &encrypted.as_slice()[header_size..encrypted.header().size as usize];
+ Owned::<MESSAGE_ALIGN>::copy_from_slice(record).into()
+ }
+
+ #[test]
+ fn polled_reply_encrypted_records_take_the_flattening_path() {
+ let encryptor =
+ EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(&[7u8; 32]).expect("valid 32B key"));
+ let fragments = PollFragments::from_iter([Fragment::whole(encrypted_record(&encryptor))]);
+ let expected = flattened_reply(fragments.clone(), Some(&encryptor));
+ let reply = vectored_reply(fragments, Some(&encryptor)).expect("decrypting reply");
+ assert_eq!(reply.fragments().len(), 1);
+ assert_eq!(reply.into_contiguous().as_slice(), expected.as_slice());
+ assert_eq!(polled_count(&expected), 2);
+ }
}
diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs
index ec7ff6ba0..c6e428b77 100644
--- a/core/server_common/src/consensus_message.rs
+++ b/core/server_common/src/consensus_message.rs
@@ -17,6 +17,7 @@
use crate::iobuf::{Frozen, Owned};
use crate::sharding::METADATA_GROUP;
+use aligned_vec::{AVec, ConstAlign};
use iggy_binary_protocol::{
Command, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader,
ForwardLogoutHeader, ForwardLogoutResultHeader, ForwardRegisterHeader,
@@ -35,6 +36,12 @@ use std::{
pub const MESSAGE_ALIGN: usize = 4096;
+/// Fragment list behind a [`ResponseBacking`]. Inline for the single-buffer
+/// frame every reply but a poll is, so the per-connection mailboxes and the
+/// inter-shard reply lane stay one word wider than a bare [`Frozen`]; a
+/// vectored poll reply spills its fragment table to the heap once.
+pub type ResponseFragments = SmallVec<[Frozen<MESSAGE_ALIGN>; 1]>;
+
pub trait MessageBacking<H>
where
H: ConsensusHeader,
@@ -73,14 +80,68 @@ pub struct RequestBacking {
owned: Owned<MESSAGE_ALIGN>,
}
+/// An outbound frame as a list of buffers written back to back: the wire bytes
+/// are the concatenation of `fragments`. Never empty; the first fragment holds
+/// the whole frame header.
#[derive(Debug, Clone)]
pub struct ResponseBacking {
- fragments: SmallVec<[Frozen<MESSAGE_ALIGN>; 4]>,
+ fragments: ResponseFragments,
}
impl RequestBackingKind for RequestBacking {}
impl ResponseBackingKind for ResponseBacking {}
+impl ResponseBacking {
+ /// The fragment carrying the frame header.
+ #[must_use]
+ pub fn first(&self) -> &Frozen<MESSAGE_ALIGN> {
+ self.fragments
+ .first()
+ .expect("response backing is never empty")
+ }
+
+ #[must_use]
+ pub fn fragments(&self) -> &[Frozen<MESSAGE_ALIGN>] {
+ &self.fragments
+ }
+
+ #[must_use]
+ pub fn into_fragments(self) -> ResponseFragments {
+ self.fragments
+ }
+
+ #[must_use]
+ pub fn total_len(&self) -> usize {
+ self.fragments.iter().map(Frozen::len).sum()
+ }
+
+ /// The frame as one buffer: the single fragment as is, or the fragments
+ /// copied back to back. For writers whose record layer needs a contiguous
+ /// payload (WebSocket frames, in-process reply decoding).
+ #[must_use]
+ pub fn into_contiguous(self) -> Frozen<MESSAGE_ALIGN> {
+ match self.fragments.as_slice() {
+ [single] => single.clone(),
+ fragments => {
+ let mut joined: AVec<u8, ConstAlign<MESSAGE_ALIGN>> =
+ AVec::with_capacity(MESSAGE_ALIGN, self.total_len());
+ for fragment in fragments {
+ joined.extend_from_slice(fragment);
+ }
+ Owned::from(joined).into()
+ }
+ }
+ }
+}
+
+impl From<Frozen<MESSAGE_ALIGN>> for ResponseBacking {
+ fn from(frozen: Frozen<MESSAGE_ALIGN>) -> Self {
+ Self {
+ fragments: smallvec::smallvec![frozen],
+ }
+ }
+}
+
impl RequestBacking {
fn into_owned(self) -> Owned<MESSAGE_ALIGN> {
self.owned
@@ -503,13 +564,13 @@ where
}
}
-impl<H> TryFrom<SmallVec<[Frozen<MESSAGE_ALIGN>; 4]>> for Message<H, ResponseBacking>
+impl<H> TryFrom<ResponseFragments> for Message<H, ResponseBacking>
where
H: ConsensusHeader,
{
type Error = ConsensusError;
- fn try_from(fragments: SmallVec<[Frozen<MESSAGE_ALIGN>; 4]>) -> Result<Self, Self::Error> {
+ fn try_from(fragments: ResponseFragments) -> Result<Self, Self::Error> {
let Some(first) = fragments.first() else {
return Err(ConsensusError::InvalidCommand {
expected: H::COMMAND,
@@ -1600,7 +1661,7 @@ mod tests {
fn response_backing_single_fragment_roundtrip() {
let owned = header_bytes(Command::Reply, 256);
let frozen: Frozen<MESSAGE_ALIGN> = owned.into();
- let fragments: smallvec::SmallVec<[Frozen<MESSAGE_ALIGN>; 4]> = smallvec![frozen];
+ let fragments: ResponseFragments = smallvec![frozen];
let msg = Message::<ReplyHeader, ResponseBacking>::try_from(fragments).expect("valid");
assert_eq!(msg.header().command, Command::Reply);
assert_eq!(msg.fragments().len(), 1);
@@ -1608,7 +1669,7 @@ mod tests {
#[test]
fn response_backing_empty_fragments_returns_err() {
- let fragments: smallvec::SmallVec<[Frozen<MESSAGE_ALIGN>; 4]> = smallvec![];
+ let fragments: ResponseFragments = smallvec![];
let result = Message::<ReplyHeader, ResponseBacking>::try_from(fragments);
assert!(matches!(result, Err(ConsensusError::InvalidCommand { .. })));
}
@@ -1617,7 +1678,7 @@ mod tests {
fn response_backing_first_fragment_too_short_returns_err() {
let owned = Owned::<MESSAGE_ALIGN>::zeroed(100);
let frozen: Frozen<MESSAGE_ALIGN> = owned.into();
- let fragments: smallvec::SmallVec<[Frozen<MESSAGE_ALIGN>; 4]> = smallvec![frozen];
+ let fragments: ResponseFragments = smallvec![frozen];
let result = Message::<ReplyHeader, ResponseBacking>::try_from(fragments);
assert!(matches!(result, Err(ConsensusError::InvalidCommand { .. })));
}
@@ -1628,7 +1689,7 @@ mod tests {
// the header; the floor must reject before any consumer slices a body.
let owned = header_bytes(Command::Reply, size_of::<ReplyHeader>() as u32 - 1);
let frozen: Frozen<MESSAGE_ALIGN> = owned.into();
- let fragments: smallvec::SmallVec<[Frozen<MESSAGE_ALIGN>; 4]> = smallvec![frozen];
+ let fragments: ResponseFragments = smallvec![frozen];
let result = Message::<ReplyHeader, ResponseBacking>::try_from(fragments);
assert!(matches!(result, Err(ConsensusError::InvalidCommand { .. })));
}
diff --git a/core/server_common/src/iobuf.rs b/core/server_common/src/iobuf.rs
index 3c793ec98..2337a8165 100644
--- a/core/server_common/src/iobuf.rs
+++ b/core/server_common/src/iobuf.rs
@@ -23,6 +23,12 @@ use std::ptr::NonNull;
use std::slice;
use std::sync::atomic::{AtomicUsize, Ordering, fence};
+/// Linux `IOV_MAX`: the most iovecs one vectored IO syscall accepts.
+/// Vectored writers must chunk their buffer lists at this many entries or
+/// the syscall fails: `EINVAL` from `writev(2)`/`pwritev(2)` (segment file
+/// writes), `EMSGSIZE` from `sendmsg(2)` (socket writes).
+pub const IOV_MAX: usize = 1024;
+
#[derive(Debug, Clone)]
pub struct Owned<const ALIGN: usize = 4096> {
inner: AVec<u8, ConstAlign<ALIGN>>,
@@ -115,6 +121,10 @@ impl<const ALIGN: usize> Owned<ALIGN> {
&mut self.inner
}
+ pub fn extend_from_slice(&mut self, bytes: &[u8]) {
+ self.inner.extend_from_slice(bytes);
+ }
+
pub fn split_at(self, split_at: usize) -> (Prefix<ALIGN>, Frozen<ALIGN>) {
assert!(split_at <= self.inner.len());
@@ -276,6 +286,13 @@ impl<const ALIGN: usize> Frozen<ALIGN> {
self.inner.as_slice()
}
+ /// Whether `self` and `other` refcount the same backing allocation,
+ /// regardless of their sliced windows. Any hit keeps the whole
+ /// allocation alive, not just the window.
+ pub fn shares_allocation(&self, other: &Self) -> bool {
+ self.inner.ctrlb == other.inner.ctrlb
+ }
+
pub fn split_at(self, split_at: usize) -> (Prefix<ALIGN>, Frozen<ALIGN>) {
assert!(split_at <= self.inner.len);
@@ -648,6 +665,16 @@ mod tests {
assert_eq!(prefix.as_slice(), b"abc");
}
+ #[test]
+ fn owned_extend_from_slice_fills_reserved_capacity_in_place() {
+ let mut o: Owned<A> = Owned::with_capacity(6);
+ let capacity = o.buf_capacity();
+ o.extend_from_slice(b"abc");
+ o.extend_from_slice(b"def");
+ assert_eq!(o.as_slice(), b"abcdef");
+ assert_eq!(o.buf_capacity(), capacity);
+ }
+
// Owned::split_at: shared control block / disjoint views
#[test]
diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs
index 16070aeed..9c7c1a55b 100644
--- a/core/server_common/src/lib.rs
+++ b/core/server_common/src/lib.rs
@@ -38,6 +38,7 @@ pub use certificates::generate_self_signed_certificate;
pub use consensus_message::{
ConsensusMessage, FragmentedBacking, MESSAGE_ALIGN, Message, MessageBacking, MessageBag,
MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking, ResponseBackingKind,
+ ResponseFragments,
};
pub use executor::create_shard_executor;
pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolSettings, memory_pool};
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index a2ffd20a3..8b56a415a 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -52,11 +52,11 @@ use iggy_common::variadic;
use iggy_common::{IggyError, IggyExpiry, IggyTimestamp};
use journal::superblock::{PingPongSuperblock, SuperblockStore};
use journal::{Journal, JournalHandle};
-use message_bus::MessageBus;
use message_bus::client_listener::RequestHandler;
use message_bus::fd_transfer::DupedFd;
use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
use message_bus::replica::listener::MessageHandler;
+use message_bus::{BusMessage, MessageBus};
use metadata::IggyMetadata;
use metadata::impls::metadata::StreamsFrontend;
use metadata::stm::StateMachine;
@@ -681,10 +681,7 @@ pub enum LifecycleFrame {
},
/// A shard that doesn't hold the client's TCP connection forwards a
/// client send to the owning shard (top 16 bits of `client_id`).
- ForwardClientSend {
- client_id: u128,
- msg: Frozen<MESSAGE_ALIGN>,
- },
+ ForwardClientSend { client_id: u128, msg: BusMessage },
/// A peer shard hands a metadata consensus submit (login/logout) to
/// shard 0, the metadata consensus owner. The committed op returns over
/// the `reply` sender carried in [`MetadataSubmit`]. Always addressed to
@@ -3391,7 +3388,7 @@ where
);
let frame = ShardFrame::lifecycle(LifecycleFrame::ForwardClientSend {
client_id: request_header.client,
- msg: reply.into_generic().into_frozen(),
+ msg: reply.into_generic().into_frozen().into(),
});
let Some(sender) = self.senders.get(self.id as usize) else {
return false;
diff --git a/core/simulator/src/bus.rs b/core/simulator/src/bus.rs
index 39ff8feb7..daa65c8dc 100644
--- a/core/simulator/src/bus.rs
+++ b/core/simulator/src/bus.rs
@@ -24,7 +24,8 @@ use message_bus::fd_transfer::DupedFd;
use message_bus::installer::conn_info::ClientConnMeta;
use message_bus::replica::listener::MessageHandler;
use message_bus::{
- ClientConnectionLostFn, ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn, SendError,
+ BusMessage, ClientConnectionLostFn, ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn,
+ SendError,
};
use server_common::{
MESSAGE_ALIGN, Message,
@@ -189,7 +190,7 @@ impl MessageBus for SimOutbox {
async fn send_to_client(
&self,
client_id: u128,
- data: Frozen<MESSAGE_ALIGN>,
+ data: impl Into<BusMessage>,
) -> Result<(), SendError> {
if !self.clients.borrow().contains(&client_id) {
return Err(SendError::ClientNotFound(client_id));
@@ -199,7 +200,7 @@ impl MessageBus for SimOutbox {
from_replica: Some(self.self_id),
to_replica: None,
to_client: Some(client_id),
- payload: EnvelopePayload::Client(frozen_to_message(&data)),
+ payload: EnvelopePayload::Client(frozen_to_message(&data.into().into_contiguous())),
});
Ok(())
@@ -264,7 +265,7 @@ impl MessageBus for SharedSimOutbox {
async fn send_to_client(
&self,
client_id: u128,
- data: Frozen<MESSAGE_ALIGN>,
+ data: impl Into<BusMessage>,
) -> Result<(), SendError> {
self.0.send_to_client(client_id, data).await
}