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 2025/04/12 17:09:10 UTC
(iggy) branch zero-copy-no-batching-rebase-3 updated (c58ef05a -> dc35cb42)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch zero-copy-no-batching-rebase-3
in repository https://gitbox.apache.org/repos/asf/iggy.git
from c58ef05a feat(server): implement zero-copy message handling
new 476d72e3 even more improvements
new b8bdbc98 implement BytesMutPool
new dc35cb42 fix base_position for IggyIndexesMut
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
Cargo.lock | 24 ++
sdk/src/http/messages.rs | 1 +
sdk/src/identifier.rs | 5 +
sdk/src/messages/partitioning.rs | 5 +
sdk/src/messages/send_messages.rs | 38 +-
sdk/src/models/messaging/messages_batch.rs | 191 +---------
server/Cargo.toml | 2 +
server/src/binary/command.rs | 2 -
.../handlers/messages/send_messages_handler.rs | 72 ++--
server/src/channels/commands/print_sysinfo.rs | 7 +-
server/src/http/jwt/storage.rs | 7 +-
server/src/http/messages.rs | 6 +-
server/src/main.rs | 3 +
.../src/streaming/segments/indexes/index_reader.rs | 24 +-
.../src/streaming/segments/indexes/index_writer.rs | 7 +-
.../src/streaming/segments/indexes/indexes_mut.rs | 53 ++-
.../streaming/segments/messages/messages_reader.rs | 39 +-
.../streaming/segments/messages/messages_writer.rs | 4 +-
server/src/streaming/segments/messages/mod.rs | 2 +-
.../streaming/segments/messages/persister_task.rs | 2 +-
.../src/streaming/segments/messages_accumulator.rs | 11 +-
server/src/streaming/segments/reading_messages.rs | 11 +-
server/src/streaming/segments/segment.rs | 6 +-
.../streaming/segments/types/messages_batch_mut.rs | 404 +++++++++++++++------
.../streaming/segments/types/messages_batch_set.rs | 92 +++--
server/src/streaming/segments/writing_messages.rs | 18 +-
server/src/streaming/systems/messages.rs | 14 +-
server/src/streaming/systems/storage.rs | 10 +-
server/src/streaming/utils/bytes_mut_pool.rs | 308 ++++++++++++++++
server/src/streaming/utils/mod.rs | 1 +
30 files changed, 883 insertions(+), 486 deletions(-)
create mode 100644 server/src/streaming/utils/bytes_mut_pool.rs
(iggy) 01/03: even more improvements
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch zero-copy-no-batching-rebase-3
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 476d72e3b001c1edce19784c1a107fdfe6424662
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Sat Apr 12 10:45:01 2025 +0200
even more improvements
---
server/src/streaming/segments/indexes/index_writer.rs | 7 +++----
server/src/streaming/segments/messages/messages_writer.rs | 4 ++--
server/src/streaming/segments/messages/mod.rs | 2 +-
server/src/streaming/segments/messages/persister_task.rs | 2 +-
server/src/streaming/segments/segment.rs | 6 ++----
server/src/streaming/segments/types/messages_batch_mut.rs | 13 -------------
server/src/streaming/segments/types/messages_batch_set.rs | 8 ++++----
server/src/streaming/segments/writing_messages.rs | 10 ++++++----
8 files changed, 19 insertions(+), 33 deletions(-)
diff --git a/server/src/streaming/segments/indexes/index_writer.rs b/server/src/streaming/segments/indexes/index_writer.rs
index 56476a3a..9fd9525e 100644
--- a/server/src/streaming/segments/indexes/index_writer.rs
+++ b/server/src/streaming/segments/indexes/index_writer.rs
@@ -104,13 +104,12 @@ impl IndexWriter {
})
.map_err(|_| IggyError::CannotSaveIndexToSegment)?;
- if self.fsync {
- let _ = self.fsync().await;
- }
-
self.index_size_bytes
.fetch_add(indexes.len() as u64, Ordering::Release);
+ if self.fsync {
+ let _ = self.fsync().await;
+ }
trace!(
"Saved {count} indexes of size {} to file: {}",
INDEX_SIZE * count,
diff --git a/server/src/streaming/segments/messages/messages_writer.rs b/server/src/streaming/segments/messages/messages_writer.rs
index 018b4075..bb51f8b4 100644
--- a/server/src/streaming/segments/messages/messages_writer.rs
+++ b/server/src/streaming/segments/messages/messages_writer.rs
@@ -107,7 +107,7 @@ impl MessagesWriter {
/// Append a batch of messages to the messages file.
pub async fn save_batch_set(
&mut self,
- batch_set: &IggyMessagesBatchSet,
+ batch_set: IggyMessagesBatchSet,
confirmation: Confirmation,
) -> Result<IggyByteSize, IggyError> {
let messages_size = batch_set.size();
@@ -147,7 +147,7 @@ impl MessagesWriter {
}
Confirmation::NoWait => {
if let Some(task) = &self.persister_task {
- task.persist(batch_set.clone()).await;
+ task.persist(batch_set).await;
} else {
panic!(
"Confirmation::NoWait is used, but MessagesPersisterTask is not set for messages file: {}",
diff --git a/server/src/streaming/segments/messages/mod.rs b/server/src/streaming/segments/messages/mod.rs
index ce102d72..7cfcdb49 100644
--- a/server/src/streaming/segments/messages/mod.rs
+++ b/server/src/streaming/segments/messages/mod.rs
@@ -34,7 +34,7 @@ pub use persister_task::PersisterTask;
async fn write_batch(
file: &mut File,
file_path: &str,
- batches: &IggyMessagesBatchSet,
+ batches: IggyMessagesBatchSet,
) -> Result<usize, IggyError> {
let mut slices: Vec<IoSlice> = batches.iter().map(|b| IoSlice::new(b)).collect();
diff --git a/server/src/streaming/segments/messages/persister_task.rs b/server/src/streaming/segments/messages/persister_task.rs
index c3482970..887913d2 100644
--- a/server/src/streaming/segments/messages/persister_task.rs
+++ b/server/src/streaming/segments/messages/persister_task.rs
@@ -175,7 +175,7 @@ impl PersisterTask {
while let Ok(request) = receiver.recv_async().await {
match request {
PersisterTaskCommand::WriteRequest(messages) => {
- match write_batch(&mut file, &file_path, &messages).await {
+ match write_batch(&mut file, &file_path, messages).await {
Ok(bytes_written) => {
if fsync {
file.sync_all()
diff --git a/server/src/streaming/segments/segment.rs b/server/src/streaming/segments/segment.rs
index e691be19..026834ae 100644
--- a/server/src/streaming/segments/segment.rs
+++ b/server/src/streaming/segments/segment.rs
@@ -31,8 +31,6 @@ use std::sync::Arc;
use tokio::fs::remove_file;
use tracing::{info, warn};
-const INDEXES_CAPACITY: usize = 20 * 1000 * 1000; // 24 MB, 15 million indexes (16 bytes each) for 15 million messages of 1kB
-
#[derive(Debug)]
pub struct Segment {
pub(super) stream_id: u32,
@@ -102,7 +100,7 @@ impl Segment {
last_index_position: 0,
max_size_bytes: config.segment.size,
message_expiry,
- indexes: IggyIndexesMut::with_capacity(INDEXES_CAPACITY, 0),
+ indexes: IggyIndexesMut::with_capacity(1_000_000, 0),
accumulator: MessagesAccumulator::default(),
is_closed: false,
messages_writer: None,
@@ -402,9 +400,9 @@ impl Segment {
self.end_offset = end_offset;
}
+ /// Explicitly drop the old indexes to ensure memory is freed
pub fn drop_indexes(&mut self) {
let old_indexes = std::mem::replace(&mut self.indexes, IggyIndexesMut::empty());
- // Explicitly drop the old indexes to ensure memory is freed
drop(old_indexes);
}
}
diff --git a/server/src/streaming/segments/types/messages_batch_mut.rs b/server/src/streaming/segments/types/messages_batch_mut.rs
index be09baa0..60af072b 100644
--- a/server/src/streaming/segments/types/messages_batch_mut.rs
+++ b/server/src/streaming/segments/types/messages_batch_mut.rs
@@ -49,19 +49,6 @@ impl Sizeable for IggyMessagesBatchMut {
}
impl IggyMessagesBatchMut {
- /// Creates an empty messages container with the specified capacity to avoid reallocations.
- ///
- /// # Arguments
- ///
- /// * `bytes_capacity` - The expected total size of all messages in bytes
- pub fn with_capacity(bytes_capacity: usize) -> Self {
- let index_capacity = bytes_capacity / INDEX_SIZE + 1; // Add 1 to avoid rounding down to 0
- Self {
- indexes: IggyIndexesMut::with_capacity(index_capacity, 0),
- messages: BytesMut::with_capacity(bytes_capacity),
- }
- }
-
/// Creates a new messages container from existing index and message buffers.
///
/// # Arguments
diff --git a/server/src/streaming/segments/types/messages_batch_set.rs b/server/src/streaming/segments/types/messages_batch_set.rs
index 6230ac5a..6f32b383 100644
--- a/server/src/streaming/segments/types/messages_batch_set.rs
+++ b/server/src/streaming/segments/types/messages_batch_set.rs
@@ -26,7 +26,7 @@ use std::ops::Index;
use tracing::trace;
/// A container for multiple IggyMessagesBatch objects
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default)]
pub struct IggyMessagesBatchSet {
/// The collection of message containers
batches: Vec<IggyMessagesBatch>,
@@ -221,8 +221,8 @@ impl IggyMessagesBatchSet {
}
let first_offset = container.first_offset();
- if first_offset.is_some()
- && first_offset.unwrap() + container.count() as u64 <= start_offset
+ if first_offset.is_none()
+ || first_offset.unwrap() + container.count() as u64 <= start_offset
{
continue;
}
@@ -256,7 +256,7 @@ impl IggyMessagesBatchSet {
}
let first_timestamp = container.first_timestamp();
- if first_timestamp.is_some() && first_timestamp.unwrap() < timestamp {
+ if first_timestamp.is_none() || first_timestamp.unwrap() < timestamp {
continue;
}
diff --git a/server/src/streaming/segments/writing_messages.rs b/server/src/streaming/segments/writing_messages.rs
index e1e52701..7e80574b 100644
--- a/server/src/streaming/segments/writing_messages.rs
+++ b/server/src/streaming/segments/writing_messages.rs
@@ -90,6 +90,8 @@ impl Segment {
None => self.config.segment.server_confirmation,
};
+ batches.extract_indexes_to(&mut self.indexes);
+
let batch_size = batches.size();
let batch_count = batches.count();
@@ -97,7 +99,7 @@ impl Segment {
.messages_writer
.as_mut()
.expect("Messages writer not initialized")
- .save_batch_set(&batches, confirmation)
+ .save_batch_set(batches, confirmation)
.await
.with_error_context(|error| {
format!(
@@ -107,8 +109,6 @@ impl Segment {
self.last_index_position += saved_bytes.as_bytes_u64() as u32;
- batches.extract_indexes_to(&mut self.indexes);
-
let unsaved_indexes_slice = self.indexes.unsaved_slice();
self.index_writer
.as_mut()
@@ -168,7 +168,9 @@ impl Segment {
current_segment_size,
max_segment_size_from_config
);
- if self.config.segment.cache_indexes == CacheIndexesConfig::OpenSegment {
+ if self.config.segment.cache_indexes == CacheIndexesConfig::OpenSegment
+ || self.config.segment.cache_indexes == CacheIndexesConfig::None
+ {
self.drop_indexes();
}
self.shutdown_writing().await;
(iggy) 03/03: fix base_position for IggyIndexesMut
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch zero-copy-no-batching-rebase-3
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit dc35cb42c31cd1023c95c1a30c3e5219b1bf4ef6
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Sat Apr 12 18:53:58 2025 +0200
fix base_position for IggyIndexesMut
---
sdk/src/models/messaging/messages_batch.rs | 28 ---------
.../src/streaming/segments/indexes/index_reader.rs | 2 +-
.../src/streaming/segments/indexes/indexes_mut.rs | 1 +
.../streaming/segments/types/messages_batch_mut.rs | 73 ++++------------------
server/src/streaming/utils/bytes_mut_pool.rs | 15 ++---
5 files changed, 18 insertions(+), 101 deletions(-)
diff --git a/sdk/src/models/messaging/messages_batch.rs b/sdk/src/models/messaging/messages_batch.rs
index 911ff723..f42883e5 100644
--- a/sdk/src/models/messaging/messages_batch.rs
+++ b/sdk/src/models/messaging/messages_batch.rs
@@ -165,34 +165,6 @@ impl IggyMessagesBatch {
None
}
}
-
- /// Consumes the batch and returns a vector of messages.
- pub fn into_messages(self) -> Vec<IggyMessage> {
- if self.is_empty() {
- return Vec::new();
- }
-
- let mut result = Vec::with_capacity(self.count as usize);
-
- for i in 0..self.count as usize {
- if let Some((start, end)) = self.get_message_boundaries(i) {
- if let Ok(message) = IggyMessage::from_bytes(self.messages.slice(start..end)) {
- result.push(message);
- }
- }
- }
-
- result
- }
-}
-
-impl IntoIterator for IggyMessagesBatch {
- type Item = IggyMessage;
- type IntoIter = std::vec::IntoIter<IggyMessage>;
-
- fn into_iter(self) -> Self::IntoIter {
- self.into_messages().into_iter()
- }
}
impl Index<usize> for IggyMessagesBatch {
diff --git a/server/src/streaming/segments/indexes/index_reader.rs b/server/src/streaming/segments/indexes/index_reader.rs
index cf0d0438..f60e0d32 100644
--- a/server/src/streaming/segments/indexes/index_reader.rs
+++ b/server/src/streaming/segments/indexes/index_reader.rs
@@ -103,7 +103,7 @@ impl IndexReader {
}
};
let index_count = file_size / INDEX_SIZE as u32;
- let indexes = IggyIndexesMut::from_bytes(buf, index_count);
+ let indexes = IggyIndexesMut::from_bytes(buf, 0);
if indexes.count() != index_count {
error!(
"Loaded {} indexes from disk, expected {}, file {} is probably corrupted!",
diff --git a/server/src/streaming/segments/indexes/indexes_mut.rs b/server/src/streaming/segments/indexes/indexes_mut.rs
index da89a334..779279fd 100644
--- a/server/src/streaming/segments/indexes/indexes_mut.rs
+++ b/server/src/streaming/segments/indexes/indexes_mut.rs
@@ -70,6 +70,7 @@ impl IggyIndexesMut {
/// Sets the base position of the indexes
pub fn set_base_position(&mut self, base_position: u32) {
+ tracing::error!("Setting base position to {base_position}");
self.base_position = base_position;
}
diff --git a/server/src/streaming/segments/types/messages_batch_mut.rs b/server/src/streaming/segments/types/messages_batch_mut.rs
index 87c87c88..b336b011 100644
--- a/server/src/streaming/segments/types/messages_batch_mut.rs
+++ b/server/src/streaming/segments/types/messages_batch_mut.rs
@@ -84,8 +84,8 @@ impl IggyMessagesBatchMut {
/// Creates a new messages container from a slice of IggyMessage objects.
///
/// # Note
- /// This function should be used only for testing purposes, because it creates
- /// deep copies of the messages.
+ /// This function should be used only for testing purposes,
+ /// because it creates deep copies of the messages.
///
/// # Arguments
///
@@ -160,7 +160,6 @@ impl IggyMessagesBatchMut {
let mut invalid_messages_indexes =
deduplicator.map(|_| Vec::with_capacity(messages_count as usize));
- // let (mut indexes, mut messages) = self.decompose();
self.indexes.set_base_position(current_position);
let mut iter: IggyMessageViewMutIterator<'_> =
IggyMessageViewMutIterator::new(&mut self.messages);
@@ -175,7 +174,7 @@ impl IggyMessagesBatchMut {
if let Some(deduplicator) = deduplicator {
if !deduplicator.try_insert(message.header().id()).await {
- warn!(
+ eprintln!(
"Detected duplicate message ID {}, removing...",
message.header().id()
);
@@ -528,7 +527,7 @@ impl IggyMessagesBatchMut {
}
let msg_count = self.count() as usize;
- if indexes_to_remove.len() >= msg_count {
+ if indexes_to_remove.len() > msg_count {
return;
}
@@ -544,6 +543,8 @@ impl IggyMessagesBatchMut {
})
.collect();
+ eprintln!("boundaries_to_remove: {boundaries_to_remove:?}");
+
assert_eq!(
boundaries_to_remove.len(),
indexes_to_remove.len(),
@@ -599,6 +600,9 @@ impl IggyMessagesBatchMut {
chunk_len,
);
}
+
+ self.messages = new_buffer;
+ self.indexes = new_indexes;
}
/// Validates that all messages in batch have correct checksums.
@@ -787,7 +791,8 @@ impl Validatable<IggyError> for IggyMessagesBatchMut {
}
self.validate_indexes_structure()?;
- self.validate_message_contents()
+ self.validate_message_contents()?;
+ Ok(())
}
}
@@ -827,59 +832,3 @@ impl Deref for IggyMessagesBatchMut {
&self.messages
}
}
-
-/// Iterator over messages batch that yields immutable message views
-pub struct IggyMessageViewIterator<'a> {
- buffer: &'a BytesMut,
- position: usize,
-}
-
-impl<'a> IggyMessageViewIterator<'a> {
- pub fn new(buffer: &'a BytesMut) -> Self {
- Self {
- buffer,
- position: 0,
- }
- }
-}
-
-impl<'a> Iterator for IggyMessageViewIterator<'a> {
- type Item = IggyMessageView<'a>;
-
- fn next(&mut self) -> Option<Self::Item> {
- if self.position >= self.buffer.len() {
- return None;
- }
-
- let remaining = &self.buffer[self.position..];
- if remaining.len() < IGGY_MESSAGE_HEADER_SIZE {
- error!(
- "Buffer too small for message header at position {}, buffer len: {}",
- self.position,
- self.buffer.len()
- );
- return None;
- }
-
- let header_view = IggyMessageHeaderView::new(remaining);
- let message_size = header_view.payload_length()
- + header_view.user_headers_length()
- + IGGY_MESSAGE_HEADER_SIZE;
-
- if message_size > remaining.len() {
- error!(
- "Message size {} exceeds remaining buffer size {} at position {}",
- message_size,
- remaining.len(),
- self.position
- );
- return None;
- }
-
- let message_view =
- IggyMessageView::new(&self.buffer[self.position..self.position + message_size]);
- self.position += message_size;
-
- Some(message_view)
- }
-}
diff --git a/server/src/streaming/utils/bytes_mut_pool.rs b/server/src/streaming/utils/bytes_mut_pool.rs
index 82a8f302..1f089cac 100644
--- a/server/src/streaming/utils/bytes_mut_pool.rs
+++ b/server/src/streaming/utils/bytes_mut_pool.rs
@@ -57,11 +57,11 @@ impl BytesMutPool {
const EXTRA_LARGE_BUFFER_SIZE: usize = 2 * 1024 * 1024;
const MAX_BUFFER_SIZE: usize = 16 * 1024 * 1024;
- const SMALL_POOL_SIZE: usize = 4096;
- const MEDIUM_POOL_SIZE: usize = 1024;
- const LARGE_POOL_SIZE: usize = 512;
- const EXTRA_LARGE_POOL_SIZE: usize = 128;
- const MAX_POOL_SIZE: usize = 64;
+ const SMALL_POOL_SIZE: usize = 16384;
+ const MEDIUM_POOL_SIZE: usize = 4096;
+ const LARGE_POOL_SIZE: usize = 1024;
+ const EXTRA_LARGE_POOL_SIZE: usize = 256;
+ const MAX_POOL_SIZE: usize = 128;
/// Initialize the bytes pool
pub fn init_pool() {
@@ -153,35 +153,30 @@ impl BytesMutPool {
trace!("Small buffer pool full, dropping buffer");
} else {
self.small_returned.fetch_add(1, Ordering::Relaxed);
- trace!("Returned small buffer to pool");
}
} else if capacity == Self::MEDIUM_BUFFER_SIZE {
if self.medium_buffers.push(buffer).is_err() {
trace!("Medium buffer pool full, dropping buffer");
} else {
self.medium_returned.fetch_add(1, Ordering::Relaxed);
- trace!("Returned medium buffer to pool");
}
} else if capacity == Self::LARGE_BUFFER_SIZE {
if self.large_buffers.push(buffer).is_err() {
trace!("Large buffer pool full, dropping buffer");
} else {
self.large_returned.fetch_add(1, Ordering::Relaxed);
- trace!("Returned large buffer to pool");
}
} else if capacity == Self::EXTRA_LARGE_BUFFER_SIZE {
if self.extra_large_buffers.push(buffer).is_err() {
trace!("Extra large buffer pool full, dropping buffer");
} else {
self.extra_large_returned.fetch_add(1, Ordering::Relaxed);
- trace!("Returned extra large buffer to pool");
}
} else if capacity == Self::MAX_BUFFER_SIZE {
if self.max_buffers.push(buffer).is_err() {
trace!("Max buffer pool full, dropping buffer");
} else {
self.max_returned.fetch_add(1, Ordering::Relaxed);
- trace!("Returned max buffer to pool");
}
} else if capacity != 0 {
trace!(
(iggy) 02/03: implement BytesMutPool
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch zero-copy-no-batching-rebase-3
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b8bdbc98ca4f134470ab1571014b1083f00c96c0
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Sat Apr 12 17:38:04 2025 +0200
implement BytesMutPool
---
Cargo.lock | 24 ++
sdk/src/http/messages.rs | 1 +
sdk/src/identifier.rs | 5 +
sdk/src/messages/partitioning.rs | 5 +
sdk/src/messages/send_messages.rs | 38 ++-
sdk/src/models/messaging/messages_batch.rs | 163 +---------
server/Cargo.toml | 2 +
server/src/binary/command.rs | 2 -
.../handlers/messages/send_messages_handler.rs | 72 +++--
server/src/channels/commands/print_sysinfo.rs | 7 +-
server/src/http/jwt/storage.rs | 7 +-
server/src/http/messages.rs | 6 +-
server/src/main.rs | 3 +
.../src/streaming/segments/indexes/index_reader.rs | 24 +-
.../src/streaming/segments/indexes/indexes_mut.rs | 52 +++-
.../streaming/segments/messages/messages_reader.rs | 39 +--
.../src/streaming/segments/messages_accumulator.rs | 11 +-
server/src/streaming/segments/reading_messages.rs | 11 +-
server/src/streaming/segments/segment.rs | 2 +-
.../streaming/segments/types/messages_batch_mut.rs | 336 +++++++++++++++++----
.../streaming/segments/types/messages_batch_set.rs | 84 +++---
server/src/streaming/segments/writing_messages.rs | 12 +-
server/src/streaming/systems/messages.rs | 14 +-
server/src/streaming/systems/storage.rs | 10 +-
server/src/streaming/utils/bytes_mut_pool.rs | 313 +++++++++++++++++++
server/src/streaming/utils/mod.rs | 1 +
26 files changed, 869 insertions(+), 375 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index fa08280f..432e33c7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1231,6 +1231,19 @@ dependencies = [
"cfg-if",
]
+[[package]]
+name = "crossbeam"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8"
+dependencies = [
+ "crossbeam-channel",
+ "crossbeam-deque",
+ "crossbeam-epoch",
+ "crossbeam-queue",
+ "crossbeam-utils",
+]
+
[[package]]
name = "crossbeam-channel"
version = "0.5.14"
@@ -1259,6 +1272,15 @@ dependencies = [
"crossbeam-utils",
]
+[[package]]
+name = "crossbeam-queue"
+version = "0.3.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
+dependencies = [
+ "crossbeam-utils",
+]
+
[[package]]
name = "crossbeam-utils"
version = "0.8.21"
@@ -4870,6 +4892,7 @@ dependencies = [
"chrono",
"clap",
"console-subscriber",
+ "crossbeam",
"dashmap",
"derive_more",
"dotenvy",
@@ -4887,6 +4910,7 @@ dependencies = [
"mockall",
"moka",
"nix",
+ "once_cell",
"openssl",
"opentelemetry",
"opentelemetry-appender-tracing",
diff --git a/sdk/src/http/messages.rs b/sdk/src/http/messages.rs
index 22fd3ede..4de7bec7 100644
--- a/sdk/src/http/messages.rs
+++ b/sdk/src/http/messages.rs
@@ -71,6 +71,7 @@ impl MessageClient for HttpClient {
self.post(
&get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
&SendMessages {
+ metadata_length: 0, // this field is used only for TCP/QUIC
stream_id: stream_id.clone(),
topic_id: topic_id.clone(),
partitioning: partitioning.clone(),
diff --git a/sdk/src/identifier.rs b/sdk/src/identifier.rs
index e66d0059..cd65e318 100644
--- a/sdk/src/identifier.rs
+++ b/sdk/src/identifier.rs
@@ -200,6 +200,11 @@ impl Identifier {
identifier.validate()?;
Ok(identifier)
}
+
+ /// Maximum size of the Identifier struct
+ pub const fn maximum_byte_size() -> usize {
+ 2 + 255
+ }
}
impl Sizeable for Identifier {
diff --git a/sdk/src/messages/partitioning.rs b/sdk/src/messages/partitioning.rs
index 8e454b38..f41b8814 100644
--- a/sdk/src/messages/partitioning.rs
+++ b/sdk/src/messages/partitioning.rs
@@ -155,6 +155,11 @@ impl Partitioning {
value,
})
}
+
+ /// Maximum size of the Partitioning struct
+ pub const fn maximum_byte_size() -> usize {
+ 2 + 255
+ }
}
impl Hash for Partitioning {
diff --git a/sdk/src/messages/send_messages.rs b/sdk/src/messages/send_messages.rs
index 22a219b9..e2952c6e 100644
--- a/sdk/src/messages/send_messages.rs
+++ b/sdk/src/messages/send_messages.rs
@@ -43,6 +43,8 @@ use super::{Partitioning, PartitioningKind};
/// - `batch` - collection of messages to be sent.
#[derive(Debug, PartialEq)]
pub struct SendMessages {
+ /// Length of stream_id, topic_id, partitioning and messages_count (4 bytes)
+ pub metadata_length: u32,
/// Unique stream ID (numeric or name).
pub stream_id: Identifier,
/// Unique topic ID (numeric or name).
@@ -60,25 +62,33 @@ impl SendMessages {
partitioning: &Partitioning,
messages: &[IggyMessage],
) -> Bytes {
- let stream_id_size = stream_id.get_buffer_size();
- let topic_id_size = topic_id.get_buffer_size();
- let partitioning_size = partitioning.get_buffer_size();
+ let stream_id_field_size = stream_id.get_buffer_size();
+ let topic_id_field_size = topic_id.get_buffer_size();
+ let partitioning_field_size = partitioning.get_buffer_size();
+ let metadata_length_field_size = std::mem::size_of::<u32>();
let messages_count = messages.len();
- let messages_count_size = std::mem::size_of::<u32>();
+ let messages_count_field_size = std::mem::size_of::<u32>();
+ let metadata_length = stream_id_field_size
+ + topic_id_field_size
+ + partitioning_field_size
+ + messages_count_field_size;
let indexes_size = messages_count * INDEX_SIZE;
-
- let total_size = stream_id_size
- + topic_id_size
- + partitioning_size
- + messages_count_size
+ let messages_size = messages
+ .iter()
+ .map(|m| m.get_size_bytes().as_bytes_usize())
+ .sum::<usize>();
+
+ let total_size = metadata_length_field_size
+ + stream_id_field_size
+ + topic_id_field_size
+ + partitioning_field_size
+ + messages_count_field_size
+ indexes_size
- + messages
- .iter()
- .map(|m| m.get_size_bytes().as_bytes_usize())
- .sum::<usize>();
+ + messages_size;
let mut bytes = BytesMut::with_capacity(total_size);
+ bytes.put_u32_le(metadata_length as u32);
stream_id.write_to_buffer(&mut bytes);
topic_id.write_to_buffer(&mut bytes);
partitioning.write_to_buffer(&mut bytes);
@@ -121,6 +131,7 @@ fn write_value_at<const N: usize>(slice: &mut [u8], value: [u8; N], position: us
impl Default for SendMessages {
fn default() -> Self {
SendMessages {
+ metadata_length: 0,
stream_id: Identifier::default(),
topic_id: Identifier::default(),
partitioning: Partitioning::default(),
@@ -327,6 +338,7 @@ impl<'de> Deserialize<'de> for SendMessages {
let batch = IggyMessagesBatch::from(&messages);
Ok(SendMessages {
+ metadata_length: 0, // this field is used only for TCP/QUIC
stream_id: Identifier::default(),
topic_id: Identifier::default(),
partitioning,
diff --git a/sdk/src/models/messaging/messages_batch.rs b/sdk/src/models/messaging/messages_batch.rs
index 31851176..911ff723 100644
--- a/sdk/src/models/messaging/messages_batch.rs
+++ b/sdk/src/models/messaging/messages_batch.rs
@@ -156,116 +156,6 @@ impl IggyMessagesBatch {
}
}
- /// Returns a contiguous slice (as a new `IggyMessagesBatch`) of up to `count` messages
- /// whose message headers have an offset greater than or equal to the provided `start_offset`.
- pub fn slice_by_offset(&self, start_offset: u64, count: u32) -> Option<Self> {
- if self.is_empty() || count == 0 {
- return None;
- }
-
- let first_offset = self.first_offset()?;
-
- if start_offset < first_offset {
- return self.slice_by_index(0, count);
- }
-
- let last_offset = self.last_offset()?;
- if start_offset > last_offset {
- return None;
- }
-
- let offset_diff = start_offset - first_offset;
- let first_message_index = offset_diff as usize;
-
- if first_message_index >= self.count() as usize {
- return None;
- }
-
- self.slice_by_index(first_message_index as u32, count)
- }
-
- /// Helper method to slice the batch starting from a specific index
- fn slice_by_index(&self, start_index: u32, count: u32) -> Option<Self> {
- if start_index >= self.count() {
- return None;
- }
-
- let last_message_index =
- std::cmp::min((start_index + count) as usize, self.count() as usize);
-
- let sub_indexes = self.indexes.slice_by_offset(
- start_index,
- (last_message_index - start_index as usize) as u32,
- )?;
-
- let first_message_position = self.message_start_position(start_index as usize);
- let last_message_position = self.message_end_position(last_message_index - 1);
-
- let sub_buffer = self
- .messages
- .slice(first_message_position..last_message_position);
-
- Some(IggyMessagesBatch {
- count: (last_message_index - start_index as usize) as u32,
- indexes: sub_indexes,
- messages: sub_buffer,
- })
- }
-
- /// Returns a contiguous slice (as a new `IggyMessagesBatch`) of up to `count` messages
- /// whose message headers have a timestamp greater than or equal to the provided `timestamp`.
- ///
- /// If no messages meet the criteria, returns `None`.
- pub fn slice_by_timestamp(&self, timestamp: u64, count: u32) -> Option<Self> {
- if self.is_empty() || count == 0 {
- return None;
- }
-
- // Use binary search to find the first message with timestamp >= the target
- let first_message_index = self.binary_search_timestamp(timestamp)?;
-
- self.slice_by_index(first_message_index, count)
- }
-
- /// Find the position of the index with timestamp closest to (but not exceeding) the target
- fn binary_search_timestamp(&self, target_timestamp: u64) -> Option<u32> {
- if self.count() == 0 {
- return None;
- }
-
- let last_timestamp = self.get(self.count() as usize - 1)?.header().timestamp();
- if target_timestamp > last_timestamp {
- return Some(self.count() - 1);
- }
-
- let first_timestamp = self.get(0)?.header().timestamp();
- if target_timestamp <= first_timestamp {
- return Some(0);
- }
-
- let mut low = 0;
- let mut high = self.count() - 1;
-
- while low <= high {
- let mid = low + (high - low) / 2;
- let mid_index = self.get(mid as usize)?;
- let mid_timestamp = mid_index.header().timestamp();
-
- match mid_timestamp.cmp(&target_timestamp) {
- std::cmp::Ordering::Equal => return Some(mid),
- std::cmp::Ordering::Less => low = mid + 1,
- std::cmp::Ordering::Greater => {
- if mid == 0 {
- break;
- }
- high = mid - 1;
- }
- }
- }
-
- Some(low)
- }
-
/// Get the message at the specified index.
/// Returns None if the index is out of bounds.
pub fn get(&self, index: usize) -> Option<IggyMessageView> {
@@ -294,58 +184,6 @@ impl IggyMessagesBatch {
result
}
-
- /// Validates that all messages in batch have correct checksums.
- pub fn validate_checksums(&self) -> Result<(), IggyError> {
- for message in self.iter() {
- let calculated_checksum = message.calculate_checksum();
- let actual_checksum = message.header().checksum();
- let offset = message.header().offset();
- if calculated_checksum != actual_checksum {
- return Err(IggyError::InvalidMessageChecksum(
- actual_checksum,
- calculated_checksum,
- offset,
- ));
- }
- }
- Ok(())
- }
-
- /// Validates that all messages have correct checksums and offsets.
- /// This function should be called after messages have been read from disk.
- ///
- /// # Arguments
- ///
- /// * `absolute_start_offset` - The absolute offset of the first message in the batch.
- ///
- /// # Returns
- ///
- /// * `Ok(())` - If all messages have correct checksums and offsets.
- /// * `Err(IggyError)` - If any message has an invalid checksum or offset.
- pub fn validate_checksums_and_offsets(
- &self,
- absolute_start_offset: u64,
- ) -> Result<(), IggyError> {
- let mut current_offset = absolute_start_offset;
- for message in self.iter() {
- let calculated_checksum = message.calculate_checksum();
- let actual_checksum = message.header().checksum();
- let offset = message.header().offset();
- if offset != current_offset {
- return Err(IggyError::InvalidOffset(offset));
- }
- if calculated_checksum != actual_checksum {
- return Err(IggyError::InvalidMessageChecksum(
- actual_checksum,
- calculated_checksum,
- offset,
- ));
- }
- current_offset += 1;
- }
- Ok(())
- }
}
impl IntoIterator for IggyMessagesBatch {
@@ -375,6 +213,7 @@ impl Index<usize> for IggyMessagesBatch {
&self.messages[start..end]
}
}
+
impl BytesSerializable for IggyMessagesBatch {
fn to_bytes(&self) -> Bytes {
panic!("should not be used");
diff --git a/server/Cargo.toml b/server/Cargo.toml
index d4483c5e..4b1ec236 100644
--- a/server/Cargo.toml
+++ b/server/Cargo.toml
@@ -56,6 +56,7 @@ bytes = "1.10.1"
chrono = "0.4.40"
clap = { version = "4.5.34", features = ["derive"] }
console-subscriber = { version = "0.4.1", optional = true }
+crossbeam = "0.8.4"
dashmap = "6.1.0"
derive_more = "2.0.1"
dotenvy = { version = "0.15.7" }
@@ -72,6 +73,7 @@ lending-iterator = "0.1.7"
mimalloc = { version = "0.1", optional = true }
moka = { version = "0.12.10", features = ["future"] }
nix = { version = "0.29", features = ["fs"] }
+once_cell = "1.21.3"
openssl = { version = "0.10.71", features = ["vendored"] }
opentelemetry = { version = "0.29.0", features = ["trace", "logs"] }
opentelemetry-appender-tracing = { version = "0.29.1", features = ["log"] }
diff --git a/server/src/binary/command.rs b/server/src/binary/command.rs
index 77e9c8a5..f138467f 100644
--- a/server/src/binary/command.rs
+++ b/server/src/binary/command.rs
@@ -69,8 +69,6 @@ use iggy::users::update_user::UpdateUser;
use strum::EnumString;
use tracing::error;
-// todo add missing commands
-
define_server_command_enum! {
Ping(Ping), PING_CODE, PING, false;
GetStats(GetStats), GET_STATS_CODE, GET_STATS, false;
diff --git a/server/src/binary/handlers/messages/send_messages_handler.rs b/server/src/binary/handlers/messages/send_messages_handler.rs
index 1c1bc2b9..a043e736 100644
--- a/server/src/binary/handlers/messages/send_messages_handler.rs
+++ b/server/src/binary/handlers/messages/send_messages_handler.rs
@@ -21,8 +21,8 @@ use crate::binary::sender::SenderKind;
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
use crate::streaming::session::Session;
use crate::streaming::systems::system::SharedSystem;
+use crate::streaming::utils::bytes_mut_pool::{BytesMutExt, BYTES_MUT_POOL};
use anyhow::Result;
-use bytes::{Buf, BytesMut};
use iggy::error::IggyError;
use iggy::identifier::Identifier;
use iggy::models::messaging::INDEX_SIZE;
@@ -49,40 +49,56 @@ impl ServerCommandHandler for SendMessages {
session: &Session,
system: &SharedSystem,
) -> Result<(), IggyError> {
- let length = length - 4;
- let mut buffer = BytesMut::with_capacity(length as usize);
- unsafe {
- buffer.set_len(length as usize);
- }
- sender.read(&mut buffer).await?;
-
- let mut element_size;
-
- let stream_id = Identifier::from_raw_bytes(buffer.chunk())?;
- element_size = stream_id.get_size_bytes().as_bytes_usize();
- buffer.advance(element_size);
- self.stream_id = stream_id;
+ let total_payload_size = length as usize - std::mem::size_of::<u32>();
+ let metadata_len_field_size = std::mem::size_of::<u32>();
- let topic_id = Identifier::from_raw_bytes(buffer.chunk())?;
- element_size = topic_id.get_size_bytes().as_bytes_usize();
- buffer.advance(element_size);
- self.topic_id = topic_id;
+ let mut metadata_length_buffer = [0u8; 4];
+ sender.read(&mut metadata_length_buffer).await?;
+ let metadata_size = u32::from_le_bytes(metadata_length_buffer);
- let partitioning = Partitioning::from_raw_bytes(buffer.chunk())?;
- element_size = partitioning.get_size_bytes().as_bytes_usize();
- buffer.advance(element_size);
- self.partitioning = partitioning;
+ let mut metadata_buffer = BYTES_MUT_POOL.get_buffer(metadata_size as usize);
+ unsafe { metadata_buffer.set_len(metadata_size as usize) };
+ sender.read(&mut metadata_buffer).await?;
- let messages_count = buffer.get_u32_le() as usize;
+ let mut element_size = 0;
- let mut indexes_and_messages = buffer.split();
+ let stream_id = Identifier::from_raw_bytes(&metadata_buffer)?;
+ element_size += stream_id.get_size_bytes().as_bytes_usize();
+ self.stream_id = stream_id;
- let messages = indexes_and_messages.split_off(messages_count * INDEX_SIZE);
- let indexes = indexes_and_messages;
+ let topic_id = Identifier::from_raw_bytes(&metadata_buffer[element_size..])?;
+ element_size += topic_id.get_size_bytes().as_bytes_usize();
+ self.topic_id = topic_id;
- let indexes = IggyIndexesMut::from_bytes(indexes);
+ let partitioning = Partitioning::from_raw_bytes(&metadata_buffer[element_size..])?;
+ element_size += partitioning.get_size_bytes().as_bytes_usize();
+ self.partitioning = partitioning;
- let batch = IggyMessagesBatchMut::from_indexes_and_messages(indexes, messages);
+ let messages_count = u32::from_le_bytes(
+ metadata_buffer[element_size..element_size + 4]
+ .try_into()
+ .unwrap(),
+ );
+ let indexes_size = messages_count as usize * INDEX_SIZE;
+
+ metadata_buffer.return_to_pool();
+
+ let mut indexes_buffer = BYTES_MUT_POOL.get_buffer(indexes_size);
+ unsafe { indexes_buffer.set_len(indexes_size) };
+ sender.read(&mut indexes_buffer).await?;
+
+ let messages_size =
+ total_payload_size - metadata_size as usize - indexes_size - metadata_len_field_size;
+ let mut messages_buffer = BYTES_MUT_POOL.get_buffer(messages_size);
+ unsafe { messages_buffer.set_len(messages_size) };
+ sender.read(&mut messages_buffer).await?;
+
+ let indexes = IggyIndexesMut::from_bytes(indexes_buffer, 0);
+ let batch = IggyMessagesBatchMut::from_indexes_and_messages(
+ messages_count,
+ indexes,
+ messages_buffer,
+ );
batch.validate()?;
diff --git a/server/src/channels/commands/print_sysinfo.rs b/server/src/channels/commands/print_sysinfo.rs
index f40bb4e0..c32c7229 100644
--- a/server/src/channels/commands/print_sysinfo.rs
+++ b/server/src/channels/commands/print_sysinfo.rs
@@ -17,8 +17,9 @@
*/
use crate::{
- channels::server_command::BackgroundServerCommand, configs::server::ServerConfig,
- streaming::systems::system::SharedSystem,
+ channels::server_command::BackgroundServerCommand,
+ configs::server::ServerConfig,
+ streaming::{systems::system::SharedSystem, utils::bytes_mut_pool::BYTES_MUT_POOL},
};
use flume::{Receiver, Sender};
use human_repr::HumanCount;
@@ -89,6 +90,8 @@ impl BackgroundServerCommand<SysInfoPrintCommand> for SysInfoPrintExecutor {
stats.read_bytes,
stats.written_bytes,
stats.run_time);
+
+ BYTES_MUT_POOL.log_stats();
}
fn start_command_sender(
diff --git a/server/src/http/jwt/storage.rs b/server/src/http/jwt/storage.rs
index 05389a9c..30884c09 100644
--- a/server/src/http/jwt/storage.rs
+++ b/server/src/http/jwt/storage.rs
@@ -17,13 +17,14 @@
*/
use crate::http::jwt::COMPONENT;
+use crate::streaming::utils::bytes_mut_pool::{BytesMutExt, BYTES_MUT_POOL};
use crate::streaming::utils::file;
use crate::{
http::jwt::json_web_token::RevokedAccessToken, streaming::persistence::persister::PersisterKind,
};
use ahash::AHashMap;
use anyhow::Context;
-use bytes::{BufMut, BytesMut};
+use bytes::BufMut;
use error_set::ErrContext;
use iggy::error::IggyError;
use std::sync::Arc;
@@ -69,7 +70,7 @@ impl TokenStorage {
})
.map_err(|_| IggyError::CannotReadFileMetadata)?
.len() as usize;
- let mut buffer = BytesMut::with_capacity(file_size);
+ let mut buffer = BYTES_MUT_POOL.get_buffer(file_size);
buffer.put_bytes(0, file_size);
file.read_exact(&mut buffer)
.await
@@ -87,6 +88,8 @@ impl TokenStorage {
.map_err(|_| IggyError::CannotDeserializeResource)?
.0;
+ buffer.return_to_pool();
+
let tokens = tokens
.into_iter()
.map(|(id, expiry)| RevokedAccessToken { id, expiry })
diff --git a/server/src/http/messages.rs b/server/src/http/messages.rs
index 17dc5ad0..1decefdb 100644
--- a/server/src/http/messages.rs
+++ b/server/src/http/messages.rs
@@ -143,6 +143,8 @@ fn make_mutable(batch: IggyMessagesBatch) -> IggyMessagesBatchMut {
let (_, indexes, messages) = batch.decompose();
let (_, indexes_buffer) = indexes.decompose();
let indexes_buffer_mut = BytesMut::from(indexes_buffer);
- let indexes_mut = IggyIndexesMut::from_bytes(indexes_buffer_mut);
- IggyMessagesBatchMut::from_indexes_and_messages(indexes_mut, messages.into())
+ let indexes_mut = IggyIndexesMut::from_bytes(indexes_buffer_mut, 0);
+ let count = indexes_mut.count();
+ let messages_mut = messages.into();
+ IggyMessagesBatchMut::from_indexes_and_messages(count, indexes_mut, messages_mut)
}
diff --git a/server/src/main.rs b/server/src/main.rs
index 7eac061e..e9a81962 100644
--- a/server/src/main.rs
+++ b/server/src/main.rs
@@ -38,6 +38,7 @@ use server::log::tokio_console::Logging;
use server::quic::quic_server;
use server::server_error::ServerError;
use server::streaming::systems::system::{SharedSystem, System};
+use server::streaming::utils::bytes_mut_pool::BytesMutPool;
use server::tcp::tcp_server;
use tokio::time::Instant;
use tracing::{info, instrument};
@@ -94,6 +95,8 @@ async fn main() -> Result<(), ServerError> {
#[cfg(not(feature = "disable-mimalloc"))]
info!("Using mimalloc allocator");
+ BytesMutPool::init_pool();
+
let system = SharedSystem::new(System::new(
config.system.clone(),
config.data_maintenance.clone(),
diff --git a/server/src/streaming/segments/indexes/index_reader.rs b/server/src/streaming/segments/indexes/index_reader.rs
index 55fc0ac5..cf0d0438 100644
--- a/server/src/streaming/segments/indexes/index_reader.rs
+++ b/server/src/streaming/segments/indexes/index_reader.rs
@@ -17,9 +17,10 @@
*/
use super::IggyIndexesMut;
+use crate::streaming::utils::bytes_mut_pool::{BytesMutExt, BYTES_MUT_POOL};
use bytes::BytesMut;
use error_set::ErrContext;
-use iggy::models::messaging::{IggyIndex, IggyIndexes, INDEX_SIZE};
+use iggy::models::messaging::{IggyIndex, INDEX_SIZE};
use iggy::{error::IggyError, models::messaging::IggyIndexView};
use std::{
fs::File as StdFile,
@@ -102,7 +103,7 @@ impl IndexReader {
}
};
let index_count = file_size / INDEX_SIZE as u32;
- let indexes = IggyIndexesMut::from_bytes(buf);
+ let indexes = IggyIndexesMut::from_bytes(buf, index_count);
if indexes.count() != index_count {
error!(
"Loaded {} indexes from disk, expected {}, file {} is probably corrupted!",
@@ -123,7 +124,7 @@ impl IndexReader {
&self,
relative_start_offset: u32,
count: u32,
- ) -> Result<Option<IggyIndexes>, IggyError> {
+ ) -> Result<Option<IggyIndexesMut>, IggyError> {
let file_size = self.file_size();
let total_indexes = file_size / INDEX_SIZE as u32;
@@ -198,8 +199,8 @@ impl IndexReader {
base_position
);
- Ok(Some(IggyIndexes::new(
- indexes_bytes.freeze(),
+ Ok(Some(IggyIndexesMut::from_bytes(
+ indexes_bytes,
base_position,
)))
}
@@ -212,7 +213,7 @@ impl IndexReader {
&self,
timestamp: u64,
count: u32,
- ) -> Result<Option<IggyIndexes>, IggyError> {
+ ) -> Result<Option<IggyIndexesMut>, IggyError> {
let file_size = self.file_size();
let total_indexes = file_size / INDEX_SIZE as u32;
@@ -281,8 +282,8 @@ impl IndexReader {
base_position
);
- Ok(Some(IggyIndexes::new(
- indexes_bytes.freeze(),
+ Ok(Some(IggyIndexesMut::from_bytes(
+ indexes_bytes,
base_position,
)))
}
@@ -354,7 +355,7 @@ impl IndexReader {
async fn read_at(&self, offset: u32, len: u32) -> Result<BytesMut, std::io::Error> {
let file = self.file.clone();
spawn_blocking(move || {
- let mut buf = BytesMut::with_capacity(len as usize);
+ let mut buf = BYTES_MUT_POOL.get_buffer(len as usize);
unsafe { buf.set_len(len as usize) };
file.read_exact_at(&mut buf, offset as u64)?;
Ok(buf)
@@ -395,6 +396,9 @@ impl IndexReader {
}
};
- Ok(Some(IggyIndexView::new(&buf).to_index()))
+ let index = IggyIndexView::new(&buf).to_index();
+ buf.return_to_pool();
+
+ Ok(Some(index))
}
}
diff --git a/server/src/streaming/segments/indexes/indexes_mut.rs b/server/src/streaming/segments/indexes/indexes_mut.rs
index 0c16cb0f..da89a334 100644
--- a/server/src/streaming/segments/indexes/indexes_mut.rs
+++ b/server/src/streaming/segments/indexes/indexes_mut.rs
@@ -16,11 +16,13 @@
* under the License.
*/
-use bytes::{BufMut, Bytes, BytesMut};
-use iggy::models::messaging::{IggyIndexView, IggyIndexes, INDEX_SIZE};
+use bytes::{BufMut, BytesMut};
+use iggy::models::messaging::{IggyIndexView, INDEX_SIZE};
use std::fmt;
use std::ops::{Deref, Index as StdIndex};
+use crate::streaming::utils::bytes_mut_pool::{BytesMutExt, BYTES_MUT_POOL};
+
/// A container for binary-encoded index data.
/// Optimized for efficient storage and I/O operations.
#[derive(Default, Clone)]
@@ -41,17 +43,24 @@ impl IggyIndexesMut {
}
/// Creates indexes from bytes
- pub fn from_bytes(indexes: BytesMut) -> Self {
+ pub fn from_bytes(indexes: BytesMut, base_position: u32) -> Self {
Self {
buffer: indexes,
saved_count: 0,
- base_position: 0,
+ base_position,
}
}
+ /// Decompose the container into its components
+ pub fn decompose(mut self) -> (u32, BytesMut) {
+ let base_position = self.base_position;
+ let buffer = std::mem::replace(&mut self.buffer, BytesMut::new());
+ (base_position, buffer)
+ }
+
/// Gets the size of all indexes messages
pub fn messages_size(&self) -> u32 {
- self.last_position()
+ self.last_position() - self.base_position
}
/// Gets the base position of the indexes
@@ -74,15 +83,15 @@ impl IggyIndexesMut {
/// Creates a new container with the specified capacity
pub fn with_capacity(capacity: usize, base_position: u32) -> Self {
Self {
- buffer: BytesMut::with_capacity(capacity * INDEX_SIZE),
+ buffer: BYTES_MUT_POOL.get_buffer(capacity * INDEX_SIZE),
saved_count: 0,
base_position,
}
}
- /// Makes the indexes immutable
- pub fn make_immutable(self) -> IggyIndexes {
- IggyIndexes::new(self.buffer.freeze(), self.base_position)
+ /// Gets the capacity of the buffer
+ pub fn capacity(&self) -> usize {
+ self.buffer.capacity()
}
/// Inserts a new index at the end of buffer
@@ -93,8 +102,8 @@ impl IggyIndexesMut {
}
/// Appends another slice of indexes to this one.
- pub fn concatenate(&mut self, other: Bytes) {
- self.buffer.put(other);
+ pub fn append_slice(&mut self, other: &[u8]) {
+ self.buffer.put_slice(other);
}
/// Gets the number of indexes in the container
@@ -225,7 +234,11 @@ impl IggyIndexesMut {
}
/// Slices the container to return a view of a specific range of indexes
- pub fn slice_by_offset(&self, relative_start_offset: u32, count: u32) -> Option<IggyIndexes> {
+ pub fn slice_by_offset(
+ &self,
+ relative_start_offset: u32,
+ count: u32,
+ ) -> Option<IggyIndexesMut> {
let available_count = self.count().saturating_sub(relative_start_offset);
let actual_count = std::cmp::min(count, available_count);
@@ -240,15 +253,15 @@ impl IggyIndexesMut {
let slice = BytesMut::from(&self.buffer[start_byte..end_byte]);
if relative_start_offset == 0 {
- Some(IggyIndexes::new(slice.freeze(), self.base_position))
+ Some(IggyIndexesMut::from_bytes(slice, self.base_position))
} else {
let position_offset = self.get(relative_start_offset - 1).unwrap().position();
- Some(IggyIndexes::new(slice.freeze(), position_offset))
+ Some(IggyIndexesMut::from_bytes(slice, position_offset))
}
}
/// Loads indexes from cache based on timestamp
- pub fn slice_by_timestamp(&self, timestamp: u64, count: u32) -> Option<IggyIndexes> {
+ pub fn slice_by_timestamp(&self, timestamp: u64, count: u32) -> Option<IggyIndexesMut> {
if self.count() == 0 {
return None;
}
@@ -274,7 +287,7 @@ impl IggyIndexesMut {
0
};
- Some(IggyIndexes::new(slice.freeze(), base_position))
+ Some(IggyIndexesMut::from_bytes(slice, base_position))
}
/// Find the position of the index with timestamp closest to (but not exceeding) the target
@@ -317,6 +330,13 @@ impl IggyIndexesMut {
}
}
+impl Drop for IggyIndexesMut {
+ fn drop(&mut self) {
+ let indexes = std::mem::replace(&mut self.buffer, BytesMut::new());
+ indexes.return_to_pool();
+ }
+}
+
impl StdIndex<usize> for IggyIndexesMut {
type Output = [u8];
diff --git a/server/src/streaming/segments/messages/messages_reader.rs b/server/src/streaming/segments/messages/messages_reader.rs
index fbff0fb0..c1d0f99b 100644
--- a/server/src/streaming/segments/messages/messages_reader.rs
+++ b/server/src/streaming/segments/messages/messages_reader.rs
@@ -16,12 +16,11 @@
* under the License.
*/
-use bytes::{Bytes, BytesMut};
+use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
+use crate::streaming::utils::bytes_mut_pool::BYTES_MUT_POOL;
+use bytes::BytesMut;
use error_set::ErrContext;
-use iggy::{
- error::IggyError,
- models::messaging::{IggyIndexes, IggyMessagesBatch},
-};
+use iggy::error::IggyError;
use std::{fs::File as StdFile, os::unix::prelude::FileExt};
use std::{
io::ErrorKind,
@@ -90,12 +89,11 @@ impl MessagesReader {
/// Loads and returns all message IDs from the messages file.
pub async fn load_all_message_ids_from_disk(
&self,
- indexes: IggyIndexes,
+ indexes: IggyIndexesMut,
messages_count: u32,
) -> Result<Vec<u128>, IggyError> {
let file_size = self.file_size();
if file_size == 0 {
- trace!("Messages file {} is empty.", self.file_path);
return Ok(vec![]);
}
@@ -113,7 +111,11 @@ impl MessagesReader {
}
};
- let messages = IggyMessagesBatch::new(indexes, messages_bytes, messages_count);
+ let messages = IggyMessagesBatchMut::from_indexes_and_messages(
+ messages_count,
+ indexes,
+ messages_bytes,
+ );
let mut ids = Vec::with_capacity(messages_count as usize);
for message in messages.iter() {
@@ -126,12 +128,11 @@ impl MessagesReader {
/// Loads and returns a batch of messages from the messages file.
pub async fn load_messages_from_disk(
&self,
- indexes: IggyIndexes,
- ) -> Result<IggyMessagesBatch, IggyError> {
+ indexes: IggyIndexesMut,
+ ) -> Result<IggyMessagesBatchMut, IggyError> {
let file_size = self.file_size();
if file_size == 0 {
- trace!("Messages file {} is empty.", self.file_path);
- return Ok(IggyMessagesBatch::empty());
+ return Ok(IggyMessagesBatchMut::empty());
}
let start_pos = indexes.base_position();
@@ -139,13 +140,13 @@ impl MessagesReader {
let messages_count = indexes.count();
if start_pos + count_bytes > file_size {
- return Ok(IggyMessagesBatch::empty());
+ return Ok(IggyMessagesBatchMut::empty());
}
let messages_bytes = match self.read_at(start_pos as u64, count_bytes).await {
Ok(buf) => buf,
Err(error) if error.kind() == ErrorKind::UnexpectedEof => {
- return Ok(IggyMessagesBatch::empty());
+ return Ok(IggyMessagesBatchMut::empty());
}
Err(error) => {
error!(
@@ -156,10 +157,10 @@ impl MessagesReader {
}
};
- Ok(IggyMessagesBatch::new(
+ Ok(IggyMessagesBatchMut::from_indexes_and_messages(
+ messages_count,
indexes,
messages_bytes,
- messages_count,
))
}
@@ -169,13 +170,13 @@ impl MessagesReader {
}
/// Reads `len` bytes from the messages file at the specified `offset`.
- async fn read_at(&self, offset: u64, len: u32) -> Result<Bytes, std::io::Error> {
+ async fn read_at(&self, offset: u64, len: u32) -> Result<BytesMut, std::io::Error> {
let file = self.file.clone();
spawn_blocking(move || {
- let mut buf = BytesMut::with_capacity(len as usize);
+ let mut buf = BYTES_MUT_POOL.get_buffer(len as usize);
unsafe { buf.set_len(len as usize) };
file.read_exact_at(&mut buf, offset)?;
- Ok(buf.freeze())
+ Ok(buf)
})
.await?
}
diff --git a/server/src/streaming/segments/messages_accumulator.rs b/server/src/streaming/segments/messages_accumulator.rs
index a8a07663..fe259c27 100644
--- a/server/src/streaming/segments/messages_accumulator.rs
+++ b/server/src/streaming/segments/messages_accumulator.rs
@@ -66,11 +66,10 @@ impl MessagesAccumulator {
segment_start_offset: u64,
segment_current_offset: u64,
segment_current_position: u32,
- batch: IggyMessagesBatchMut,
+ mut batch: IggyMessagesBatchMut,
deduplicator: Option<&MessageDeduplicator>,
) {
let batch_messages_count = batch.count();
-
if batch_messages_count == 0 {
return;
}
@@ -85,7 +84,7 @@ impl MessagesAccumulator {
self.initialize_or_update_offsets(segment_current_offset, segment_current_position);
- let prepared_batch = batch
+ batch
.prepare_for_persistence(
segment_start_offset,
self.current_offset,
@@ -94,9 +93,9 @@ impl MessagesAccumulator {
)
.await;
- let batch_size = prepared_batch.size();
+ let batch_size = batch.size();
- self.batches.add_batch(prepared_batch);
+ self.batches.add_batch(batch);
self.messages_count += batch_messages_count;
self.current_offset = self.base_offset + self.messages_count as u64 - 1;
@@ -151,7 +150,7 @@ impl MessagesAccumulator {
/// Checks if the accumulator is empty (has no messages).
pub fn is_empty(&self) -> bool {
- self.batches.is_empty() || self.messages_count == 0
+ self.messages_count == 0
}
/// Returns the number of messages in the accumulator that have not been persisted.
diff --git a/server/src/streaming/segments/reading_messages.rs b/server/src/streaming/segments/reading_messages.rs
index 34fbc44d..7ac79bf1 100644
--- a/server/src/streaming/segments/reading_messages.rs
+++ b/server/src/streaming/segments/reading_messages.rs
@@ -16,10 +16,9 @@
* under the License.
*/
-use super::IggyMessagesBatchSet;
+use super::{IggyIndexesMut, IggyMessagesBatchMut, IggyMessagesBatchSet};
use crate::streaming::segments::segment::Segment;
use error_set::ErrContext;
-use iggy::models::messaging::{IggyIndexes, IggyMessagesBatch};
use iggy::prelude::*;
use std::sync::atomic::Ordering;
use tracing::trace;
@@ -276,7 +275,7 @@ impl Segment {
&self,
relative_start_offset: u32,
count: u32,
- ) -> Result<Option<IggyIndexes>, IggyError> {
+ ) -> Result<Option<IggyIndexesMut>, IggyError> {
let indexes = if !self.indexes.is_empty() {
self.indexes.slice_by_offset(relative_start_offset, count)
} else {
@@ -293,7 +292,7 @@ impl Segment {
&self,
timestamp: u64,
count: u32,
- ) -> Result<Option<IggyIndexes>, IggyError> {
+ ) -> Result<Option<IggyIndexesMut>, IggyError> {
let indexes = if !self.indexes.is_empty() {
self.indexes.slice_by_timestamp(timestamp, count)
} else {
@@ -359,7 +358,7 @@ impl Segment {
&self,
timestamp: u64,
count: u32,
- ) -> Result<IggyMessagesBatch, IggyError> {
+ ) -> Result<IggyMessagesBatchMut, IggyError> {
tracing::trace!(
"Loading {count} messages from disk, timestamp: {timestamp}, current_timestamp: {}...",
self.end_timestamp
@@ -368,7 +367,7 @@ impl Segment {
let indexes_to_read = self.load_indexes_by_timestamp(timestamp, count).await?;
if indexes_to_read.is_none() {
- return Ok(IggyMessagesBatch::empty());
+ return Ok(IggyMessagesBatchMut::empty());
}
let indexes_to_read = indexes_to_read.unwrap();
diff --git a/server/src/streaming/segments/segment.rs b/server/src/streaming/segments/segment.rs
index 026834ae..fcb5bb90 100644
--- a/server/src/streaming/segments/segment.rs
+++ b/server/src/streaming/segments/segment.rs
@@ -100,7 +100,7 @@ impl Segment {
last_index_position: 0,
max_size_bytes: config.segment.size,
message_expiry,
- indexes: IggyIndexesMut::with_capacity(1_000_000, 0),
+ indexes: IggyIndexesMut::with_capacity(1024 * 1024, 0),
accumulator: MessagesAccumulator::default(),
is_closed: false,
messages_writer: None,
diff --git a/server/src/streaming/segments/types/messages_batch_mut.rs b/server/src/streaming/segments/types/messages_batch_mut.rs
index 60af072b..87c87c88 100644
--- a/server/src/streaming/segments/types/messages_batch_mut.rs
+++ b/server/src/streaming/segments/types/messages_batch_mut.rs
@@ -19,14 +19,15 @@
use super::message_view_mut::IggyMessageViewMutIterator;
use crate::streaming::deduplication::message_deduplicator::MessageDeduplicator;
use crate::streaming::segments::indexes::IggyIndexesMut;
+use crate::streaming::utils::bytes_mut_pool::{BytesMutExt, BYTES_MUT_POOL};
use crate::streaming::utils::random_id;
use bytes::{BufMut, BytesMut};
use iggy::messages::MAX_PAYLOAD_SIZE;
-use iggy::models::messaging::{IggyIndexView, IggyMessagesBatch, INDEX_SIZE};
+use iggy::models::messaging::{IggyIndexView, INDEX_SIZE};
use iggy::prelude::*;
use iggy::utils::timestamp::IggyTimestamp;
use lending_iterator::prelude::*;
-use std::ops::Deref;
+use std::ops::{Deref, Index};
use tracing::{error, warn};
/// A container for mutable messages that are being prepared for persistence.
@@ -35,6 +36,9 @@ use tracing::{error, warn};
/// and the corresponding index data that allows for efficient message lookup.
#[derive(Debug, Default)]
pub struct IggyMessagesBatchMut {
+ /// The number of messages in the batch
+ count: u32,
+
/// The index data for all messages in the buffer
indexes: IggyIndexesMut,
@@ -49,14 +53,32 @@ impl Sizeable for IggyMessagesBatchMut {
}
impl IggyMessagesBatchMut {
+ /// Creates a new empty messages container
+ pub fn empty() -> Self {
+ Self {
+ count: 0,
+ indexes: IggyIndexesMut::empty(),
+ messages: BytesMut::new(),
+ }
+ }
+
/// Creates a new messages container from existing index and message buffers.
///
/// # Arguments
///
/// * `indexes` - Preprocessed index data
/// * `messages` - Serialized message data
- pub fn from_indexes_and_messages(indexes: IggyIndexesMut, messages: BytesMut) -> Self {
- Self { indexes, messages }
+ /// * `count` - Number of messages in the batch
+ pub fn from_indexes_and_messages(
+ count: u32,
+ indexes: IggyIndexesMut,
+ messages: BytesMut,
+ ) -> Self {
+ Self {
+ count,
+ indexes,
+ messages,
+ }
}
/// Creates a new messages container from a slice of IggyMessage objects.
@@ -70,7 +92,7 @@ impl IggyMessagesBatchMut {
/// * `messages` - Slice of message objects to store
/// * `messages_size` - Total size of all messages in bytes
pub fn from_messages(messages: &[IggyMessage], messages_size: u32) -> Self {
- let mut messages_buffer = BytesMut::with_capacity(messages_size as usize);
+ let mut messages_buffer = BYTES_MUT_POOL.get_buffer(messages_size as usize);
let mut indexes_buffer = IggyIndexesMut::with_capacity(messages.len(), 0);
let mut position = 0;
@@ -81,7 +103,7 @@ impl IggyMessagesBatchMut {
indexes_buffer.insert(0, position, 0);
}
- Self::from_indexes_and_messages(indexes_buffer, messages_buffer)
+ Self::from_indexes_and_messages(messages.len() as u32, indexes_buffer, messages_buffer)
}
/// Creates a lending iterator that yields mutable views of messages.
@@ -117,15 +139,15 @@ impl IggyMessagesBatchMut {
///
/// An immutable `IggyMessagesBatch` ready for persistence
pub async fn prepare_for_persistence(
- self,
+ &mut self,
start_offset: u64,
base_offset: u64,
current_position: u32,
deduplicator: Option<&MessageDeduplicator>,
- ) -> IggyMessagesBatch {
+ ) {
let messages_count = self.count();
if messages_count == 0 {
- return IggyMessagesBatch::empty();
+ return;
}
let mut curr_abs_offset = base_offset;
@@ -138,9 +160,10 @@ impl IggyMessagesBatchMut {
let mut invalid_messages_indexes =
deduplicator.map(|_| Vec::with_capacity(messages_count as usize));
- let (mut indexes, mut messages) = self.decompose();
- indexes.set_base_position(current_position);
- let mut iter = IggyMessageViewMutIterator::new(&mut messages);
+ // let (mut indexes, mut messages) = self.decompose();
+ self.indexes.set_base_position(current_position);
+ let mut iter: IggyMessageViewMutIterator<'_> =
+ IggyMessageViewMutIterator::new(&mut self.messages);
let timestamp = IggyTimestamp::now().as_micros();
while let Some(mut message) = iter.next() {
@@ -169,9 +192,9 @@ impl IggyMessagesBatchMut {
curr_position += message_size;
let relative_offset = (curr_abs_offset - start_offset) as u32;
- indexes.set_offset_at(curr_rel_offset, relative_offset);
- indexes.set_position_at(curr_rel_offset, curr_position);
- indexes.set_timestamp_at(curr_rel_offset, timestamp);
+ self.indexes.set_offset_at(curr_rel_offset, relative_offset);
+ self.indexes.set_position_at(curr_rel_offset, curr_position);
+ self.indexes.set_timestamp_at(curr_rel_offset, timestamp);
curr_abs_offset += 1;
curr_rel_offset += 1;
@@ -179,48 +202,53 @@ impl IggyMessagesBatchMut {
if let Some(invalid_messages_indexes) = invalid_messages_indexes {
if invalid_messages_indexes.is_empty() {
- return IggyMessagesBatch::new(
- indexes.make_immutable(),
- messages.freeze(),
- messages_count,
- );
+ return;
}
- let batch = IggyMessagesBatchMut::from_indexes_and_messages(indexes, messages)
- .remove_messages(&invalid_messages_indexes, current_position);
-
- let messages_count = batch.count();
-
- let (indexes, messages) = batch.decompose();
-
- return IggyMessagesBatch::new(
- indexes.make_immutable(),
- messages.freeze(),
- messages_count,
- );
+ self.remove_messages(&invalid_messages_indexes, current_position);
}
+ }
- IggyMessagesBatch::new(indexes.make_immutable(), messages.freeze(), messages_count)
+ /// Returns the first offset in the batch
+ pub fn first_offset(&self) -> Option<u64> {
+ if self.is_empty() {
+ return None;
+ }
+ Some(IggyMessageView::new(&self.messages).header().offset())
}
/// Returns the first timestamp in the batch
- pub fn first_timestamp(&self) -> u64 {
- IggyMessageView::new(&self.messages).header().timestamp()
+ pub fn first_timestamp(&self) -> Option<u64> {
+ if self.is_empty() {
+ return None;
+ }
+ Some(IggyMessageView::new(&self.messages).header().timestamp())
}
/// Returns the last timestamp in the batch
- pub fn last_timestamp(&self) -> u64 {
+ pub fn last_timestamp(&self) -> Option<u64> {
if self.is_empty() {
- return 0;
+ return None;
}
let last_index = self.count() as usize - 1;
- self.get_message_boundaries(last_index)
- .map(|(start, _)| {
- IggyMessageView::new(&self.messages[start..])
- .header()
- .timestamp()
- })
- .unwrap_or(0)
+ self.get_message_boundaries(last_index).map(|(start, _)| {
+ IggyMessageView::new(&self.messages[start..])
+ .header()
+ .timestamp()
+ })
+ }
+
+ /// Returns the last offset in the batch
+ pub fn last_offset(&self) -> Option<u64> {
+ if self.is_empty() {
+ return None;
+ }
+ let last_index = self.count() as usize - 1;
+ self.get_message_boundaries(last_index).map(|(start, _)| {
+ IggyMessageView::new(&self.messages[start..])
+ .header()
+ .offset()
+ })
}
/// Checks if the batch is empty.
@@ -229,8 +257,21 @@ impl IggyMessagesBatchMut {
}
/// Decomposes the batch into its constituent parts.
- pub fn decompose(self) -> (IggyIndexesMut, BytesMut) {
- (self.indexes, self.messages)
+ pub fn decompose(mut self) -> (IggyIndexesMut, BytesMut) {
+ let indexes = std::mem::replace(&mut self.indexes, IggyIndexesMut::empty());
+ let messages = std::mem::replace(&mut self.messages, BytesMut::new());
+
+ (indexes, messages)
+ }
+
+ /// Take the indexes from the batch
+ pub fn take_indexes(&mut self) -> IggyIndexesMut {
+ std::mem::take(&mut self.indexes)
+ }
+
+ /// Borrows the indexes from the batch
+ pub fn indexes(&self) -> &IggyIndexesMut {
+ &self.indexes
}
/// Get message position from the indexes at the given index
@@ -266,6 +307,152 @@ impl IggyMessagesBatchMut {
}
}
+ /// Returns a contiguous slice (as a new `IggyMessagesBatch`) of up to `count` messages
+ /// whose message headers have an offset greater than or equal to the provided `start_offset`.
+ pub fn slice_by_offset(&self, start_offset: u64, count: u32) -> Option<Self> {
+ if self.is_empty() || count == 0 {
+ return None;
+ }
+
+ let first_offset = self.first_offset()?;
+
+ if start_offset < first_offset {
+ return self.slice_by_index(0, count);
+ }
+
+ let last_offset = self.last_offset()?;
+ if start_offset > last_offset {
+ return None;
+ }
+
+ let offset_diff = start_offset - first_offset;
+ let first_message_index = offset_diff as usize;
+
+ if first_message_index >= self.count() as usize {
+ return None;
+ }
+
+ self.slice_by_index(first_message_index as u32, count)
+ }
+
+ /// Helper method to slice the batch starting from a specific index
+ fn slice_by_index(&self, start_index: u32, count: u32) -> Option<Self> {
+ if start_index >= self.count() {
+ return None;
+ }
+
+ let last_message_index =
+ std::cmp::min((start_index + count) as usize, self.count() as usize);
+
+ let sub_indexes = self.indexes.slice_by_offset(
+ start_index,
+ (last_message_index - start_index as usize) as u32,
+ )?;
+
+ let first_message_position = self.message_start_position(start_index as usize)?;
+ let last_message_position = self.message_end_position(last_message_index - 1)?;
+
+ // TODO(hubcio): messages from accumulator unfortunately are deep-copied
+ let mut sub_buffer =
+ BYTES_MUT_POOL.get_buffer(last_message_position - first_message_position);
+ sub_buffer.put_slice(&self.messages[first_message_position..last_message_position]);
+
+ Some(IggyMessagesBatchMut {
+ count: sub_indexes.count(),
+ indexes: sub_indexes,
+ messages: sub_buffer,
+ })
+ }
+
+ /// Returns a contiguous slice (as a new `IggyMessagesBatch`) of up to `count` messages
+ /// whose message headers have a timestamp greater than or equal to the provided `timestamp`.
+ ///
+ /// If no messages meet the criteria, returns `None`.
+ pub fn slice_by_timestamp(&self, timestamp: u64, count: u32) -> Option<Self> {
+ if self.is_empty() || count == 0 {
+ return None;
+ }
+
+ // Use binary search to find the first message with timestamp >= the target
+ let first_message_index = self.binary_search_timestamp(timestamp)?;
+
+ self.slice_by_index(first_message_index, count)
+ }
+
+ /// Find the position of the index with timestamp closest to (but not exceeding) the target
+ fn binary_search_timestamp(&self, target_timestamp: u64) -> Option<u32> {
+ if self.count() == 0 {
+ return None;
+ }
+
+ let last_timestamp = self.get(self.count() as usize - 1)?.header().timestamp();
+ if target_timestamp > last_timestamp {
+ return Some(self.count() - 1);
+ }
+
+ let first_timestamp = self.get(0)?.header().timestamp();
+ if target_timestamp <= first_timestamp {
+ return Some(0);
+ }
+
+ let mut low = 0;
+ let mut high = self.count() - 1;
+
+ while low <= high {
+ let mid = low + (high - low) / 2;
+ let mid_index = self.get(mid as usize)?;
+ let mid_timestamp = mid_index.header().timestamp();
+
+ match mid_timestamp.cmp(&target_timestamp) {
+ std::cmp::Ordering::Equal => return Some(mid),
+ std::cmp::Ordering::Less => low = mid + 1,
+ std::cmp::Ordering::Greater => {
+ if mid == 0 {
+ break;
+ }
+ high = mid - 1;
+ }
+ }
+ }
+
+ Some(low)
+ }
+
+ /// Validates that all messages have correct checksums and offsets.
+ /// This function should be called after messages have been read from disk.
+ ///
+ /// # Arguments
+ ///
+ /// * `absolute_start_offset` - The absolute offset of the first message in the batch.
+ ///
+ /// # Returns
+ ///
+ /// * `Ok(())` - If all messages have correct checksums and offsets.
+ /// * `Err(IggyError)` - If any message has an invalid checksum or offset.
+ pub fn validate_checksums_and_offsets(
+ &self,
+ absolute_start_offset: u64,
+ ) -> Result<(), IggyError> {
+ let mut current_offset = absolute_start_offset;
+ for message in self.iter() {
+ let calculated_checksum = message.calculate_checksum();
+ let actual_checksum = message.header().checksum();
+ let offset = message.header().offset();
+ if offset != current_offset {
+ return Err(IggyError::InvalidOffset(offset));
+ }
+ if calculated_checksum != actual_checksum {
+ return Err(IggyError::InvalidMessageChecksum(
+ actual_checksum,
+ calculated_checksum,
+ offset,
+ ));
+ }
+ current_offset += 1;
+ }
+ Ok(())
+ }
+
/// Gets the byte range for a message at the given index
fn get_message_boundaries(&self, index: usize) -> Option<(usize, usize)> {
let start = self.message_start_position(index)?;
@@ -325,7 +512,7 @@ impl IggyMessagesBatchMut {
/// # Returns
///
/// A new `IggyMessagesBatchMut` with the specified messages removed
- pub fn remove_messages(self, indexes_to_remove: &[u32], current_position: u32) -> Self {
+ pub fn remove_messages(&mut self, indexes_to_remove: &[u32], current_position: u32) {
/*
A temporary list of message boundaries is first collected for each index
that should be removed. Chunks of data that are not removed are appended
@@ -336,14 +523,13 @@ impl IggyMessagesBatchMut {
This allows for avoiding copying unnecessary data and ensures that indexes
match the newly constructed buffer.
*/
-
if indexes_to_remove.is_empty() || self.is_empty() {
- return self;
+ return;
}
let msg_count = self.count() as usize;
if indexes_to_remove.len() >= msg_count {
- return IggyMessagesBatchMut::default();
+ return;
}
let current_size = self.size();
@@ -369,11 +555,11 @@ impl IggyMessagesBatchMut {
let new_size = current_size - size_to_remove;
let new_message_count = msg_count as u32 - indexes_to_remove.len() as u32;
- let mut new_buffer = BytesMut::with_capacity(new_size as usize);
+ let mut new_buffer = BYTES_MUT_POOL.get_buffer(new_size as usize);
let mut new_indexes =
IggyIndexesMut::with_capacity(new_message_count as usize, current_position);
- let mut source = self.messages;
+ let mut source = std::mem::take(&mut self.messages);
let mut last_pos = 0_usize;
let mut new_pos = current_position;
@@ -413,8 +599,23 @@ impl IggyMessagesBatchMut {
chunk_len,
);
}
+ }
- IggyMessagesBatchMut::from_indexes_and_messages(new_indexes, new_buffer)
+ /// Validates that all messages in batch have correct checksums.
+ pub fn validate_checksums(&self) -> Result<(), IggyError> {
+ for message in self.iter() {
+ let calculated_checksum = message.calculate_checksum();
+ let actual_checksum = message.header().checksum();
+ let offset = message.header().offset();
+ if calculated_checksum != actual_checksum {
+ return Err(IggyError::InvalidMessageChecksum(
+ actual_checksum,
+ calculated_checksum,
+ offset,
+ ));
+ }
+ }
+ Ok(())
}
/// Validates the structure of the indexes (sizes, counts, etc.)
@@ -590,6 +791,35 @@ impl Validatable<IggyError> for IggyMessagesBatchMut {
}
}
+impl Index<usize> for IggyMessagesBatchMut {
+ type Output = [u8];
+
+ fn index(&self, index: usize) -> &Self::Output {
+ if index >= self.count as usize {
+ panic!(
+ "Index out of bounds: the len is {} but the index is {}",
+ self.count, index
+ );
+ }
+
+ let (start, end) = self
+ .get_message_boundaries(index)
+ .expect("Invalid message boundaries");
+
+ &self.messages[start..end]
+ }
+}
+
+impl Drop for IggyMessagesBatchMut {
+ fn drop(&mut self) {
+ let indexes = std::mem::replace(&mut self.indexes, IggyIndexesMut::empty());
+ let messages = std::mem::replace(&mut self.messages, BytesMut::new());
+
+ messages.return_to_pool();
+ drop(indexes);
+ }
+}
+
impl Deref for IggyMessagesBatchMut {
type Target = BytesMut;
diff --git a/server/src/streaming/segments/types/messages_batch_set.rs b/server/src/streaming/segments/types/messages_batch_set.rs
index 6f32b383..32ecac29 100644
--- a/server/src/streaming/segments/types/messages_batch_set.rs
+++ b/server/src/streaming/segments/types/messages_batch_set.rs
@@ -18,18 +18,20 @@
use crate::binary::handlers::messages::poll_messages_handler::IggyPollMetadata;
use crate::streaming::segments::IggyIndexesMut;
+use crate::streaming::utils::bytes_mut_pool::BytesMutExt;
use bytes::Bytes;
use iggy::models::messaging::IggyMessageView;
-use iggy::models::messaging::IggyMessagesBatch;
use iggy::prelude::*;
use std::ops::Index;
use tracing::trace;
+use super::IggyMessagesBatchMut;
+
/// A container for multiple IggyMessagesBatch objects
#[derive(Debug, Default)]
pub struct IggyMessagesBatchSet {
/// The collection of message containers
- batches: Vec<IggyMessagesBatch>,
+ batches: Vec<IggyMessagesBatchMut>,
/// Total number of messages across all containers
count: u32,
/// Total size in bytes across all containers
@@ -56,7 +58,7 @@ impl IggyMessagesBatchSet {
}
/// Create a batch set from an existing vector of IggyMessages
- pub fn from_vec(messages: Vec<IggyMessagesBatch>) -> Self {
+ pub fn from_vec(messages: Vec<IggyMessagesBatchMut>) -> Self {
let mut batch = Self::with_capacity(messages.len());
for msg in messages {
batch.add_batch(msg);
@@ -65,25 +67,25 @@ impl IggyMessagesBatchSet {
}
/// Add a message container to the batch
- pub fn add_batch(&mut self, messages: IggyMessagesBatch) {
+ pub fn add_batch(&mut self, messages: IggyMessagesBatchMut) {
self.count += messages.count();
self.size += messages.size();
self.batches.push(messages);
}
/// Add another batch of messages to the batch
- pub fn add_batch_set(&mut self, other: IggyMessagesBatchSet) {
+ pub fn add_batch_set(&mut self, mut other: IggyMessagesBatchSet) {
self.count += other.count();
self.size += other.size();
- self.batches.extend(other.batches);
+ let other_batches = std::mem::take(&mut other.batches);
+ self.batches.extend(other_batches);
}
/// Extract indexes from all batches in the set
- pub fn extract_indexes_to(&mut self, target: &mut IggyIndexesMut) {
- for batch in self.iter_mut() {
- let indexes = batch.take_indexes();
- let (_, indexes_buffer) = indexes.decompose();
- target.concatenate(indexes_buffer);
+ pub fn append_indexes_to(&self, target: &mut IggyIndexesMut) {
+ for batch in self.iter() {
+ let indexes = batch.indexes();
+ target.append_slice(indexes);
}
}
@@ -109,16 +111,25 @@ impl IggyMessagesBatchSet {
/// Get timestamp of first message in first batch
pub fn first_timestamp(&self) -> Option<u64> {
+ if self.is_empty() {
+ return None;
+ }
self.batches.first().map(|batch| batch.first_timestamp())?
}
/// Get offset of first message in first batch
pub fn first_offset(&self) -> Option<u64> {
+ if self.is_empty() {
+ return None;
+ }
self.batches.first().map(|batch| batch.first_offset())?
}
/// Get timestamp of last message in last batch
pub fn last_timestamp(&self) -> Option<u64> {
+ if self.is_empty() {
+ return None;
+ }
self.batches.last().map(|batch| batch.last_timestamp())?
}
@@ -128,22 +139,22 @@ impl IggyMessagesBatchSet {
}
/// Get a reference to the underlying vector of message containers
- pub fn inner(&self) -> &Vec<IggyMessagesBatch> {
+ pub fn inner(&self) -> &Vec<IggyMessagesBatchMut> {
&self.batches
}
/// Consume the batch, returning the underlying vector of message containers
- pub fn into_inner(self) -> Vec<IggyMessagesBatch> {
- self.batches
+ pub fn into_inner(mut self) -> Vec<IggyMessagesBatchMut> {
+ std::mem::take(&mut self.batches)
}
/// Iterate over all message containers in the batch
- pub fn iter(&self) -> impl Iterator<Item = &IggyMessagesBatch> {
+ pub fn iter(&self) -> impl Iterator<Item = &IggyMessagesBatchMut> {
self.batches.iter()
}
/// Iterate over all mutable message containers in the batch
- pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut IggyMessagesBatch> {
+ pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut IggyMessagesBatchMut> {
self.batches.iter_mut()
}
@@ -334,31 +345,44 @@ impl Sizeable for IggyMessagesBatchSet {
}
}
-impl From<Vec<IggyMessagesBatch>> for IggyMessagesBatchSet {
- fn from(messages: Vec<IggyMessagesBatch>) -> Self {
+impl From<Vec<IggyMessagesBatchMut>> for IggyMessagesBatchSet {
+ fn from(messages: Vec<IggyMessagesBatchMut>) -> Self {
Self::from_vec(messages)
}
}
-impl From<IggyMessagesBatch> for IggyMessagesBatchSet {
- fn from(messages: IggyMessagesBatch) -> Self {
+impl From<IggyMessagesBatchMut> for IggyMessagesBatchSet {
+ fn from(messages: IggyMessagesBatchMut) -> Self {
Self::from_vec(vec![messages])
}
}
+impl Drop for IggyMessagesBatchSet {
+ fn drop(&mut self) {
+ for batch in self.batches.iter_mut() {
+ let batch = std::mem::take(batch);
+ let (indexes, messages) = batch.decompose();
+ messages.return_to_pool();
+ drop(indexes);
+ }
+ }
+}
+
/// Iterator that consumes an IggyMessagesBatchSet and yields batches
pub struct IggyMessagesBatchSetIntoIter {
- batches: Vec<IggyMessagesBatch>,
+ batches: Vec<IggyMessagesBatchMut>,
position: usize,
}
impl Iterator for IggyMessagesBatchSetIntoIter {
- type Item = IggyMessagesBatch;
+ type Item = IggyMessagesBatchMut;
fn next(&mut self) -> Option<Self::Item> {
if self.position < self.batches.len() {
- let batch =
- std::mem::replace(&mut self.batches[self.position], IggyMessagesBatch::empty());
+ let batch = std::mem::replace(
+ &mut self.batches[self.position],
+ IggyMessagesBatchMut::empty(),
+ );
self.position += 1;
Some(batch)
} else {
@@ -371,15 +395,3 @@ impl Iterator for IggyMessagesBatchSetIntoIter {
(remaining, Some(remaining))
}
}
-
-impl IntoIterator for IggyMessagesBatchSet {
- type Item = IggyMessagesBatch;
- type IntoIter = IggyMessagesBatchSetIntoIter;
-
- fn into_iter(self) -> Self::IntoIter {
- IggyMessagesBatchSetIntoIter {
- batches: self.batches,
- position: 0,
- }
- }
-}
diff --git a/server/src/streaming/segments/writing_messages.rs b/server/src/streaming/segments/writing_messages.rs
index 7e80574b..3a40e3e1 100644
--- a/server/src/streaming/segments/writing_messages.rs
+++ b/server/src/streaming/segments/writing_messages.rs
@@ -84,17 +84,17 @@ impl Segment {
let accumulator = std::mem::take(&mut self.accumulator);
- let mut batches = accumulator.into_batch_set();
+ let batches = accumulator.into_batch_set();
let confirmation = match confirmation {
Some(val) => val,
None => self.config.segment.server_confirmation,
};
- batches.extract_indexes_to(&mut self.indexes);
-
let batch_size = batches.size();
let batch_count = batches.count();
+ batches.append_indexes_to(&mut self.indexes);
+
let saved_bytes = self
.messages_writer
.as_mut()
@@ -130,17 +130,15 @@ impl Segment {
self.check_and_handle_segment_full().await?;
- let saved_messages_count = unsaved_messages_count;
-
trace!(
"Saved {} messages on disk in segment with start offset: {} for partition with ID: {}, total bytes written: {}.",
- saved_messages_count,
+ unsaved_messages_count,
self.start_offset,
self.partition_id,
saved_bytes
);
- Ok(saved_messages_count)
+ Ok(unsaved_messages_count)
}
fn update_counters(&mut self, messages_size: u64, messages_count: u64) {
diff --git a/server/src/streaming/systems/messages.rs b/server/src/streaming/systems/messages.rs
index 2b758079..1716dd26 100644
--- a/server/src/streaming/systems/messages.rs
+++ b/server/src/streaming/systems/messages.rs
@@ -21,11 +21,10 @@ use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut, IggyMessa
use crate::streaming::session::Session;
use crate::streaming::systems::system::System;
use crate::streaming::systems::COMPONENT;
-use bytes::BytesMut;
+use crate::streaming::utils::bytes_mut_pool::BYTES_MUT_POOL;
use error_set::ErrContext;
use iggy::confirmation::Confirmation;
use iggy::consumer::Consumer;
-use iggy::models::messaging::IggyMessagesBatch;
use iggy::prelude::*;
use iggy::utils::crypto::EncryptorKind;
use iggy::{error::IggyError, identifier::Identifier};
@@ -164,7 +163,7 @@ impl System {
let count = batch.count();
let mut indexes = IggyIndexesMut::with_capacity(batch.count() as usize, 0);
- let mut decrypted_messages = BytesMut::with_capacity(batch.size() as usize);
+ let mut decrypted_messages = BYTES_MUT_POOL.get_buffer(batch.size() as usize);
let mut position = 0;
for message in batch.iter() {
@@ -185,9 +184,8 @@ impl System {
}
}
}
- let indexes = indexes.make_immutable();
- let decrypted_messages = decrypted_messages.freeze();
- let decrypted_batch = IggyMessagesBatch::new(indexes, decrypted_messages, count);
+ let decrypted_batch =
+ IggyMessagesBatchMut::from_indexes_and_messages(count, indexes, decrypted_messages);
decrypted_batches.push(decrypted_batch);
}
@@ -199,7 +197,8 @@ impl System {
batch: IggyMessagesBatchMut,
encryptor: &EncryptorKind,
) -> Result<IggyMessagesBatchMut, IggyError> {
- let mut encrypted_messages = BytesMut::with_capacity(batch.size() as usize * 2);
+ let mut encrypted_messages = BYTES_MUT_POOL.get_buffer(batch.size() as usize * 2);
+ let count = batch.count();
let mut indexes = IggyIndexesMut::with_capacity(batch.count() as usize, 0);
let mut position = 0;
@@ -230,6 +229,7 @@ impl System {
}
Ok(IggyMessagesBatchMut::from_indexes_and_messages(
+ count,
indexes,
encrypted_messages,
))
diff --git a/server/src/streaming/systems/storage.rs b/server/src/streaming/systems/storage.rs
index 49649b5f..ff1cc3f2 100644
--- a/server/src/streaming/systems/storage.rs
+++ b/server/src/streaming/systems/storage.rs
@@ -16,13 +16,16 @@
* under the License.
*/
-use crate::streaming::persistence::persister::PersisterKind;
use crate::streaming::storage::SystemInfoStorage;
use crate::streaming::systems::info::SystemInfo;
use crate::streaming::systems::COMPONENT;
+use crate::streaming::utils::bytes_mut_pool::BytesMutExt;
use crate::streaming::utils::file;
+use crate::streaming::{
+ persistence::persister::PersisterKind, utils::bytes_mut_pool::BYTES_MUT_POOL,
+};
use anyhow::Context;
-use bytes::{BufMut, BytesMut};
+use bytes::BufMut;
use error_set::ErrContext;
use iggy::error::IggyError;
use std::sync::Arc;
@@ -60,7 +63,7 @@ impl SystemInfoStorage for FileSystemInfoStorage {
})
.map_err(|_| IggyError::CannotReadFileMetadata)?
.len() as usize;
- let mut buffer = BytesMut::with_capacity(file_size);
+ let mut buffer = BYTES_MUT_POOL.get_buffer(file_size);
buffer.put_bytes(0, file_size);
file.read_exact(&mut buffer)
.await
@@ -75,6 +78,7 @@ impl SystemInfoStorage for FileSystemInfoStorage {
bincode::serde::decode_from_slice(&buffer, bincode::config::standard())
.with_context(|| "Failed to deserialize system info")
.map_err(|_| IggyError::CannotDeserializeResource)?;
+ buffer.return_to_pool();
Ok(system_info)
}
diff --git a/server/src/streaming/utils/bytes_mut_pool.rs b/server/src/streaming/utils/bytes_mut_pool.rs
new file mode 100644
index 00000000..82a8f302
--- /dev/null
+++ b/server/src/streaming/utils/bytes_mut_pool.rs
@@ -0,0 +1,313 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use bytes::BytesMut;
+use crossbeam::queue::ArrayQueue;
+use iggy::prelude::IggyByteSize;
+use once_cell::sync::Lazy;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use tracing::{debug, info, trace};
+
+pub static BYTES_MUT_POOL: Lazy<BytesMutPool> = Lazy::new(BytesMutPool::default);
+
+/// A pool for reusing BytesMut buffers
+#[derive(Clone)]
+pub struct BytesMutPool {
+ // Buffers
+ small_buffers: Arc<ArrayQueue<BytesMut>>,
+ medium_buffers: Arc<ArrayQueue<BytesMut>>,
+ large_buffers: Arc<ArrayQueue<BytesMut>>,
+ extra_large_buffers: Arc<ArrayQueue<BytesMut>>,
+ max_buffers: Arc<ArrayQueue<BytesMut>>,
+
+ // Stats
+ small_created: Arc<AtomicUsize>,
+ medium_created: Arc<AtomicUsize>,
+ large_created: Arc<AtomicUsize>,
+ extra_large_created: Arc<AtomicUsize>,
+ max_created: Arc<AtomicUsize>,
+ small_returned: Arc<AtomicUsize>,
+ medium_returned: Arc<AtomicUsize>,
+ large_returned: Arc<AtomicUsize>,
+ extra_large_returned: Arc<AtomicUsize>,
+ max_returned: Arc<AtomicUsize>,
+}
+
+impl BytesMutPool {
+ // TODO(hubcio): make BytesMutPool bucket sizes configurable via server.toml
+ const SMALL_BUFFER_SIZE: usize = 4 * 1024;
+ const MEDIUM_BUFFER_SIZE: usize = 32 * 1024;
+ const LARGE_BUFFER_SIZE: usize = 512 * 1024;
+ const EXTRA_LARGE_BUFFER_SIZE: usize = 2 * 1024 * 1024;
+ const MAX_BUFFER_SIZE: usize = 16 * 1024 * 1024;
+
+ const SMALL_POOL_SIZE: usize = 4096;
+ const MEDIUM_POOL_SIZE: usize = 1024;
+ const LARGE_POOL_SIZE: usize = 512;
+ const EXTRA_LARGE_POOL_SIZE: usize = 128;
+ const MAX_POOL_SIZE: usize = 64;
+
+ /// Initialize the bytes pool
+ pub fn init_pool() {
+ Lazy::force(&BYTES_MUT_POOL);
+ }
+
+ /// Get a buffer with at least the specified capacity
+ pub fn get_buffer(&self, capacity: usize) -> BytesMut {
+ if capacity <= Self::SMALL_BUFFER_SIZE {
+ if let Some(mut buffer) = self.small_buffers.pop() {
+ buffer.clear();
+ trace!("Reused small buffer with capacity: {}", buffer.capacity());
+ return buffer;
+ }
+ self.small_created.fetch_add(1, Ordering::Relaxed);
+ let buffer = BytesMut::with_capacity(Self::SMALL_BUFFER_SIZE);
+ trace!(
+ "Created new small buffer with capacity: {}",
+ buffer.capacity()
+ );
+ buffer
+ } else if capacity <= Self::MEDIUM_BUFFER_SIZE {
+ if let Some(mut buffer) = self.medium_buffers.pop() {
+ buffer.clear();
+ trace!("Reused medium buffer with capacity: {}", buffer.capacity());
+ return buffer;
+ }
+ self.medium_created.fetch_add(1, Ordering::Relaxed);
+ let buffer = BytesMut::with_capacity(Self::MEDIUM_BUFFER_SIZE);
+ trace!(
+ "Created new medium buffer with capacity: {}",
+ buffer.capacity()
+ );
+ buffer
+ } else if capacity <= Self::LARGE_BUFFER_SIZE {
+ if let Some(mut buffer) = self.large_buffers.pop() {
+ buffer.clear();
+ trace!("Reused large buffer with capacity: {}", buffer.capacity());
+ return buffer;
+ }
+ self.large_created.fetch_add(1, Ordering::Relaxed);
+ let buffer = BytesMut::with_capacity(Self::LARGE_BUFFER_SIZE);
+ trace!(
+ "Created new large buffer with capacity: {}",
+ buffer.capacity()
+ );
+ buffer
+ } else if capacity <= Self::EXTRA_LARGE_BUFFER_SIZE {
+ if let Some(mut buffer) = self.extra_large_buffers.pop() {
+ buffer.clear();
+ trace!(
+ "Reused extra large buffer with capacity: {}",
+ buffer.capacity()
+ );
+ return buffer;
+ }
+ self.extra_large_created.fetch_add(1, Ordering::Relaxed);
+ let buffer = BytesMut::with_capacity(Self::EXTRA_LARGE_BUFFER_SIZE);
+ trace!(
+ "Created new extra large buffer with capacity: {}",
+ buffer.capacity()
+ );
+ buffer
+ } else if capacity <= Self::MAX_BUFFER_SIZE {
+ if let Some(mut buffer) = self.max_buffers.pop() {
+ buffer.clear();
+ trace!("Reused max buffer with capacity: {}", buffer.capacity());
+ return buffer;
+ }
+ self.max_created.fetch_add(1, Ordering::Relaxed);
+ let buffer = BytesMut::with_capacity(Self::MAX_BUFFER_SIZE);
+ trace!(
+ "Created new max buffer with capacity: {}",
+ buffer.capacity()
+ );
+ buffer
+ } else {
+ // For very large buffers that exceed our max size, just allocate directly
+ debug!("Created oversized buffer with capacity: {} B", capacity);
+ BytesMut::with_capacity(capacity)
+ }
+ }
+
+ /// Return a buffer to the pool
+ fn return_buffer(&self, buffer: BytesMut) {
+ let capacity = buffer.capacity();
+ if capacity == Self::SMALL_BUFFER_SIZE {
+ if self.small_buffers.push(buffer).is_err() {
+ trace!("Small buffer pool full, dropping buffer");
+ } else {
+ self.small_returned.fetch_add(1, Ordering::Relaxed);
+ trace!("Returned small buffer to pool");
+ }
+ } else if capacity == Self::MEDIUM_BUFFER_SIZE {
+ if self.medium_buffers.push(buffer).is_err() {
+ trace!("Medium buffer pool full, dropping buffer");
+ } else {
+ self.medium_returned.fetch_add(1, Ordering::Relaxed);
+ trace!("Returned medium buffer to pool");
+ }
+ } else if capacity == Self::LARGE_BUFFER_SIZE {
+ if self.large_buffers.push(buffer).is_err() {
+ trace!("Large buffer pool full, dropping buffer");
+ } else {
+ self.large_returned.fetch_add(1, Ordering::Relaxed);
+ trace!("Returned large buffer to pool");
+ }
+ } else if capacity == Self::EXTRA_LARGE_BUFFER_SIZE {
+ if self.extra_large_buffers.push(buffer).is_err() {
+ trace!("Extra large buffer pool full, dropping buffer");
+ } else {
+ self.extra_large_returned.fetch_add(1, Ordering::Relaxed);
+ trace!("Returned extra large buffer to pool");
+ }
+ } else if capacity == Self::MAX_BUFFER_SIZE {
+ if self.max_buffers.push(buffer).is_err() {
+ trace!("Max buffer pool full, dropping buffer");
+ } else {
+ self.max_returned.fetch_add(1, Ordering::Relaxed);
+ trace!("Returned max buffer to pool");
+ }
+ } else if capacity != 0 {
+ trace!(
+ "Returned buffer to pool with unknown capacity: {}",
+ capacity
+ );
+ }
+ }
+
+ /// Log stats about buffer allocation and reuse
+ pub fn log_stats(&self) {
+ let sm_created = self.small_created.load(Ordering::Relaxed);
+ let md_created = self.medium_created.load(Ordering::Relaxed);
+ let lg_created = self.large_created.load(Ordering::Relaxed);
+ let xl_created = self.extra_large_created.load(Ordering::Relaxed);
+ let mx_created = self.max_created.load(Ordering::Relaxed);
+
+ let sm_returned = self.small_returned.load(Ordering::Relaxed);
+ let md_returned = self.medium_returned.load(Ordering::Relaxed);
+ let lg_returned = self.large_returned.load(Ordering::Relaxed);
+ let xl_returned = self.extra_large_returned.load(Ordering::Relaxed);
+ let mx_returned = self.max_returned.load(Ordering::Relaxed);
+
+ let sm_pool = self.small_buffers.len();
+ let md_pool = self.medium_buffers.len();
+ let lg_pool = self.large_buffers.len();
+ let xl_pool = self.extra_large_buffers.len();
+ let mx_pool = self.max_buffers.len();
+
+ let sm_size = IggyByteSize::from((sm_pool * Self::SMALL_BUFFER_SIZE) as u64);
+ let md_size = IggyByteSize::from((md_pool * Self::MEDIUM_BUFFER_SIZE) as u64);
+ let lg_size = IggyByteSize::from((lg_pool * Self::LARGE_BUFFER_SIZE) as u64);
+ let xl_size = IggyByteSize::from((xl_pool * Self::EXTRA_LARGE_BUFFER_SIZE) as u64);
+ let mx_size = IggyByteSize::from((mx_pool * Self::MAX_BUFFER_SIZE) as u64);
+
+ let total_size = IggyByteSize::from(
+ (sm_pool * Self::SMALL_BUFFER_SIZE
+ + md_pool * Self::MEDIUM_BUFFER_SIZE
+ + lg_pool * Self::LARGE_BUFFER_SIZE
+ + xl_pool * Self::EXTRA_LARGE_BUFFER_SIZE
+ + mx_pool * Self::MAX_BUFFER_SIZE) as u64,
+ );
+
+ let sm_reuse = if sm_created > 0 {
+ sm_returned as f64 / sm_created as f64 * 100.0
+ } else {
+ 0.0
+ };
+ let md_reuse = if md_created > 0 {
+ md_returned as f64 / md_created as f64 * 100.0
+ } else {
+ 0.0
+ };
+ let lg_reuse = if lg_created > 0 {
+ lg_returned as f64 / lg_created as f64 * 100.0
+ } else {
+ 0.0
+ };
+ let xl_reuse = if xl_created > 0 {
+ xl_returned as f64 / xl_created as f64 * 100.0
+ } else {
+ 0.0
+ };
+ let mx_reuse = if mx_created > 0 {
+ mx_returned as f64 / mx_created as f64 * 100.0
+ } else {
+ 0.0
+ };
+
+ let sm_util = sm_pool as f64 / Self::SMALL_POOL_SIZE as f64 * 100.0;
+ let md_util = md_pool as f64 / Self::MEDIUM_POOL_SIZE as f64 * 100.0;
+ let lg_util = lg_pool as f64 / Self::LARGE_POOL_SIZE as f64 * 100.0;
+ let xl_util = xl_pool as f64 / Self::EXTRA_LARGE_POOL_SIZE as f64 * 100.0;
+ let mx_util = mx_pool as f64 / Self::MAX_POOL_SIZE as f64 * 100.0;
+
+ info!("BytesPool: Small[{}/{}|{:.1}%|{:.1}%|{}] Medium[{}/{}|{:.1}%|{:.1}%|{}] Large[{}/{}|{:.1}%|{:.1}%|{}]",
+ sm_returned, sm_created, sm_reuse, sm_util, sm_size,
+ md_returned, md_created, md_reuse, md_util, md_size,
+ lg_returned, lg_created, lg_reuse, lg_util, lg_size);
+
+ info!(
+ "BytesPool: XLarge[{}/{}|{:.1}%|{:.1}%|{}] Max[{}/{}|{:.1}%|{:.1}%|{}] Total: {}",
+ xl_returned,
+ xl_created,
+ xl_reuse,
+ xl_util,
+ xl_size,
+ mx_returned,
+ mx_created,
+ mx_reuse,
+ mx_util,
+ mx_size,
+ total_size
+ );
+ }
+}
+
+impl Default for BytesMutPool {
+ fn default() -> Self {
+ Self {
+ small_buffers: Arc::new(ArrayQueue::new(Self::SMALL_POOL_SIZE)),
+ medium_buffers: Arc::new(ArrayQueue::new(Self::MEDIUM_POOL_SIZE)),
+ large_buffers: Arc::new(ArrayQueue::new(Self::LARGE_POOL_SIZE)),
+ extra_large_buffers: Arc::new(ArrayQueue::new(Self::EXTRA_LARGE_POOL_SIZE)),
+ max_buffers: Arc::new(ArrayQueue::new(Self::MAX_POOL_SIZE)),
+ small_created: Arc::new(AtomicUsize::new(0)),
+ medium_created: Arc::new(AtomicUsize::new(0)),
+ large_created: Arc::new(AtomicUsize::new(0)),
+ extra_large_created: Arc::new(AtomicUsize::new(0)),
+ max_created: Arc::new(AtomicUsize::new(0)),
+ small_returned: Arc::new(AtomicUsize::new(0)),
+ medium_returned: Arc::new(AtomicUsize::new(0)),
+ large_returned: Arc::new(AtomicUsize::new(0)),
+ extra_large_returned: Arc::new(AtomicUsize::new(0)),
+ max_returned: Arc::new(AtomicUsize::new(0)),
+ }
+ }
+}
+
+/// Extension trait for more ergonomic buffer return
+pub trait BytesMutExt {
+ fn return_to_pool(self);
+}
+
+impl BytesMutExt for BytesMut {
+ fn return_to_pool(self) {
+ BYTES_MUT_POOL.return_buffer(self);
+ }
+}
diff --git a/server/src/streaming/utils/mod.rs b/server/src/streaming/utils/mod.rs
index 33481414..25fc7d43 100644
--- a/server/src/streaming/utils/mod.rs
+++ b/server/src/streaming/utils/mod.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+pub mod bytes_mut_pool;
pub mod crypto;
pub mod file;
pub mod hash;