mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
refactor(ecstore): add per-instance InstanceContext, migrate erasure setup type (#4413)
* refactor(ecstore): add per-instance InstanceContext, migrate erasure setup type Phase 5 of the global-singleton consolidation (backlog#939): begin moving runtime identity state out of process globals so multiple ECStore instances can coexist in one process. Isolation is carried by the object graph (ECStore -> Sets -> SetDisks holding an Arc<InstanceContext>), not a task-local, which does not propagate across the many internal tokio::spawn boundaries in the data/background paths. This first slice migrates the erasure setup type -- previously three independent process-global bools -- into a single per-instance RwLock<SetupType> that derives is_erasure / is_dist_erasure / is_erasure_sd, removing a triple source of truth that could drift out of sync. - New runtime::instance module: InstanceContext + process bootstrap context. - The legacy free-function facade (is_erasure/update_erasure_type/...) keeps its signatures and forwards to the current instance's context, falling back to the bootstrap context before a store is published. - ECStore gains a pub(crate) ctx field and setup_is_* accessors; its constructors adopt the bootstrap context (never mint a fresh one) so startup writes and post-construction reads share one cell -- single-instance behavior is byte-for-byte unchanged. Tests: erasure predicate derivation vs the legacy behavior, object-graph carrier isolation across two ECStore instances, and bootstrap adoption. Refs: backlog#939 (Phase 5, Slice 1), backlog#653 (item 8) * refactor(ecstore): thread InstanceContext down the object graph (Phase 5 Slice 2) (#4415) * refactor(ecstore): source the namespace lock manager per-instance (#4417) refactor(ecstore): source the namespace lock manager per-instance (Phase 5 Slice 3) Phase 5 Slice 3 (backlog#939): give each instance its own lock namespace by sourcing SetDisks' lock manager from the instance context instead of the process singleton. This removes the false cross-instance mutual exclusion (and attendant ABBA risk) that a shared GlobalLockManager would cause once multiple instances coexist. - InstanceContext gains a `lock_manager: Arc<GlobalLockManager>`. `new()` mints a fresh manager (independent per-instance); `bootstrap_ctx()` aliases the process singleton via get_global_lock_manager(), so a single-instance deployment keeps exactly one shared namespace. - SetDisks::new sources `local_lock_manager` from `ctx.lock_manager()` (the ctx it already adopts), not `runtime_sources::global_lock_manager()`. Single instance: same Arc as before, so behavior is unchanged. - Remove the now-unused `runtime_sources::global_lock_manager()` wrapper. Tests: bootstrap lock manager aliases the process singleton; two fresh contexts own distinct managers; a SetDisks' lock manager is the one from its context and aliases the global singleton in a single-instance build. Verification: cargo test -p rustfs-ecstore (10 Phase 5 + set_disk locking regressions green), cargo clippy -p rustfs-ecstore --all-targets (clean), make pre-commit (pass). Refs: backlog#939 (Phase 5, Slice 3). Stacked on #4415 (Slice 2).
This commit is contained in:
@@ -33,6 +33,7 @@ use crate::{
|
||||
error::StorageError,
|
||||
layout::endpoints::{Endpoints, PoolEndpoints},
|
||||
object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader},
|
||||
runtime::instance::{InstanceContext, bootstrap_ctx},
|
||||
runtime::sources as runtime_sources,
|
||||
set_disk::SetDisks,
|
||||
store::init_format::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file},
|
||||
@@ -77,6 +78,12 @@ pub struct Sets {
|
||||
pub default_parity_count: usize,
|
||||
pub distribution_algo: DistributionAlgoVersion,
|
||||
exit_signal: Option<Sender<()>>,
|
||||
/// Per-instance runtime context (Phase 5, backlog#939).
|
||||
///
|
||||
/// Carried down the object graph (ECStore → Sets → SetDisks) so that
|
||||
/// instance-scoped state resolves through the owning instance rather than a
|
||||
/// process global. Consumed starting Slice 3 (lock-namespace isolation).
|
||||
ctx: Arc<InstanceContext>,
|
||||
}
|
||||
|
||||
impl Drop for Sets {
|
||||
@@ -186,6 +193,10 @@ impl Sets {
|
||||
default_parity_count: parity_count,
|
||||
distribution_algo: fm.erasure.distribution_algo.clone(),
|
||||
exit_signal: Some(tx),
|
||||
// Single-instance: same bootstrap context the owning ECStore adopts
|
||||
// (constructed before the store, so sourced here directly). Slice 8
|
||||
// threads a per-instance context in for true multi-instance.
|
||||
ctx: bootstrap_ctx(),
|
||||
});
|
||||
|
||||
let asets = sets.clone();
|
||||
@@ -200,6 +211,12 @@ impl Sets {
|
||||
self.set_drive_count
|
||||
}
|
||||
|
||||
/// This pool's per-instance runtime context (Phase 5, backlog#939).
|
||||
#[allow(dead_code)] // Consumed starting Slice 3 (lock-namespace isolation).
|
||||
pub(crate) fn instance_ctx(&self) -> &Arc<InstanceContext> {
|
||||
&self.ctx
|
||||
}
|
||||
|
||||
pub async fn monitor_and_connect_endpoints(&self, mut rx: Receiver<()>) {
|
||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||
|
||||
@@ -1165,6 +1182,7 @@ mod tests {
|
||||
default_parity_count: 1,
|
||||
distribution_algo: DistributionAlgoVersion::V1,
|
||||
exit_signal: None,
|
||||
ctx: bootstrap_ctx(),
|
||||
};
|
||||
|
||||
let result = sets
|
||||
@@ -1244,6 +1262,7 @@ mod tests {
|
||||
default_parity_count: 1,
|
||||
distribution_algo: DistributionAlgoVersion::V1,
|
||||
exit_signal: None,
|
||||
ctx: bootstrap_ctx(),
|
||||
});
|
||||
|
||||
let bucket = format!("bucket-{}", Uuid::new_v4().simple());
|
||||
@@ -1290,6 +1309,7 @@ mod tests {
|
||||
default_parity_count: 0,
|
||||
distribution_algo: DistributionAlgoVersion::V1,
|
||||
exit_signal: None,
|
||||
ctx: bootstrap_ctx(),
|
||||
};
|
||||
|
||||
let err = sets
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::instance::{InstanceContext, bootstrap_ctx};
|
||||
use crate::bucket::bandwidth::monitor::Monitor;
|
||||
use crate::{
|
||||
bucket::lifecycle::bucket_lifecycle_ops::LifecycleSys,
|
||||
@@ -42,16 +43,17 @@ pub const DISK_RESERVE_FRACTION: f64 = 0.15;
|
||||
// These should be migrated to AppContext over time.
|
||||
// See issue #730 for migration plan.
|
||||
//
|
||||
// Tier A (needs migration): GLOBAL_OBJECT_API, GLOBAL_IS_ERASURE*, GLOBAL_LOCAL_DISK_*,
|
||||
// Tier A (needs migration): GLOBAL_OBJECT_API, GLOBAL_LOCAL_DISK_*,
|
||||
// GLOBAL_ROOT_DISK_THRESHOLD, GLOBAL_LIFECYCLE_SYS, GLOBAL_EVENT_NOTIFIER, etc.
|
||||
// Tier B (keep as static): GLOBAL_RUSTFS_PORT, GLOBAL_REGION, env var caches, etc.
|
||||
//
|
||||
// Phase 5 (backlog#939): the erasure setup type moved into the per-instance
|
||||
// `InstanceContext` (see `super::instance`); the erasure predicates below now
|
||||
// forward to the current instance's context.
|
||||
lazy_static! {
|
||||
static ref GLOBAL_RUSTFS_PORT: OnceLock<u16> = OnceLock::new();
|
||||
static ref GLOBAL_DEPLOYMENT_ID: OnceLock<Uuid> = OnceLock::new();
|
||||
pub static ref GLOBAL_OBJECT_API: OnceLock<Arc<ECStore>> = OnceLock::new();
|
||||
pub static ref GLOBAL_IS_ERASURE: RwLock<bool> = RwLock::new(false);
|
||||
pub static ref GLOBAL_IS_DIST_ERASURE: RwLock<bool> = RwLock::new(false);
|
||||
pub static ref GLOBAL_IS_ERASURE_SD: RwLock<bool> = RwLock::new(false);
|
||||
pub static ref GLOBAL_LOCAL_DISK_MAP: Arc<RwLock<HashMap<String, Option<DiskStore>>>> = Arc::new(RwLock::new(HashMap::new()));
|
||||
pub static ref GLOBAL_LOCAL_DISK_ID_MAP: Arc<RwLock<HashMap<Uuid, String>>> = Arc::new(RwLock::new(HashMap::new()));
|
||||
pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
|
||||
@@ -207,6 +209,19 @@ pub fn resolve_object_store_handle() -> Option<Arc<ECStore>> {
|
||||
.or_else(new_object_layer_fn)
|
||||
}
|
||||
|
||||
/// Resolve the instance context for the legacy free-function facade.
|
||||
///
|
||||
/// Prefers the currently-published `ECStore`'s own context; before any store
|
||||
/// is published (e.g. during storage startup, or in unit tests) it falls back
|
||||
/// to the process-level [`bootstrap_ctx`]. Because `ECStore::new` adopts the
|
||||
/// bootstrap `Arc`, single-instance callers always observe one and the same
|
||||
/// context — behavior is unchanged from the previous process-global bools.
|
||||
pub(crate) fn current_ctx() -> Arc<InstanceContext> {
|
||||
resolve_object_store_handle()
|
||||
.map(|store| store.ctx.clone())
|
||||
.unwrap_or_else(bootstrap_ctx)
|
||||
}
|
||||
|
||||
/// Set the global object layer
|
||||
///
|
||||
/// # Arguments
|
||||
@@ -226,8 +241,7 @@ pub async fn set_object_layer(o: Arc<ECStore>) {
|
||||
/// * `bool` - True if the setup type is distributed erasure coding, false otherwise
|
||||
///
|
||||
pub async fn is_dist_erasure() -> bool {
|
||||
let lock = GLOBAL_IS_DIST_ERASURE.read().await;
|
||||
*lock
|
||||
current_ctx().is_dist_erasure().await
|
||||
}
|
||||
|
||||
/// Check if the setup type is erasure coding with single data center
|
||||
@@ -236,8 +250,7 @@ pub async fn is_dist_erasure() -> bool {
|
||||
/// * `bool` - True if the setup type is erasure coding with single data center, false otherwise
|
||||
///
|
||||
pub async fn is_erasure_sd() -> bool {
|
||||
let lock = GLOBAL_IS_ERASURE_SD.read().await;
|
||||
*lock
|
||||
current_ctx().is_erasure_sd().await
|
||||
}
|
||||
|
||||
/// Check if the setup type is erasure coding
|
||||
@@ -246,8 +259,7 @@ pub async fn is_erasure_sd() -> bool {
|
||||
/// * `bool` - True if the setup type is erasure coding, false otherwise
|
||||
///
|
||||
pub async fn is_erasure() -> bool {
|
||||
let lock = GLOBAL_IS_ERASURE.read().await;
|
||||
*lock
|
||||
current_ctx().is_erasure().await
|
||||
}
|
||||
|
||||
/// Update the global erasure type based on the setup type
|
||||
@@ -258,18 +270,7 @@ pub async fn is_erasure() -> bool {
|
||||
/// # Returns
|
||||
/// * None
|
||||
pub async fn update_erasure_type(setup_type: SetupType) {
|
||||
let mut is_erasure = GLOBAL_IS_ERASURE.write().await;
|
||||
*is_erasure = setup_type == SetupType::Erasure;
|
||||
|
||||
let mut is_dist_erasure = GLOBAL_IS_DIST_ERASURE.write().await;
|
||||
*is_dist_erasure = setup_type == SetupType::DistErasure;
|
||||
|
||||
if *is_dist_erasure {
|
||||
*is_erasure = true
|
||||
}
|
||||
|
||||
let mut is_erasure_sd = GLOBAL_IS_ERASURE_SD.write().await;
|
||||
*is_erasure_sd = setup_type == SetupType::ErasureSD;
|
||||
current_ctx().update_erasure_type(setup_type).await;
|
||||
}
|
||||
|
||||
// pub fn is_legacy() -> bool {
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
// 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.
|
||||
|
||||
//! Per-instance runtime state ("instance context").
|
||||
//!
|
||||
//! Phase 5 of the global-singleton consolidation (backlog#939). State that
|
||||
//! carries an `ECStore` instance's identity/runtime is being moved out of the
|
||||
//! process-level statics in [`super::global`] into this context so that
|
||||
//! multiple `ECStore` instances can coexist in one process without
|
||||
//! cross-contaminating each other's state.
|
||||
//!
|
||||
//! ## Isolation carrier
|
||||
//!
|
||||
//! Multi-instance disambiguation is carried by the **object graph**
|
||||
//! (`ECStore` → `Vec<Arc<Sets>>` → `SetDisks`), each holding an
|
||||
//! `Arc<InstanceContext>`. Instance-scoped reads resolve through `self.ctx`,
|
||||
//! so no call-site signatures change. (A `task_local` was rejected during
|
||||
//! design review: it does not propagate across the many internal
|
||||
//! `tokio::spawn` boundaries in the data/background paths.)
|
||||
//!
|
||||
//! ## Backward compatibility
|
||||
//!
|
||||
//! The legacy free-function facade in [`super::global`]
|
||||
//! (`is_erasure`/`update_erasure_type`/…) resolves the *current* instance via
|
||||
//! the object-store resolver and falls back to the process-level
|
||||
//! [`bootstrap_ctx`] when no store is published yet. Because `ECStore::new`
|
||||
//! **adopts** the same bootstrap `Arc` (rather than minting a fresh one),
|
||||
//! startup writes and post-construction reads hit the same cell and
|
||||
//! single-instance behavior is byte-for-byte unchanged.
|
||||
|
||||
use crate::layout::endpoints::SetupType;
|
||||
use rustfs_lock::{GlobalLockManager, get_global_lock_manager};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
/// Runtime state owned by a single `ECStore` instance.
|
||||
///
|
||||
/// This is intentionally minimal in the first migration slice; subsequent
|
||||
/// slices move additional identity/runtime state (topology, disk registry,
|
||||
/// service handles, cancellation token) into this struct.
|
||||
#[derive(Debug)]
|
||||
pub struct InstanceContext {
|
||||
/// The deployment's erasure setup type.
|
||||
///
|
||||
/// Single source of truth for the derived `is_erasure` /
|
||||
/// `is_dist_erasure` / `is_erasure_sd` predicates, replacing the three
|
||||
/// previously-independent process-global erasure-mode bools (which could
|
||||
/// drift out of sync). Stored as one value so the three predicates can
|
||||
/// never observe a torn intermediate state.
|
||||
erasure_kind: RwLock<SetupType>,
|
||||
/// This instance's namespace lock manager (Phase 5 Slice 3, backlog#939).
|
||||
///
|
||||
/// Owned per-instance so two instances no longer share a lock namespace
|
||||
/// (which would make same-named objects in different instances falsely
|
||||
/// contend). Single-instance aliases the process singleton, so behavior is
|
||||
/// unchanged; see [`bootstrap_ctx`].
|
||||
lock_manager: Arc<GlobalLockManager>,
|
||||
}
|
||||
|
||||
impl InstanceContext {
|
||||
/// Create a fresh instance context in the initial [`SetupType::Unknown`]
|
||||
/// state with its own lock manager — byte-for-byte equivalent to the old
|
||||
/// all-`false` erasure globals.
|
||||
pub fn new() -> Self {
|
||||
Self::with_lock_manager(Arc::new(GlobalLockManager::new()))
|
||||
}
|
||||
|
||||
/// Build a context bound to a specific lock manager.
|
||||
///
|
||||
/// [`bootstrap_ctx`] uses this to alias the process-global lock manager so
|
||||
/// single-instance deployments keep one shared namespace; `new` mints a
|
||||
/// fresh manager for a genuinely independent instance.
|
||||
fn with_lock_manager(lock_manager: Arc<GlobalLockManager>) -> Self {
|
||||
Self {
|
||||
erasure_kind: RwLock::new(SetupType::Unknown),
|
||||
lock_manager,
|
||||
}
|
||||
}
|
||||
|
||||
/// This instance's namespace lock manager.
|
||||
pub fn lock_manager(&self) -> Arc<GlobalLockManager> {
|
||||
self.lock_manager.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;
|
||||
}
|
||||
|
||||
/// Whether this instance uses erasure coding.
|
||||
///
|
||||
/// True for both single-node ([`SetupType::Erasure`]) and distributed
|
||||
/// ([`SetupType::DistErasure`]) erasure setups, matching the original
|
||||
/// `update_erasure_type` derivation where `DistErasure` implied
|
||||
/// `is_erasure == true`.
|
||||
pub async fn is_erasure(&self) -> bool {
|
||||
matches!(*self.erasure_kind.read().await, SetupType::Erasure | SetupType::DistErasure)
|
||||
}
|
||||
|
||||
/// Whether this instance uses distributed erasure coding.
|
||||
pub async fn is_dist_erasure(&self) -> bool {
|
||||
*self.erasure_kind.read().await == SetupType::DistErasure
|
||||
}
|
||||
|
||||
/// Whether this instance uses single-drive erasure coding.
|
||||
pub async fn is_erasure_sd(&self) -> bool {
|
||||
*self.erasure_kind.read().await == SetupType::ErasureSD
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for InstanceContext {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
/// Process-level bootstrap instance context.
|
||||
///
|
||||
/// Storage startup writes erasure state before any `ECStore` exists (see
|
||||
/// `init_startup_storage_foundation`), so the context must exist ahead of the
|
||||
/// object graph. `ECStore::new` adopts this same `Arc` via [`bootstrap_ctx`],
|
||||
/// guaranteeing that startup writes and post-construction reads share one cell.
|
||||
static BOOTSTRAP_CTX: OnceLock<Arc<InstanceContext>> = OnceLock::new();
|
||||
|
||||
/// Return the process-level bootstrap instance context, creating it on first
|
||||
/// access.
|
||||
///
|
||||
/// This is the single-instance default that the legacy free-function facade
|
||||
/// falls back to before an `ECStore` is published, and the `Arc` that
|
||||
/// `ECStore::new` adopts as its own `ctx`.
|
||||
pub fn bootstrap_ctx() -> Arc<InstanceContext> {
|
||||
BOOTSTRAP_CTX
|
||||
.get_or_init(|| Arc::new(InstanceContext::with_lock_manager(get_global_lock_manager())))
|
||||
.clone()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
// The SetupType inputs must derive the exact (is_erasure,
|
||||
// is_dist_erasure, is_erasure_sd) triples that the original three
|
||||
// process-global erasure bools produced via update_erasure_type().
|
||||
#[tokio::test]
|
||||
async fn erasure_predicates_match_legacy_derivation() {
|
||||
let cases = [
|
||||
// (input, is_erasure, is_dist_erasure, is_erasure_sd)
|
||||
(SetupType::Unknown, false, false, false),
|
||||
(SetupType::FS, false, false, false),
|
||||
(SetupType::Erasure, true, false, false),
|
||||
// DistErasure implies is_erasure == true (legacy global.rs:267-269).
|
||||
(SetupType::DistErasure, true, true, false),
|
||||
(SetupType::ErasureSD, false, false, true),
|
||||
];
|
||||
|
||||
for (input, want_erasure, want_dist, want_sd) in cases {
|
||||
let ctx = InstanceContext::new();
|
||||
ctx.update_erasure_type(input.clone()).await;
|
||||
assert_eq!(ctx.is_erasure().await, want_erasure, "is_erasure for {input:?}");
|
||||
assert_eq!(ctx.is_dist_erasure().await, want_dist, "is_dist_erasure for {input:?}");
|
||||
assert_eq!(ctx.is_erasure_sd().await, want_sd, "is_erasure_sd for {input:?}");
|
||||
}
|
||||
}
|
||||
|
||||
// A fresh context (before any update) reflects the initial all-false state.
|
||||
#[tokio::test]
|
||||
async fn fresh_context_is_all_false() {
|
||||
let ctx = InstanceContext::new();
|
||||
assert!(!ctx.is_erasure().await);
|
||||
assert!(!ctx.is_dist_erasure().await);
|
||||
assert!(!ctx.is_erasure_sd().await);
|
||||
}
|
||||
|
||||
// bootstrap_ctx() is a stable process singleton: repeated calls return the
|
||||
// same Arc, so a startup write is visible to a later read through it.
|
||||
#[tokio::test]
|
||||
async fn bootstrap_ctx_is_stable_singleton() {
|
||||
let a = bootstrap_ctx();
|
||||
let b = bootstrap_ctx();
|
||||
assert!(Arc::ptr_eq(&a, &b), "bootstrap_ctx must return the same Arc");
|
||||
}
|
||||
|
||||
// Two independent contexts do not share erasure state — the property that
|
||||
// lets two ECStore instances (each carrying its own ctx) stay isolated.
|
||||
#[tokio::test]
|
||||
async fn distinct_contexts_do_not_share_state() {
|
||||
let ctx_a = Arc::new(InstanceContext::new());
|
||||
let ctx_b = Arc::new(InstanceContext::new());
|
||||
ctx_a.update_erasure_type(SetupType::DistErasure).await;
|
||||
ctx_b.update_erasure_type(SetupType::ErasureSD).await;
|
||||
|
||||
assert!(ctx_a.is_dist_erasure().await && ctx_a.is_erasure().await);
|
||||
assert!(!ctx_a.is_erasure_sd().await);
|
||||
|
||||
assert!(ctx_b.is_erasure_sd().await);
|
||||
assert!(!ctx_b.is_erasure().await && !ctx_b.is_dist_erasure().await);
|
||||
}
|
||||
|
||||
// Single-instance zero-change: the bootstrap context's lock manager is the
|
||||
// very same Arc the process singleton hands out, so the lock namespace is
|
||||
// unchanged from before Slice 3.
|
||||
#[tokio::test]
|
||||
async fn bootstrap_lock_manager_aliases_process_singleton() {
|
||||
assert!(
|
||||
Arc::ptr_eq(&bootstrap_ctx().lock_manager(), &get_global_lock_manager()),
|
||||
"bootstrap context must alias the process lock-manager singleton"
|
||||
);
|
||||
}
|
||||
|
||||
// Two independent contexts own distinct lock managers — the property that
|
||||
// stops two instances from falsely contending on same-named objects.
|
||||
#[tokio::test]
|
||||
async fn distinct_contexts_have_distinct_lock_managers() {
|
||||
let ctx_a = InstanceContext::new();
|
||||
let ctx_b = InstanceContext::new();
|
||||
assert!(
|
||||
!Arc::ptr_eq(&ctx_a.lock_manager(), &ctx_b.lock_manager()),
|
||||
"fresh contexts must not share a lock manager"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -16,4 +16,5 @@
|
||||
#![allow(dead_code)]
|
||||
|
||||
pub(crate) mod global;
|
||||
pub(crate) mod instance;
|
||||
pub(crate) mod sources;
|
||||
|
||||
@@ -32,13 +32,13 @@ use crate::{
|
||||
error::Result,
|
||||
layout::endpoints::{EndpointServerPools, SetupType},
|
||||
runtime::global::{
|
||||
GLOBAL_BOOT_TIME, GLOBAL_EVENT_NOTIFIER, GLOBAL_IS_ERASURE_SD, 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,
|
||||
GLOBAL_TIER_CONFIG_MGR, 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_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_EVENT_NOTIFIER, 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, GLOBAL_TIER_CONFIG_MGR,
|
||||
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,
|
||||
@@ -148,7 +148,7 @@ pub async fn setup_is_dist_erasure() -> bool {
|
||||
}
|
||||
|
||||
pub async fn setup_is_erasure_sd() -> bool {
|
||||
*GLOBAL_IS_ERASURE_SD.read().await
|
||||
is_erasure_sd().await
|
||||
}
|
||||
|
||||
pub(crate) async fn current_setup_type() -> SetupType {
|
||||
@@ -215,7 +215,7 @@ pub(crate) async fn scanner_init_time() -> Option<chrono::DateTime<chrono::Utc>>
|
||||
}
|
||||
|
||||
pub(crate) async fn root_disk_threshold_for_erasure_disk() -> Option<u64> {
|
||||
if *GLOBAL_IS_ERASURE_SD.read().await {
|
||||
if is_erasure_sd().await {
|
||||
None
|
||||
} else {
|
||||
Some(*GLOBAL_ROOT_DISK_THRESHOLD.read().await)
|
||||
@@ -288,10 +288,6 @@ pub(crate) fn ensure_deployment_id(deployment_id: Uuid) {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn global_lock_manager() -> Arc<rustfs_lock::GlobalLockManager> {
|
||||
rustfs_lock::get_global_lock_manager()
|
||||
}
|
||||
|
||||
pub fn global_lock_client() -> Option<Arc<dyn LockClient>> {
|
||||
get_global_lock_client()
|
||||
}
|
||||
|
||||
@@ -2421,6 +2421,7 @@ async fn test_init_and_start_rebalance_rejects_second_start_after_gate() {
|
||||
decommission_cancelers: tokio::sync::RwLock::new(Vec::new()),
|
||||
start_gate: tokio::sync::Mutex::new(()),
|
||||
pool_meta_save_gate: tokio::sync::Mutex::new(()),
|
||||
ctx: crate::runtime::instance::bootstrap_ctx(),
|
||||
});
|
||||
|
||||
let err = store
|
||||
|
||||
@@ -81,6 +81,7 @@ use crate::error::{GenericError, ObjectApiError, is_err_object_not_found};
|
||||
use crate::io_support::bitrot::{create_bitrot_reader, create_bitrot_reader_from_bytes, create_bitrot_writer};
|
||||
use crate::object_api::ObjectOptions;
|
||||
use crate::object_api::get_object_body_cache_hook;
|
||||
use crate::runtime::instance::{InstanceContext, bootstrap_ctx};
|
||||
use crate::runtime::sources as runtime_sources;
|
||||
use crate::services::batch_processor::AsyncBatchProcessor;
|
||||
use crate::storage_api_contracts::{
|
||||
@@ -1571,6 +1572,12 @@ pub struct SetDisks {
|
||||
get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
|
||||
pub lockers: Vec<Arc<dyn LockClient>>,
|
||||
local_lock_manager: Arc<rustfs_lock::GlobalLockManager>,
|
||||
/// Per-instance runtime context (Phase 5, backlog#939).
|
||||
///
|
||||
/// The leaf of the object-graph ctx plumbing (ECStore → Sets → SetDisks).
|
||||
/// Slice 3 sources `local_lock_manager` from this context to give each
|
||||
/// instance its own lock namespace.
|
||||
ctx: Arc<InstanceContext>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
|
||||
@@ -1716,6 +1723,10 @@ impl SetDisks {
|
||||
format: FormatV3,
|
||||
lockers: Vec<Arc<dyn LockClient>>,
|
||||
) -> Arc<Self> {
|
||||
// Single-instance sources the process bootstrap context (the one the
|
||||
// owning ECStore adopts). Slice 8 threads a per-instance context in for
|
||||
// true multi-instance.
|
||||
let ctx = bootstrap_ctx();
|
||||
Arc::new(SetDisks {
|
||||
locker_owner,
|
||||
disks,
|
||||
@@ -1731,10 +1742,26 @@ impl SetDisks {
|
||||
.time_to_live(GET_OBJECT_METADATA_CACHE_TTL)
|
||||
.build(),
|
||||
lockers,
|
||||
local_lock_manager: runtime_sources::global_lock_manager(),
|
||||
// Sourced from the instance context so each instance owns its lock
|
||||
// namespace (Phase 5 Slice 3). Single-instance: ctx aliases the
|
||||
// process lock-manager singleton, so this is unchanged.
|
||||
local_lock_manager: ctx.lock_manager(),
|
||||
ctx,
|
||||
})
|
||||
}
|
||||
|
||||
/// This set's per-instance runtime context (Phase 5, backlog#939).
|
||||
#[allow(dead_code)] // Read by tests; consumed by later slices.
|
||||
pub(crate) fn instance_ctx(&self) -> &Arc<InstanceContext> {
|
||||
&self.ctx
|
||||
}
|
||||
|
||||
/// The lock manager this set actually uses (test-only; Phase 5 Slice 3).
|
||||
#[cfg(test)]
|
||||
pub(crate) fn local_lock_manager_for_test(&self) -> &Arc<rustfs_lock::GlobalLockManager> {
|
||||
&self.local_lock_manager
|
||||
}
|
||||
|
||||
// async fn cached_disk_health(&self, index: usize) -> Option<bool> {
|
||||
// let cache = self.disk_health_cache.read().await;
|
||||
// cache
|
||||
|
||||
@@ -334,6 +334,9 @@ impl ECStore {
|
||||
decommission_cancelers,
|
||||
start_gate: Mutex::new(()),
|
||||
pool_meta_save_gate: Mutex::new(()),
|
||||
// Adopt the process bootstrap context so startup writes (erasure
|
||||
// type recorded before this point) and later reads share one cell.
|
||||
ctx: crate::runtime::instance::bootstrap_ctx(),
|
||||
});
|
||||
|
||||
// Only set it when the global deployment ID is not yet configured
|
||||
|
||||
@@ -44,6 +44,7 @@ use crate::error::{
|
||||
is_err_read_quorum, is_err_version_not_found, to_object_err,
|
||||
};
|
||||
use crate::runtime::global::DISK_RESERVE_FRACTION;
|
||||
use crate::runtime::instance::InstanceContext;
|
||||
use crate::runtime::sources as runtime_sources;
|
||||
use crate::services::rebalance::RebalanceMeta;
|
||||
use crate::storage_api_contracts::{
|
||||
@@ -187,6 +188,14 @@ pub struct ECStore {
|
||||
/// The saver then clones the latest `pool_meta` under a short read lock and
|
||||
/// releases it before awaiting disk writes.
|
||||
pub(crate) pool_meta_save_gate: Mutex<()>,
|
||||
/// Per-instance runtime state (Phase 5, backlog#939).
|
||||
///
|
||||
/// Carries this instance's identity/runtime out of the process globals so
|
||||
/// multiple instances can coexist without cross-contamination. `new`
|
||||
/// adopts the process bootstrap context (never mints a fresh one) so that
|
||||
/// startup writes and post-construction reads share one cell — single
|
||||
/// instance behavior is unchanged.
|
||||
pub(crate) ctx: Arc<InstanceContext>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for ECStore {
|
||||
@@ -285,6 +294,29 @@ impl ECStore {
|
||||
}
|
||||
}
|
||||
|
||||
/// Phase 5: Per-instance erasure setup accessors (backlog#939)
|
||||
///
|
||||
/// These read this instance's own [`InstanceContext`] rather than a process
|
||||
/// global, so two instances carrying different contexts stay isolated. The
|
||||
/// legacy free-function facade (`runtime::global::is_erasure` etc.) forwards to
|
||||
/// the current instance's context, preserving single-instance behavior.
|
||||
impl ECStore {
|
||||
/// Whether this instance uses erasure coding (single-node or distributed).
|
||||
pub async fn setup_is_erasure(&self) -> bool {
|
||||
self.ctx.is_erasure().await
|
||||
}
|
||||
|
||||
/// Whether this instance uses distributed erasure coding.
|
||||
pub async fn setup_is_dist_erasure(&self) -> bool {
|
||||
self.ctx.is_dist_erasure().await
|
||||
}
|
||||
|
||||
/// Whether this instance uses single-drive erasure coding.
|
||||
pub async fn setup_is_erasure_sd(&self) -> bool {
|
||||
self.ctx.is_erasure_sd().await
|
||||
}
|
||||
}
|
||||
|
||||
// impl Clone for ECStore {
|
||||
// fn clone(&self) -> Self {
|
||||
// let pool_meta = match self.pool_meta.read() {
|
||||
@@ -783,7 +815,7 @@ impl crate::storage_api_contracts::admin::StorageAdminApi for ECStore {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::layout::endpoints::{Endpoints, PoolEndpoints};
|
||||
use crate::layout::endpoints::{Endpoints, PoolEndpoints, SetupType};
|
||||
use crate::runtime::global::reset_local_disk_test_state;
|
||||
use crate::runtime::sources::{clear_local_disk_id_map_for_test, local_disk_path_by_id};
|
||||
use crate::store::init_format::{connect_load_init_formats, init_disks};
|
||||
@@ -800,6 +832,60 @@ mod tests {
|
||||
assert!(infos.iter().all(|info| info.is_none()));
|
||||
}
|
||||
|
||||
// Build a minimal ECStore carrying an explicit instance context. Empty
|
||||
// pools/disks are sufficient: the Phase 5 accessors read only `self.ctx`.
|
||||
fn build_store_with_ctx(ctx: Arc<InstanceContext>) -> ECStore {
|
||||
let endpoint_pools = EndpointServerPools::default();
|
||||
ECStore {
|
||||
id: uuid::Uuid::new_v4(),
|
||||
disk_map: std::collections::HashMap::new(),
|
||||
pools: Vec::new(),
|
||||
peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools),
|
||||
pool_meta: RwLock::new(PoolMeta::default()),
|
||||
rebalance_meta: RwLock::new(None),
|
||||
decommission_cancelers: RwLock::new(Vec::new()),
|
||||
start_gate: Mutex::new(()),
|
||||
pool_meta_save_gate: Mutex::new(()),
|
||||
ctx,
|
||||
}
|
||||
}
|
||||
|
||||
// The object graph is the isolation carrier: two ECStore instances holding
|
||||
// distinct contexts report independent erasure state through their real
|
||||
// `&self` accessors — no cross-contamination.
|
||||
#[tokio::test]
|
||||
async fn instance_context_carrier_isolates_two_stores() {
|
||||
let ctx_a = Arc::new(InstanceContext::new());
|
||||
let ctx_b = Arc::new(InstanceContext::new());
|
||||
ctx_a.update_erasure_type(SetupType::DistErasure).await;
|
||||
ctx_b.update_erasure_type(SetupType::ErasureSD).await;
|
||||
|
||||
let store_a = build_store_with_ctx(ctx_a);
|
||||
let store_b = build_store_with_ctx(ctx_b);
|
||||
|
||||
// store_a: distributed erasure (implies is_erasure), not single-drive.
|
||||
assert!(store_a.setup_is_erasure().await);
|
||||
assert!(store_a.setup_is_dist_erasure().await);
|
||||
assert!(!store_a.setup_is_erasure_sd().await);
|
||||
|
||||
// store_b: single-drive erasure only.
|
||||
assert!(store_b.setup_is_erasure_sd().await);
|
||||
assert!(!store_b.setup_is_erasure().await);
|
||||
assert!(!store_b.setup_is_dist_erasure().await);
|
||||
}
|
||||
|
||||
// The production/test constructors ADOPT the process bootstrap context
|
||||
// (same Arc), so a startup write recorded before the store existed is
|
||||
// visible through the store afterward — single-instance behavior preserved.
|
||||
#[tokio::test]
|
||||
async fn store_adopts_bootstrap_context() {
|
||||
let store = build_store_with_ctx(crate::runtime::instance::bootstrap_ctx());
|
||||
assert!(
|
||||
Arc::ptr_eq(&store.ctx, &crate::runtime::instance::bootstrap_ctx()),
|
||||
"store built via adoption must share the bootstrap context Arc"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_has_space_for() {
|
||||
let disk_infos = vec![None, None]; // No actual disk info
|
||||
|
||||
@@ -1967,9 +1967,49 @@ mod tests {
|
||||
decommission_cancelers: RwLock::new(Vec::new()),
|
||||
start_gate: Mutex::new(()),
|
||||
pool_meta_save_gate: Mutex::new(()),
|
||||
ctx: crate::runtime::instance::bootstrap_ctx(),
|
||||
}
|
||||
}
|
||||
|
||||
// Phase 5 Slice 2 (backlog#939): the instance context flows down the whole
|
||||
// object graph — ECStore, its Sets, and their SetDisks must all carry the
|
||||
// same `Arc<InstanceContext>` in a single-instance deployment.
|
||||
#[tokio::test]
|
||||
async fn instance_context_flows_through_object_graph() {
|
||||
let store = new_read_lock_test_store().await;
|
||||
|
||||
let sets = store.pools.first().expect("test store has one pool");
|
||||
assert!(
|
||||
std::sync::Arc::ptr_eq(&store.ctx, sets.instance_ctx()),
|
||||
"Sets must carry the store's instance context"
|
||||
);
|
||||
|
||||
let set_disks = sets.disk_set.first().expect("pool has one set");
|
||||
assert!(
|
||||
std::sync::Arc::ptr_eq(sets.instance_ctx(), set_disks.instance_ctx()),
|
||||
"SetDisks must carry the Sets' instance context"
|
||||
);
|
||||
}
|
||||
|
||||
// Phase 5 Slice 3 (backlog#939): a SetDisks sources its lock manager from
|
||||
// its instance context (not an independent process lookup), and in a
|
||||
// single-instance build that context aliases the process lock-manager
|
||||
// singleton — so the lock namespace is unchanged.
|
||||
#[tokio::test]
|
||||
async fn set_disks_lock_manager_comes_from_instance_context() {
|
||||
let store = new_read_lock_test_store().await;
|
||||
let set_disks = store.pools[0].disk_set.first().expect("pool has one set");
|
||||
|
||||
assert!(
|
||||
std::sync::Arc::ptr_eq(set_disks.local_lock_manager_for_test(), &set_disks.instance_ctx().lock_manager()),
|
||||
"SetDisks lock manager must be sourced from its instance context"
|
||||
);
|
||||
assert!(
|
||||
std::sync::Arc::ptr_eq(set_disks.local_lock_manager_for_test(), &rustfs_lock::get_global_lock_manager()),
|
||||
"single-instance lock manager must alias the process singleton"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn acquired_read_lock_marks_metadata_cache_safe_for_set_layer() {
|
||||
let store = new_read_lock_test_store().await;
|
||||
|
||||
Reference in New Issue
Block a user