mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-25 21:46:50 +00:00
fix(ecstore): guard transition worker max (#3003)
* fix(ecstore): guard transition worker max * test: clarify env mutation safety * test(ecstore): document env mutation concurrency contract
This commit is contained in:
@@ -112,11 +112,20 @@ fn resolve_transition_worker_count() -> (i64, i64, i64) {
|
|||||||
.filter(|value| *value > 0)
|
.filter(|value| *value > 0)
|
||||||
.unwrap_or(fallback);
|
.unwrap_or(fallback);
|
||||||
let mut effective = configured;
|
let mut effective = configured;
|
||||||
let absolute_max = get_env_i64(ENV_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX);
|
let absolute_max = resolve_transition_workers_absolute_max();
|
||||||
effective = std::cmp::min(effective, absolute_max);
|
effective = std::cmp::min(effective, absolute_max);
|
||||||
(configured, absolute_max, effective)
|
(configured, absolute_max, effective)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn resolve_transition_workers_absolute_max() -> i64 {
|
||||||
|
let absolute_max = get_env_i64(ENV_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX);
|
||||||
|
if absolute_max > 0 {
|
||||||
|
absolute_max
|
||||||
|
} else {
|
||||||
|
DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn resolve_transition_queue_capacity() -> usize {
|
fn resolve_transition_queue_capacity() -> usize {
|
||||||
get_env_usize(ENV_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_CAPACITY).max(1)
|
get_env_usize(ENV_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_CAPACITY).max(1)
|
||||||
}
|
}
|
||||||
@@ -942,7 +951,7 @@ impl TransitionState {
|
|||||||
n = effective;
|
n = effective;
|
||||||
}
|
}
|
||||||
// Allow environment override of maximum workers
|
// Allow environment override of maximum workers
|
||||||
let absolute_max = get_env_i64(ENV_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX);
|
let absolute_max = resolve_transition_workers_absolute_max();
|
||||||
n = std::cmp::min(n, absolute_max);
|
n = std::cmp::min(n, absolute_max);
|
||||||
|
|
||||||
let previous_num_workers = GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst);
|
let previous_num_workers = GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst);
|
||||||
@@ -2294,6 +2303,9 @@ mod tests {
|
|||||||
assert!(opts.skip_decommissioned);
|
assert!(opts.skip_decommissioned);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SAFETY: this helper is only used from `#[serial]` tests and those tests run under a
|
||||||
|
// single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access
|
||||||
|
// process environment while `env::set_var`/`env::remove_var` is active.
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
fn with_transition_worker_env<F>(transition: Option<&str>, absolute: Option<&str>, test_fn: F)
|
fn with_transition_worker_env<F>(transition: Option<&str>, absolute: Option<&str>, test_fn: F)
|
||||||
where
|
where
|
||||||
@@ -2343,6 +2355,9 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SAFETY: this helper is only used from `#[serial]` tests and those tests run under a
|
||||||
|
// single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access
|
||||||
|
// process environment while `env::set_var`/`env::remove_var` is active.
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
async fn with_transition_worker_env_async<F, Fut>(transition: Option<&str>, absolute: Option<&str>, test_fn: F)
|
async fn with_transition_worker_env_async<F, Fut>(transition: Option<&str>, absolute: Option<&str>, test_fn: F)
|
||||||
where
|
where
|
||||||
@@ -2393,6 +2408,9 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SAFETY: this helper is only used from `#[serial]` tests and those tests run under a
|
||||||
|
// single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access
|
||||||
|
// process environment while `env::set_var`/`env::remove_var` is active.
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
fn with_transition_queue_env<F>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
|
fn with_transition_queue_env<F>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
|
||||||
where
|
where
|
||||||
@@ -2442,6 +2460,9 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SAFETY: this helper is only used from `#[serial]` tests and those tests run under a
|
||||||
|
// single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access
|
||||||
|
// process environment while `env::set_var`/`env::remove_var` is active.
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
|
async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
|
||||||
where
|
where
|
||||||
@@ -2554,6 +2575,26 @@ mod tests {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
#[serial]
|
||||||
|
fn resolve_transition_worker_count_ignores_non_positive_absolute_max() {
|
||||||
|
with_transition_worker_env(Some("4"), Some("0"), || {
|
||||||
|
let (configured, absolute_max, effective) = resolve_transition_worker_count();
|
||||||
|
|
||||||
|
assert_eq!(configured, 4);
|
||||||
|
assert_eq!(absolute_max, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX);
|
||||||
|
assert_eq!(effective, 4);
|
||||||
|
});
|
||||||
|
|
||||||
|
with_transition_worker_env(Some("4"), Some("-1"), || {
|
||||||
|
let (configured, absolute_max, effective) = resolve_transition_worker_count();
|
||||||
|
|
||||||
|
assert_eq!(configured, 4);
|
||||||
|
assert_eq!(absolute_max, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX);
|
||||||
|
assert_eq!(effective, 4);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
#[serial]
|
#[serial]
|
||||||
fn resolve_transition_worker_count_falls_back_for_zero_value() {
|
fn resolve_transition_worker_count_falls_back_for_zero_value() {
|
||||||
|
|||||||
@@ -807,6 +807,9 @@ async fn wait_for_transition(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
|
||||||
|
// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
|
||||||
|
// `env::set_var` / `env::remove_var` window.
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
async fn with_forced_immediate_enqueue_timeout<F, Fut>(test_fn: F)
|
async fn with_forced_immediate_enqueue_timeout<F, Fut>(test_fn: F)
|
||||||
where
|
where
|
||||||
|
|||||||
@@ -327,6 +327,9 @@ async fn wait_for_transition(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
|
||||||
|
// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
|
||||||
|
// `env::set_var` / `env::remove_var` window.
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
async fn with_forced_immediate_enqueue_timeout<F, Fut>(test_fn: F)
|
async fn with_forced_immediate_enqueue_timeout<F, Fut>(test_fn: F)
|
||||||
where
|
where
|
||||||
|
|||||||
Reference in New Issue
Block a user