diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs index ba43afac3..40bd14f50 100644 --- a/crates/ecstore/src/runtime/global.rs +++ b/crates/ecstore/src/runtime/global.rs @@ -53,9 +53,6 @@ pub const DISK_RESERVE_FRACTION: f64 = 0.15; lazy_static! { static ref GLOBAL_RUSTFS_PORT: OnceLock = OnceLock::new(); pub static ref GLOBAL_OBJECT_API: OnceLock> = OnceLock::new(); - pub static ref GLOBAL_LOCAL_DISK_MAP: Arc>>> = Arc::new(RwLock::new(HashMap::new())); - pub static ref GLOBAL_LOCAL_DISK_ID_MAP: Arc>> = Arc::new(RwLock::new(HashMap::new())); - pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc> = Arc::new(RwLock::new(Vec::new())); pub static ref GLOBAL_ROOT_DISK_THRESHOLD: RwLock = RwLock::new(0); pub static ref GLOBAL_LIFECYCLE_SYS: Arc = LifecycleSys::new(); pub static ref GLOBAL_BOOT_TIME: OnceCell = OnceCell::new(); @@ -158,9 +155,10 @@ pub fn get_global_endpoints_opt() -> Option { #[cfg(test)] pub async fn reset_local_disk_test_state() { - GLOBAL_LOCAL_DISK_MAP.write().await.clear(); - GLOBAL_LOCAL_DISK_ID_MAP.write().await.clear(); - GLOBAL_LOCAL_DISK_SET_DRIVES.write().await.clear(); + let ctx = current_ctx(); + ctx.local_disk_map().write().await.clear(); + ctx.local_disk_id_map().write().await.clear(); + ctx.local_disk_set_drives().write().await.clear(); } pub async fn is_first_cluster_node_local() -> bool { diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index d3f950bc9..68d60a4ce 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -39,14 +39,17 @@ //! startup writes and post-construction reads hit the same cell and //! single-instance behavior is byte-for-byte unchanged. +use super::global::TypeLocalDiskSetDrives; use crate::bucket::bandwidth::monitor::Monitor; use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState}; use crate::bucket::replication::{DynReplicationPool, ReplicationStats}; +use crate::disk::DiskStore; use crate::layout::endpoints::{EndpointServerPools, SetupType}; use crate::services::event_notification::EventNotifier; use crate::services::tier::tier::TierConfigMgr; use rustfs_lock::{GlobalLockManager, get_global_lock_manager}; use s3s::region::Region; +use std::collections::HashMap; use std::sync::{Arc, OnceLock}; use tokio::sync::{OnceCell, RwLock}; use uuid::Uuid; @@ -125,6 +128,16 @@ pub struct InstanceContext { /// Async set-once, built with the live storage during init. Replaces the /// process-global static. replication_pool: OnceCell>, + /// This instance's local disk registry (Phase 5 Slice 12, backlog#939). + /// + /// Populated during storage startup (before the `ECStore` exists) via the + /// bootstrap context that the store then adopts, so startup writes and later + /// reads share one registry. Replaces the three `GLOBAL_LOCAL_DISK_*` + /// statics: the path→disk map, the disk-id→endpoint map, and the + /// pool/set/drive layout. + local_disk_map: Arc>>>, + local_disk_id_map: Arc>>, + local_disk_set_drives: Arc>, } impl InstanceContext { @@ -154,6 +167,9 @@ impl InstanceContext { bucket_monitor: OnceLock::new(), replication_stats: OnceCell::new(), replication_pool: OnceCell::new(), + local_disk_map: Arc::new(RwLock::new(HashMap::new())), + local_disk_id_map: Arc::new(RwLock::new(HashMap::new())), + local_disk_set_drives: Arc::new(RwLock::new(Vec::new())), } } @@ -270,6 +286,21 @@ impl InstanceContext { &self.replication_pool } + /// Handle to this instance's local path→disk map. + pub(crate) fn local_disk_map(&self) -> Arc>>> { + self.local_disk_map.clone() + } + + /// Handle to this instance's local disk-id→endpoint map. + pub(crate) fn local_disk_id_map(&self) -> Arc>> { + self.local_disk_id_map.clone() + } + + /// Handle to this instance's local pool/set/drive layout. + pub(crate) fn local_disk_set_drives(&self) -> Arc> { + self.local_disk_set_drives.clone() + } + /// Update this instance's erasure setup type. pub async fn update_erasure_type(&self, setup_type: SetupType) { *self.erasure_kind.write().await = setup_type; diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 0e0c2d20c..dc1137427 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -29,13 +29,12 @@ use crate::{ error::Result, layout::endpoints::{EndpointServerPools, SetupType}, runtime::global::{ - GLOBAL_BOOT_TIME, GLOBAL_LIFECYCLE_SYS, GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, - GLOBAL_LOCAL_NODE_NAME_FALLBACK, GLOBAL_ROOT_DISK_THRESHOLD, TypeLocalDiskSetDrives, - get_background_services_cancel_token, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints, - get_global_endpoints_opt, get_global_lock_client, 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_erasure_sd, is_first_cluster_node_local, - resolve_object_store_handle, set_global_deployment_id, set_global_lock_client, set_global_lock_clients, set_object_layer, - update_erasure_type, + GLOBAL_BOOT_TIME, GLOBAL_LIFECYCLE_SYS, GLOBAL_LOCAL_NODE_NAME_FALLBACK, GLOBAL_ROOT_DISK_THRESHOLD, + TypeLocalDiskSetDrives, get_background_services_cancel_token, get_global_bucket_monitor, get_global_deployment_id, + get_global_endpoints, get_global_endpoints_opt, get_global_lock_client, 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_erasure_sd, + is_first_cluster_node_local, resolve_object_store_handle, set_global_deployment_id, set_global_lock_client, + set_global_lock_clients, set_object_layer, update_erasure_type, }, services::batch_processor::{GlobalBatchProcessors, get_global_processors}, services::event_notification::EventNotifier, @@ -371,7 +370,7 @@ pub fn bucket_monitor() -> Option> { /// `InstanceContext` (Phase 5 disk-registry migration, backlog#939) without a /// cross-crate signature change. Single-instance behavior is unchanged. pub async fn local_disk_map_read() -> OwnedRwLockReadGuard>> { - GLOBAL_LOCAL_DISK_MAP.clone().read_owned().await + local_disk_map_handle().read_owned().await } pub(crate) fn init_bucket_monitor_for_current_endpoints() { @@ -380,15 +379,15 @@ pub(crate) fn init_bucket_monitor_for_current_endpoints() { } pub(crate) fn local_disk_map_handle() -> Arc>>> { - GLOBAL_LOCAL_DISK_MAP.clone() + crate::runtime::global::current_ctx().local_disk_map() } pub(crate) fn local_disk_id_map_handle() -> Arc>> { - GLOBAL_LOCAL_DISK_ID_MAP.clone() + crate::runtime::global::current_ctx().local_disk_id_map() } pub(crate) fn local_disk_set_drives_handle() -> Arc> { - GLOBAL_LOCAL_DISK_SET_DRIVES.clone() + crate::runtime::global::current_ctx().local_disk_set_drives() } pub(crate) fn tier_config_mgr_handle() -> Arc> { @@ -408,24 +407,25 @@ pub(crate) fn event_notifier_handle() -> Arc> { } pub(crate) async fn local_disk_by_path(path: &str) -> Option { - GLOBAL_LOCAL_DISK_MAP.read().await.get(path).cloned().flatten() + local_disk_map_handle().read().await.get(path).cloned().flatten() } pub(crate) async fn local_disk_path_by_id(disk_id: &Uuid) -> Option { - GLOBAL_LOCAL_DISK_ID_MAP.read().await.get(disk_id).cloned() + local_disk_id_map_handle().read().await.get(disk_id).cloned() } #[cfg(test)] pub(crate) async fn clear_local_disk_id_map_for_test() { - GLOBAL_LOCAL_DISK_ID_MAP.write().await.clear(); + local_disk_id_map_handle().write().await.clear(); } pub(crate) async fn record_local_disk_id(disk_id: Uuid, endpoint: String) { - GLOBAL_LOCAL_DISK_ID_MAP.write().await.insert(disk_id, endpoint); + local_disk_id_map_handle().write().await.insert(disk_id, endpoint); } pub(crate) async fn replace_local_disk_id(previous: Option, current: Option, endpoint: String) { - let mut disk_id_map = GLOBAL_LOCAL_DISK_ID_MAP.write().await; + let id_map = local_disk_id_map_handle(); + let mut disk_id_map = id_map.write().await; if let Some(previous_id) = previous { disk_id_map.remove(&previous_id); } @@ -435,7 +435,8 @@ pub(crate) async fn replace_local_disk_id(previous: Option, current: Optio } pub(crate) async fn record_local_disks(disks: Vec) { - let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; + let map = local_disk_map_handle(); + let mut global_local_disk_map = map.write().await; for disk in disks { let path = disk.endpoint().to_string(); global_local_disk_map.insert(path, Some(disk.clone())); @@ -443,13 +444,14 @@ pub(crate) async fn record_local_disks(disks: Vec) { } pub(crate) async fn local_disk_set_drive(pool_idx: usize, set_idx: usize, disk_idx: usize) -> Option { - GLOBAL_LOCAL_DISK_SET_DRIVES.read().await[pool_idx][set_idx][disk_idx].clone() + local_disk_set_drives_handle().read().await[pool_idx][set_idx][disk_idx].clone() } pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option { - let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await; + let set_drives = local_disk_set_drives_handle(); + let global_set_drives = set_drives.read().await; if global_set_drives.is_empty() { - return GLOBAL_LOCAL_DISK_MAP + return local_disk_map_handle() .read() .await .get(&endpoint.to_string()) @@ -470,11 +472,11 @@ pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option Vec { - GLOBAL_LOCAL_DISK_MAP.read().await.keys().cloned().collect() + local_disk_map_handle().read().await.keys().cloned().collect() } pub(crate) async fn local_disks() -> Vec { - GLOBAL_LOCAL_DISK_MAP + local_disk_map_handle() .read() .await .values() @@ -483,11 +485,12 @@ pub(crate) async fn local_disks() -> Vec { } pub(crate) async fn local_disk_entries() -> Vec> { - GLOBAL_LOCAL_DISK_MAP.read().await.values().cloned().collect() + local_disk_map_handle().read().await.values().cloned().collect() } pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPools, opt: &DiskOption) -> Result<()> { - let mut global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.write().await; + let set_drives = local_disk_set_drives_handle(); + let mut global_set_drives = set_drives.write().await; for pool_eps in endpoint_pools.as_ref().iter() { let mut set_count_drives = Vec::with_capacity(pool_eps.set_count); for _ in 0..pool_eps.set_count { @@ -497,7 +500,8 @@ pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPoo global_set_drives.push(set_count_drives); } - let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; + let map = local_disk_map_handle(); + let mut global_local_disk_map = map.write().await; for pool_eps in endpoint_pools.as_ref().iter() { for ep in pool_eps.endpoints.as_ref().iter() {