gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-10 14:16:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): use pool lock quorum for set writes

Browse Source 
Use the pool-wide namespace lock client domain for every set in a pool so degraded EC8+4 multi-set writes are gated by node-level lock quorum instead of the narrower per-set endpoint host slice.

Keep namespace-lock domain deduplication tied to both the pool namespace and shared clients, and add regression coverage for three-locker degraded writes plus cross-pool domain separation.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-10 16:25:45 +08:00
parent 3391528025
commit 18bcd4fbb2
3 changed files with 121 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/sets.rs
+5 -6
View File
@@ -219,7 +219,10 @@ impl Sets {
let mut disk_set = Vec::with_capacity(set_count);
let lock_registry = runtime_sources::lock_registry();
let pool_lockers = runtime_sources::lock_registry()
.as_ref()
.map(|registry| registry.clients_for_endpoints(endpoints.endpoints.as_ref()))
.unwrap_or_default();
for i in 0..set_count {
let mut set_drive = Vec::with_capacity(set_drive_count);
@@ -270,10 +273,6 @@ impl Sets {
}
}
let lockers = lock_registry
.as_ref()
.map(|registry| registry.clients_for_endpoints(&set_endpoints))
.unwrap_or_default();
let set_disks = SetDisks::new_with_instance_ctx(
runtime_sources::local_node_name().await,
Arc::new(RwLock::new(set_drive)),
@@ -283,7 +282,7 @@ impl Sets {
pool_idx,
set_endpoints,
fm.clone(),
lockers,
pool_lockers.clone(),
instance_ctx.clone(),
)
.await;
crates/ecstore/src/set_disk/mod.rs
+78 -4
View File
@@ -3852,7 +3852,7 @@ pub struct SetDisks {
pub default_parity_count: usize,
pub set_index: usize,
pub pool_index: usize,
/// Stable namespace shared by every object lock created for this set.
/// Stable namespace shared by every object lock created for this pool.
set_lock_namespace: Arc<str>,
pub format: FormatV3,
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
@@ -4491,7 +4491,7 @@ impl SetDisks {
instance_ctx: Arc<InstanceContext>,
) -> Arc<Self> {
let ctx = instance_ctx;
let set_lock_namespace: Arc<str> = format!("set-{pool_index}-{set_index}").into();
let set_lock_namespace: Arc<str> = format!("pool-{pool_index}").into();
let shared_lockers = Arc::from(lockers.to_vec());
Arc::new(SetDisks {
locker_owner,
@@ -4605,7 +4605,9 @@ impl SetDisks {
pub(crate) async fn shares_namespace_lock_domain(&self, other: &Self) -> bool {
match (self.ctx.is_dist_erasure().await, other.ctx.is_dist_erasure().await) {
(false, false) => Arc::ptr_eq(&self.local_lock_manager, &other.local_lock_manager),
(true, true) => same_distributed_lock_domain(&self.lockers, &other.lockers),
(true, true) => {
self.set_lock_namespace == other.set_lock_namespace && same_distributed_lock_domain(&self.lockers, &other.lockers)
}
_ => false,
}
}
@@ -7123,7 +7125,7 @@ mod tests {
ctx.update_erasure_type(SetupType::Erasure).await;
let set = make_test_set_disks_with_ctx(Vec::new(), ctx).await;
assert_eq!(&*set.set_lock_namespace, "set-0-0");
assert_eq!(&*set.set_lock_namespace, "pool-0");
let before = Arc::strong_count(&set.set_lock_namespace);
let lock = set
.new_ns_lock("bucket", "object")
@@ -8348,6 +8350,78 @@ mod tests {
);
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn test_new_ns_lock_distributed_write_succeeds_with_three_lockers_one_offline() {
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
let manager_a = Arc::new(rustfs_lock::GlobalLockManager::new());
let manager_b = Arc::new(rustfs_lock::GlobalLockManager::new());
let healthy_a: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager_a));
let healthy_b: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager_b));
let failing_client: Arc<dyn LockClient> = Arc::new(FailingClient);
let set_disks = make_test_set_disks(vec![healthy_a, failing_client, healthy_b]).await;
let guard = set_disks
.new_ns_lock("bucket", "object")
.await
.expect("namespace lock should be created")
.get_write_lock(Duration::from_millis(500))
.await
.expect("two healthy lockers should satisfy the three-locker write quorum");
match guard {
NamespaceLockGuard::Standard(_) => {}
NamespaceLockGuard::Fast(_) => panic!("Expected distributed guard for dist-erasure"),
}
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn namespace_lock_domain_includes_pool_namespace() {
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
let first: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
let second: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
let lockers = vec![first, second];
let same_pool_first_set = make_test_set_disks_with_ctx(lockers.clone(), bootstrap_ctx()).await;
let same_pool_second_set = SetDisks::new_with_instance_ctx(
"test-owner".to_string(),
Arc::new(RwLock::new(vec![None, None])),
2,
1,
1,
0,
same_pool_first_set.set_endpoints.clone(),
FormatV3::new(2, 2),
lockers.clone(),
bootstrap_ctx(),
)
.await;
let other_pool_set = SetDisks::new_with_instance_ctx(
"test-owner".to_string(),
Arc::new(RwLock::new(vec![None, None])),
2,
1,
0,
1,
same_pool_first_set.set_endpoints.clone(),
FormatV3::new(1, 2),
lockers,
bootstrap_ctx(),
)
.await;
assert!(
same_pool_first_set.shares_namespace_lock_domain(&same_pool_second_set).await,
"sets in the same pool share the object namespace lock domain"
);
assert!(
!same_pool_first_set.shares_namespace_lock_domain(&other_pool_set).await,
"different pool namespaces must not be deduplicated solely by identical clients"
);
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn streaming_reader_holds_read_lock_until_eof() {
crates/ecstore/src/store/object.rs
+38 -18
View File
@@ -50,7 +50,7 @@ use crate::services::notification_sys::{
use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata};
use crate::set_disk::{
SetDisks, get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold,
is_lock_optimization_enabled, is_object_lock_diag_enabled, same_distributed_lock_domain,
is_lock_optimization_enabled, is_object_lock_diag_enabled,
};
use crate::storage_api_contracts::{
list::ListOperations as _,
@@ -3440,10 +3440,15 @@ impl ECStore {
for pool in &self.pools {
let hashed_set = pool.get_disks_by_key(object);
let lock_domain_already_held = !distributed
|| locked_sets
.iter()
.any(|locked_set| same_distributed_lock_domain(&locked_set.lockers, &hashed_set.lockers));
let mut lock_domain_already_held = !distributed;
if !lock_domain_already_held {
for locked_set in &locked_sets {
if locked_set.shares_namespace_lock_domain(&hashed_set).await {
lock_domain_already_held = true;
break;
}
}
}
if lock_domain_already_held {
continue;
}
@@ -3500,10 +3505,15 @@ impl ECStore {
let mut locked_sets = vec![fixed_set];
for pool in &self.pools {
for set in &pool.disk_set {
let lock_domain_already_held = !distributed
|| locked_sets
.iter()
.any(|locked_set| same_distributed_lock_domain(&locked_set.lockers, &set.lockers));
let mut lock_domain_already_held = !distributed;
if !lock_domain_already_held {
for locked_set in &locked_sets {
if locked_set.shares_namespace_lock_domain(set).await {
lock_domain_already_held = true;
break;
}
}
}
if lock_domain_already_held {
continue;
}
@@ -3581,10 +3591,15 @@ impl ECStore {
let mut locked_sets = vec![fixed_set];
for pool in &self.pools {
for set in &pool.disk_set {
let lock_domain_already_held = !distributed
|| locked_sets
.iter()
.any(|locked_set| same_distributed_lock_domain(&locked_set.lockers, &set.lockers));
let mut lock_domain_already_held = !distributed;
if !lock_domain_already_held {
for locked_set in &locked_sets {
if locked_set.shares_namespace_lock_domain(set).await {
lock_domain_already_held = true;
break;
}
}
}
if lock_domain_already_held {
continue;
}
@@ -3667,11 +3682,15 @@ impl ECStore {
.get(pool_idx)
.ok_or_else(|| Error::other(format!("invalid data movement publication pool {pool_idx}")))?;
let set = pool.get_disks_by_key(object);
let lock_domain_already_held = !locked_sets.is_empty()
&& (!distributed
|| locked_sets.iter().any(|locked_set: &Arc<crate::set_disk::SetDisks>| {
same_distributed_lock_domain(&locked_set.lockers, &set.lockers)
}));
let mut lock_domain_already_held = !locked_sets.is_empty() && !distributed;
if !lock_domain_already_held {
for locked_set in &locked_sets {
if locked_set.shares_namespace_lock_domain(&set).await {
lock_domain_already_held = true;
break;
}
}
}
if lock_domain_already_held {
continue;
}
@@ -5717,6 +5736,7 @@ mod tests {
GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup, GetObjectBodySource, clear_get_object_body_cache_hook,
lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
};
use crate::set_disk::same_distributed_lock_domain;
use crate::set_disk::{SetDisks, disk_call_counters};
use crate::storage_api_contracts::bucket::MakeBucketOptions;
use crate::storage_api_contracts::lifecycle::TransitionedObject;