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/03/01 10:39:09 UTC
(iggy) branch zero_copy_schema updated: varint
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
The following commit(s) were added to refs/heads/zero_copy_schema by this push:
new c6930c09 varint
c6930c09 is described below
commit c6930c09fdc1bfffb49f494cf32a0e07df00a298
Author: numinex <g....@gmail.com>
AuthorDate: Sat Mar 1 11:35:00 2025 +0100
varint
---
configs/server.toml | 2 +-
sdk/src/messages/send_messages.rs | 20 ++++++++-----
sdk/src/models/batch.rs | 17 ++++++-----
sdk/src/models/header.rs | 2 +-
sdk/src/utils/duration.rs | 5 ++--
sdk/src/utils/mod.rs | 1 +
sdk/src/utils/varint.rs | 61 +++++++++++++++++++++++++++++++++++++++
7 files changed, 89 insertions(+), 19 deletions(-)
diff --git a/configs/server.toml b/configs/server.toml
index 0ea8afe9..470c64bb 100644
--- a/configs/server.toml
+++ b/configs/server.toml
@@ -374,7 +374,7 @@ sysinfo_print_interval = "10 s"
# Enables or disables the system cache.
# `true` activates caching for frequently accessed data.
# `false` disables caching, data is always read from the source.
-enabled = true
+enabled = false
# Maximum size of the cache, e.g. "4GB".
size = "4 GB"
diff --git a/sdk/src/messages/send_messages.rs b/sdk/src/messages/send_messages.rs
index c3f6896c..d5760f97 100644
--- a/sdk/src/messages/send_messages.rs
+++ b/sdk/src/messages/send_messages.rs
@@ -8,8 +8,9 @@ use crate::models::header;
use crate::models::header::{HeaderKey, HeaderValue};
use crate::utils::byte_size::IggyByteSize;
use crate::utils::sizeable::Sizeable;
+use crate::utils::varint::{encode_var, varint_size};
use crate::validatable::Validatable;
-use bytes::{BufMut, Bytes, BytesMut};
+use bytes::{Buf, BufMut, Bytes, BytesMut};
use serde::{Deserialize, Serialize};
use serde_with::base64::Base64;
use serde_with::serde_as;
@@ -468,21 +469,26 @@ pub(crate) fn as_bytes(
let headers_len = header::get_headers_size_bytes(&message.headers).as_bytes_u64();
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;
+ //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
- bytes.extend_from_slice(&total_len.to_le_bytes());
- bytes.extend_from_slice(&payload_len.to_le_bytes());
bytes.extend_from_slice(&message.id.to_le_bytes());
+
+ let mut position = bytes.len();
+ let payload_size = varint_size(payload_len);
+ let headers_size = varint_size(headers_len);
+ unsafe {bytes.set_len(position + payload_size + headers_size)};
+
+ //TODO: Replace this with varint
+ let advanced = encode_var(payload_len, &mut bytes[position..]);
+ position += advanced;
+ encode_var(headers_len, &mut bytes[position..]);
bytes.extend_from_slice(&message.payload);
if let Some(headers) = &message.headers {
let headers_bytes = headers.to_bytes();
- bytes.put_u32_le(headers_bytes.len() as u32);
bytes.put_slice(&headers_bytes);
- } else {
- bytes.put_u32_le(0);
}
}
diff --git a/sdk/src/models/batch.rs b/sdk/src/models/batch.rs
index 3c7095e8..6dff5a23 100644
--- a/sdk/src/models/batch.rs
+++ b/sdk/src/models/batch.rs
@@ -5,7 +5,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::{
binary::messages,
- utils::{byte_size::IggyByteSize, timestamp::IggyTimestamp},
+ utils::{byte_size::IggyByteSize, timestamp::IggyTimestamp, varint::decode_var},
};
pub const IGGY_BATCH_OVERHEAD: u64 = 16 + 16 + 16 + 16 + 8 + 8 + 8 + 8 + 4 + 4 + 4 + 4 + 1;
@@ -14,6 +14,8 @@ 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...
+ // TODO Fix me: Reorganize those fields, such that the most frequently accessed fields
+ // are at the top in the same cache line.
pub payload_type: u8,
pub last_offset_delta: u32,
pub last_timestamp_delta: u32,
@@ -284,11 +286,12 @@ impl IggyMutableBatch {
last_timestamp_delta = (IggyTimestamp::now().as_micros() - base_timestamp) as u32;
let timestamp_delta_value = last_timestamp_delta.to_le_bytes();
write_value_at(&mut self.batch, timestamp_delta_value, position);
-
position += 4;
- let total_len =
- u64::from_le_bytes(self.batch[position..position + 8].try_into().unwrap());
- position += 8;
+ // Skip the Id
+ position += 16;
+ let payload_len = decode_var(&self.batch[position..], &mut position);
+ let headers_len = decode_var(&self.batch[position..], &mut position);
+ let total_len = payload_len + headers_len;
position += total_len as usize;
messages_count += 1;
}
@@ -419,14 +422,14 @@ impl<'msg> Iterator for IggyMessageIterator<'msg> {
pos += 4;
let timestamp_delta = u32::from_le_bytes(data[pos..pos + 4].try_into().ok()?);
pos += 4;
+ let id = u128::from_le_bytes(data[pos..pos + 16].try_into().ok()?);
+ pos += 16;
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 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..pos + payload_length as usize];
diff --git a/sdk/src/models/header.rs b/sdk/src/models/header.rs
index fd3c9ccc..0adc8de9 100644
--- a/sdk/src/models/header.rs
+++ b/sdk/src/models/header.rs
@@ -659,7 +659,7 @@ impl BytesSerializable for HashMap<HeaderKey, HeaderValue> {
/// Returns the size in bytes of the specified headers.
pub fn get_headers_size_bytes(headers: &Option<HashMap<HeaderKey, HeaderValue>>) -> IggyByteSize {
// Headers length field
- let mut size = 4;
+ let mut size = 0;
if let Some(headers) = headers {
for (key, value) in headers {
// Key length + Key + Kind + Value length + Value
diff --git a/sdk/src/utils/duration.rs b/sdk/src/utils/duration.rs
index 7cdc44f9..471e7e2f 100644
--- a/sdk/src/utils/duration.rs
+++ b/sdk/src/utils/duration.rs
@@ -57,11 +57,10 @@ impl IggyDuration {
self.duration.as_secs() == 0
}
- //TODO(numinex) - Replace it with std::time::Duration::abs_diff when it becomes stable
pub fn abs_diff(&self, other: IggyDuration) -> IggyDuration {
- let diff = self.duration.as_nanos().abs_diff(other.duration.as_nanos());
+ let diff = self.duration.abs_diff(other.duration);
IggyDuration {
- duration: Duration::from_nanos(diff as u64),
+ duration: diff
}
}
}
diff --git a/sdk/src/utils/mod.rs b/sdk/src/utils/mod.rs
index 7516f405..6b08c873 100644
--- a/sdk/src/utils/mod.rs
+++ b/sdk/src/utils/mod.rs
@@ -8,3 +8,4 @@ pub mod sizeable;
pub mod text;
pub mod timestamp;
pub mod topic_size;
+pub mod varint;
\ No newline at end of file
diff --git a/sdk/src/utils/varint.rs b/sdk/src/utils/varint.rs
new file mode 100644
index 00000000..04936208
--- /dev/null
+++ b/sdk/src/utils/varint.rs
@@ -0,0 +1,61 @@
+const MSB: u8 = 0b1000_0000;
+const DROP_MSB: u8 = 0b0111_1111;
+
+#[inline]
+pub fn encode_var(value: u64, dst: &mut [u8]) -> usize {
+ let mut n = value;
+ let mut i = 0;
+
+ while n >= 0x80 {
+ let dst_i = unsafe { dst.get_unchecked_mut(i) };
+ *dst_i = MSB | (n as u8);
+ i += 1;
+ n >>= 7;
+ }
+
+ let dst_i = unsafe { dst.get_unchecked_mut(i) };
+ *dst_i = n as u8;
+ i + 1
+}
+
+#[inline]
+pub fn varint_size(value: u64) -> usize {
+ let mut n = value;
+ let mut i = 0;
+
+ while n >= 0x80 {
+ i += 1;
+ n >>= 7;
+ }
+
+ i + 1
+}
+
+#[inline]
+pub fn decode_var(src: &[u8], position: &mut usize) -> u64 {
+ let mut result: u64 = 0;
+ let mut shift = 0;
+ let mut idx = 0;
+
+ #[allow(unused_assignments)]
+ let mut success = false;
+ loop {
+ let b = *unsafe { src.get_unchecked(idx)};
+ let msb_dropped = b & DROP_MSB;
+ result |= (msb_dropped as u64) << shift;
+ shift += 7;
+ idx += 1;
+ *position += 1;
+
+ if b & MSB == 0 || shift > (9 * 7) {
+ success = b & MSB == 0;
+ break;
+ }
+ }
+
+ if success {
+ result
+ } else {
+ panic!("Overflow while decoding varint: {}", result);
+ }
+}