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/09/29 21:03:27 UTC
(iggy) branch io_uring_tpc_szpont created (now aca52ca3)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch io_uring_tpc_szpont
in repository https://gitbox.apache.org/repos/asf/iggy.git
at aca52ca3 szpontsoft
This branch includes the following new commits:
new aca52ca3 szpontsoft
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.
(iggy) 01/01: szpontsoft
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch io_uring_tpc_szpont
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit aca52ca306760e6a948c24ea70369c69c7316b96
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Sep 29 23:03:14 2025 +0200
szpontsoft
---
core/server/src/slab/async_safe.rs | 41 +++++++++++++++++++
core/server/src/slab/mod.rs | 2 +
core/server/src/slab/streams.rs | 61 ++++++++++++++++++++--------
core/server/src/slab/try_borrow.rs | 82 ++++++++++++++++++++++++++++++++++++++
4 files changed, 170 insertions(+), 16 deletions(-)
diff --git a/core/server/src/slab/async_safe.rs b/core/server/src/slab/async_safe.rs
new file mode 100644
index 00000000..94b8968c
--- /dev/null
+++ b/core/server/src/slab/async_safe.rs
@@ -0,0 +1,41 @@
+use crate::slab::try_borrow::BorrowResult;
+use std::cell::RefCell;
+use std::time::Duration;
+use tracing::trace;
+
+/// Helper for async-safe RefCell operations in single-threaded contexts.
+/// When a borrow conflict occurs, yields to the runtime to let other tasks complete.
+pub struct AsyncSafe;
+
+impl AsyncSafe {
+ /// Yield to the runtime by sleeping for zero duration.
+ /// This allows other tasks to run and release their borrows.
+ async fn yield_now() {
+ compio::time::sleep(Duration::ZERO).await;
+ }
+
+ /// Try to perform an operation with automatic retry on borrow conflicts.
+ /// Yields to runtime when conflicts occur, allowing other tasks to complete.
+ pub async fn with_retry<T, F, R>(max_retries: usize, context: &str, mut f: F) -> BorrowResult<R>
+ where
+ F: FnMut() -> Result<R, T>,
+ {
+ for attempt in 0..max_retries {
+ match f() {
+ Ok(result) => return BorrowResult::Success(result),
+ Err(_) => {
+ if attempt == 0 {
+ trace!("RefCell borrow conflict in {}, yielding...", context);
+ }
+ Self::yield_now().await;
+ }
+ }
+ }
+
+ trace!(
+ "RefCell borrow still conflicted after {} retries in {}, skipping operation",
+ max_retries, context
+ );
+ BorrowResult::Skipped("max retries exceeded")
+ }
+}
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index 13c8f472..6089f4e1 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -1,9 +1,11 @@
+pub mod async_safe;
pub mod consumer_groups;
pub mod helpers;
pub mod partitions;
pub mod streams;
pub mod topics;
pub mod traits_ext;
+pub mod try_borrow;
use std::fmt::{Debug, Display};
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 7fadce50..1595ebaf 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -1,4 +1,6 @@
use crate::shard::task_registry::TaskRegistry;
+use crate::slab::async_safe::AsyncSafe;
+use crate::slab::try_borrow::BorrowResult;
use crate::streaming::partitions as streaming_partitions;
use crate::{
binary::handlers::messages::poll_messages_handler::IggyPollMetadata,
@@ -837,12 +839,35 @@ impl Streams {
return Ok(0);
}
- let batches = self.with_partition_by_id_mut(
- stream_id,
- topic_id,
- partition_id,
- streaming_partitions::helpers::commit_journal(),
- );
+ // Try to commit journal with retry on conflicts
+ const MAX_RETRIES: usize = 3;
+ let batches_result =
+ AsyncSafe::with_retry(MAX_RETRIES, "persist_messages:commit_journal", || {
+ // Try to get the lock and commit
+ // Return Result<T, ()> to satisfy the signature
+ Ok::<_, ()>(self.with_partition_by_id_mut(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::commit_journal(),
+ ))
+ })
+ .await;
+
+ let batches = match batches_result {
+ BorrowResult::Success(b) => b,
+ BorrowResult::Skipped(reason) => {
+ // If we can't get the lock after retries, skip this persist
+ // The data is still in the journal and will be persisted next time
+ tracing::debug!(
+ "Skipping persist for partition {} ({}): {}",
+ partition_id,
+ reason,
+ "will retry on next interval"
+ );
+ return Ok(0);
+ }
+ };
let (saved, batch_count) = self
.with_partition_by_id_async(
@@ -860,16 +885,20 @@ impl Streams {
)
.await?;
- self.with_partition_by_id_mut(
- stream_id,
- topic_id,
- partition_id,
- streaming_partitions::helpers::update_index_and_increment_stats(
- saved,
- batch_count,
- config,
- ),
- );
+ // Try to update stats with retry
+ let _ = AsyncSafe::with_retry(MAX_RETRIES, "persist_messages:update_stats", || {
+ Ok::<_, ()>(self.with_partition_by_id_mut(
+ stream_id,
+ topic_id,
+ partition_id,
+ streaming_partitions::helpers::update_index_and_increment_stats(
+ saved,
+ batch_count,
+ config,
+ ),
+ ))
+ })
+ .await;
Ok(batch_count)
}
diff --git a/core/server/src/slab/try_borrow.rs b/core/server/src/slab/try_borrow.rs
new file mode 100644
index 00000000..aaf6ca98
--- /dev/null
+++ b/core/server/src/slab/try_borrow.rs
@@ -0,0 +1,82 @@
+use std::cell::{BorrowError, BorrowMutError, Ref, RefCell, RefMut};
+use tracing::{debug, trace};
+
+/// Helper for safe borrowing that uses try_borrow to avoid panics.
+/// In single-threaded async, if a borrow fails, it means another task
+/// is currently using it, so we handle it gracefully.
+pub struct TryBorrow;
+
+impl TryBorrow {
+ /// Try to borrow, returning None if already borrowed.
+ /// This prevents panics in single-threaded async contexts.
+ pub fn borrow<T>(cell: &RefCell<T>) -> Option<Ref<'_, T>> {
+ match cell.try_borrow() {
+ Ok(borrowed) => Some(borrowed),
+ Err(e) => {
+ trace!("RefCell borrow failed (already borrowed): {:?}", e);
+ None
+ }
+ }
+ }
+
+ /// Try to borrow_mut, returning None if already borrowed.
+ pub fn borrow_mut<T>(cell: &RefCell<T>) -> Option<RefMut<'_, T>> {
+ match cell.try_borrow_mut() {
+ Ok(borrowed) => Some(borrowed),
+ Err(e) => {
+ trace!("RefCell borrow_mut failed (already borrowed): {:?}", e);
+ None
+ }
+ }
+ }
+
+ /// Try to borrow with a descriptive context for debugging.
+ pub fn borrow_with_context<'a, T>(
+ cell: &'a RefCell<T>,
+ context: &str,
+ ) -> Result<Ref<'a, T>, BorrowError> {
+ cell.try_borrow().map_err(|e| {
+ debug!("RefCell borrow failed in {}: {:?}", context, e);
+ e
+ })
+ }
+
+ /// Try to borrow_mut with a descriptive context for debugging.
+ pub fn borrow_mut_with_context<'a, T>(
+ cell: &'a RefCell<T>,
+ context: &str,
+ ) -> Result<RefMut<'a, T>, BorrowMutError> {
+ cell.try_borrow_mut().map_err(|e| {
+ debug!("RefCell borrow_mut failed in {}: {:?}", context, e);
+ e
+ })
+ }
+}
+
+/// Result type for operations that might skip due to borrow conflicts
+pub enum BorrowResult<T> {
+ /// Operation succeeded
+ Success(T),
+ /// Operation skipped because resource was busy
+ Skipped(&'static str),
+}
+
+impl<T> BorrowResult<T> {
+ pub fn is_success(&self) -> bool {
+ matches!(self, BorrowResult::Success(_))
+ }
+
+ pub fn is_skipped(&self) -> bool {
+ matches!(self, BorrowResult::Skipped(_))
+ }
+
+ pub fn unwrap_or_default(self) -> T
+ where
+ T: Default,
+ {
+ match self {
+ BorrowResult::Success(val) => val,
+ BorrowResult::Skipped(_) => T::default(),
+ }
+ }
+}