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

refactor: centralize ecstore batch sources and aliases (#3817)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-24 14:12:48 +08:00
committed by GitHub
parent cbf526df87
commit 94034ef406
11 changed files with 85 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/config/com.rs
+6 -14
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::config::{audit, notify, oidc, set_global_storage_class, storageclass};
use crate::config::{audit, notify, oidc, storageclass};
use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET};
use crate::error::{Error, Result};
use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
@@ -1288,7 +1288,7 @@ where
match storageclass::lookup_config(&kvs, *count) {
Ok(res) => {
if i == 0 {
set_global_storage_class(res);
runtime_sources::set_storage_class_config(res);
}
}
Err(err) => {
@@ -1312,8 +1312,8 @@ mod tests {
use crate::disk::endpoint::Endpoint;
use crate::endpoints::SetupType;
use crate::error::{Error, Result};
use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type};
use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
use crate::runtime_sources;
use crate::set_disk::SetDisks;
use http::HeaderMap;
use rustfs_config::audit::{AUDIT_AMQP_SUB_SYS, AUDIT_KAFKA_SUB_SYS, AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS};
@@ -1413,7 +1413,7 @@ mod tests {
impl SetupTypeGuard {
async fn switch_to(next: SetupType) -> Self {
let previous = current_setup_type().await;
update_erasure_type(next).await;
runtime_sources::set_setup_type(next).await;
Self { previous }
}
}
@@ -1424,22 +1424,14 @@ mod tests {
let handle = tokio::runtime::Handle::current();
tokio::task::block_in_place(|| {
handle.block_on(async move {
update_erasure_type(previous).await;
runtime_sources::set_setup_type(previous).await;
});
});
}
}
async fn current_setup_type() -> SetupType {
if is_dist_erasure().await {
SetupType::DistErasure
} else if is_erasure_sd().await {
SetupType::ErasureSD
} else if is_erasure().await {
SetupType::Erasure
} else {
SetupType::Unknown
}
runtime_sources::current_setup_type().await
}
impl LockingConfigStorage {
crates/ecstore/src/runtime_sources.rs
+23 -1
View File
@@ -17,12 +17,14 @@ use std::{collections::HashMap, sync::Arc, time::SystemTime};
use crate::bucket::bandwidth::monitor::Monitor;
use crate::disk::endpoint::Endpoint;
use crate::{
batch_processor::{GlobalBatchProcessors, get_global_processors},
bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, GLOBAL_ExpiryState, GLOBAL_TransitionState, TransitionState},
bucket::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys},
bucket::replication::{DynReplicationPool, GLOBAL_REPLICATION_POOL, GLOBAL_REPLICATION_STATS, ReplicationStats},
config::{get_global_storage_class, set_global_storage_class, storageclass},
disk::{DiskAPI, DiskOption, DiskStore, new_disk},
endpoints::EndpointServerPools,
endpoints::SetupType,
error::Result,
event_notification::EventNotifier,
global::{
@@ -31,7 +33,7 @@ use crate::{
TypeLocalDiskSetDrives, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints,
get_global_endpoints_opt, get_global_lock_clients, get_global_region, get_global_tier_config_mgr, global_rustfs_port,
init_global_bucket_monitor, is_dist_erasure, is_erasure, is_first_cluster_node_local, resolve_object_store_handle,
set_global_deployment_id, set_global_lock_client, set_global_lock_clients, set_object_layer,
set_global_deployment_id, set_global_lock_client, set_global_lock_clients, set_object_layer, update_erasure_type,
},
notification_sys::{NotificationSys, get_global_notification_sys},
store::ECStore,
@@ -95,6 +97,22 @@ pub(crate) async fn setup_is_erasure_sd() -> bool {
*GLOBAL_IsErasureSD.read().await
}
pub(crate) async fn current_setup_type() -> SetupType {
if setup_is_dist_erasure().await {
SetupType::DistErasure
} else if setup_is_erasure_sd().await {
SetupType::ErasureSD
} else if setup_is_erasure().await {
SetupType::Erasure
} else {
SetupType::Unknown
}
}
pub(crate) async fn set_setup_type(setup_type: SetupType) {
update_erasure_type(setup_type).await;
}
pub(crate) async fn local_node_name() -> String {
GLOBAL_LOCAL_NODE_NAME.read().await.clone()
}
@@ -244,6 +262,10 @@ pub(crate) fn set_storage_class_config(config: storageclass::Config) {
set_global_storage_class(config);
}
pub(crate) fn batch_processors() -> &'static GlobalBatchProcessors {
get_global_processors()
}
pub(crate) fn global_tier_config_mgr() -> Arc<RwLock<TierConfigMgr>> {
get_global_tier_config_mgr()
}
crates/ecstore/src/set_disk.rs
+4 -13
View File
@@ -15,7 +15,7 @@
#![allow(unused_imports)]
#![allow(unused_variables)]
use crate::batch_processor::{AsyncBatchProcessor, get_global_processors};
use crate::batch_processor::AsyncBatchProcessor;
use crate::bitrot::{create_bitrot_reader, create_bitrot_writer};
use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
use crate::bucket::metadata_sys;
@@ -5306,7 +5306,6 @@ mod tests {
use crate::disk::error::DiskError;
use crate::disk::health_state::RuntimeDriveHealthState;
use crate::endpoints::SetupType;
use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type};
use crate::object_api::ObjectInfo;
use crate::store_init::save_format_file;
use crate::store_list_objects::ListPathOptions;
@@ -5447,7 +5446,7 @@ mod tests {
impl SetupTypeGuard {
async fn switch_to(next: SetupType) -> Self {
let previous = current_setup_type().await;
update_erasure_type(next).await;
runtime_sources::set_setup_type(next).await;
Self { previous }
}
}
@@ -5458,22 +5457,14 @@ mod tests {
let handle = tokio::runtime::Handle::current();
tokio::task::block_in_place(|| {
handle.block_on(async move {
update_erasure_type(previous).await;
runtime_sources::set_setup_type(previous).await;
});
});
}
}
async fn current_setup_type() -> SetupType {
if is_dist_erasure().await {
SetupType::DistErasure
} else if is_erasure_sd().await {
SetupType::ErasureSD
} else if is_erasure().await {
SetupType::Erasure
} else {
SetupType::Unknown
}
runtime_sources::current_setup_type().await
}
async fn make_formatted_local_disk_for_info_test(disk_idx: usize, format: &FormatV3) -> (TempDir, Endpoint, DiskStore) {
crates/ecstore/src/set_disk/lock.rs
+1 -1
View File
@@ -166,7 +166,7 @@ impl SetDisks {
});
}
let processor = get_global_processors().metadata_processor();
let processor = runtime_sources::batch_processors().metadata_processor();
let results = processor.execute_batch(futures).await;
for (submitted_idx, result) in results.into_iter().enumerate() {
crates/ecstore/src/set_disk/read.rs
+1 -1
View File
@@ -307,7 +307,7 @@ impl SetDisks {
let version_id = version_id.to_string();
let opts = opts.clone();
let processor = get_global_processors().read_processor();
let processor = runtime_sources::batch_processors().read_processor();
let tasks: Vec<_> = disks
.iter()
.take(required_reads + 2) // Read a few extra for reliability
crates/ecstore/src/set_disk/write.rs
+1 -1
View File
@@ -270,7 +270,7 @@ impl SetDisks {
let mut errs = Vec::with_capacity(disks.len());
// Use improved simple batch processor instead of join_all for better performance
let processor = get_global_processors().write_processor();
let processor = runtime_sources::batch_processors().write_processor();
let tasks: Vec<_> = disks
.iter()