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/24 05:59:41 UTC
(iggy) branch zero_copy_schema updated (aef657a2 -> ebb4a79a)
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 aef657a2 part 1 - sending
new ebb4a79a part 1 - sending
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 (aef657a2)
\
N -- N -- N refs/heads/zero_copy_schema (ebb4a79a)
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:
server/src/tcp/connection_handler.rs | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
(iggy) 01/01: part 1 - sending
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 ebb4a79a2fbb9232a2dcfe17f3bfb2f95030a7b4
Author: numinex <g....@gmail.com>
AuthorDate: Sun Feb 23 21:28:14 2025 +0100
part 1 - sending
---
bench/src/actors/consumer.rs | 12 ++-
bench/src/actors/producing_consumer.rs | 10 ++-
bench/src/actors/utils.rs | 5 +-
integration/tests/streaming/stream.rs | 3 +
sdk/src/messages/mod.rs | 1 -
sdk/src/messages/send_messages.rs | 4 +-
sdk/src/messages/send_messages_server.rs | 67 ---------------
sdk/src/models/batch.rs | 36 +++++----
server/src/binary/command.rs | 5 +-
server/src/binary/handlers/messages/mod.rs | 1 -
.../handlers/messages/send_messages_handler.rs | 38 ---------
server/src/command.rs | 13 ++-
.../src/compat/index_rebuilding/index_rebuilder.rs | 8 +-
server/src/http/messages.rs | 4 +
server/src/quic/listener.rs | 4 +
server/src/streaming/batching/batch_accumulator.rs | 94 +++++++++-------------
server/src/streaming/partitions/messages.rs | 10 +--
server/src/streaming/partitions/partition.rs | 2 +-
server/src/streaming/segments/logs/log_reader.rs | 36 ++++++---
server/src/streaming/segments/logs/log_writer.rs | 34 +++++---
.../src/streaming/segments/logs/persister_task.rs | 55 ++++++++-----
server/src/streaming/segments/reading_messages.rs | 16 ++--
server/src/streaming/segments/segment.rs | 4 +
server/src/streaming/segments/writing_messages.rs | 31 +++----
server/src/streaming/systems/messages.rs | 8 +-
server/src/streaming/systems/system.rs | 4 +
server/src/streaming/topics/messages.rs | 23 ++++--
server/src/tcp/connection_handler.rs | 14 ++--
28 files changed, 255 insertions(+), 287 deletions(-)
diff --git a/bench/src/actors/consumer.rs b/bench/src/actors/consumer.rs
index ce2b3e83..2fe93471 100644
--- a/bench/src/actors/consumer.rs
+++ b/bench/src/actors/consumer.rs
@@ -1,4 +1,4 @@
-use super::utils::calculate_latency_from_first_message;
+//use super::utils::calculate_latency_from_first_message;
use crate::analytics::metrics::individual::from_records;
use crate::analytics::record::BenchmarkRecord;
use crate::rate_limiter::RateLimiter;
@@ -129,6 +129,7 @@ impl Consumer {
self.consumer_id, self.warmup_time
);
}
+ /*
let warmup_end = Instant::now() + self.warmup_time.get_duration();
while Instant::now() < warmup_end {
let offset = current_iteration * messages_per_batch as u64;
@@ -161,6 +162,7 @@ impl Consumer {
}
current_iteration += 1;
}
+ */
}
if let Some(cg_id) = self.consumer_group_id {
@@ -226,6 +228,7 @@ impl Consumer {
continue;
}
+ /*
let polled_messages = polled_messages.unwrap();
if polled_messages.messages.is_empty() {
// Store initial poll timestamp if this is the first attempt
@@ -272,9 +275,12 @@ impl Consumer {
continue;
}
+ */
let latency = if self.calculate_latency_from_message_payload {
- calculate_latency_from_first_message(&polled_messages.messages[0])
+ //TODO: Fix me
+ //calculate_latency_from_first_message(&polled_messages.messages[0])
+ initial_poll_timestamp.unwrap_or(before_poll).elapsed()
} else {
initial_poll_timestamp.unwrap_or(before_poll).elapsed()
};
@@ -283,6 +289,7 @@ impl Consumer {
self.batches_left_to_receive.fetch_sub(1, Ordering::AcqRel);
+ /*
received_messages += polled_messages.messages.len() 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
@@ -293,6 +300,7 @@ impl Consumer {
polled_messages.messages[0].get_size_bytes().as_bytes_u64()
* messages_per_batch as u64;
}
+ */
total_user_data_bytes += IggyByteSize::from(batch_user_size_bytes);
total_bytes += IggyByteSize::from(batch_size_total_bytes);
diff --git a/bench/src/actors/producing_consumer.rs b/bench/src/actors/producing_consumer.rs
index 7865e6d8..fac44f9c 100644
--- a/bench/src/actors/producing_consumer.rs
+++ b/bench/src/actors/producing_consumer.rs
@@ -1,4 +1,4 @@
-use crate::actors::utils::{calculate_latency_from_first_message, put_timestamp_in_first_message};
+use crate::actors::utils::put_timestamp_in_first_message;
use crate::analytics::metrics::individual::from_records;
use crate::analytics::record::BenchmarkRecord;
use crate::rate_limiter::RateLimiter;
@@ -172,6 +172,7 @@ impl ProducingConsumer {
self.polling_kind
),
};
+ /*
let polled_messages = client
.poll_messages(
&stream_id,
@@ -205,6 +206,7 @@ impl ProducingConsumer {
continue;
}
+ */
current_offset += messages_per_batch as u64;
}
}
@@ -253,6 +255,7 @@ impl ProducingConsumer {
)
.await?;
+ /*
if polled_messages.messages.len() != messages_per_batch as usize {
let should_warn = last_warning_time
.map(|t| t.elapsed() >= Duration::from_secs(1))
@@ -275,10 +278,13 @@ impl ProducingConsumer {
continue;
}
+ */
// Extract send timestamp from first message in batch
let latency = if self.calculate_latency_from_timestamp_in_first_message {
- calculate_latency_from_first_message(&polled_messages.messages[0])
+ //TODO: Fix me
+ //calculate_latency_from_first_message(&polled_messages.messages[0])
+ before_send.elapsed()
} else {
before_send.elapsed()
};
diff --git a/bench/src/actors/utils.rs b/bench/src/actors/utils.rs
index 90edf19b..90dce53d 100644
--- a/bench/src/actors/utils.rs
+++ b/bench/src/actors/utils.rs
@@ -1,8 +1,9 @@
use std::time::Duration;
use bytes::Bytes;
-use iggy::{messages::send_messages::Message, models::messages::PolledMessage};
+use iggy::messages::send_messages::Message;
+//TODO: Fix me
pub fn put_timestamp_in_first_message(message: &mut Message) {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
@@ -27,6 +28,7 @@ pub fn put_timestamp_in_first_message(message: &mut Message) {
message.payload = Bytes::from(payload_vec);
}
+/*
pub fn calculate_latency_from_first_message(message: &PolledMessage) -> Duration {
let send_timestamp = u64::from_le_bytes(message.payload[0..8].try_into().unwrap());
let now = std::time::SystemTime::now()
@@ -35,3 +37,4 @@ pub fn calculate_latency_from_first_message(message: &PolledMessage) -> Duration
.as_micros() as u64;
Duration::from_micros(now - send_timestamp)
}
+*/
diff --git a/integration/tests/streaming/stream.rs b/integration/tests/streaming/stream.rs
index 0e65f621..3ea4d999 100644
--- a/integration/tests/streaming/stream.rs
+++ b/integration/tests/streaming/stream.rs
@@ -94,6 +94,8 @@ async fn should_delete_existing_stream_from_disk() {
}
}
+//TODO: Fix me
+/*
#[tokio::test]
async fn should_purge_existing_stream_on_disk() {
let setup = TestSetup::init().await;
@@ -163,6 +165,7 @@ async fn should_purge_existing_stream_on_disk() {
assert!(loaded_messages.messages.is_empty());
}
}
+*/
async fn assert_persisted_stream(stream_path: &str, topics_directory: &str) {
let stream_metadata = fs::metadata(stream_path).await.unwrap();
diff --git a/sdk/src/messages/mod.rs b/sdk/src/messages/mod.rs
index 9a15f8b5..8a588efc 100644
--- a/sdk/src/messages/mod.rs
+++ b/sdk/src/messages/mod.rs
@@ -1,7 +1,6 @@
pub mod flush_unsaved_buffer;
pub mod poll_messages;
pub mod send_messages;
-pub mod send_messages_server;
const MAX_HEADERS_SIZE: u32 = 100 * 1000;
pub const MAX_PAYLOAD_SIZE: u32 = 10 * 1000 * 1000;
diff --git a/sdk/src/messages/send_messages.rs b/sdk/src/messages/send_messages.rs
index a337d27d..93bcc446 100644
--- a/sdk/src/messages/send_messages.rs
+++ b/sdk/src/messages/send_messages.rs
@@ -9,7 +9,7 @@ use crate::models::header::{HeaderKey, HeaderValue};
use crate::utils::byte_size::IggyByteSize;
use crate::utils::sizeable::Sizeable;
use crate::validatable::Validatable;
-use bytes::{Buf, BufMut, Bytes, BytesMut};
+use bytes::{BufMut, Bytes, BytesMut};
use serde::{Deserialize, Serialize};
use serde_with::base64::Base64;
use serde_with::serde_as;
@@ -27,6 +27,8 @@ const EMPTY_KEY_VALUE: Vec<u8> = vec![];
/// - `topic_id` - unique topic ID (numeric or name).
/// - `partitioning` - to which partition the messages should be sent - either provided by the client or calculated by the server.
/// - `messages` - collection of messages to be sent.
+// TODO: Fix me, this struct should have `batch` containing IggyBatch instead of `messages` field
+// so it's compatible with the server side deserialization
#[derive(Debug, Serialize, Deserialize, PartialEq)]
pub struct SendMessages {
/// Unique stream ID (numeric or name).
diff --git a/sdk/src/messages/send_messages_server.rs b/sdk/src/messages/send_messages_server.rs
deleted file mode 100644
index ee72403e..00000000
--- a/sdk/src/messages/send_messages_server.rs
+++ /dev/null
@@ -1,67 +0,0 @@
-use super::send_messages::Partitioning;
-use crate::{
- bytes_serializable::BytesSerializable,
- command::{Command, SEND_MESSAGES_CODE},
- error::IggyError,
- identifier::Identifier,
- models::batch::IggyBatch,
- utils::sizeable::Sizeable,
- validatable::Validatable,
-};
-use bytes::Bytes;
-
-#[derive(Debug, PartialEq, Default)]
-pub struct SendMessages {
- pub stream_id: Identifier,
- pub topic_id: Identifier,
- pub partitioning: Partitioning,
- pub batch: IggyBatch,
-}
-
-impl std::fmt::Display for SendMessages {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- write!(
- f,
- "{}|{}|{}|batch_size:{}",
- self.stream_id,
- self.topic_id,
- self.partitioning,
- self.batch.payload.len()
- )
- }
-}
-
-impl Command for SendMessages {
- fn code(&self) -> u32 {
- SEND_MESSAGES_CODE
- }
-}
-
-impl Validatable<IggyError> for SendMessages {
- fn validate(&self) -> Result<(), IggyError> {
- Ok(())
- }
-}
-
-impl BytesSerializable for SendMessages {
- fn from_bytes(bytes: Bytes) -> Result<Self, IggyError> {
- if bytes.len() < 11 {
- return Err(IggyError::InvalidCommand);
- }
-
- todo!();
- /*
- let command = Self {
- stream_id,
- topic_id,
- partitioning: key,
- batch,
- };
- Ok(command)
- */
- }
-
- fn to_bytes(&self) -> Bytes {
- todo!()
- }
-}
diff --git a/sdk/src/models/batch.rs b/sdk/src/models/batch.rs
index e6e485b1..25a3d270 100644
--- a/sdk/src/models/batch.rs
+++ b/sdk/src/models/batch.rs
@@ -10,19 +10,19 @@ pub const IGGY_BATCH_OVERHEAD: u64 = 16 + 16 + 16 + 16 + 8 + 8 + 8 + 8 + 4 + 4 +
pub struct IggyHeader {
// TODO: replace timestamp with `IggyTimestamp`
// and maybe impl `IggyOffset` and other wrapper structs for things like checksums, attributes etc...
- payload_type: u8,
+ pub payload_type: u8,
pub last_offset_delta: u32,
- last_timestamp_delta: u32,
- release: u32,
- attributes: u32,
+ pub last_timestamp_delta: u32,
+ pub release: u32,
+ pub attributes: u32,
pub base_offset: u64,
pub batch_length: u64,
- origin_timestamp: u64,
+ pub origin_timestamp: u64,
pub base_timestamp: u64,
- reserved_nonce: u128,
- parent: u128,
- checksum_body: u128,
- checksum: u128,
+ pub reserved_nonce: u128,
+ pub parent: u128,
+ pub checksum_body: u128,
+ pub checksum: u128,
}
impl IggyHeader {
@@ -251,14 +251,15 @@ impl IggyBatch {
// TODO: Impl iterators (mutable and non-mutable) instead of this.
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 timestamp_delta = base_timestamp - IggyTimestamp::now();
- let timestamp_delta = timestamp_delta.as_micros() as u32;
- let timestamp_delta = timestamp_delta.to_le_bytes();
- write_value_at(batch, timestamp_delta, position);
+ 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());
@@ -266,6 +267,7 @@ impl IggyBatch {
messages_count += 1;
}
header.last_offset_delta = messages_count;
+ header.last_timestamp_delta = timestamp_delta;
fn write_value_at<const N: usize>(slice: &mut [u8], value: [u8; N], position: usize) {
let slice = &mut slice[position..position + N];
@@ -284,10 +286,14 @@ impl IggyBatch {
writer.write_all(&self.batch).await.unwrap();
}
- pub fn get_size_bytes(&self) -> IggyByteSize {
- let header_size = IggyByteSize::from(IGGY_BATCH_OVERHEAD);
+ 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
}
}
diff --git a/server/src/binary/command.rs b/server/src/binary/command.rs
index 6569c3de..d2b4b73e 100644
--- a/server/src/binary/command.rs
+++ b/server/src/binary/command.rs
@@ -125,8 +125,9 @@ async fn try_handle(
ServerCommand::LoginWithPersonalAccessToken(command) => {
login_with_personal_access_token_handler::handle(command, sender, session, system).await
}
- ServerCommand::SendMessages(command) => {
- send_messages_handler::handle(command, sender, session, system).await
+ ServerCommand::SendMessages() => {
+ //TODO: Fix me.
+ Ok(())
}
ServerCommand::PollMessages(command) => {
poll_messages_handler::handle(command, sender, session, system).await
diff --git a/server/src/binary/handlers/messages/mod.rs b/server/src/binary/handlers/messages/mod.rs
index 8c73185b..e9a721d5 100644
--- a/server/src/binary/handlers/messages/mod.rs
+++ b/server/src/binary/handlers/messages/mod.rs
@@ -1,5 +1,4 @@
pub mod flush_unsaved_buffer_handler;
pub mod poll_messages_handler;
-pub mod send_messages_handler;
pub const COMPONENT: &str = "MESSAGE_HANDLER";
diff --git a/server/src/binary/handlers/messages/send_messages_handler.rs b/server/src/binary/handlers/messages/send_messages_handler.rs
deleted file mode 100644
index 3903b444..00000000
--- a/server/src/binary/handlers/messages/send_messages_handler.rs
+++ /dev/null
@@ -1,38 +0,0 @@
-use crate::binary::handlers::messages::COMPONENT;
-use crate::binary::sender::SenderKind;
-use crate::streaming::session::Session;
-use crate::streaming::systems::system::SharedSystem;
-use anyhow::Result;
-use error_set::ErrContext;
-use iggy::error::IggyError;
-use tracing::debug;
-
-pub async fn handle(
- command: iggy::messages::send_messages_server::SendMessages,
- sender: &mut SenderKind,
- session: &Session,
- system: &SharedSystem,
-) -> Result<(), IggyError> {
- //TODO: Fix me
- /*
- debug!("session: {session}, command: {command}");
- let system = system.read().await;
- let stream_id = command.stream_id.clone();
- let topic_id = command.topic_id.clone();
- let partitioning = command.partitioning.clone();
- let messages = command.messages;
- // TODO(haze): Add confirmation level after testing is complete
- system
- .append_messages(session, stream_id, topic_id, partitioning, messages, None)
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to append messages for stream_id: {}, topic_id: {}, partitioning: {}, session: {}",
- command.stream_id, command.topic_id, command.partitioning, session
- )
- })?;
- sender.send_empty_ok_response().await?;
- Ok(())
- */
- todo!();
-}
diff --git a/server/src/command.rs b/server/src/command.rs
index d202a514..94d6c1f9 100644
--- a/server/src/command.rs
+++ b/server/src/command.rs
@@ -73,7 +73,7 @@ pub enum ServerCommand {
CreatePersonalAccessToken(CreatePersonalAccessToken),
DeletePersonalAccessToken(DeletePersonalAccessToken),
LoginWithPersonalAccessToken(LoginWithPersonalAccessToken),
- SendMessages(iggy::messages::send_messages_server::SendMessages),
+ SendMessages(),
PollMessages(PollMessages),
FlushUnsavedBuffer(FlushUnsavedBuffer),
GetConsumerOffset(GetConsumerOffset),
@@ -123,7 +123,8 @@ impl ServerCommand {
ServerCommand::CreatePersonalAccessToken(payload) => as_bytes(payload),
ServerCommand::DeletePersonalAccessToken(payload) => as_bytes(payload),
ServerCommand::LoginWithPersonalAccessToken(payload) => as_bytes(payload),
- ServerCommand::SendMessages(payload) => as_bytes(payload),
+ //TODO: Fix me, I have no idea how XD
+ ServerCommand::SendMessages() => Bytes::new(),
ServerCommand::PollMessages(payload) => as_bytes(payload),
ServerCommand::StoreConsumerOffset(payload) => as_bytes(payload),
ServerCommand::DeleteConsumerOffset(payload) => as_bytes(payload),
@@ -187,9 +188,7 @@ impl ServerCommand {
LoginWithPersonalAccessToken::from_bytes(payload)?,
))
}
- SEND_MESSAGES_CODE => Ok(ServerCommand::SendMessages(
- iggy::messages::send_messages_server::SendMessages::from_bytes(payload)?,
- )),
+ SEND_MESSAGES_CODE => Ok(ServerCommand::SendMessages()),
POLL_MESSAGES_CODE => Ok(ServerCommand::PollMessages(PollMessages::from_bytes(
payload,
)?)),
@@ -295,7 +294,7 @@ impl Validatable<IggyError> for ServerCommand {
ServerCommand::CreatePersonalAccessToken(command) => command.validate(),
ServerCommand::DeletePersonalAccessToken(command) => command.validate(),
ServerCommand::LoginWithPersonalAccessToken(command) => command.validate(),
- ServerCommand::SendMessages(command) => command.validate(),
+ ServerCommand::SendMessages() => Ok(()),
ServerCommand::PollMessages(command) => command.validate(),
ServerCommand::StoreConsumerOffset(command) => command.validate(),
ServerCommand::DeleteConsumerOffset(command) => command.validate(),
@@ -378,7 +377,7 @@ impl Display for ServerCommand {
write!(formatter, "{DELETE_PARTITIONS}|{payload}")
}
ServerCommand::PollMessages(payload) => write!(formatter, "{POLL_MESSAGES}|{payload}"),
- ServerCommand::SendMessages(payload) => write!(formatter, "{SEND_MESSAGES}|{payload}"),
+ ServerCommand::SendMessages() => write!(formatter, "Noop"),
ServerCommand::StoreConsumerOffset(payload) => {
write!(formatter, "{STORE_CONSUMER_OFFSET}|{payload}")
}
diff --git a/server/src/compat/index_rebuilding/index_rebuilder.rs b/server/src/compat/index_rebuilding/index_rebuilder.rs
index ce24f72c..a3f3c777 100644
--- a/server/src/compat/index_rebuilding/index_rebuilder.rs
+++ b/server/src/compat/index_rebuilding/index_rebuilder.rs
@@ -1,7 +1,5 @@
+use crate::server_error::CompatError;
use crate::streaming::utils::file;
-use crate::{
- server_error::CompatError, streaming::batching::message_batch::RETAINED_BATCH_HEADER_LEN,
-};
use std::io::SeekFrom;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, BufWriter};
@@ -64,6 +62,8 @@ impl IndexRebuilder {
}
pub async fn rebuild(&self) -> Result<(), CompatError> {
+ //TODO: Fix me
+ /*
let mut reader = BufReader::new(file::open(&self.log_path).await?);
let mut writer = BufWriter::new(file::overwrite(&self.index_path).await?);
let mut position = 0;
@@ -92,5 +92,7 @@ impl IndexRebuilder {
writer.flush().await?;
Ok(())
+ */
+ todo!()
}
}
diff --git a/server/src/http/messages.rs b/server/src/http/messages.rs
index 21a14591..9ae2bf4f 100644
--- a/server/src/http/messages.rs
+++ b/server/src/http/messages.rs
@@ -71,6 +71,8 @@ async fn send_messages(
Path((stream_id, topic_id)): Path<(String, String)>,
Json(mut command): Json<SendMessages>,
) -> Result<StatusCode, CustomError> {
+ //TODO: Fix me
+ /*
command.stream_id = Identifier::from_str_value(&stream_id)?;
command.topic_id = Identifier::from_str_value(&topic_id)?;
command.partitioning.length = command.partitioning.value.len() as u8;
@@ -104,6 +106,8 @@ async fn send_messages(
)
})?;
Ok(StatusCode::CREATED)
+ */
+ todo!()
}
#[instrument(skip_all, name = "trace_flush_unsaved_buffer", fields(iggy_user_id = identity.user_id, iggy_stream_id = stream_id, iggy_topic_id = topic_id, iggy_partition_id = partition_id, iggy_fsync = fsync))]
diff --git a/server/src/quic/listener.rs b/server/src/quic/listener.rs
index 98affb5e..e4e3cf4c 100644
--- a/server/src/quic/listener.rs
+++ b/server/src/quic/listener.rs
@@ -100,6 +100,8 @@ async fn handle_stream(
system: SharedSystem,
session: impl AsRef<Session>,
) -> anyhow::Result<()> {
+ // TODO: Fix me
+ /*
let (send_stream, mut recv_stream) = stream;
// TODO: read to BytesMut instead of Vec<u8>
let request = recv_stream
@@ -132,4 +134,6 @@ async fn handle_stream(
command::handle(command, &mut sender, session.as_ref(), system.clone())
.await
.with_context(|| "Error when handling the QUIC request.")
+ */
+ todo!()
}
diff --git a/server/src/streaming/batching/batch_accumulator.rs b/server/src/streaming/batching/batch_accumulator.rs
index 0233cc32..b1be9e8d 100644
--- a/server/src/streaming/batching/batch_accumulator.rs
+++ b/server/src/streaming/batching/batch_accumulator.rs
@@ -1,6 +1,5 @@
-use bytes::BytesMut;
+use iggy::models::batch::{IggyBatch, IggyHeader};
use iggy::utils::byte_size::IggyByteSize;
-use iggy::utils::sizeable::Sizeable;
use std::sync::Arc;
#[derive(Debug)]
@@ -9,7 +8,8 @@ pub struct BatchAccumulator {
current_size: IggyByteSize,
current_offset: u64,
current_timestamp: u64,
- messages: Vec<Arc<RetainedMessage>>,
+ leading_header: Option<IggyHeader>,
+ batches: Vec<IggyBatch>,
}
impl BatchAccumulator {
@@ -19,23 +19,27 @@ impl BatchAccumulator {
current_size: IggyByteSize::from(0),
current_offset: 0,
current_timestamp: 0,
- messages: Vec::with_capacity(capacity),
+ leading_header: None,
+ batches: Vec::with_capacity(capacity),
}
}
- pub fn append(&mut self, batch_size: IggyByteSize, items: &[Arc<RetainedMessage>]) {
- assert!(!items.is_empty());
+ pub fn append(&mut self, batch_size: IggyByteSize, batch: IggyBatch) {
+ let header = batch.header;
+ if self.leading_header.is_none() {
+ self.base_offset = header.base_offset;
+ self.leading_header = Some(header);
+ }
+ self.current_timestamp = header.base_timestamp + header.last_timestamp_delta as u64;
+ self.current_offset = header.base_offset + header.last_offset_delta as u64;
self.current_size += batch_size;
- self.current_offset = items.last().unwrap().offset;
- self.current_timestamp = items.last().unwrap().timestamp;
- self.messages.extend(items.iter().cloned());
+ self.batches.push(batch);
+ //TODO: Fix me
}
- pub fn get_messages_by_offset(
- &self,
- start_offset: u64,
- end_offset: u64,
- ) -> Vec<Arc<RetainedMessage>> {
+ pub fn get_messages_by_offset(&self, start_offset: u64, end_offset: u64) -> Vec<Arc<()>> {
+ //TODO: Fix me
+ /*
let start_idx = self
.messages
.partition_point(|msg| msg.offset < start_offset);
@@ -43,26 +47,33 @@ impl BatchAccumulator {
.messages
.partition_point(|msg| msg.offset <= end_offset);
self.messages[start_idx..end_idx].to_vec()
+ */
+ todo!()
}
- pub fn get_messages_by_timestamp(
- &self,
- start_timestamp: u64,
- count: usize,
- ) -> Vec<Arc<RetainedMessage>> {
+ pub fn get_messages_by_timestamp(&self, start_timestamp: u64, count: usize) -> Vec<Arc<()>> {
+ //TODO: Fix me
+ /*
let start_idx = self
.messages
.partition_point(|msg| msg.timestamp < start_timestamp);
let end_idx = std::cmp::min(start_idx + count, self.messages.len());
self.messages[start_idx..end_idx].to_vec()
+ */
+ todo!()
}
pub fn is_empty(&self) -> bool {
- self.messages.is_empty()
+ self.batches.is_empty()
}
pub fn unsaved_messages_count(&self) -> usize {
- self.messages.len()
+ //TODO: Fix me, replace this with leading_header last_offset_delta + 1.
+ self.batches
+ .iter()
+ .map(|b| b.header)
+ .map(|h| (h.last_offset_delta + 1) as usize)
+ .sum()
}
pub fn batch_max_offset(&self) -> u64 {
@@ -77,42 +88,11 @@ impl BatchAccumulator {
self.base_offset
}
- pub fn materialize_batch_and_update_state(&mut self) -> RetainedMessageBatch {
- let batch_base_offset = self.base_offset;
- let batch_last_offset_delta = (self.current_offset - self.base_offset) as u32;
-
- let num_messages = self.messages.len();
- let last_batch_timestamp = if num_messages > 0 {
- self.messages[num_messages - 1].timestamp
- } else {
- 0
- };
-
- let messages = std::mem::take(&mut self.messages);
- let mut bytes = BytesMut::with_capacity(self.current_size.as_bytes_u64() as usize);
- for message in messages {
- message.extend(&mut bytes);
- }
-
- self.base_offset = 0;
- self.current_size = IggyByteSize::from(0);
- self.current_offset = 0;
- self.current_timestamp = 0;
-
- let batch_payload = bytes.freeze();
- let batch_payload_len = IggyByteSize::from(batch_payload.len() as u64);
- RetainedMessageBatch::new(
- batch_base_offset,
- batch_last_offset_delta,
- last_batch_timestamp,
- batch_payload_len,
- batch_payload,
+ pub fn materialize(self) -> (IggyHeader, Vec<IggyBatch>) {
+ (
+ self.leading_header
+ .expect("No leading header when materializing batch"),
+ self.batches,
)
}
}
-
-impl Sizeable for BatchAccumulator {
- fn get_size_bytes(&self) -> IggyByteSize {
- self.current_size + RETAINED_BATCH_HEADER_LEN.into()
- }
-}
diff --git a/server/src/streaming/partitions/messages.rs b/server/src/streaming/partitions/messages.rs
index 368e18a9..b4603b50 100644
--- a/server/src/streaming/partitions/messages.rs
+++ b/server/src/streaming/partitions/messages.rs
@@ -1,5 +1,4 @@
use crate::streaming::batching::appendable_batch_info::AppendableBatchInfo;
-use crate::streaming::batching::iterator::IntoMessagesIterator;
use crate::streaming::partitions::partition::Partition;
use crate::streaming::partitions::COMPONENT;
use crate::streaming::polling_consumer::PollingConsumer;
@@ -389,6 +388,7 @@ impl Partition {
}
}
+ let mut batch = batch;
let batch_size = appendable_batch_info.batch_size;
let base_offset = if !self.should_increment_offset {
0
@@ -442,17 +442,15 @@ impl Partition {
}
{
- /*
- let last_segment = self.segments.last_mut().ok_or(IggyError::SegmentNotFound)?;
- last_segment
- .append_batch(batch_size, messages_count, &retained_messages)
+ let last_segment = self.segments.last_mut().ok_or(IggyError::SegmentNotFound)?;
+ last_segment
+ .append_batch(batch_size, messages_count, batch)
.await
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to append batch into last segment: {last_segment}",
)
})?;
- */
}
/*
diff --git a/server/src/streaming/partitions/partition.rs b/server/src/streaming/partitions/partition.rs
index 6e495448..8cf4628c 100644
--- a/server/src/streaming/partitions/partition.rs
+++ b/server/src/streaming/partitions/partition.rs
@@ -92,7 +92,7 @@ impl Partition {
let consumer_group_offsets_path =
config.get_consumer_group_offsets_path(stream_id, topic_id, partition_id);
//TODO: Fix me
- /*
+ /*
let (cached_memory_tracker, messages) = match config.cache.enabled {
false => (None, None),
true => (
diff --git a/server/src/streaming/segments/logs/log_reader.rs b/server/src/streaming/segments/logs/log_reader.rs
index e90db334..bfa8b79e 100644
--- a/server/src/streaming/segments/logs/log_reader.rs
+++ b/server/src/streaming/segments/logs/log_reader.rs
@@ -1,10 +1,4 @@
-use crate::streaming::{
- batching::{
- iterator::IntoMessagesIterator,
- message_batch::{RetainedMessageBatch, RETAINED_BATCH_HEADER_LEN},
- },
- segments::indexes::IndexRange,
-};
+use crate::streaming::segments::indexes::IndexRange;
use bytes::BytesMut;
use error_set::ErrContext;
use iggy::{error::IggyError, utils::byte_size::IggyByteSize};
@@ -76,7 +70,9 @@ impl SegmentLogReader {
pub async fn load_batches_by_range_impl(
&self,
index_range: &IndexRange,
- ) -> Result<Vec<RetainedMessageBatch>, IggyError> {
+ ) -> Result<Vec<()>, IggyError> {
+ //TODO: Fix me
+ /*
let mut file_size = self.file_size();
if file_size == 0 {
trace!("Log file {} is empty.", self.file_path);
@@ -108,6 +104,8 @@ impl SegmentLogReader {
trace!("Loaded {} message batches.", batches.len());
Ok(batches)
+ */
+ todo!()
}
/// Loads and returns all message IDs from the log file.
@@ -117,6 +115,8 @@ impl SegmentLogReader {
trace!("Log file {} is empty.", self.file_path);
return Ok(Vec::new());
}
+ //TODO: Fix me
+ /*
let mut offset = 0_u64;
let mut message_ids = Vec::new();
@@ -139,6 +139,8 @@ impl SegmentLogReader {
trace!("Loaded {} message IDs from the log.", message_ids.len());
Ok(message_ids)
+ */
+ todo!()
}
/// Loads message batches given an index range and calls the provided callback for each batch.
@@ -148,8 +150,10 @@ impl SegmentLogReader {
mut on_batch: F,
) -> Result<(), IggyError>
where
- F: FnMut(RetainedMessageBatch) -> Result<(), IggyError>,
+ F: FnMut(()) -> Result<(), IggyError>,
{
+ //TODO: Fix me
+ /*
let mut file_size = self.file_size();
if file_size == 0 {
trace!("Log file {} is empty.", self.file_path);
@@ -178,6 +182,8 @@ impl SegmentLogReader {
}
Ok(())
+ */
+ todo!()
}
/// Loads message batches up to a given size and calls the provided callback for each batch
@@ -185,8 +191,10 @@ impl SegmentLogReader {
pub async fn load_batches_by_size_with_callback(
&self,
bytes_to_load: u64,
- mut on_batch: impl FnMut(RetainedMessageBatch) -> Result<(), IggyError>,
+ mut on_batch: impl FnMut(()) -> Result<(), IggyError>,
) -> Result<(), IggyError> {
+ // TODO: Fix me
+ /*
let mut file_size = self.file_size();
if file_size == 0 {
trace!("Log file {} is empty.", self.file_path);
@@ -214,13 +222,17 @@ impl SegmentLogReader {
}
Ok(())
+ */
+ todo!()
}
async fn read_next_batch(
&self,
offset: u64,
file_size: u64,
- ) -> Result<Option<(RetainedMessageBatch, u64)>, IggyError> {
+ ) -> Result<Option<((), u64)>, IggyError> {
+ //TODO: Fix me
+ /*
let batch_header_size = RETAINED_BATCH_HEADER_LEN;
if offset + batch_header_size > file_size {
return Ok(None);
@@ -322,6 +334,8 @@ impl SegmentLogReader {
);
Ok(Some((batch, bytes_read)))
+ */
+ todo!()
}
fn file_size(&self) -> u64 {
diff --git a/server/src/streaming/segments/logs/log_writer.rs b/server/src/streaming/segments/logs/log_writer.rs
index 5b6811b7..fb35a0cc 100644
--- a/server/src/streaming/segments/logs/log_writer.rs
+++ b/server/src/streaming/segments/logs/log_writer.rs
@@ -1,9 +1,9 @@
use super::PersisterTask;
-use crate::streaming::batching::message_batch::RetainedMessageBatch;
use error_set::ErrContext;
use iggy::{
confirmation::Confirmation,
error::IggyError,
+ models::batch::{IggyBatch, IggyHeader, IGGY_BATCH_OVERHEAD},
utils::{byte_size::IggyByteSize, duration::IggyDuration, sizeable::Sizeable},
};
use std::{
@@ -94,15 +94,16 @@ impl SegmentLogWriter {
/// Append a message batch to the log file.
pub async fn save_batches(
&mut self,
- batch: RetainedMessageBatch,
+ header: IggyHeader,
+ batches: Vec<IggyBatch>,
confirmation: Confirmation,
) -> Result<IggyByteSize, IggyError> {
- let batch_size = batch.get_size_bytes();
+ let batch_size =
+ IGGY_BATCH_OVERHEAD + batches.iter().map(|b| b.batch.len() as u64).sum::<u64>();
match confirmation {
Confirmation::Wait => {
- self.write_batch(batch).await?;
- self.log_size_bytes
- .fetch_add(batch_size.as_bytes_u64(), Ordering::AcqRel);
+ self.write_batch(header, batches).await?;
+ self.log_size_bytes.fetch_add(batch_size, Ordering::AcqRel);
trace!(
"Written batch of size {batch_size} bytes to log file: {}",
self.file_path
@@ -113,7 +114,7 @@ impl SegmentLogWriter {
}
Confirmation::NoWait => {
if let Some(task) = &self.persister_task {
- task.persist(batch).await;
+ task.persist(batch_size, header, batches).await;
} else {
panic!(
"Confirmation::NoWait is used, but LogPersisterTask is not set for log file: {}",
@@ -123,15 +124,23 @@ impl SegmentLogWriter {
}
}
- Ok(batch_size)
+ Ok(batch_size.into())
}
/// Write a batch of bytes to the log file and return the new file position.
- async fn write_batch(&mut self, batch_to_write: RetainedMessageBatch) -> Result<(), IggyError> {
+ async fn write_batch(
+ &mut self,
+ header: IggyHeader,
+ batches: Vec<IggyBatch>,
+ ) -> Result<(), IggyError> {
if let Some(ref mut file) = self.file {
- let header = batch_to_write.header_as_bytes();
- let batch_bytes = batch_to_write.bytes;
- let slices = [IoSlice::new(&header), IoSlice::new(&batch_bytes)];
+ let mut slices = Vec::new();
+
+ let header = header.as_bytes();
+ slices.push(IoSlice::new(&header));
+ batches.iter().for_each(|b| {
+ slices.push(IoSlice::new(&b.batch));
+ });
file.write_vectored(&slices)
.await
@@ -139,7 +148,6 @@ impl SegmentLogWriter {
format!("Failed to log to file: {}. {error}", self.file_path)
})
.map_err(|_| IggyError::CannotWriteToFile)?;
-
Ok(())
} else {
error!("File handle is not available for synchronous write.");
diff --git a/server/src/streaming/segments/logs/persister_task.rs b/server/src/streaming/segments/logs/persister_task.rs
index a9503740..116460f3 100644
--- a/server/src/streaming/segments/logs/persister_task.rs
+++ b/server/src/streaming/segments/logs/persister_task.rs
@@ -1,6 +1,9 @@
-use crate::streaming::batching::message_batch::{RetainedMessageBatch, RETAINED_BATCH_HEADER_LEN};
use flume::{unbounded, Receiver};
-use iggy::{error::IggyError, utils::duration::IggyDuration};
+use iggy::{
+ error::IggyError,
+ models::batch::{IggyBatch, IggyHeader, IGGY_BATCH_OVERHEAD},
+ utils::duration::IggyDuration,
+};
use std::{
io::IoSlice,
sync::{
@@ -15,7 +18,7 @@ use tracing::{error, trace, warn};
#[derive(Debug)]
/// A command to the persister task.
enum PersisterTaskCommand {
- WriteRequest(RetainedMessageBatch),
+ WriteRequest(u64, IggyHeader, Vec<IggyBatch>),
Shutdown,
}
@@ -38,32 +41,34 @@ impl PersisterTask {
retry_delay: IggyDuration,
) -> Self {
let (sender, receiver) = unbounded();
- let log_file_size_clone = log_file_size.clone();
+ let log_file_size = log_file_size.clone();
let file_path_clone = file_path.clone();
let handle = tokio::spawn(async move {
Self::run(
file,
- file_path,
+ file_path_clone,
receiver,
fsync,
max_retries,
retry_delay,
- log_file_size_clone,
+ log_file_size,
)
.await;
});
Self {
sender,
- file_path: file_path_clone,
+ file_path,
_handle: handle,
}
}
/// Sends the batch bytes to the persister task (fire-and-forget).
- pub async fn persist(&self, batch_to_write: RetainedMessageBatch) {
+ pub async fn persist(&self, batch_size: u64, header: IggyHeader, batches: Vec<IggyBatch>) {
if let Err(e) = self
.sender
- .send_async(PersisterTaskCommand::WriteRequest(batch_to_write))
+ .send_async(PersisterTaskCommand::WriteRequest(
+ batch_size, header, batches,
+ ))
.await
{
error!(
@@ -173,11 +178,13 @@ impl PersisterTask {
) {
while let Ok(request) = receiver.recv_async().await {
match request {
- PersisterTaskCommand::WriteRequest(batch_to_write) => {
+ PersisterTaskCommand::WriteRequest(batch_size, header, batches) => {
match Self::write_with_retries(
&mut file,
&file_path,
- batch_to_write,
+ batch_size,
+ header,
+ batches,
fsync,
max_retries,
retry_delay,
@@ -214,23 +221,35 @@ impl PersisterTask {
async fn write_with_retries(
file: &mut File,
file_path: &str,
- batch_to_write: RetainedMessageBatch,
+ batch_size: u64,
+ header: IggyHeader,
+ batches: Vec<IggyBatch>,
fsync: bool,
max_retries: u32,
retry_delay: IggyDuration,
) -> Result<u64, IggyError> {
- let header = batch_to_write.header_as_bytes();
- let batch_bytes = batch_to_write.bytes;
- let slices = [IoSlice::new(&header), IoSlice::new(&batch_bytes)];
- let bytes_written = RETAINED_BATCH_HEADER_LEN + batch_bytes.len() as u64;
+ // TODO: Fix me, this logic is repeated, maybe could be encapsulated inside of the batch_accumulator
+ // that yields accumulated batches and leading header.
+ let mut slices = Vec::new();
+ let header = header.as_bytes();
+ slices.push(IoSlice::new(&header));
+ batches.iter().for_each(|b| {
+ slices.push(IoSlice::new(&b.batch));
+ });
let mut attempts = 0;
+ // TODO: This "retry" logic should be rewritten.
+ // There are certain kind of errors whom retyring is considered harmful, for example fsync failure
+ // (https://www.usenix.org/system/files/atc20-rebello.pdf)
+ // There are few errors which are worth retrying such as LSE (Latent sector error), as the file system
+ // might be able to realocate a new sector for the data and recover, but not every file system supports that.
+ // In general this topic should be furthered researched rather than just naively retry when the write fails.
loop {
match file.write_vectored(&slices).await {
Ok(_) => {
if fsync {
match file.sync_all().await {
- Ok(_) => return Ok(bytes_written),
+ Ok(_) => return Ok(batch_size),
Err(e) => {
attempts += 1;
error!(
@@ -240,7 +259,7 @@ impl PersisterTask {
}
}
} else {
- return Ok(bytes_written);
+ return Ok(batch_size);
}
}
Err(e) => {
diff --git a/server/src/streaming/segments/reading_messages.rs b/server/src/streaming/segments/reading_messages.rs
index 7e0f58e0..ede1e535 100644
--- a/server/src/streaming/segments/reading_messages.rs
+++ b/server/src/streaming/segments/reading_messages.rs
@@ -8,7 +8,6 @@ use iggy::{
use std::sync::Arc;
use tracing::{trace, warn};
-const EMPTY_MESSAGES: Vec<RetainedMessage> = vec![];
const COMPONENT: &str = "STREAMING_SEGMENT";
impl Segment {
@@ -62,7 +61,7 @@ impl Segment {
count: u32,
) -> Result<Vec<Arc<()>>, IggyError> {
//TODO: Fix me
- /*
+ /*
if count == 0 {
return Ok(EMPTY_MESSAGES.into_iter().map(Arc::new).collect());
}
@@ -126,21 +125,18 @@ impl Segment {
self.get_messages_by_offset(self.start_offset, self.get_messages_count() as u32)
.await
*/
- todo!()
+ todo!()
}
pub async fn get_all_batches(&self) -> Result<Vec<()>, IggyError> {
//TODO: Fix me
- /*
+ /*
self.load_batches_by_range(&IndexRange::max_range()).await
*/
todo!()
}
- pub async fn get_newest_batches_by_size(
- &self,
- size_bytes: u64,
- ) -> Result<Vec<()>, IggyError> {
+ pub async fn get_newest_batches_by_size(&self, size_bytes: u64) -> Result<Vec<()>, IggyError> {
//TODO: Fix me
/*
let mut batches = Vec::new();
@@ -278,6 +274,8 @@ impl Segment {
/// Loads and verifies message checksums from the log file.
pub async fn load_message_checksums(&self) -> Result<(), IggyError> {
+ //TODO: Fix me
+ /*
self.log_reader
.as_ref()
.unwrap()
@@ -305,6 +303,8 @@ impl Segment {
format!("Failed to load batches by max range for {}. {error}", self)
})?;
Ok(())
+ */
+ todo!()
}
/// Loads and returns all message IDs from the log file.
diff --git a/server/src/streaming/segments/segment.rs b/server/src/streaming/segments/segment.rs
index 9bd0d4b4..95eefb80 100644
--- a/server/src/streaming/segments/segment.rs
+++ b/server/src/streaming/segments/segment.rs
@@ -249,6 +249,8 @@ impl Segment {
}
pub async fn is_expired(&self, now: IggyTimestamp) -> bool {
+ //TODO: Fix me
+ /*
if !self.is_closed {
return false;
}
@@ -272,6 +274,8 @@ impl Segment {
last_message_timestamp + expiry.as_micros() <= now.as_micros()
}
}
+ */
+ todo!()
}
pub async fn shutdown_reading(&mut self) {
diff --git a/server/src/streaming/segments/writing_messages.rs b/server/src/streaming/segments/writing_messages.rs
index 626cb288..d80988a3 100644
--- a/server/src/streaming/segments/writing_messages.rs
+++ b/server/src/streaming/segments/writing_messages.rs
@@ -1,12 +1,12 @@
use super::indexes::*;
use crate::streaming::batching::batch_accumulator::BatchAccumulator;
-use crate::streaming::batching::message_batch::RETAINED_BATCH_HEADER_LEN;
use crate::streaming::segments::segment::Segment;
use error_set::ErrContext;
-use iggy::{confirmation::Confirmation, models::batch::IggyBatch};
use iggy::error::IggyError;
+use iggy::models::batch::IGGY_BATCH_OVERHEAD;
use iggy::utils::byte_size::IggyByteSize;
use iggy::utils::sizeable::Sizeable;
+use iggy::{confirmation::Confirmation, models::batch::IggyBatch};
use std::sync::atomic::Ordering;
use tracing::{info, trace};
@@ -23,10 +23,9 @@ impl Segment {
self.partition_id,
));
}
- let messages_count = batch.header.last_offset_delta as usize;
let messages_cap = std::cmp::max(
self.config.partition.messages_required_to_save as usize,
- messages_count
+ messages_count as usize,
);
if self.current_offset == 0 {
self.start_timestamp = batch.header.base_timestamp;
@@ -104,11 +103,7 @@ impl Segment {
self.partition_id
);
- let batch = batch_accumulator.materialize_batch_and_update_state();
- let batch_size = batch.get_size_bytes();
- if batch_size > 0 {
- self.unsaved_messages = Some(batch_accumulator);
- }
+ let (header, batches) = batch_accumulator.materialize();
let confirmation = match confirmation {
Some(val) => val,
None => self.config.segment.server_confirmation,
@@ -117,12 +112,13 @@ impl Segment {
.log_writer
.as_mut()
.unwrap()
- .save_batches(batch, confirmation)
+ .save_batches(header, batches, confirmation)
.await
- .with_error_context(|error| {
- format!("Failed to save batch of size {batch_size} for {self}. {error}",)
- })?;
+ .with_error_context(
+ |error| format!("Failed to save batch for seg: {self}. {error}",),
+ )?;
+ self.last_index_position += saved_bytes.as_bytes_u64() as u32;
self.index_writer
.as_mut()
.unwrap()
@@ -130,14 +126,13 @@ impl Segment {
.await
.with_error_context(|error| format!("Failed to save index for {self}. {error}"))?;
- self.last_index_position += batch_size.as_bytes_u64() as u32;
- self.size_bytes += IggyByteSize::from(RETAINED_BATCH_HEADER_LEN);
+ self.size_bytes += IggyByteSize::from(IGGY_BATCH_OVERHEAD);
self.size_of_parent_stream
- .fetch_add(RETAINED_BATCH_HEADER_LEN, Ordering::AcqRel);
+ .fetch_add(IGGY_BATCH_OVERHEAD, Ordering::AcqRel);
self.size_of_parent_topic
- .fetch_add(RETAINED_BATCH_HEADER_LEN, Ordering::AcqRel);
+ .fetch_add(IGGY_BATCH_OVERHEAD, Ordering::AcqRel);
self.size_of_parent_partition
- .fetch_add(RETAINED_BATCH_HEADER_LEN, Ordering::AcqRel);
+ .fetch_add(IGGY_BATCH_OVERHEAD, Ordering::AcqRel);
trace!(
"Saved {} messages on disk in segment with start offset: {} for partition with ID: {}, total bytes written: {}.",
diff --git a/server/src/streaming/systems/messages.rs b/server/src/streaming/systems/messages.rs
index f2625d90..e94fb23f 100644
--- a/server/src/streaming/systems/messages.rs
+++ b/server/src/streaming/systems/messages.rs
@@ -110,9 +110,9 @@ impl System {
pub async fn append_messages(
&self,
session: &Session,
- stream_id: Identifier,
- topic_id: Identifier,
- partitioning: Partitioning,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partitioning: &Partitioning,
batch: IggyBatch,
confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
@@ -163,7 +163,7 @@ impl System {
}
}
*/
- let batch_size_bytes = batch.get_size_bytes();
+ let batch_size_bytes = batch.get_size();
topic
.append_messages(batch_size_bytes, partitioning, batch, confirmation)
.await?;
diff --git a/server/src/streaming/systems/system.rs b/server/src/streaming/systems/system.rs
index 7323a424..2a068f5f 100644
--- a/server/src/streaming/systems/system.rs
+++ b/server/src/streaming/systems/system.rs
@@ -269,6 +269,8 @@ impl System {
}
pub async fn clean_cache(&self, size_to_clean: IggyByteSize) {
+ //TODO: Fix me
+ /*
for stream in self.streams.values() {
for topic in stream.get_topics() {
for partition in topic.get_partitions().into_iter() {
@@ -285,5 +287,7 @@ impl System {
}
}
}
+ */
+ todo!()
}
}
diff --git a/server/src/streaming/topics/messages.rs b/server/src/streaming/topics/messages.rs
index c7d9b71b..4ce8bac2 100644
--- a/server/src/streaming/topics/messages.rs
+++ b/server/src/streaming/topics/messages.rs
@@ -1,5 +1,4 @@
use crate::streaming::batching::appendable_batch_info::AppendableBatchInfo;
-use crate::streaming::models::messages::RetainedMessage;
use crate::streaming::polling_consumer::PollingConsumer;
use crate::streaming::topics::topic::Topic;
use crate::streaming::topics::COMPONENT;
@@ -13,7 +12,6 @@ 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::messages::PolledMessages;
use iggy::utils::byte_size::IggyByteSize;
use iggy::utils::expiry::IggyExpiry;
use iggy::utils::sizeable::Sizeable;
@@ -33,7 +31,9 @@ impl Topic {
partition_id: u32,
strategy: PollingStrategy,
count: u32,
- ) -> Result<PolledMessages, IggyError> {
+ ) -> Result<(), IggyError> {
+ //TODO: Fix me
+ /*
if !self.has_partitions() {
return Err(IggyError::NoPartitions(self.topic_id, self.stream_id));
}
@@ -72,12 +72,14 @@ impl Topic {
current_offset: partition.current_offset,
messages,
})
+ */
+ todo!()
}
pub async fn append_messages(
&self,
batch_size: IggyByteSize,
- partitioning: Partitioning,
+ partitioning: &Partitioning,
batch: IggyBatch,
confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
@@ -187,6 +189,8 @@ impl Topic {
}
pub(crate) async fn load_messages_from_disk_to_cache(&mut self) -> Result<(), IggyError> {
+ //TODO: Fix me
+ /*
if !self.config.cache.enabled {
return Ok(());
}
@@ -264,9 +268,13 @@ impl Topic {
}
Ok(())
+ */
+ todo!()
}
- fn cache_integrity_check(cache: &[Arc<RetainedMessage>]) -> bool {
+ fn cache_integrity_check(cache: ()) -> bool {
+ //TODO: Fix me
+ /*
if cache.is_empty() {
warn!("Cache is empty!");
return false;
@@ -293,6 +301,8 @@ impl Topic {
}
true
+ */
+ todo!()
}
pub async fn get_expired_segments_start_offsets_per_partition(
@@ -315,6 +325,8 @@ impl Topic {
#[cfg(test)]
mod tests {
+ //TODO: Fix me
+ /*
use super::*;
use crate::configs::system::SystemConfig;
use crate::streaming::persistence::persister::FileWithSyncPersister;
@@ -469,4 +481,5 @@ mod tests {
topic.persist().await.unwrap();
topic
}
+ */
}
diff --git a/server/src/tcp/connection_handler.rs b/server/src/tcp/connection_handler.rs
index c761c648..cb0a4ba6 100644
--- a/server/src/tcp/connection_handler.rs
+++ b/server/src/tcp/connection_handler.rs
@@ -47,9 +47,9 @@ pub(crate) async fn handle_connection(
}
let length = u32::from_le_bytes(initial_buffer) - 4;
- sender.read(&mut code_buffer);
+ sender.read(&mut code_buffer).await?;
let code = u32::from_le_bytes(code_buffer);
- // TODO: Refactor the way the command is structured
+ // TODO: Fix me - Refactor the way the command is structured
// `ServerCommnad` struct should have a method that takes the `sender`
// and uses it to construct the command, rather than eagerly reading all of the data from the network.
if code == SEND_MESSAGES_CODE {
@@ -58,7 +58,7 @@ pub(crate) async fn handle_connection(
let metadata_len = u32::from_le_bytes(metadata_len);
let mut metadata_buf = BytesMut::with_capacity(metadata_len as _);
unsafe { metadata_buf.set_len(metadata_len as _) };
- sender.read(&mut metadata_buf);
+ sender.read(&mut metadata_buf).await?;
let metadata_buf = metadata_buf.freeze();
// TODO: This is disgusting, fuck `Bytes`.
@@ -72,7 +72,7 @@ pub(crate) async fn handle_connection(
sender.read(&mut header_buffer).await?;
let header = IggyHeader::from_bytes(&header_buffer);
- let batch_length = metadata_len - length - 4 - IGGY_BATCH_OVERHEAD as u32;
+ let batch_length = length - metadata_len - IGGY_BATCH_OVERHEAD as u32 - 4;
let mut batch_buffer = Vec::with_capacity(batch_length as _);
unsafe { batch_buffer.set_len(batch_length as _) };
sender.read(&mut batch_buffer).await?;
@@ -80,7 +80,7 @@ pub(crate) async fn handle_connection(
let system = system.read().await;
{
system
- .append_messages(&session, stream_id, topic_id, partitioning, batch, None)
+ .append_messages(&session, &stream_id, &topic_id, &partitioning, batch, None)
.await
.with_error_context(|error| {
format!(
@@ -94,7 +94,9 @@ pub(crate) async fn handle_connection(
debug!("Received a TCP request, length: {length}");
let mut command_buffer = BytesMut::with_capacity(length as usize);
command_buffer.put_bytes(0, length as usize);
- sender.read(&mut command_buffer).await?;
+ if length > 0 {
+ sender.read(&mut command_buffer).await?;
+ }
let command = ServerCommand::from_code_and_payload(code, command_buffer.freeze());
if command.is_err() {
sender