You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/08/12 03:50:09 UTC
(iggy) 02/10: perf(metadata): decode ops off the buffer instead of copying the body
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch vsr-cleanup
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6423f2227b4ae506dd139592d0ebc9c8ba6ba981
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Aug 11 16:53:50 2026 +0530
perf(metadata): decode ops off the buffer instead of copying the body
---
core/metadata/src/stm/authz.rs | 3 +--
core/metadata/src/stm/mod.rs | 25 ++++++++++-------
core/server_common/src/consensus_message.rs | 42 +++++++++++++++++++++++++++++
3 files changed, 59 insertions(+), 11 deletions(-)
diff --git a/core/metadata/src/stm/authz.rs b/core/metadata/src/stm/authz.rs
index 00ea17416..bbd09fa52 100644
--- a/core/metadata/src/stm/authz.rs
+++ b/core/metadata/src/stm/authz.rs
@@ -56,7 +56,6 @@ use iggy_binary_protocol::requests::users::ChangePasswordRequest;
use iggy_binary_protocol::{Operation, PrepareHeader, WireDecode, WireIdentifier};
use iggy_common::{IggyError, variadic};
use server_common::Message;
-use std::mem::size_of;
/// Gate a committed prepare, then apply it. A denial commits as an
/// `Unauthorized` no-op (the gate never mutates state); an allow proceeds to
@@ -137,7 +136,7 @@ pub(crate) fn authorize(
if user_id == ROOT_USER_ID {
return None;
}
- let body = &prepare.as_slice()[size_of::<PrepareHeader>()..header.size as usize];
+ let body = prepare.body();
match header.operation {
// Streams. `create_stream` is unscoped; the rest resolve the stream id.
diff --git a/core/metadata/src/stm/mod.rs b/core/metadata/src/stm/mod.rs
index 698dbd3c0..cd363cb24 100644
--- a/core/metadata/src/stm/mod.rs
+++ b/core/metadata/src/stm/mod.rs
@@ -412,18 +412,25 @@ macro_rules! collect_handlers {
fn parse(input: Self::Input) -> Result<::iggy_common::Either<Self::Cmd, Self::Input>, Self::Error> {
use ::iggy_binary_protocol::WireDecode;
use ::iggy_common::Either;
- use ::iggy_binary_protocol::{Operation, PrepareHeader};
- match input.header().operation {
+ use ::iggy_binary_protocol::Operation;
+
+ // Both scalars copied out of one header read. `header()`
+ // re-validates the bit pattern on each call, and the borrow has
+ // to end before the pass-through arm can move `input` on.
+ let (operation, timestamp) = {
+ let header = input.header();
+ (header.operation, header.timestamp)
+ };
+
+ match operation {
$(
Operation::$operation => {
- // TODO: FIXME, zero allocation operation construction.
- let header = *input.header();
- let body = ::bytes::Bytes::copy_from_slice(
- &input.as_slice()[core::mem::size_of::<PrepareHeader>()..header.size as usize]
- );
- let cmd = [<$operation Request>]::decode_from(&body)
+ // Decoded straight off the backing buffer. Every field a
+ // request keeps is owned, so nothing borrows past this
+ // call and the body never needs a copy of its own.
+ let cmd = [<$operation Request>]::decode_from(input.body())
.map_err(|_| ::iggy_common::IggyError::InvalidCommand)?;
- let ts = ::iggy_common::IggyTimestamp::from(header.timestamp);
+ let ts = ::iggy_common::IggyTimestamp::from(timestamp);
Ok(Either::Left([<$state Command>]::$operation(cmd, ts)))
},
)*
diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs
index 4fe65638b..5bf824fd8 100644
--- a/core/server_common/src/consensus_message.rs
+++ b/core/server_common/src/consensus_message.rs
@@ -326,6 +326,24 @@ where
<RequestBacking as MutableBacking<H>>::as_slice(&self.backing)
}
+ /// The frame body: the bytes after the header, up to the header's `size`.
+ /// Empty for a header-only frame.
+ ///
+ /// Borrowed from the backing buffer rather than copied out of it. A
+ /// `WireDecode` parse takes `&[u8]` and keeps only the fields it decodes, so
+ /// copying the body first buys nothing and costs a memcpy per frame on paths
+ /// that run per replicated op (`metadata::stm`'s apply, `metadata::stm::authz`).
+ ///
+ /// # Panics
+ /// If `size` does not span the header or overruns the buffer.
+ /// [`TryFrom<Owned>`](Message::try_from) rejects both, so every received frame
+ /// satisfies this; a buffer from [`Message::new`] must have its header stamped
+ /// first.
+ #[must_use]
+ pub fn body(&self) -> &[u8] {
+ &self.as_slice()[size_of::<H>()..self.header().size() as usize]
+ }
+
pub fn as_mut_slice(&mut self) -> &mut [u8] {
<RequestBacking as MutableBacking<H>>::as_mut_slice(&mut self.backing)
}
@@ -795,6 +813,30 @@ mod tests {
assert_eq!(generic.total_len(), 256);
}
+ // body(): the slice every wire decode reads from
+
+ #[test]
+ fn body_is_the_bytes_between_the_header_and_the_frame_size() {
+ // A 512-byte allocation holding a 260-byte frame: the accessor follows the
+ // header's `size`, never the buffer that happens to hold it.
+ const BODY: [u8; 4] = [1, 2, 3, 4];
+ let frame_size = size_of::<GenericHeader>() + BODY.len();
+ let mut owned = header_bytes_sized(Command2::Prepare, frame_size as u32, 512);
+ owned.as_mut_slice()[size_of::<GenericHeader>()..frame_size].copy_from_slice(&BODY);
+
+ let message = Message::<GenericHeader>::try_from(owned).expect("valid generic");
+ assert_eq!(message.body(), BODY);
+ }
+
+ #[test]
+ fn body_is_empty_for_a_header_only_frame() {
+ // Header-only commands go through the same accessor, so `size` equal to the
+ // header must yield an empty slice rather than an inverted-range panic.
+ let owned = header_bytes_sized(Command2::Prepare, size_of::<GenericHeader>() as u32, 512);
+ let message = Message::<GenericHeader>::try_from(owned).expect("valid generic");
+ assert!(message.body().is_empty());
+ }
+
// try_as_typed: validation gates the unsafe ptr-cast reborrow
#[test]