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/04 14:42:06 UTC
(iggy) branch zero-copy-no-batching-rebase updated: minor imprs
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
The following commit(s) were added to refs/heads/zero-copy-no-batching-rebase by this push:
new 42a00b74 minor imprs
42a00b74 is described below
commit 42a00b744d4fbba9613bf6c65e6de6e0d6f1856c
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 | 26 -----
.../streaming/segments/messages/messages_reader.rs | 107 ++++++++++++++++++---
server/src/streaming/segments/messages/mod.rs | 33 ++++---
.../streaming/segments/types/messages_batch_set.rs | 2 +-
7 files changed, 118 insertions(+), 62 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..ee25f160 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,
diff --git a/server/src/streaming/segments/messages/messages_reader.rs b/server/src/streaming/segments/messages/messages_reader.rs
index 39cdaad6..f5796df8 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 = 10;
+ 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/messages/mod.rs b/server/src/streaming/segments/messages/mod.rs
index 0642fdbb..08a5a985 100644
--- a/server/src/streaming/segments/messages/mod.rs
+++ b/server/src/streaming/segments/messages/mod.rs
@@ -18,24 +18,35 @@ async fn write_batch(
file_path: &str,
batches: IggyMessagesBatchSet,
) -> Result<usize, IggyError> {
- let mut slices: Vec<IoSlice> = batches.iter().map(|b| IoSlice::new(b)).collect();
+ // let mut slices: Vec<IoSlice> = batches.iter().map(|b| IoSlice::new(b)).collect();
- let slices = &mut slices.as_mut_slice();
- let mut total_written = 0;
+ // let slices = &mut slices.as_mut_slice();
+ // let mut total_written = 0;
- while !slices.is_empty() {
- let bytes_written = file
- .write_vectored(slices)
+ // while !slices.is_empty() {
+ // let bytes_written = file
+ // .write_vectored(slices)
+ // .await
+ // .with_error_context(|error| {
+ // format!("Failed to write messages to file: {file_path}, error: {error}",)
+ // })
+ // .map_err(|_| IggyError::CannotWriteToFile)?;
+
+ // total_written += bytes_written;
+
+ // IoSlice::advance_slices(slices, bytes_written);
+ // }
+
+ let mut bytes_written = 0;
+ for batch in batches.iter() {
+ bytes_written += batch.size() as usize;
+ file.write_all(batch)
.await
.with_error_context(|error| {
format!("Failed to write messages to file: {file_path}, error: {error}",)
})
.map_err(|_| IggyError::CannotWriteToFile)?;
-
- total_written += bytes_written;
-
- IoSlice::advance_slices(slices, bytes_written);
}
- Ok(total_written)
+ Ok(bytes_written)
}
diff --git a/server/src/streaming/segments/types/messages_batch_set.rs b/server/src/streaming/segments/types/messages_batch_set.rs
index 28f1225e..7b28e2d2 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>,