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/05 10:51:47 UTC
(iggy) branch zero-copy-no-batching-rebase updated (0acf84bc -> 2d9cfdad)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch zero-copy-no-batching-rebase
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 0acf84bc minor imprs
new 2d9cfdad minor imprs
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 (0acf84bc)
\
N -- N -- N refs/heads/zero-copy-no-batching-rebase (2d9cfdad)
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/streaming/segments/indexes/index_writer.rs | 6 +++++-
server/src/streaming/segments/types/messages_batch_set.rs | 2 ++
server/src/streaming/segments/writing_messages.rs | 1 +
3 files changed, 8 insertions(+), 1 deletion(-)
(iggy) 01/01: minor imprs
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
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 2d9cfdaded7dcccfce51516928008418ab96c04d
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Apr 4 09:57:38 2025 +0200
minor imprs
---
Cargo.lock | 10 --
profile.json.gz | Bin 0 -> 868218 bytes
sdk/Cargo.toml | 2 -
sdk/src/models/messaging/messages_batch.rs | 27 ------
server/src/streaming/partitions/storage.rs | 4 -
.../src/streaming/segments/indexes/index_writer.rs | 6 +-
.../streaming/segments/messages/messages_reader.rs | 107 ++++++++++++++++++---
.../streaming/segments/types/messages_batch_set.rs | 4 +-
server/src/streaming/segments/writing_messages.rs | 1 +
9 files changed, 104 insertions(+), 57 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 78a0738a..223d750c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2009,15 +2009,6 @@ version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a8d1add55171497b4705a648c6b583acafb01d58050a51727785f0b2c8e0a2b2"
-[[package]]
-name = "gxhash"
-version = "3.4.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a197c9b654827513cf53842c5c6d3da2b4b35a785f8e0eff78bdf8e445aba1bb"
-dependencies = [
- "rustversion",
-]
-
[[package]]
name = "h2"
version = "0.4.8"
@@ -2523,7 +2514,6 @@ dependencies = [
"flume",
"futures",
"futures-util",
- "gxhash",
"humantime",
"keyring",
"passterm",
diff --git a/profile.json.gz b/profile.json.gz
new file mode 100644
index 00000000..e5e1f0fe
Binary files /dev/null and b/profile.json.gz differ
diff --git a/sdk/Cargo.toml b/sdk/Cargo.toml
index e1e0445e..217aa874 100644
--- a/sdk/Cargo.toml
+++ b/sdk/Cargo.toml
@@ -37,8 +37,6 @@ fast-async-mutex = { version = "0.6.7", optional = true }
flume = "0.11.1"
futures = "0.3.31"
futures-util = "0.3.31"
-# gxhash = { version = "3.4.1", features = ["hybrid"] }
-gxhash = "3.4.1"
humantime = "2.1.0"
keyring = { version = "3.6.1", optional = true, features = [
"sync-secret-service",
diff --git a/sdk/src/models/messaging/messages_batch.rs b/sdk/src/models/messaging/messages_batch.rs
index 88220615..b717d6dc 100644
--- a/sdk/src/models/messaging/messages_batch.rs
+++ b/sdk/src/models/messaging/messages_batch.rs
@@ -23,32 +23,6 @@ pub struct IggyMessagesBatch {
impl IggyMessagesBatch {
/// Create a batch from indexes buffer and messages buffer
pub fn new(indexes: IggyIndexes, messages: Bytes, count: u32) -> Self {
- #[cfg(debug_assertions)]
- {
- debug_assert_eq!(indexes.len() % INDEX_SIZE, 0);
- let indexes_count = indexes.len() / INDEX_SIZE;
- debug_assert_eq!(indexes_count, count as usize);
-
- if count > 0 {
- let first_position = indexes.get(0).unwrap().position() as usize;
-
- let iter = IggyMessageViewIterator::new(&messages);
- let mut current_position = first_position;
-
- for (i, message) in iter.enumerate().skip(1) {
- if i < count as usize {
- current_position += message.size();
- let expected_position = indexes.get(i as u32).unwrap().position() as usize;
- debug_assert_eq!(
- current_position, expected_position,
- "Message {} position mismatch: {} != {}",
- i, current_position, expected_position
- );
- }
- }
- }
- }
-
Self {
count,
indexes,
@@ -427,7 +401,6 @@ impl Index<usize> for IggyMessagesBatch {
self.position_at(index as u32 - 1) as usize
};
- // todo check if OK, compare with messages_batch_mut.rs
let end_position = if index == self.count as usize - 1 {
self.messages.len()
} else {
diff --git a/server/src/streaming/partitions/storage.rs b/server/src/streaming/partitions/storage.rs
index cfa4b5a6..a8ce8f66 100644
--- a/server/src/streaming/partitions/storage.rs
+++ b/server/src/streaming/partitions/storage.rs
@@ -197,12 +197,8 @@ impl PartitionStorage for FilePartitionStorage {
CacheIndexesConfig::OpenSegment
) && !partition.segments.is_empty()
{
- tracing::error!("Clearing all segment indexes ({} segments) except the last one for partition with ID: {} and topic with ID: {}",
- segments_count, partition.partition_id, partition.topic_id);
-
let segments_count = partition.segments.len();
for i in 0..segments_count - 1 {
- tracing::error!("Dropping indexes for segment with ID: {} for partition with ID: {} and topic with ID: {}", partition.segments[i].start_offset(), partition.partition_id, partition.topic_id);
partition.segments[i].drop_indexes();
}
}
diff --git a/server/src/streaming/segments/indexes/index_writer.rs b/server/src/streaming/segments/indexes/index_writer.rs
index d3094a9f..527ab050 100644
--- a/server/src/streaming/segments/indexes/index_writer.rs
+++ b/server/src/streaming/segments/indexes/index_writer.rs
@@ -87,7 +87,11 @@ impl IndexWriter {
self.index_size_bytes
.fetch_add(indexes.len() as u64, Ordering::Release);
- trace!("Saved {count} indexes to file: {}", self.file_path);
+ trace!(
+ "Saved {count} indexes of size {} to file: {}",
+ INDEX_SIZE * count,
+ self.file_path
+ );
Ok(())
}
diff --git a/server/src/streaming/segments/messages/messages_reader.rs b/server/src/streaming/segments/messages/messages_reader.rs
index 39cdaad6..650a3980 100644
--- a/server/src/streaming/segments/messages/messages_reader.rs
+++ b/server/src/streaming/segments/messages/messages_reader.rs
@@ -3,10 +3,12 @@ use error_set::ErrContext;
use iggy::{
error::IggyError,
models::messaging::{IggyIndexes, IggyMessagesBatch},
+ prelude::IggyTimestamp,
};
use std::{
fs::{File, OpenOptions},
- os::unix::prelude::FileExt,
+ os::{fd::AsRawFd, unix::prelude::FileExt},
+ time::Duration,
};
use std::{
io::ErrorKind,
@@ -15,8 +17,11 @@ use std::{
Arc,
},
};
-use tokio::task::spawn_blocking;
-use tracing::{error, trace};
+use tokio::{
+ task::{spawn_blocking, JoinHandle},
+ time::interval,
+};
+use tracing::{debug, error, trace};
/// A dedicated struct for reading from the log file.
#[derive(Debug)]
@@ -24,6 +29,8 @@ pub struct MessagesReader {
file_path: String,
file: Arc<File>,
log_size_bytes: Arc<AtomicU64>,
+ last_access: Arc<AtomicU64>,
+ cache_advisor_task: Option<JoinHandle<()>>,
}
impl MessagesReader {
@@ -63,11 +70,16 @@ impl MessagesReader {
trace!("Opened messages file for reading: {file_path}, size: {actual_log_size}");
- Ok(Self {
+ let mut reader = Self {
file_path: file_path.to_string(),
file: Arc::new(file),
log_size_bytes,
- })
+ last_access: Arc::new(AtomicU64::new(IggyTimestamp::now().as_micros())),
+ cache_advisor_task: None,
+ };
+
+ reader.start_cache_advisor_task();
+ Ok(reader)
}
/// Loads and returns all message IDs from the log file.
@@ -119,10 +131,7 @@ impl MessagesReader {
let count_bytes = indexes.messages_size();
let messages_count = indexes.count();
- let messages_bytes = match self
- .read_bytes_at(start_pos as u64, count_bytes as u64)
- .await
- {
+ let messages_bytes = match self.read_bytes_at(start_pos as u64, count_bytes).await {
Ok(buf) => buf,
Err(error) if error.kind() == ErrorKind::UnexpectedEof => {
return Ok(IggyMessagesBatch::empty())
@@ -144,12 +153,14 @@ impl MessagesReader {
}
/// Returns the size of the log file in bytes.
- fn file_size(&self) -> u64 {
- self.log_size_bytes.load(Ordering::Acquire)
+ pub fn file_size(&self) -> u32 {
+ self.log_size_bytes.load(Ordering::Acquire) as u32
}
/// Reads `len` bytes from the log file at the specified `offset`.
- async fn read_bytes_at(&self, offset: u64, len: u64) -> Result<Bytes, std::io::Error> {
+ async fn read_bytes_at(&self, offset: u64, len: u32) -> Result<Bytes, std::io::Error> {
+ self.last_access
+ .store(IggyTimestamp::now().as_micros(), Ordering::Release);
let file = self.file.clone();
spawn_blocking(move || {
let mut buf = BytesMut::with_capacity(len as usize);
@@ -159,4 +170,76 @@ impl MessagesReader {
})
.await?
}
+
+ /// Starts a background task that manages file caching based on access patterns
+ fn start_cache_advisor_task(&mut self) {
+ let file = self.file.clone();
+ let file_path = self.file_path.clone();
+ let last_access = self.last_access.clone();
+ let log_size_bytes = self.log_size_bytes.clone();
+
+ const SEQUENTIAL_ACCESS_THRESHOLD_SECS: u64 = 1;
+ const DONTNEED_THRESHOLD_SECS: u64 = 5;
+ const CHECK_INTERVAL_SECS: u64 = 1;
+
+ let task = tokio::spawn(async move {
+ let mut last_advice: Option<nix::fcntl::PosixFadviseAdvice> = None;
+ let mut interval = interval(Duration::from_secs(CHECK_INTERVAL_SECS));
+
+ #[cfg(not(target_os = "macos"))]
+ loop {
+ interval.tick().await;
+
+ let now = IggyTimestamp::now().as_micros();
+ let last_access_time = last_access.load(Ordering::Acquire);
+ let elapsed_secs = (now - last_access_time) / 1_000_000;
+
+ let fd = file.as_raw_fd();
+ let file_size = log_size_bytes.load(Ordering::Acquire);
+
+ let advice = if elapsed_secs < SEQUENTIAL_ACCESS_THRESHOLD_SECS {
+ Some(nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL)
+ } else if elapsed_secs > DONTNEED_THRESHOLD_SECS {
+ Some(nix::fcntl::PosixFadviseAdvice::POSIX_FADV_DONTNEED)
+ } else {
+ None
+ };
+
+ if let Some(adv) = advice {
+ if last_advice != Some(adv) {
+ match nix::fcntl::posix_fadvise(fd, 0, file_size as i64, adv) {
+ Ok(_) => {
+ if adv == nix::fcntl::PosixFadviseAdvice::POSIX_FADV_DONTNEED {
+ trace!(
+ "Set DONTNEED advice on file: {}, no access for {}s",
+ file_path,
+ elapsed_secs
+ );
+ } else if adv
+ == nix::fcntl::PosixFadviseAdvice::POSIX_FADV_SEQUENTIAL
+ {
+ trace!(
+ "Set SEQUENTIAL advice on file: {}, recent access {}s ago",
+ file_path,
+ elapsed_secs
+ );
+ }
+ last_advice = Some(adv);
+ }
+ Err(e) => {
+ debug!("Failed to set posix_fadvise on file {}: {}", file_path, e);
+ }
+ }
+ }
+ }
+ }
+
+ #[cfg(target_os = "macos")]
+ loop {
+ interval.tick().await;
+ }
+ });
+
+ self.cache_advisor_task = Some(task);
+ }
}
diff --git a/server/src/streaming/segments/types/messages_batch_set.rs b/server/src/streaming/segments/types/messages_batch_set.rs
index 28f1225e..5d3d7210 100644
--- a/server/src/streaming/segments/types/messages_batch_set.rs
+++ b/server/src/streaming/segments/types/messages_batch_set.rs
@@ -7,7 +7,7 @@ use std::ops::Index;
use tracing::trace;
/// A container for multiple IggyMessagesBatch objects
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Default)]
pub struct IggyMessagesBatchSet {
/// The collection of message containers
batches: Vec<IggyMessagesBatch>,
@@ -59,11 +59,13 @@ impl IggyMessagesBatchSet {
self.batches.extend(other.batches);
}
+ // todo usize
/// Get the total number of messages in the batch
pub fn count(&self) -> u32 {
self.count
}
+ // todo usize
/// Get the total size of all messages in bytes
pub fn size(&self) -> u32 {
self.size
diff --git a/server/src/streaming/segments/writing_messages.rs b/server/src/streaming/segments/writing_messages.rs
index 7a5bed27..d43f65c1 100644
--- a/server/src/streaming/segments/writing_messages.rs
+++ b/server/src/streaming/segments/writing_messages.rs
@@ -58,6 +58,7 @@ impl Segment {
let accumulator = std::mem::take(&mut self.accumulator);
+ // todo change to put
accumulator.update_indexes(&mut self.indexes);
let batches = accumulator.into_batch_set();