You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/02/26 15:24:01 UTC
(iggy) branch zero_copy_schema updated (605f1457 -> 62376591)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch zero_copy_schema
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 605f1457 send response vectored
new 62376591 send response vectored
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (605f1457)
\
N -- N -- N refs/heads/zero_copy_schema (62376591)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 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:
bench/src/actors/consumer.rs | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
(iggy) 01/01: send response vectored
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch zero_copy_schema
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 62376591abe5b3b1d9d99f60c1a9d0ceec532875
Author: numinex <g....@gmail.com>
AuthorDate: Wed Feb 26 09:12:24 2025 +0100
send response vectored
---
bench/src/actors/consumer.rs | 20 +++--
sdk/src/binary/messages.rs | 10 ++-
sdk/src/clients/client.rs | 9 +--
sdk/src/messages/send_messages.rs | 5 +-
sdk/src/models/batch.rs | 92 ++++++++++++++--------
.../handlers/messages/poll_messages_handler.rs | 20 ++++-
server/src/binary/sender.rs | 7 ++
server/src/quic/quic_sender.rs | 12 +++
server/src/streaming/batching/batch_accumulator.rs | 8 +-
server/src/streaming/partitions/messages.rs | 7 +-
server/src/streaming/segments/fetch_result.rs | 11 ++-
server/src/streaming/segments/logs/log_reader.rs | 15 +++-
server/src/streaming/segments/logs/log_writer.rs | 6 +-
.../src/streaming/segments/logs/persister_task.rs | 8 +-
server/src/streaming/segments/mod.rs | 4 +-
server/src/streaming/segments/reading_messages.rs | 2 +-
server/src/streaming/segments/writing_messages.rs | 4 +-
server/src/streaming/systems/messages.rs | 8 +-
server/src/streaming/topics/messages.rs | 12 +--
server/src/tcp/connection_handler.rs | 6 +-
server/src/tcp/sender.rs | 21 ++++-
server/src/tcp/tcp_sender.rs | 8 ++
server/src/tcp/tcp_tls_sender.rs | 8 ++
23 files changed, 206 insertions(+), 97 deletions(-)
diff --git a/bench/src/actors/consumer.rs b/bench/src/actors/consumer.rs
index 5a92bfb2..ebe56478 100644
--- a/bench/src/actors/consumer.rs
+++ b/bench/src/actors/consumer.rs
@@ -15,6 +15,7 @@ use iggy_bench_report::actor_kind::ActorKind;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::{login_root, ClientFactory};
+use std::hint::black_box;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -204,7 +205,7 @@ impl Consumer {
),
};
let before_poll = Instant::now();
- let polled_messages = client
+ let batch = client
.poll_messages(
&stream_id,
&topic_id,
@@ -215,7 +216,7 @@ impl Consumer {
auto_commit,
)
.await;
- if let Err(e) = polled_messages {
+ if let Err(e) = batch {
if matches!(e, IggyError::TopicIdNotFound(_, _)) {
topic_not_found_counter += 1;
if topic_not_found_counter > 1000 {
@@ -228,9 +229,8 @@ impl Consumer {
continue;
}
- /*
- let polled_messages = polled_messages.unwrap();
- if polled_messages.messages.is_empty() {
+ let batch = batch.unwrap();
+ if batch.batch.is_empty() {
// Store initial poll timestamp if this is the first attempt
if initial_poll_timestamp.is_none() {
initial_poll_timestamp = Some(before_poll);
@@ -253,6 +253,7 @@ impl Consumer {
continue;
}
+ /*
if polled_messages.messages.len() != messages_per_batch as usize {
let should_warn = last_warning_time
.map(|t| t.elapsed() >= Duration::from_secs(1))
@@ -289,18 +290,15 @@ impl Consumer {
self.batches_left_to_receive.fetch_sub(1, Ordering::AcqRel);
- /*
- received_messages += polled_messages.messages.len() as u64;
+ received_messages += batch.iter().count() as u64;
// We don't need to calculate the size whole batch every time by iterating over it - just always use the size of the first message
if batch_user_size_bytes == 0 || batch_size_total_bytes == 0 {
batch_user_size_bytes =
- polled_messages.messages[0].payload.len() as u64 * messages_per_batch as u64;
+ batch.batch.len() as u64;
batch_size_total_bytes =
- polled_messages.messages[0].get_size_bytes().as_bytes_u64()
- * messages_per_batch as u64;
+ batch.batch.len() as u64;
}
- */
total_user_data_bytes += IggyByteSize::from(batch_user_size_bytes);
total_bytes += IggyByteSize::from(batch_size_total_bytes);
diff --git a/sdk/src/binary/messages.rs b/sdk/src/binary/messages.rs
index 89413979..bbffecd2 100644
--- a/sdk/src/binary/messages.rs
+++ b/sdk/src/binary/messages.rs
@@ -9,7 +9,7 @@ use crate::messages::flush_unsaved_buffer::FlushUnsavedBuffer;
use crate::messages::poll_messages::PollingStrategy;
use crate::messages::send_messages::{Message, Partitioning};
use crate::messages::{poll_messages, send_messages};
-use crate::models::batch::IggyBatch;
+use crate::models::batch::{IggyBatch, IggyHeader, IGGY_BATCH_OVERHEAD};
#[async_trait::async_trait]
impl<B: BinaryClient> MessageClient for B {
@@ -38,8 +38,12 @@ impl<B: BinaryClient> MessageClient for B {
),
)
.await?;
- //TODO: Fix me
- todo!()
+ let mut position = 0;
+ let header = IggyHeader::from_bytes(&response[..IGGY_BATCH_OVERHEAD as usize]);
+ position += IGGY_BATCH_OVERHEAD as usize;
+ let bytes = response.slice(position..);
+ let batch = IggyBatch::new(header, bytes);
+ Ok(batch)
}
async fn send_messages(
diff --git a/sdk/src/clients/client.rs b/sdk/src/clients/client.rs
index ccacea25..5e2f359b 100644
--- a/sdk/src/clients/client.rs
+++ b/sdk/src/clients/client.rs
@@ -544,12 +544,11 @@ impl MessageClient for IggyClient {
count: u32,
auto_commit: bool,
) -> Result<IggyBatch, IggyError> {
- /*
if count == 0 {
return Err(IggyError::InvalidMessagesCount);
}
- let mut polled_messages = self
+ let batch = self
.client
.read()
.await
@@ -564,6 +563,8 @@ impl MessageClient for IggyClient {
)
.await?;
+ // TODO: Fix me
+ /*
if let Some(ref encryptor) = self.encryptor {
for message in &mut polled_messages.messages {
let payload = encryptor.decrypt(&message.payload)?;
@@ -571,10 +572,8 @@ impl MessageClient for IggyClient {
message.length = IggyByteSize::from(message.payload.len() as u64);
}
}
-
- Ok(polled_messages)
*/
- todo!();
+ Ok(batch)
}
async fn send_messages(
diff --git a/sdk/src/messages/send_messages.rs b/sdk/src/messages/send_messages.rs
index c2185107..6c642794 100644
--- a/sdk/src/messages/send_messages.rs
+++ b/sdk/src/messages/send_messages.rs
@@ -467,8 +467,9 @@ pub(crate) fn as_bytes(
// TODO: create a writer method on the `Message` and in the future
// once the `Message` struct is dropped on the `IggyMessage`.
let headers_len = header::get_headers_size_bytes(&message.headers).as_bytes_u64();
- let payload_len = message.payload.len() as u64 + 16;
- let total_len = headers_len + payload_len;
+ let payload_len = message.payload.len() as u64;
+ // 8 for the payload_len and 16 for the id
+ let total_len = headers_len + payload_len + 8 + 16;
bytes.extend_from_slice(&0u32.to_le_bytes()); // offset_delta
bytes.extend_from_slice(&0u32.to_le_bytes()); // timestamp_delta
diff --git a/sdk/src/models/batch.rs b/sdk/src/models/batch.rs
index e388e971..f4b691d1 100644
--- a/sdk/src/models/batch.rs
+++ b/sdk/src/models/batch.rs
@@ -1,3 +1,4 @@
+use bytes::{Bytes, BytesMut};
use serde_with::serde_as;
use std::{io::Write, ops::Range};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -225,15 +226,25 @@ impl<const N: usize> std::io::Write for HeaderWriter<N> {
}
}
+// TODO: THIS IS SHIIIIIIIIIT, fuck Bytes
#[serde_as]
#[derive(Debug, Default, PartialEq)]
pub struct IggyBatch {
pub header: IggyHeader,
- pub batch: Vec<u8>,
+ //TODO: Once the dependency on `Bytes` is dropped, change it to Vec<u8>.
+ pub batch: Bytes,
}
-impl IggyBatch {
- pub fn new(header: IggyHeader, batch: Vec<u8>) -> Self {
+#[serde_as]
+#[derive(Debug, Default, PartialEq)]
+pub struct IggyMutableBatch {
+ pub header: IggyHeader,
+ //TODO: Once the dependency on `Bytes` is dropped, change it to Vec<u8>.
+ pub batch: BytesMut,
+}
+
+impl IggyMutableBatch {
+ pub fn new(header: IggyHeader, batch: BytesMut) -> Self {
Self { header, batch }
}
@@ -243,7 +254,6 @@ impl IggyBatch {
base_timestamp: IggyTimestamp,
) {
let header = &mut self.header;
- let batch = &mut self.batch;
header.base_offset = base_offset;
header.base_timestamp = base_timestamp.into();
@@ -252,23 +262,24 @@ impl IggyBatch {
let mut position = 0;
let mut messages_count = 0u32;
let mut timestamp_delta = 0;
- while position < batch.len() {
+ while position < self.batch.len() {
let offset_delta = messages_count.to_le_bytes();
- write_value_at(batch, offset_delta, position);
+ write_value_at(&mut self.batch, offset_delta, position);
position += 4;
let duration_delta = IggyTimestamp::now() - base_timestamp;
timestamp_delta = duration_delta.as_micros() as u32;
let timestamp_delta_value = timestamp_delta.to_le_bytes();
- write_value_at(batch, timestamp_delta_value, position);
+ write_value_at(&mut self.batch, timestamp_delta_value, position);
position += 4;
- let msg_len = u64::from_le_bytes(batch[position..position + 8].try_into().unwrap()) + 8;
- position += 8 + msg_len as usize;
+ let total_len = u64::from_le_bytes(self.batch[position..position + 8].try_into().unwrap());
+ position += 8;
+ position += total_len as usize;
messages_count += 1;
}
header.last_offset_delta = messages_count - 1;
header.last_timestamp_delta = timestamp_delta;
- header.batch_length = batch.len() as u64;
+ header.batch_length = self.batch.len() as u64;
fn write_value_at<const N: usize>(slice: &mut [u8], value: [u8; N], position: usize) {
let slice = &mut slice[position..position + N];
@@ -299,6 +310,31 @@ impl IggyBatch {
}
}
+impl IggyBatch {
+ pub fn new(header: IggyHeader, batch: Bytes) -> Self {
+ Self { header, batch }
+ }
+
+ pub fn write_header<const N: usize>(&self, writer: &mut HeaderWriter<N>) {
+ self.header.write_header(writer);
+ }
+
+ pub async fn write_messages<W: AsyncWriteExt + Unpin>(&self, writer: &mut W) {
+ writer.write_all(&self.batch).await.unwrap();
+ }
+
+ pub fn get_size(&self) -> IggyByteSize {
+ let batch_size = IggyByteSize::from(self.batch.len() as u64);
+ batch_size
+ }
+
+ pub fn get_size_w_header(&self) -> IggyByteSize {
+ let header_size = IggyByteSize::from(IGGY_BATCH_OVERHEAD);
+ let batch_size = IggyByteSize::from(self.batch.len() as u64);
+ header_size + batch_size
+ }
+}
+
#[derive(Debug, PartialEq)]
pub struct IggyMessageView<'msg> {
pub id: u128,
@@ -339,7 +375,7 @@ impl<'batch> IntoIterator for &'batch IggyBatch {
}
impl<'batch> IggyBatch {
- pub fn iter(&self) -> IggyMessageIterator<'batch> {
+ pub fn iter(&'batch self) -> IggyMessageIterator<'batch> {
IggyMessageIterator {
batch: self,
position: 0,
@@ -356,47 +392,33 @@ impl<'msg> Iterator for IggyMessageIterator<'msg> {
type Item = (Range<usize>, IggyMessageView<'msg>);
fn next(&mut self) -> Option<Self::Item> {
- let data = &self.batch.batch;
+ if self.position >= self.batch.batch.len() {
+ return None
+ }
+ let data = self.batch.batch.as_ref();
let start_pos = self.position;
let mut pos = self.position;
- /*
- let mut position = 0;
- let mut messages_count = 0u32;
- let mut timestamp_delta = 0;
- while position < batch.len() {
- let offset_delta = messages_count.to_le_bytes();
- write_value_at(batch, offset_delta, position);
- position += 4;
- let duration_delta = IggyTimestamp::now() - base_timestamp;
- timestamp_delta = duration_delta.as_micros() as u32;
- let timestamp_delta_value = timestamp_delta.to_le_bytes();
- write_value_at(batch, timestamp_delta_value, position);
-
- position += 4;
- let msg_len = u64::from_le_bytes(batch[position..position + 8].try_into().unwrap()) + 8;
- position += 8 + msg_len as usize;
- messages_count += 1;
- }
- */
//TODO: replace all of those oks with an Faillable iterator I guess ?
let offset_delta = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
pos += 4;
let timestamp_delta = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
pos += 4;
let total_length = u64::from_le_bytes(data[pos..pos + 8].try_into().ok()?);
+ // Subtract the id and payload_length field
+ let msg_len = total_length - 16 - 8;
pos += 8;
let payload_length = u64::from_le_bytes(data[pos..pos + 8].try_into().ok()?);
pos += 8;
- let headers_length = total_length - payload_length;
let id = u128::from_le_bytes(data[pos..pos + 16].try_into().ok()?);
pos += 16;
+ let headers_length = msg_len - payload_length;
- let payload = &data[pos..payload_length as usize];
+ let payload = &data[pos..pos + payload_length as usize];
pos += payload_length as usize;
- let headers = &data[pos..headers_length as usize];
+ let headers = &data[pos..pos + headers_length as usize];
pos += headers_length as usize;
- self.position += pos;
+ self.position = pos;
let offset = self.batch.header.base_offset + u64::from(offset_delta);
let timestamp = self.batch.header.base_timestamp + timestamp_delta as u64;
diff --git a/server/src/binary/handlers/messages/poll_messages_handler.rs b/server/src/binary/handlers/messages/poll_messages_handler.rs
index d2342589..059bb52a 100644
--- a/server/src/binary/handlers/messages/poll_messages_handler.rs
+++ b/server/src/binary/handlers/messages/poll_messages_handler.rs
@@ -1,3 +1,5 @@
+use std::io::IoSlice;
+
use crate::binary::handlers::messages::COMPONENT;
use crate::binary::mapper;
use crate::binary::sender::SenderKind;
@@ -8,6 +10,7 @@ use anyhow::Result;
use error_set::ErrContext;
use iggy::error::IggyError;
use iggy::messages::poll_messages::PollMessages;
+use iggy::models::batch::IGGY_BATCH_OVERHEAD;
use tracing::debug;
pub async fn handle(
@@ -32,6 +35,21 @@ pub async fn handle(
"{COMPONENT} (error: {error}) - failed to poll messages for consumer: {}, stream_id: {}, topic_id: {}, partition_id: {:?}, session: {}.",
command.consumer, command.stream_id, command.topic_id, command.partition_id, session
))?;
- sender.send_ok_response(&[]).await?;
+ let length = result.slices.iter().map(|s| s.range.len() as u32).sum::<u32>() + IGGY_BATCH_OVERHEAD as u32;
+ let length = length.to_le_bytes();
+ // Adding 1 for the header and 2 for the prefix required by `send_ok_response`.
+ let mut slices = Vec::with_capacity(1 + result.slices.len() + 2);
+ let header = result.header.as_bytes();
+ slices.push(IoSlice::new(&header));
+ for slice in result.slices.iter() {
+ // Ergh... it is kinda ugly that we have to index this way, rather than just passing `Range`,
+ // but borrow checker gets really anal about it, since the range is field of `IggyBatchSlice`
+ // that we borrow inside of that loop.
+ // It is fixable by decoupling `Range` from `Bytes`, but w/e.
+ let slice = &slice.bytes[slice.range.start..slice.range.end];
+ slices.push(IoSlice::new(slice));
+ }
+
+ sender.send_ok_response_vectored(&length, slices).await?;
Ok(())
}
diff --git a/server/src/binary/sender.rs b/server/src/binary/sender.rs
index 0c1291e9..965efe71 100644
--- a/server/src/binary/sender.rs
+++ b/server/src/binary/sender.rs
@@ -1,4 +1,5 @@
use std::future::Future;
+use std::io::IoSlice;
use crate::tcp::tcp_sender::TcpSender;
use crate::tcp::tcp_tls_sender::TcpTlsSender;
@@ -35,6 +36,11 @@ pub trait Sender {
&mut self,
payload: &[u8],
) -> impl Future<Output = Result<(), IggyError>> + Send;
+ fn send_ok_response_vectored(
+ &mut self,
+ length: &[u8],
+ slices: Vec<IoSlice<'_>>,
+ ) -> impl Future<Output = Result<(), IggyError>> + Send;
fn send_error_response(
&mut self,
error: IggyError,
@@ -68,6 +74,7 @@ impl SenderKind {
async fn read(&mut self, buffer: &mut [u8]) -> Result<usize, IggyError>;
async fn send_empty_ok_response(&mut self) -> Result<(), IggyError>;
async fn send_ok_response(&mut self, payload: &[u8]) -> Result<(), IggyError>;
+ async fn send_ok_response_vectored(&mut self, length: &[u8], slices: Vec<IoSlice<'_>>) -> Result<(), IggyError>;
async fn send_error_response(&mut self, error: IggyError) -> Result<(), IggyError>;
async fn shutdown(&mut self) -> Result<(), ServerError>;
}
diff --git a/server/src/quic/quic_sender.rs b/server/src/quic/quic_sender.rs
index 7e7be69b..7e43965b 100644
--- a/server/src/quic/quic_sender.rs
+++ b/server/src/quic/quic_sender.rs
@@ -1,8 +1,11 @@
+use std::io::IoSlice;
+
use crate::quic::COMPONENT;
use crate::{binary::sender::Sender, server_error::ServerError};
use error_set::ErrContext;
use iggy::error::IggyError;
use quinn::{RecvStream, SendStream};
+use tokio::io::AsyncWriteExt;
use tracing::{debug, error};
const STATUS_OK: &[u8] = &[0; 4];
@@ -39,6 +42,15 @@ impl Sender for QuicSender {
async fn shutdown(&mut self) -> Result<(), ServerError> {
Ok(())
}
+
+ async fn send_ok_response_vectored(
+ &mut self,
+ length: &[u8],
+ slices: Vec<IoSlice<'_>>,
+ ) -> Result<(), IggyError> {
+ //TODO: Fix me, quinn seems to have support for write vectored.
+ todo!()
+ }
}
impl QuicSender {
diff --git a/server/src/streaming/batching/batch_accumulator.rs b/server/src/streaming/batching/batch_accumulator.rs
index b1be9e8d..4ac636b6 100644
--- a/server/src/streaming/batching/batch_accumulator.rs
+++ b/server/src/streaming/batching/batch_accumulator.rs
@@ -1,4 +1,4 @@
-use iggy::models::batch::{IggyBatch, IggyHeader};
+use iggy::models::batch::{IggyBatch, IggyHeader, IggyMutableBatch};
use iggy::utils::byte_size::IggyByteSize;
use std::sync::Arc;
@@ -9,7 +9,7 @@ pub struct BatchAccumulator {
current_offset: u64,
current_timestamp: u64,
leading_header: Option<IggyHeader>,
- batches: Vec<IggyBatch>,
+ batches: Vec<IggyMutableBatch>,
}
impl BatchAccumulator {
@@ -24,7 +24,7 @@ impl BatchAccumulator {
}
}
- pub fn append(&mut self, batch_size: IggyByteSize, batch: IggyBatch) {
+ pub fn append(&mut self, batch_size: IggyByteSize, batch: IggyMutableBatch) {
let header = batch.header;
if self.leading_header.is_none() {
self.base_offset = header.base_offset;
@@ -88,7 +88,7 @@ impl BatchAccumulator {
self.base_offset
}
- pub fn materialize(self) -> (IggyHeader, Vec<IggyBatch>) {
+ pub fn materialize(self) -> (IggyHeader, Vec<IggyMutableBatch>) {
(
self.leading_header
.expect("No leading header when materializing batch"),
diff --git a/server/src/streaming/partitions/messages.rs b/server/src/streaming/partitions/messages.rs
index 3fbd226d..ab3a1b8c 100644
--- a/server/src/streaming/partitions/messages.rs
+++ b/server/src/streaming/partitions/messages.rs
@@ -6,6 +6,7 @@ use crate::streaming::segments::*;
use error_set::ErrContext;
use iggy::error::IggyError;
use iggy::messages::send_messages::Message;
+use iggy::models::batch::IggyMutableBatch;
use iggy::utils::timestamp::IggyTimestamp;
use iggy::{confirmation::Confirmation, models::batch::IggyBatch};
use std::sync::{atomic::Ordering, Arc};
@@ -98,7 +99,9 @@ impl Partition {
.await
}
//_ => Self::get_messages_from_segments(segments, start_offset, count).await,
- _ => panic!("todo, the method from above is already implemented, need to flatten the results")
+ _ => panic!(
+ "todo, the method from above is already implemented, need to flatten the results"
+ ),
}
}
@@ -367,7 +370,7 @@ impl Partition {
pub async fn append_messages(
&mut self,
appendable_batch_info: AppendableBatchInfo,
- batch: IggyBatch,
+ batch: IggyMutableBatch,
confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
{
diff --git a/server/src/streaming/segments/fetch_result.rs b/server/src/streaming/segments/fetch_result.rs
index 8f9ed869..03d76e65 100644
--- a/server/src/streaming/segments/fetch_result.rs
+++ b/server/src/streaming/segments/fetch_result.rs
@@ -1,3 +1,4 @@
+use bytes::Bytes;
use iggy::models::batch::IggyHeader;
use std::ops::Range;
@@ -5,11 +6,11 @@ use std::ops::Range;
pub struct IggyBatchSlice {
pub range: Range<usize>,
pub header: IggyHeader,
- pub bytes: Vec<u8>,
+ pub bytes: Bytes,
}
impl IggyBatchSlice {
- pub fn new(range: Range<usize>, header: IggyHeader, bytes: Vec<u8>) -> Self {
+ pub fn new(range: Range<usize>, header: IggyHeader, bytes: Bytes) -> Self {
Self {
range,
header,
@@ -27,6 +28,10 @@ pub struct IggyBatchFetchResult {
impl IggyBatchFetchResult {
pub fn new(slices: Vec<IggyBatchSlice>, msg_count: u32, header: IggyHeader) -> Self {
- Self { slices, msg_count, header }
+ Self {
+ slices,
+ msg_count,
+ header,
+ }
}
}
diff --git a/server/src/streaming/segments/logs/log_reader.rs b/server/src/streaming/segments/logs/log_reader.rs
index 33a0e5cb..eaab2863 100644
--- a/server/src/streaming/segments/logs/log_reader.rs
+++ b/server/src/streaming/segments/logs/log_reader.rs
@@ -1,5 +1,5 @@
use crate::streaming::segments::indexes::IndexRange;
-use bytes::BytesMut;
+use bytes::{Bytes, BytesMut};
use error_set::ErrContext;
use iggy::{
error::IggyError,
@@ -320,7 +320,7 @@ impl SegmentLogReader {
return Ok((None, 0));
}
- let batch_bytes = match self.read_at(batch_position, batch_length as u64).await {
+ let batch_bytes = match self.read_bytes_at(batch_position, batch_length as u64).await {
Ok(buf) => buf,
Err(error) if error.kind() == ErrorKind::UnexpectedEof => return Ok((None, 0)),
Err(error) => {
@@ -351,4 +351,15 @@ impl SegmentLogReader {
})
.await?
}
+
+ async fn read_bytes_at(&self, offset: u64, len: u64) -> Result<Bytes, std::io::Error> {
+ let file = self.file.clone();
+ spawn_blocking(move || {
+ let mut buf = BytesMut::with_capacity(len as usize);
+ unsafe {buf.set_len(len as usize)};
+ file.read_exact_at(&mut buf, offset)?;
+ Ok(buf.freeze())
+ })
+ .await?
+ }
}
diff --git a/server/src/streaming/segments/logs/log_writer.rs b/server/src/streaming/segments/logs/log_writer.rs
index fb35a0cc..a2f6fd44 100644
--- a/server/src/streaming/segments/logs/log_writer.rs
+++ b/server/src/streaming/segments/logs/log_writer.rs
@@ -3,7 +3,7 @@ use error_set::ErrContext;
use iggy::{
confirmation::Confirmation,
error::IggyError,
- models::batch::{IggyBatch, IggyHeader, IGGY_BATCH_OVERHEAD},
+ models::batch::{IggyBatch, IggyHeader, IggyMutableBatch, IGGY_BATCH_OVERHEAD},
utils::{byte_size::IggyByteSize, duration::IggyDuration, sizeable::Sizeable},
};
use std::{
@@ -95,7 +95,7 @@ impl SegmentLogWriter {
pub async fn save_batches(
&mut self,
header: IggyHeader,
- batches: Vec<IggyBatch>,
+ batches: Vec<IggyMutableBatch>,
confirmation: Confirmation,
) -> Result<IggyByteSize, IggyError> {
let batch_size =
@@ -131,7 +131,7 @@ impl SegmentLogWriter {
async fn write_batch(
&mut self,
header: IggyHeader,
- batches: Vec<IggyBatch>,
+ batches: Vec<IggyMutableBatch>,
) -> Result<(), IggyError> {
if let Some(ref mut file) = self.file {
let mut slices = Vec::new();
diff --git a/server/src/streaming/segments/logs/persister_task.rs b/server/src/streaming/segments/logs/persister_task.rs
index 116460f3..47808593 100644
--- a/server/src/streaming/segments/logs/persister_task.rs
+++ b/server/src/streaming/segments/logs/persister_task.rs
@@ -1,7 +1,7 @@
use flume::{unbounded, Receiver};
use iggy::{
error::IggyError,
- models::batch::{IggyBatch, IggyHeader, IGGY_BATCH_OVERHEAD},
+ models::batch::{IggyBatch, IggyHeader, IggyMutableBatch, IGGY_BATCH_OVERHEAD},
utils::duration::IggyDuration,
};
use std::{
@@ -18,7 +18,7 @@ use tracing::{error, trace, warn};
#[derive(Debug)]
/// A command to the persister task.
enum PersisterTaskCommand {
- WriteRequest(u64, IggyHeader, Vec<IggyBatch>),
+ WriteRequest(u64, IggyHeader, Vec<IggyMutableBatch>),
Shutdown,
}
@@ -63,7 +63,7 @@ impl PersisterTask {
}
/// Sends the batch bytes to the persister task (fire-and-forget).
- pub async fn persist(&self, batch_size: u64, header: IggyHeader, batches: Vec<IggyBatch>) {
+ pub async fn persist(&self, batch_size: u64, header: IggyHeader, batches: Vec<IggyMutableBatch>) {
if let Err(e) = self
.sender
.send_async(PersisterTaskCommand::WriteRequest(
@@ -223,7 +223,7 @@ impl PersisterTask {
file_path: &str,
batch_size: u64,
header: IggyHeader,
- batches: Vec<IggyBatch>,
+ batches: Vec<IggyMutableBatch>,
fsync: bool,
max_retries: u32,
retry_delay: IggyDuration,
diff --git a/server/src/streaming/segments/mod.rs b/server/src/streaming/segments/mod.rs
index 06208b3c..324a19f4 100644
--- a/server/src/streaming/segments/mod.rs
+++ b/server/src/streaming/segments/mod.rs
@@ -1,13 +1,13 @@
+mod fetch_result;
mod indexes;
mod logs;
mod reading_messages;
mod segment;
mod writing_messages;
-mod fetch_result;
+pub use fetch_result::*;
pub use indexes::Index;
pub use segment::Segment;
-pub use fetch_result::*;
pub const LOG_EXTENSION: &str = "log";
pub const INDEX_EXTENSION: &str = "index";
diff --git a/server/src/streaming/segments/reading_messages.rs b/server/src/streaming/segments/reading_messages.rs
index 1fc488e3..aa9c3f9d 100644
--- a/server/src/streaming/segments/reading_messages.rs
+++ b/server/src/streaming/segments/reading_messages.rs
@@ -212,7 +212,7 @@ impl Segment {
let slices = batches
.into_iter()
.map(|batch| {
- let mut ranges = batch.into_iter().filter_map(|(range, msg)| {
+ let mut ranges = batch.iter().filter_map(|(range, msg)| {
if msg.offset >= start_offset && msg.offset <= end_offset {
msg_count += 1;
Some(range)
diff --git a/server/src/streaming/segments/writing_messages.rs b/server/src/streaming/segments/writing_messages.rs
index a083dc7d..d4a7b5fc 100644
--- a/server/src/streaming/segments/writing_messages.rs
+++ b/server/src/streaming/segments/writing_messages.rs
@@ -3,7 +3,7 @@ use crate::streaming::batching::batch_accumulator::BatchAccumulator;
use crate::streaming::segments::segment::Segment;
use error_set::ErrContext;
use iggy::error::IggyError;
-use iggy::models::batch::IGGY_BATCH_OVERHEAD;
+use iggy::models::batch::{IggyMutableBatch, IGGY_BATCH_OVERHEAD};
use iggy::utils::byte_size::IggyByteSize;
use iggy::utils::sizeable::Sizeable;
use iggy::{confirmation::Confirmation, models::batch::IggyBatch};
@@ -15,7 +15,7 @@ impl Segment {
&mut self,
batch_size: IggyByteSize,
messages_count: u32,
- batch: IggyBatch,
+ batch: IggyMutableBatch,
) -> Result<(), IggyError> {
if self.is_closed {
return Err(IggyError::SegmentClosed(
diff --git a/server/src/streaming/systems/messages.rs b/server/src/streaming/systems/messages.rs
index beb6a057..874b0096 100644
--- a/server/src/streaming/systems/messages.rs
+++ b/server/src/streaming/systems/messages.rs
@@ -9,7 +9,7 @@ use iggy::confirmation::Confirmation;
use iggy::consumer::Consumer;
use iggy::messages::poll_messages::PollingStrategy;
use iggy::messages::send_messages::Partitioning;
-use iggy::models::batch::IggyBatch;
+use iggy::models::batch::{IggyBatch, IggyMutableBatch};
use iggy::utils::byte_size::IggyByteSize;
use iggy::utils::sizeable::Sizeable;
use iggy::{error::IggyError, identifier::Identifier};
@@ -66,7 +66,7 @@ impl System {
.await?;
// TODO: Fix me
- /*
+ /*
if polled_messages.messages.is_empty() {
return Ok(polled_messages);
}
@@ -86,7 +86,7 @@ impl System {
if self.encryptor.is_none() {
return Ok(result);
- }
+ }
// TODO: Fix me
/*
let encryptor = self.encryptor.as_ref().unwrap();
@@ -124,7 +124,7 @@ impl System {
stream_id: &Identifier,
topic_id: &Identifier,
partitioning: &Partitioning,
- batch: IggyBatch,
+ batch: IggyMutableBatch,
confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
diff --git a/server/src/streaming/topics/messages.rs b/server/src/streaming/topics/messages.rs
index 774c9e33..6ccbe7a1 100644
--- a/server/src/streaming/topics/messages.rs
+++ b/server/src/streaming/topics/messages.rs
@@ -12,7 +12,7 @@ use iggy::error::IggyError;
use iggy::locking::IggySharedMutFn;
use iggy::messages::poll_messages::{PollingKind, PollingStrategy};
use iggy::messages::send_messages::{Message, Partitioning, PartitioningKind};
-use iggy::models::batch::IggyBatch;
+use iggy::models::batch::{IggyBatch, IggyMutableBatch};
use iggy::utils::byte_size::IggyByteSize;
use iggy::utils::expiry::IggyExpiry;
use iggy::utils::sizeable::Sizeable;
@@ -59,11 +59,11 @@ impl Topic {
.await
.with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to get messages by timestamp: {value}, count: {count}"))
*/
- todo!()
+ todo!()
}
PollingKind::First => todo!(), //partition.get_first_messages(count).await,
- PollingKind::Last => todo!(), //partition.get_last_messages(count).await,
- PollingKind::Next => todo!(), //partition.get_next_messages(consumer, count).await,
+ PollingKind::Last => todo!(), //partition.get_last_messages(count).await,
+ PollingKind::Next => todo!(), //partition.get_next_messages(consumer, count).await,
}?;
/*
@@ -80,7 +80,7 @@ impl Topic {
&self,
batch_size: IggyByteSize,
partitioning: &Partitioning,
- batch: IggyBatch,
+ batch: IggyMutableBatch,
confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
if !self.has_partitions() {
@@ -137,7 +137,7 @@ impl Topic {
async fn append_messages_to_partition(
&self,
appendable_batch_info: AppendableBatchInfo,
- batch: IggyBatch,
+ batch: IggyMutableBatch,
confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
let partition = self.partitions.get(&appendable_batch_info.partition_id);
diff --git a/server/src/tcp/connection_handler.rs b/server/src/tcp/connection_handler.rs
index cb0a4ba6..705758f8 100644
--- a/server/src/tcp/connection_handler.rs
+++ b/server/src/tcp/connection_handler.rs
@@ -10,7 +10,7 @@ use iggy::command::SEND_MESSAGES_CODE;
use iggy::error::IggyError;
use iggy::identifier::Identifier;
use iggy::messages::send_messages::Partitioning;
-use iggy::models::batch::{IggyBatch, IggyHeader, IGGY_BATCH_OVERHEAD};
+use iggy::models::batch::{IggyBatch, IggyHeader, IggyMutableBatch, IGGY_BATCH_OVERHEAD};
use iggy::utils::sizeable::Sizeable;
use iggy::validatable::Validatable;
use std::io::ErrorKind;
@@ -73,10 +73,10 @@ pub(crate) async fn handle_connection(
let header = IggyHeader::from_bytes(&header_buffer);
let batch_length = length - metadata_len - IGGY_BATCH_OVERHEAD as u32 - 4;
- let mut batch_buffer = Vec::with_capacity(batch_length as _);
+ let mut batch_buffer = BytesMut::with_capacity(batch_length as _);
unsafe { batch_buffer.set_len(batch_length as _) };
sender.read(&mut batch_buffer).await?;
- let batch = IggyBatch::new(header, batch_buffer);
+ let batch = IggyMutableBatch::new(header, batch_buffer);
let system = system.read().await;
{
system
diff --git a/server/src/tcp/sender.rs b/server/src/tcp/sender.rs
index c6ab4f59..68750fe9 100644
--- a/server/src/tcp/sender.rs
+++ b/server/src/tcp/sender.rs
@@ -37,6 +37,17 @@ where
send_response(stream, STATUS_OK, payload).await
}
+pub(crate) async fn send_ok_response_vectored<T>(
+ stream: &mut T,
+ length: &[u8],
+ slices: Vec<IoSlice<'_>>,
+) -> Result<(), IggyError>
+where
+ T: AsyncRead + AsyncWrite + Unpin,
+{
+ send_response_vectored(stream, STATUS_OK, length, slices).await
+}
+
pub(crate) async fn send_error_response<T>(
stream: &mut T,
error: IggyError,
@@ -68,15 +79,17 @@ where
pub(crate) async fn send_response_vectored<T>(
stream: &mut T,
status: &[u8],
- slices: [IoSlice]
+ length: &[u8],
+ mut slices: Vec<IoSlice<'_>>,
) -> Result<(), IggyError>
where
- T: AsyncRead + AsyncWrite + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
{
debug!("Sending response with status: {:?}...", status);
- let length = (payload.len() as u32).to_le_bytes();
+ let prefix = [IoSlice::new(status), IoSlice::new(length)];
+ slices.splice(0..0, prefix);
stream
- .write_all(&[status, &length, payload].as_slice().concat())
+ .write_vectored(&slices)
.await
.map_err(|_| IggyError::TcpError)?;
debug!("Sent response with status: {:?}", status);
diff --git a/server/src/tcp/tcp_sender.rs b/server/src/tcp/tcp_sender.rs
index a87325eb..66e82da0 100644
--- a/server/src/tcp/tcp_sender.rs
+++ b/server/src/tcp/tcp_sender.rs
@@ -36,4 +36,12 @@ impl Sender for TcpSender {
})
.map_err(ServerError::IoError)
}
+
+ async fn send_ok_response_vectored(
+ &mut self,
+ length: &[u8],
+ slices: Vec<std::io::IoSlice<'_>>,
+ ) -> Result<(), IggyError> {
+ sender::send_ok_response_vectored(&mut self.stream, length, slices).await
+ }
}
diff --git a/server/src/tcp/tcp_tls_sender.rs b/server/src/tcp/tcp_tls_sender.rs
index 34356b8a..4d331f7d 100644
--- a/server/src/tcp/tcp_tls_sender.rs
+++ b/server/src/tcp/tcp_tls_sender.rs
@@ -38,4 +38,12 @@ impl Sender for TcpTlsSender {
})
.map_err(ServerError::IoError)
}
+
+ async fn send_ok_response_vectored(
+ &mut self,
+ length: &[u8],
+ slices: Vec<std::io::IoSlice<'_>>,
+ ) -> Result<(), IggyError> {
+ sender::send_ok_response_vectored(&mut self.stream, length, slices).await
+ }
}