From de8b465c414344e33ee30d47381d00bc47689a75 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 25 Oct 2026 20:15:49 +0800 Subject: [PATCH] fix(ecstore): batch concurrent fresh authority reads --- crates/ecstore/src/bucket/artifact.rs | 16 +- .../artifact/ordinary_classification.rs | 287 ++++++++++++++++++ crates/ecstore/src/core/pools.rs | 1 + crates/ecstore/src/core/pools_test.rs | 1 + crates/ecstore/src/data_usage/mod.rs | 1 + crates/ecstore/src/services/rebalance/mod.rs | 2 + .../rebalance/rebalance_unit_tests.rs | 1 + crates/ecstore/src/set_disk/ops/object.rs | 1 + crates/ecstore/src/store/heal.rs | 2 + crates/ecstore/src/store/init.rs | 1 + crates/ecstore/src/store/mod.rs | 3 + crates/ecstore/src/store/multipart.rs | 1 + crates/ecstore/src/store/object.rs | 2 + 13 files changed, 315 insertions(+), 4 deletions(-) create mode 100644 crates/ecstore/src/bucket/artifact/ordinary_classification.rs diff --git a/crates/ecstore/src/bucket/artifact.rs b/crates/ecstore/src/bucket/artifact.rs index b0f41811c..57d747ede 100644 --- a/crates/ecstore/src/bucket/artifact.rs +++ b/crates/ecstore/src/bucket/artifact.rs @@ -18,8 +18,10 @@ mod backup_retention; mod delete; pub mod inventory; +mod ordinary_classification; #[doc(hidden)] pub use delete::ArtifactDeleteTarget; +pub(crate) use ordinary_classification::OrdinaryClassificationRegistry; use super::versioning::VersioningApi as _; use super::{metadata::BucketMetadata, metadata_sys}; @@ -437,10 +439,16 @@ impl ECStore { } // Receiver classification must not migrate a legacy incarnation or // reacquire the bucket namespace lock held by a DeleteBucket caller. - let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket) - .await - .map_err(|_| DiskError::FileAccessDenied)?; - if !persisted || metadata.artifact_base_prefix.is_some() { + let allowed = self + .ordinary_classification + .classify(bucket, async { + match metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await { + Ok((metadata, persisted)) => persisted && metadata.artifact_base_prefix.is_none(), + Err(_) => false, + } + }) + .await; + if !allowed { return Err(DiskError::FileAccessDenied); } Ok(()) diff --git a/crates/ecstore/src/bucket/artifact/ordinary_classification.rs b/crates/ecstore/src/bucket/artifact/ordinary_classification.rs new file mode 100644 index 000000000..c327b657d --- /dev/null +++ b/crates/ecstore/src/bucket/artifact/ordinary_classification.rs @@ -0,0 +1,287 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; +use std::future::Future; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex, Weak}; + +/// Shares only reads started after a caller arrives; retains no idle results. +#[derive(Default)] +pub(crate) struct OrdinaryClassificationRegistry { + gates: Mutex>>, + #[cfg(test)] + after_unregister: Option, +} + +#[derive(Default)] +struct Gate { + started_generation: AtomicU64, + completed: tokio::sync::Mutex>, +} + +struct CompletedRead { + generation: u64, + allowed: bool, +} + +struct Registration<'a> { + registry: &'a OrdinaryClassificationRegistry, + bucket: &'a str, + gate: Option>, + observed_generation: u64, +} + +impl OrdinaryClassificationRegistry { + pub(crate) async fn classify(&self, bucket: &str, read: impl Future) -> bool { + self.register(bucket).classify(read).await + } + + fn register<'a>(&'a self, bucket: &'a str) -> Registration<'a> { + // Callers requiring physical admission must register after acquiring it. + // Release the registry lock before awaiting the gate, then acquire the + // existing metadata locks. + let mut gates = self.gates.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + let gate = gates.get(bucket).and_then(Weak::upgrade).unwrap_or_else(|| { + let gate = Arc::new(Gate::default()); + gates.insert(bucket.to_owned(), Arc::downgrade(&gate)); + gate + }); + let observed_generation = gate.started_generation.load(Ordering::Acquire); + Registration { + registry: self, + bucket, + gate: Some(gate), + observed_generation, + } + } +} + +impl Registration<'_> { + async fn classify(self, read: impl Future) -> bool { + let Some(gate) = self.gate.as_ref() else { + return false; + }; + let mut completed = gate.completed.lock().await; + // Only a read started after this caller registered can authorize it. + // Arrivals during a read therefore wait for the next fresh read. + if let Some(result) = completed + .as_ref() + .filter(|result| result.generation > self.observed_generation) + { + return result.allowed; + } + let Ok(previous) = gate + .started_generation + .try_update(Ordering::AcqRel, Ordering::Acquire, |generation| generation.checked_add(1)) + else { + return false; + }; + // Keep the gate through the original bounded metadata read. Cancellation + // releases it without publishing a result; another caller can read again. + let allowed = read.await; + *completed = Some(CompletedRead { + generation: previous + 1, + allowed, + }); + allowed + } +} + +impl Drop for Registration<'_> { + fn drop(&mut self) { + let Some(gate) = self.gate.take() else { + return; + }; + let mut gates = self.registry.gates.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + // Strong references are acquired and released under this registry lock, + // so concurrent final callers cannot both miss the last reference. + if Arc::strong_count(&gate) == 1 + && gates + .get(self.bucket) + .is_some_and(|registered| registered.ptr_eq(&Arc::downgrade(&gate))) + { + gates.remove(self.bucket); + } + drop(gate); + drop(gates); + #[cfg(test)] + if let Some(barrier) = &self.registry.after_unregister { + barrier.wait(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::poll; + use std::task::Poll; + use tokio::sync::oneshot; + + fn assert_empty(registry: &OrdinaryClassificationRegistry) { + assert!(registry.gates.lock().expect("classification registry").is_empty()); + } + + #[tokio::test] + async fn late_arrivals_share_only_the_next_fail_closed_read() { + let registry = OrdinaryClassificationRegistry::default(); + let reads = AtomicU64::new(0); + let (release, waiting) = oneshot::channel(); + let mut first = Box::pin(registry.classify("bucket", async { + reads.fetch_add(1, Ordering::Relaxed); + waiting.await.expect("release first read"); + true + })); + assert!(poll!(first.as_mut()).is_pending()); + let mut followers = (0..16) + .map(|_| { + Box::pin(registry.classify("bucket", async { + reads.fetch_add(1, Ordering::Relaxed); + false + })) + }) + .collect::>(); + for follower in &mut followers { + assert!(poll!(follower.as_mut()).is_pending()); + } + release.send(()).expect("first reader still alive"); + assert_eq!(poll!(first.as_mut()), Poll::Ready(true)); + for follower in &mut followers { + assert_eq!(poll!(follower.as_mut()), Poll::Ready(false)); + } + assert_eq!(reads.load(Ordering::Relaxed), 2); + assert_empty(®istry); + } + + #[tokio::test] + async fn successive_waves_always_read_again() { + let registry = OrdinaryClassificationRegistry::default(); + let reads = AtomicU64::new(0); + for expected in [true, false, true] { + assert_eq!( + registry + .classify("bucket", async { + reads.fetch_add(1, Ordering::Relaxed); + expected + }) + .await, + expected + ); + assert_empty(®istry); + } + assert_eq!(reads.load(Ordering::Relaxed), 3); + } + + #[tokio::test] + async fn canceled_leader_does_not_publish_or_strand_waiters() { + let registry = OrdinaryClassificationRegistry::default(); + let reads = AtomicU64::new(0); + let old_registration = registry.register("bucket"); + assert!( + registry + .classify("bucket", async { + reads.fetch_add(1, Ordering::Relaxed); + true + }) + .await + ); + let mut leader = Box::pin(registry.classify("bucket", async { + reads.fetch_add(1, Ordering::Relaxed); + std::future::pending::().await + })); + assert!(poll!(leader.as_mut()).is_pending()); + let mut old_waiter = + Box::pin(old_registration.classify(async { panic!("old waiter must reuse the completed post-registration read") })); + assert!(poll!(old_waiter.as_mut()).is_pending()); + let mut late_waiter = Box::pin(registry.classify("bucket", async { + reads.fetch_add(1, Ordering::Relaxed); + false + })); + assert!(poll!(late_waiter.as_mut()).is_pending()); + drop(leader); + assert_eq!(poll!(old_waiter.as_mut()), Poll::Ready(true)); + assert_eq!(poll!(late_waiter.as_mut()), Poll::Ready(false)); + assert_eq!(reads.load(Ordering::Relaxed), 3); + assert_empty(®istry); + } + + #[tokio::test] + async fn canceled_follower_does_not_retain_the_gate() { + let registry = OrdinaryClassificationRegistry::default(); + let (release, waiting) = oneshot::channel(); + let mut leader = Box::pin(registry.classify("bucket", async { + waiting.await.expect("release leader"); + true + })); + assert!(poll!(leader.as_mut()).is_pending()); + let mut follower = Box::pin(registry.classify("bucket", async { panic!("canceled follower must not read") })); + assert!(poll!(follower.as_mut()).is_pending()); + drop(follower); + assert_eq!(registry.gates.lock().expect("classification registry").len(), 1); + release.send(()).expect("leader still alive"); + assert_eq!(poll!(leader.as_mut()), Poll::Ready(true)); + assert_empty(®istry); + } + + #[test] + fn concurrent_final_registrations_remove_the_gate() { + let registry = OrdinaryClassificationRegistry { + after_unregister: Some(std::sync::Barrier::new(2)), + ..Default::default() + }; + let first = registry.register("bucket"); + let second = registry.register("bucket"); + // Both destructors reach the unlocked barrier before either returns. + // Moving the Arc release after this barrier leaves a stale registry key. + std::thread::scope(|scope| { + scope.spawn(move || drop(first)); + scope.spawn(move || drop(second)); + }); + assert_empty(®istry); + } + + #[tokio::test] + async fn buckets_and_stores_do_not_share_gates() { + let first_store = OrdinaryClassificationRegistry::default(); + let second_store = OrdinaryClassificationRegistry::default(); + let mut blocked = Box::pin(first_store.classify("first", std::future::pending())); + assert!(poll!(blocked.as_mut()).is_pending()); + assert_eq!(poll!(Box::pin(first_store.classify("second", async { true }))), Poll::Ready(true)); + assert_eq!(poll!(Box::pin(second_store.classify("first", async { false }))), Poll::Ready(false)); + assert_empty(&second_store); + drop(blocked); + assert_empty(&first_store); + } + + #[tokio::test] + async fn generation_exhaustion_fails_closed_without_reading() { + let registry = OrdinaryClassificationRegistry::default(); + let registration = registry.register("bucket"); + let gate = registration.gate.as_ref().expect("registered gate"); + gate.started_generation.store(u64::MAX, Ordering::Release); + *gate.completed.lock().await = Some(CompletedRead { + generation: u64::MAX, + allowed: true, + }); + assert!( + !registry + .classify("bucket", async { panic!("exhausted generation must deny without reading") }) + .await + ); + assert_eq!(gate.started_generation.load(Ordering::Acquire), u64::MAX); + drop(registration); + assert_empty(®istry); + } +} diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 544922627..df94915c1 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -29378,6 +29378,7 @@ mod pools_tests { pool_meta_save_gate: tokio::sync::Mutex::new(super::PoolMetaWriteState::for_test_bootstrap()), ctx, bucket_fence_registry: Arc::default(), + ordinary_classification: Default::default(), }) } diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 9223e4af1..28ada3180 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -499,6 +499,7 @@ mod decommission_lock_order_tests { ), ctx, bucket_fence_registry: Arc::default(), + ordinary_classification: Default::default(), }); crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await; (store, refresh_calls) diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index e4cae9e17..8fe7a8ef9 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -3334,6 +3334,7 @@ mod tests { pool_meta_save_gate: TokioMutex::default(), ctx, bucket_fence_registry: Arc::default(), + ordinary_classification: Default::default(), }) } diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index a6001be0e..1d4cef735 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/crates/ecstore/src/services/rebalance/mod.rs @@ -88,6 +88,7 @@ pub async fn test_store_with_persisted_rebalance_meta( pool_meta_save_gate: tokio::sync::Mutex::default(), ctx, bucket_fence_registry: std::sync::Arc::default(), + ordinary_classification: Default::default(), }); (temp_dirs, store) } @@ -266,6 +267,7 @@ async fn test_pool_stores_with_contexts( pool_meta_save_gate: tokio::sync::Mutex::new(pool_meta_write_state.independent_clone_for_test()), ctx: store_ctx, bucket_fence_registry: std::sync::Arc::default(), + ordinary_classification: Default::default(), }) }; let store = make_store(ctx); diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index 8951222a1..61edbe322 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -3415,6 +3415,7 @@ fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc, + pub(crate) ordinary_classification: crate::bucket::artifact::OrdinaryClassificationRegistry, } const METRIC_SCANNER_DATA_MOVEMENT_PAUSED: &str = "rustfs_scanner_data_movement_paused"; @@ -1904,6 +1905,7 @@ mod tests { pool_meta_save_gate: Mutex::default(), ctx, bucket_fence_registry: Arc::default(), + ordinary_classification: Default::default(), }; let rendered = format!("{store:?}"); @@ -1992,6 +1994,7 @@ mod tests { pool_meta_save_gate: Mutex::default(), ctx, bucket_fence_registry: Arc::default(), + ordinary_classification: Default::default(), }) } diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index c10c24215..6b95f4096 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -1663,6 +1663,7 @@ mod tests { pool_meta_save_gate: Mutex::default(), ctx: crate::runtime::instance::bootstrap_ctx(), bucket_fence_registry: std::sync::Arc::default(), + ordinary_classification: Default::default(), } } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 14334f4c5..ce1612d68 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -9762,6 +9762,7 @@ mod tests { pool_meta_save_gate: Mutex::default(), ctx: crate::runtime::instance::bootstrap_ctx(), bucket_fence_registry: std::sync::Arc::default(), + ordinary_classification: Default::default(), } } @@ -9871,6 +9872,7 @@ mod tests { pool_meta_save_gate: Mutex::default(), ctx, bucket_fence_registry: std::sync::Arc::default(), + ordinary_classification: Default::default(), } }