From beb807ae2bc7af6cf669d3dab8584242ae39420b Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 24 Jul 2026 19:54:14 +0800 Subject: [PATCH] perf(protos): cache internode msgpack-only flag (#5191) Cache the internode msgpack-only env gate after the first read so metadata RPC send paths avoid repeated environment parsing. Keep a reset hook for tests that intentionally change the env in-process. Co-authored-by: heihutu --- Cargo.lock | 1 + crates/ecstore/src/cluster/rpc/remote_disk.rs | 29 +++++++-- crates/protos/Cargo.toml | 1 + crates/protos/src/lib.rs | 59 ++++++++++++++++++- rustfs/src/storage/rpc/node_service/disk.rs | 13 +++- 5 files changed, 92 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a73b2b96f..ac996ae88 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9748,6 +9748,7 @@ dependencies = [ "rustfs-utils", "serde", "serde_json", + "temp-env", "tokio", "tonic", "tonic-prost", diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index cd8b44f7c..9f91c7e20 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -3018,15 +3018,32 @@ mod tests { #[test] fn compat_json_dual_writes_by_default() { - let resp = sample_read_multiple_resp("file", b"data"); - let json = compat_json(&resp).expect("compat_json should encode"); - assert!(!json.is_empty()); - assert_eq!(json, serde_json::to_string(&resp).expect("json should encode")); + with_internode_msgpack_env( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, None::<&str>), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>), + ], + || { + let resp = sample_read_multiple_resp("file", b"data"); + let json = compat_json(&resp).expect("compat_json should encode"); + assert!(!json.is_empty()); + assert_eq!(json, serde_json::to_string(&resp).expect("json should encode")); + }, + ); + } + + fn with_internode_msgpack_env(vars: [(&'static str, Option<&'static str>); 2], f: impl FnOnce() -> R) -> R { + temp_env::with_vars(vars, || { + rustfs_protos::reset_internode_rpc_msgpack_only_cache(); + let result = f(); + rustfs_protos::reset_internode_rpc_msgpack_only_cache(); + result + }) } #[test] fn compat_json_keeps_json_when_msgpack_only_lacks_fleet_confirmation() { - temp_env::with_vars( + with_internode_msgpack_env( [ (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>), @@ -3043,7 +3060,7 @@ mod tests { #[test] fn compat_json_omits_json_only_after_fleet_confirmation() { - temp_env::with_vars( + with_internode_msgpack_env( [ (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), diff --git a/crates/protos/Cargo.toml b/crates/protos/Cargo.toml index 5c568d0c5..3f57bbc6b 100644 --- a/crates/protos/Cargo.toml +++ b/crates/protos/Cargo.toml @@ -54,3 +54,4 @@ doctest = false [dev-dependencies] serde_json = { workspace = true } +temp-env = { workspace = true } diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index d10dd066c..5c88c7045 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -25,7 +25,7 @@ use std::{ collections::HashMap, error::Error, sync::LazyLock, - sync::atomic::{AtomicUsize, Ordering}, + sync::atomic::{AtomicU8, AtomicUsize, Ordering}, time::{Duration, Instant}, }; use tokio::sync::Mutex; @@ -54,7 +54,11 @@ pub const DEFAULT_GRPC_SERVER_MESSAGE_LEN: usize = 100 * 1024 * 1024; /// Default value: https:// const RUSTFS_HTTPS_PREFIX: &str = "https://"; const TLS_GENERATION_CACHE_MAX_SIZE: usize = 512; +const INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET: u8 = 0; +const INTERNODE_RPC_MSGPACK_ONLY_CACHE_FALSE: u8 = 1; +const INTERNODE_RPC_MSGPACK_ONLY_CACHE_TRUE: u8 = 2; static TLS_GENERATION_CACHE: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +static INTERNODE_RPC_MSGPACK_ONLY_CACHE: AtomicU8 = AtomicU8::new(INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET); fn enforce_tls_generation_cache_bound(generation_cache: &mut HashMap, generation: u64, addr: &str) { if generation_cache.len() < TLS_GENERATION_CACHE_MAX_SIZE || generation_cache.contains_key(addr) { @@ -642,13 +646,33 @@ mod tier_mutation_rpc_tests { /// decode the compatibility field. Operators must also set the fleet-confirmed guard after the /// convergence runbook proves every peer supports `_bin` and rollback. pub fn internode_rpc_msgpack_only() -> bool { - rustfs_utils::get_env_bool( + match INTERNODE_RPC_MSGPACK_ONLY_CACHE.load(Ordering::Acquire) { + INTERNODE_RPC_MSGPACK_ONLY_CACHE_FALSE => return false, + INTERNODE_RPC_MSGPACK_ONLY_CACHE_TRUE => return true, + _ => {} + } + + let enabled = rustfs_utils::get_env_bool( rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, rustfs_config::DEFAULT_INTERNODE_RPC_MSGPACK_ONLY, ) && rustfs_utils::get_env_bool( rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, rustfs_config::DEFAULT_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, - ) + ); + INTERNODE_RPC_MSGPACK_ONLY_CACHE.store( + if enabled { + INTERNODE_RPC_MSGPACK_ONLY_CACHE_TRUE + } else { + INTERNODE_RPC_MSGPACK_ONLY_CACHE_FALSE + }, + Ordering::Release, + ); + enabled +} + +#[doc(hidden)] +pub fn reset_internode_rpc_msgpack_only_cache() { + INTERNODE_RPC_MSGPACK_ONLY_CACHE.store(INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET, Ordering::Release); } /// Consecutive-failure threshold after which an internode peer is marked offline (grpc-optimization @@ -951,6 +975,35 @@ mod tests { assert!(bulk_channel_pool_size() >= 1); } + #[test] + fn internode_rpc_msgpack_only_reuses_cached_env_until_reset() { + reset_internode_rpc_msgpack_only_cache(); + + temp_env::with_vars( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")), + ], + || { + assert!(internode_rpc_msgpack_only()); + }, + ); + + temp_env::with_vars( + [ + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, None::<&str>), + (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>), + ], + || { + assert!(internode_rpc_msgpack_only(), "cached value should avoid repeated env reads"); + reset_internode_rpc_msgpack_only_cache(); + assert!(!internode_rpc_msgpack_only(), "reset must reload current env values"); + }, + ); + + reset_internode_rpc_msgpack_only_cache(); + } + #[tokio::test] async fn get_channel_for_class_bulk_reuses_control_cache_when_isolation_disabled() { // Isolation defaults off, so a Bulk request must reuse the control channel keyed by the diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 10292c9c3..e2bc29a8f 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -1215,9 +1215,18 @@ mod tests { ); } + fn with_internode_msgpack_env(vars: [(&'static str, Option<&'static str>); 2], f: impl FnOnce() -> R) -> R { + temp_env::with_vars(vars, || { + rustfs_protos::reset_internode_rpc_msgpack_only_cache(); + let result = f(); + rustfs_protos::reset_internode_rpc_msgpack_only_cache(); + result + }) + } + #[test] fn compat_response_json_keeps_json_when_msgpack_only_lacks_fleet_confirmation() { - temp_env::with_vars( + with_internode_msgpack_env( [ (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, None::<&str>), @@ -1237,7 +1246,7 @@ mod tests { #[test] fn compat_response_json_omits_json_only_after_fleet_confirmation() { - temp_env::with_vars( + with_internode_msgpack_env( [ (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, Some("true")), (rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY_FLEET_CONFIRMED, Some("true")),