gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-31 09:18:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(storage-api): remove namespace lock from StorageAPI (#3365)

Browse Source 
* refactor(storage-api): remove namespace lock from StorageAPI

* test(scanner): wait for runtime budget cancellation

---------

Co-authored-by: loverustfs <hello@rustfs.com>
This commit is contained in:
安正超
2026-06-11 22:42:12 +08:00
committed by GitHub
parent 82af181dcf
commit a85cc0354c
14 changed files with 126 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/replication/replication_pool.rs
+11 -6
View File
@@ -27,7 +27,7 @@ use crate::bucket::replication::replication_state::ReplicationStats;
use crate::config::com::read_config;
use crate::disk::BUCKET_META_PREFIX;
use crate::error::Error as EcstoreError;
use crate::store_api::{ObjectIO, ObjectInfo};
use crate::store_api::{NamespaceLocking, ObjectIO, ObjectInfo};
use lazy_static::lazy_static;
use rustfs_filemeta::MrfReplicateEntry;
use rustfs_filemeta::ReplicateDecision;
@@ -205,7 +205,7 @@ impl Default for ReplicationPoolOpts {
}
/// Main replication pool structure
#[derive(Debug)]
pub struct ReplicationPool<S: StorageAPI> {
pub struct ReplicationPool<S: StorageAPI + NamespaceLocking> {
// Atomic counters for active workers
active_workers: Arc<AtomicI32>,
active_lrg_workers: Arc<AtomicI32>,
@@ -245,7 +245,7 @@ pub struct ReplicationPool<S: StorageAPI> {
resyncer: Arc<ReplicationResyncer>,
}
impl<S: StorageAPI> ReplicationPool<S> {
impl<S: StorageAPI + NamespaceLocking> ReplicationPool<S> {
/// Creates a new replication pool with specified options
pub async fn new(opts: ReplicationPoolOpts, stats: Arc<ReplicationStats>, storage: Arc<S>) -> Arc<Self> {
let max_workers = opts.max_workers.unwrap_or(WORKER_MAX_LIMIT);
@@ -1093,7 +1093,7 @@ pub trait ReplicationPoolTrait: std::fmt::Debug {
// Implement the trait for ReplicationPool
#[async_trait::async_trait]
impl<S: StorageAPI> ReplicationPoolTrait for ReplicationPool<S> {
impl<S: StorageAPI + NamespaceLocking> ReplicationPoolTrait for ReplicationPool<S> {
fn active_workers(&self) -> i32 {
ReplicationPool::<S>::active_workers(self)
}
@@ -1145,7 +1145,7 @@ lazy_static! {
}
/// Initializes background replication with the given options
pub async fn init_background_replication<S: StorageAPI>(storage: Arc<S>) {
pub async fn init_background_replication<S: StorageAPI + NamespaceLocking>(storage: Arc<S>) {
let stats = GLOBAL_REPLICATION_STATS
.get_or_init(|| async {
let stats = Arc::new(ReplicationStats::new());
@@ -1169,7 +1169,12 @@ pub fn get_global_replication_pool() -> Option<Arc<DynReplicationPool>> {
GLOBAL_REPLICATION_POOL.get().cloned()
}
pub async fn schedule_replication<S: StorageAPI>(oi: ObjectInfo, o: Arc<S>, dsc: ReplicateDecision, op_type: ReplicationType) {
pub async fn schedule_replication<S: StorageAPI + NamespaceLocking>(
oi: ObjectInfo,
o: Arc<S>,
dsc: ReplicateDecision,
op_type: ReplicationType,
) {
let tgt_statuses = replication_statuses_map(&oi.replication_status_internal.clone().unwrap_or_default());
let purge_statuses = version_purge_statuses_map(&oi.version_purge_status_internal.clone().unwrap_or_default());
let tm = get_str(&oi.user_defined, SUFFIX_REPLICATION_TIMESTAMP)
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+9 -4
View File
@@ -32,7 +32,9 @@ use crate::event_notification::{EventArgs, send_event};
use crate::global::GLOBAL_LocalNodeName;
use crate::global::get_global_bucket_monitor;
use crate::set_disk::get_lock_acquire_timeout;
use crate::store_api::{DeletedObject, HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOptions, ObjectToDelete, WalkOptions};
use crate::store_api::{
DeletedObject, HTTPRangeSpec, NamespaceLocking, ObjectIO, ObjectInfo, ObjectOptions, ObjectToDelete, WalkOptions,
};
use crate::{StorageAPI, new_object_layer_fn};
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput};
@@ -700,7 +702,7 @@ impl ReplicationResyncer {
}
#[instrument(skip(cancellation_token, storage))]
pub async fn resync_bucket<S: StorageAPI>(
pub async fn resync_bucket<S: StorageAPI + NamespaceLocking>(
self: Arc<Self>,
cancellation_token: CancellationToken,
storage: Arc<S>,
@@ -1668,7 +1670,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti
dsc
}
pub async fn replicate_delete<S: StorageAPI>(dobj: DeletedObjectReplicationInfo, storage: Arc<S>) {
pub async fn replicate_delete<S: StorageAPI + NamespaceLocking>(dobj: DeletedObjectReplicationInfo, storage: Arc<S>) {
if dobj.delete_object.force_delete {
replicate_force_delete_to_targets(&dobj, storage).await;
return;
@@ -2170,7 +2172,10 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb
}
}
async fn replicate_force_delete_to_targets<S: StorageAPI>(dobj: &DeletedObjectReplicationInfo, storage: Arc<S>) {
async fn replicate_force_delete_to_targets<S: StorageAPI + NamespaceLocking>(
dobj: &DeletedObjectReplicationInfo,
storage: Arc<S>,
) {
let bucket = &dobj.bucket;
let object_name = &dobj.delete_object.object_name;