This is an automated email from the ASF dual-hosted git repository.
numinnex 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 90ce2da8e perf(common): write message header directly into target
buffer (#3707)
90ce2da8e is described below
commit 90ce2da8e5a7c89e9b6a5a8eb2303cc089019bd0
Author: Ryan Huang <[email protected]>
AuthorDate: Tue Jul 21 01:07:51 2026 +0800
perf(common): write message header directly into target buffer (#3707)
---
core/common/src/types/message/iggy_message.rs | 46 +++++++++++++-
core/common/src/types/message/message_header.rs | 80 ++++++++++++++++++++++---
2 files changed, 115 insertions(+), 11 deletions(-)
diff --git a/core/common/src/types/message/iggy_message.rs
b/core/common/src/types/message/iggy_message.rs
index 645f6ec79..7fe3a3a7c 100644
--- a/core/common/src/types/message/iggy_message.rs
+++ b/core/common/src/types/message/iggy_message.rs
@@ -359,8 +359,7 @@ impl IggyMessage {
pub fn to_bytes(&self) -> Bytes {
let mut bytes =
BytesMut::with_capacity(self.get_size_bytes().as_bytes_usize());
- let message_header = self.header.to_bytes();
- bytes.put_slice(&message_header);
+ self.header.write_to(&mut bytes);
bytes.put_slice(&self.payload);
if let Some(user_headers) = &self.user_headers {
bytes.put_slice(user_headers);
@@ -418,8 +417,13 @@ impl IggyMessage {
})
}
+ /// Serializes the message into `buf`.
+ ///
+ /// `buf` need not be preallocated: the whole message is reserved up front
so
+ /// an empty buffer grows once rather than once per appended field.
pub fn write_to_buffer(&self, buf: &mut BytesMut) {
- buf.put_slice(&self.header.to_bytes());
+ buf.reserve(self.get_size_bytes().as_bytes_usize());
+ self.header.write_to(buf);
buf.put_slice(&self.payload);
if let Some(user_headers) = &self.user_headers {
buf.put_slice(user_headers);
@@ -844,4 +848,40 @@ mod tests {
deserialized_map.get(&HeaderKey::try_from("correlation-id").unwrap())
);
}
+
+ #[test]
+ fn given_empty_destination_buffer_when_written_should_grow_exactly_once() {
+ let message = IggyMessage::builder()
+ .payload(Bytes::from(vec![b'x'; 1024]))
+ .build()
+ .unwrap();
+ let size = message.get_size_bytes().as_bytes_usize();
+
+ let mut buf = BytesMut::new();
+ message.write_to_buffer(&mut buf);
+
+ assert_eq!(buf.len(), size);
+ // A single reservation covers the whole message, so the buffer never
+ // reallocates part-way through the individually appended header
fields.
+ assert!(buf.capacity() >= size);
+
+ let mut second = BytesMut::new();
+ message.write_to_buffer(&mut second);
+ assert_eq!(&buf[..], &second[..]);
+ }
+
+ #[test]
+ fn
given_buffer_with_existing_content_when_written_should_append_without_disturbing_it()
{
+ let message = IggyMessage::builder()
+ .payload(Bytes::from_static(b"payload"))
+ .build()
+ .unwrap();
+
+ let mut buf = BytesMut::new();
+ buf.put_slice(b"existing");
+ message.write_to_buffer(&mut buf);
+
+ assert_eq!(&buf[..8], b"existing");
+ assert_eq!(&buf[8..], &message.to_bytes()[..]);
+ }
}
diff --git a/core/common/src/types/message/message_header.rs
b/core/common/src/types/message/message_header.rs
index cf2a66829..fa776bbb7 100644
--- a/core/common/src/types/message/message_header.rs
+++ b/core/common/src/types/message/message_header.rs
@@ -107,16 +107,25 @@ impl IggyMessageHeader {
})
}
+ /// Serializes the header directly into `buf`, without an intermediate
allocation.
+ ///
+ /// Reserving up front keeps an unsized `buf` to a single growth, since the
+ /// fields are appended individually.
+ pub fn write_to(&self, buf: &mut BytesMut) {
+ buf.reserve(IGGY_MESSAGE_HEADER_SIZE);
+ buf.put_u64_le(self.checksum);
+ buf.put_u128_le(self.id);
+ buf.put_u64_le(self.offset);
+ buf.put_u64_le(self.timestamp);
+ buf.put_u64_le(self.origin_timestamp);
+ buf.put_u32_le(self.user_headers_length);
+ buf.put_u32_le(self.payload_length);
+ buf.put_u64_le(self.reserved);
+ }
+
pub fn to_bytes(&self) -> Bytes {
let mut bytes =
BytesMut::with_capacity(self.get_size_bytes().as_bytes_usize());
- bytes.put_u64_le(self.checksum);
- bytes.put_u128_le(self.id);
- bytes.put_u64_le(self.offset);
- bytes.put_u64_le(self.timestamp);
- bytes.put_u64_le(self.origin_timestamp);
- bytes.put_u32_le(self.user_headers_length);
- bytes.put_u32_le(self.payload_length);
- bytes.put_u64_le(self.reserved);
+ self.write_to(&mut bytes);
bytes.freeze()
}
@@ -214,6 +223,61 @@ mod tests {
assert_eq!(header, deserialized);
}
+ #[test]
+ fn
given_populated_header_when_written_into_shared_buffer_should_append_exact_layout()
{
+ let header = IggyMessageHeader {
+ checksum: u64::MAX,
+ id: u128::MAX,
+ offset: 42,
+ timestamp: 7,
+ origin_timestamp: 9,
+ user_headers_length: u32::MAX,
+ payload_length: 13,
+ reserved: 0,
+ };
+
+ // Built independently of `to_bytes`, which now shares `write_to`'s
+ // implementation and so cannot serve as an oracle for it.
+ let mut expected = Vec::new();
+ expected.extend_from_slice(&header.checksum.to_le_bytes());
+ expected.extend_from_slice(&header.id.to_le_bytes());
+ expected.extend_from_slice(&header.offset.to_le_bytes());
+ expected.extend_from_slice(&header.timestamp.to_le_bytes());
+ expected.extend_from_slice(&header.origin_timestamp.to_le_bytes());
+ expected.extend_from_slice(&header.user_headers_length.to_le_bytes());
+ expected.extend_from_slice(&header.payload_length.to_le_bytes());
+ expected.extend_from_slice(&header.reserved.to_le_bytes());
+
+ let mut buf = BytesMut::new();
+ buf.put_slice(b"prefix");
+ header.write_to(&mut buf);
+
+ assert_eq!(&buf[..6], b"prefix");
+ assert_eq!(&buf[6..], &expected[..]);
+ assert_eq!(expected.len(), IGGY_MESSAGE_HEADER_SIZE);
+ }
+
+ #[test]
+ fn given_header_written_at_offset_when_parsed_back_should_round_trip() {
+ let header = IggyMessageHeader {
+ checksum: 1,
+ id: 2,
+ offset: 3,
+ timestamp: 4,
+ origin_timestamp: 5,
+ user_headers_length: 6,
+ payload_length: 7,
+ reserved: 0,
+ };
+
+ let mut buf = BytesMut::new();
+ header.write_to(&mut buf);
+ header.write_to(&mut buf);
+
+ let second = Bytes::copy_from_slice(&buf[IGGY_MESSAGE_HEADER_SIZE..]);
+ assert_eq!(IggyMessageHeader::from_bytes(second).unwrap(), header);
+ }
+
#[test]
fn should_serialize_header_to_correct_size() {
let header = IggyMessageHeader::default();
