gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-29 08:27:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: replace jemalloc with mimalloc (#4174)

Browse Source 
* feat: replace jemalloc with mimalloc

* docs: record allocator rounds5 retest

* fix(replication): satisfy clippy unwrap lints

* docs: keep allocator migration plan local only

* feat(profiling): rely on pyroscope cpu profiling

* refactor(profiling): centralize unsupported pprof responses

* chore(deps): update s3s revision
This commit is contained in:
houseme
2026-07-02 19:03:38 +08:00
committed by GitHub
parent 54496f2796
commit 70e1d79dfd
17 changed files with 217 additions and 1529 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/Cargo.toml
+2 -11
View File
@@ -186,25 +186,16 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true }
# Data structures
hashbrown = { workspace = true }
mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
# io-uring is Linux-only. Scope the feature to Linux targets.
tokio = { workspace = true, features = ["io-uring"] }
[target.'cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))'.dependencies]
mimalloc = { workspace = true }
[target.'cfg(not(target_os = "windows"))'.dependencies]
libmimalloc-sys = { version = "0.1.49", features = ["extended"] }
[target.'cfg(any(target_os = "linux", target_os = "macos"))'.dependencies]
pprof = { workspace = true }
# jemalloc-based allocator and memory profiling are only enabled on validated linux-gnu-x86_64 targets.
[target.'cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))'.dependencies]
tikv-jemallocator = { workspace = true }
tikv-jemalloc-ctl = { workspace = true }
jemalloc_pprof = { workspace = true }
[dev-dependencies]
uuid = { workspace = true, features = ["v4"] }
serial_test = { workspace = true }
rustfs/src/admin/handlers/profile.rs
+8 -47
View File
@@ -15,11 +15,7 @@
use crate::admin::{auth::validate_admin_request, router::Operation};
use crate::auth::{check_key_valid, get_session_token};
use crate::server::RemoteAddr;
#[cfg(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
use http::HeaderMap;
use http::StatusCode;
#[cfg(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
use http::header::CONTENT_TYPE;
use matchit::Params;
use rustfs_policy::policy::action::{Action, AdminAction};
use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
@@ -45,6 +41,10 @@ pub(super) async fn authorize_profile_request(req: &S3Request<Body>) -> S3Result
.await
}
pub(super) fn profile_not_implemented_response(message: String) -> S3Response<(StatusCode, Body)> {
S3Response::new((StatusCode::NOT_IMPLEMENTED, Body::from(message)))
}
pub struct TriggerProfileCPU {}
#[async_trait::async_trait]
impl Operation for TriggerProfileCPU {
@@ -52,30 +52,8 @@ impl Operation for TriggerProfileCPU {
authorize_profile_request(&req).await?;
info!("Triggering CPU profile dump via S3 request...");
#[cfg(not(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))))]
{
return Ok(S3Response::new((
StatusCode::NOT_IMPLEMENTED,
Body::from(
crate::profiling::dump_cpu_pprof_for(std::time::Duration::from_secs(0))
.await
.unwrap_err(),
),
)));
}
#[cfg(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
{
let dur = std::time::Duration::from_secs(60);
match crate::profiling::dump_cpu_pprof_for(dur).await {
Ok(path) => {
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/html".parse().expect("operation should succeed"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(path.display().to_string())), header))
}
Err(e) => Err(s3s::s3_error!(InternalError, "{}", format!("Failed to dump CPU profile: {e}"))),
}
}
crate::profiling::log_cpu_pprof_dump_skipped();
Ok(profile_not_implemented_response(crate::profiling::local_cpu_pprof_unsupported_message()))
}
}
@@ -86,25 +64,8 @@ impl Operation for TriggerProfileMemory {
authorize_profile_request(&req).await?;
info!("Triggering Memory profile dump via S3 request...");
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
{
return Ok(S3Response::new((
StatusCode::NOT_IMPLEMENTED,
Body::from(crate::profiling::dump_memory_pprof_now().await.unwrap_err()),
)));
}
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
{
match crate::profiling::dump_memory_pprof_now().await {
Ok(path) => {
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/html".parse().expect("operation should succeed"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(path.display().to_string())), header))
}
Err(e) => Err(s3s::s3_error!(InternalError, "{}", format!("Failed to dump Memory profile: {e}"))),
}
}
crate::profiling::log_memory_pprof_dump_skipped();
Ok(profile_not_implemented_response(crate::profiling::memory_pprof_unsupported_message()))
}
}
rustfs/src/admin/handlers/profile_admin.rs
+21 -169
View File
@@ -12,28 +12,23 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::profile::authorize_profile_request;
use super::profile::{authorize_profile_request, profile_not_implemented_response};
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::server::ADMIN_PREFIX;
use http::{HeaderMap, HeaderValue, Uri};
use http::{HeaderMap, HeaderValue};
use hyper::{Method, StatusCode};
use matchit::Params;
use s3s::header::CONTENT_TYPE;
use s3s::{Body, S3Request, S3Response, S3Result};
use std::collections::HashMap;
use serde::Serialize;
use tracing::error;
#[allow(dead_code)]
fn extract_query_params(uri: &Uri) -> HashMap<String, String> {
let mut params = HashMap::new();
if let Some(query) = uri.query() {
for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
params.insert(key.into_owned(), value.into_owned());
}
}
params
#[derive(Serialize)]
struct ProfileStatus {
enabled: &'static str,
status: &'static str,
platform: &'static str,
message: &'static str,
}
pub fn register_profiling_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
@@ -54,114 +49,17 @@ pub fn register_profiling_route(r: &mut S3Router<AdminOperation>) -> std::io::Re
pub struct ProfileHandler {}
#[allow(dead_code)]
fn map_cpu_profile_collect_error_message(err: &str) -> (StatusCode, String) {
if err.contains("start running cpu profiler error") {
return (
StatusCode::CONFLICT,
"CPU profiler is already running. Disable RUSTFS_OBS_PROFILING_EXPORT_ENABLED or retry later.".to_string(),
);
}
(StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to collect CPU profile: {err}"))
}
#[async_trait::async_trait]
impl Operation for ProfileHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
authorize_profile_request(&req).await?;
#[cfg(not(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))))]
{
let requested_url = req.uri.to_string();
let target_os = std::env::consts::OS;
let target_arch = std::env::consts::ARCH;
let target_env = option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown");
let msg = format!(
"CPU profiling is not supported on this platform. target_os={target_os}, target_env={target_env}, target_arch={target_arch}, requested_url={requested_url}"
);
return Ok(S3Response::new((StatusCode::NOT_IMPLEMENTED, Body::from(msg))));
}
#[cfg(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
{
use rustfs_config::{DEFAULT_CPU_FREQ, ENV_CPU_FREQ};
use rustfs_utils::get_env_usize;
let queries = extract_query_params(&req.uri);
let seconds = queries.get("seconds").and_then(|s| s.parse::<u64>().ok()).unwrap_or(30);
let format = queries.get("format").cloned().unwrap_or_else(|| "protobuf".to_string());
if seconds > 300 {
return Ok(S3Response::new((
StatusCode::BAD_REQUEST,
Body::from("Profile duration cannot exceed 300 seconds".to_string()),
)));
}
match format.as_str() {
"protobuf" | "pb" => match crate::profiling::dump_cpu_pprof_for(std::time::Duration::from_secs(seconds)).await {
Ok(path) => match tokio::fs::read(&path).await {
Ok(bytes) => {
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/octet-stream"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(bytes)), headers))
}
Err(e) => {
error!("Failed to read profile file {}: {}", path.display(), e);
Ok(S3Response::new((
StatusCode::INTERNAL_SERVER_ERROR,
Body::from(format!("Failed to read profile file: {e}")),
)))
}
},
Err(e) => {
let (status, message) = map_cpu_profile_collect_error_message(&e);
error!("CPU protobuf profile collection failed: {}", e);
Ok(S3Response::new((status, Body::from(message))))
}
},
"flamegraph" | "svg" => {
let freq = get_env_usize(ENV_CPU_FREQ, DEFAULT_CPU_FREQ) as i32;
let guard = match pprof::ProfilerGuard::new(freq) {
Ok(g) => g,
Err(e) => {
return Ok(S3Response::new((
StatusCode::INTERNAL_SERVER_ERROR,
Body::from(format!("Failed to create profiler: {e}")),
)));
}
};
tokio::time::sleep(std::time::Duration::from_secs(seconds)).await;
let report = match guard.report().build() {
Ok(r) => r,
Err(e) => {
return Ok(S3Response::new((
StatusCode::INTERNAL_SERVER_ERROR,
Body::from(format!("Failed to build profile report: {e}")),
)));
}
};
let mut flamegraph_buf = Vec::new();
if let Err(e) = report.flamegraph(&mut flamegraph_buf) {
return Ok(S3Response::new((
StatusCode::INTERNAL_SERVER_ERROR,
Body::from(format!("Failed to generate flamegraph: {e}")),
)));
}
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, HeaderValue::from_static("image/svg+xml"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(flamegraph_buf)), headers))
}
_ => Ok(S3Response::new((
StatusCode::BAD_REQUEST,
Body::from("Unsupported format. Use 'protobuf' or 'flamegraph'".to_string()),
))),
}
}
let requested_url = req.uri.to_string();
crate::profiling::log_cpu_pprof_dump_skipped();
Ok(profile_not_implemented_response(format!(
"{}; requested_url={requested_url}",
crate::profiling::local_cpu_pprof_unsupported_message()
)))
}
}
@@ -172,37 +70,11 @@ impl Operation for ProfileStatusHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
authorize_profile_request(&req).await?;
#[cfg(not(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))))]
let message = format!("CPU profiling is not supported on {} platform", std::env::consts::OS);
#[cfg(not(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))))]
let status = HashMap::from([
("enabled", "false"),
("status", "not_supported"),
("platform", std::env::consts::OS),
("message", message.as_str()),
]);
#[cfg(any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
let status = {
use rustfs_config::{DEFAULT_ENABLE_PROFILING, ENV_ENABLE_PROFILING};
use rustfs_utils::get_env_bool;
let enabled = get_env_bool(ENV_ENABLE_PROFILING, DEFAULT_ENABLE_PROFILING);
if enabled {
HashMap::from([
("enabled", "true"),
("status", "running"),
("supported_formats", "protobuf, flamegraph"),
("max_duration_seconds", "300"),
("endpoint", "/rustfs/admin/debug/pprof/profile"),
])
} else {
HashMap::from([
("enabled", "false"),
("status", "disabled"),
("message", "Set RUSTFS_ENABLE_PROFILING=true to enable profiling"),
])
}
let status = ProfileStatus {
enabled: "false",
status: "not_supported",
platform: std::env::consts::OS,
message: crate::profiling::LOCAL_CPU_PPROF_UNSUPPORTED_SUMMARY,
};
match serde_json::to_string(&status) {
@@ -224,11 +96,10 @@ impl Operation for ProfileStatusHandler {
#[cfg(test)]
mod tests {
use super::{ProfileHandler, ProfileStatusHandler, extract_query_params};
use super::{ProfileHandler, ProfileStatusHandler};
use crate::admin::router::Operation;
use http::{Extensions, HeaderMap, Uri};
use hyper::Method;
use hyper::StatusCode;
use matchit::Params;
use s3s::{Body, S3ErrorCode, S3Request};
@@ -246,17 +117,6 @@ mod tests {
}
}
#[test]
fn test_extract_query_params_decodes_percent_encoded_values() {
let uri: Uri = "/rustfs/admin/debug/pprof/profile?format=flamegraph&note=a%2Bb+value"
.parse()
.expect("uri should parse");
let params = extract_query_params(&uri);
assert_eq!(params.get("format"), Some(&"flamegraph".to_string()));
assert_eq!(params.get("note"), Some(&"a+b value".to_string()));
}
#[tokio::test]
async fn profile_handler_rejects_missing_credentials() {
let result = ProfileHandler {}
@@ -284,12 +144,4 @@ mod tests {
assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
assert_eq!(err.message(), Some("Signature is required"));
}
#[test]
fn cpu_profile_collect_error_maps_profiler_conflict_to_409() {
let (status, message) =
super::map_cpu_profile_collect_error_message("create profiler failed: start running cpu profiler error");
assert_eq!(status, StatusCode::CONFLICT);
assert!(message.contains("CPU profiler is already running"));
}
}
rustfs/src/allocator_reclaim.rs
+5 -30
View File
@@ -111,15 +111,7 @@ impl AllocatorReclaimController {
}
pub fn allocator_backend() -> &'static str {
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
{
"jemalloc"
}
#[cfg(all(
not(target_os = "windows"),
not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))
))]
#[cfg(not(target_os = "windows"))]
{
"mimalloc"
}
@@ -206,7 +198,7 @@ fn configured_allocator_reclaim_interval_secs() -> u64 {
}
fn effective_allocator_reclaim_force(backend: &str, configured_force: bool) -> bool {
configured_force && backend != "jemalloc"
configured_force && backend != "mimalloc-windows"
}
fn build_allocator_reclaim_desired_snapshot(
@@ -302,10 +294,7 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
)
}
#[cfg(all(
not(target_os = "windows"),
not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))
))]
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn collect_allocator_memory(force: bool) -> Result<(), String> {
// SAFETY: `mi_collect` is provided by the active global allocator backend
@@ -317,13 +306,6 @@ fn collect_allocator_memory(force: bool) -> Result<(), String> {
Ok(())
}
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
fn collect_allocator_memory(_force: bool) -> Result<(), String> {
let _ = tikv_jemalloc_ctl::background_thread::write(true);
tikv_jemalloc_ctl::epoch::advance().map_err(|err| err.to_string())?;
Ok(())
}
#[cfg(target_os = "windows")]
fn collect_allocator_memory(_force: bool) -> Result<(), String> {
Err("allocator reclaim is not supported on Windows".to_string())
@@ -368,13 +350,6 @@ pub fn init_allocator_reclaim(ctx: CancellationToken) {
}
let configured_force = configured_allocator_reclaim_force();
if backend == "jemalloc" && configured_force {
warn!(
backend,
env = rustfs_config::ENV_ALLOCATOR_RECLAIM_FORCE,
"allocator reclaim force mode is not supported on jemalloc backend; ignoring configured force flag"
);
}
let force = effective_allocator_reclaim_force(backend, configured_force);
let idle_intervals = configured_allocator_reclaim_idle_intervals();
let interval = Duration::from_secs(configured_allocator_reclaim_interval_secs());
@@ -473,8 +448,8 @@ mod tests {
}
#[test]
fn allocator_reclaim_force_preserves_jemalloc_override() {
assert!(!effective_allocator_reclaim_force("jemalloc", true));
fn allocator_reclaim_force_is_disabled_only_on_windows_backend() {
assert!(!effective_allocator_reclaim_force("mimalloc-windows", true));
assert!(effective_allocator_reclaim_force("mimalloc", true));
assert!(!effective_allocator_reclaim_force("mimalloc", false));
}
rustfs/src/main.rs
-5
View File
@@ -12,11 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
#[global_allocator]
static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
rustfs/src/profiling.rs
+73 -638
View File
@@ -12,647 +12,82 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
mod unsupported_impl {
use std::path::PathBuf;
use std::time::Duration;
use tracing::{debug, info};
use tracing::{debug, info};
const LOG_COMPONENT_PROFILING: &str = "profiling";
const LOG_SUBSYSTEM_PLATFORM: &str = "platform";
const LOG_COMPONENT_PROFILING: &str = "profiling";
const LOG_SUBSYSTEM_CPU: &str = "cpu";
const LOG_SUBSYSTEM_MEMORY: &str = "memory";
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
const LOCAL_CPU_PPROF_UNSUPPORTED_REASON: &str = "local_cpu_pprof_unsupported";
const MEMORY_PPROF_UNSUPPORTED_REASON: &str = "mimalloc_memory_pprof_unsupported";
pub const LOCAL_CPU_PPROF_UNSUPPORTED_SUMMARY: &str = "local CPU pprof dumps are not supported; use Pyroscope export instead";
pub const MEMORY_PPROF_UNSUPPORTED_SUMMARY: &str = "memory pprof dumps are not supported with the mimalloc allocator";
pub async fn init_from_env() {
let target_env = option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown");
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_PLATFORM,
event = "profiling_runtime_skipped",
reason = "unsupported_platform",
target_os = std::env::consts::OS,
target_env,
target_arch = std::env::consts::ARCH,
"Profiling runtime skipped"
);
}
/// Stop all background profiling tasks
pub fn shutdown_profiling() {
let target_env = option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown");
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_PLATFORM,
event = "profiling_shutdown_skipped",
reason = "unsupported_platform",
target_os = std::env::consts::OS,
target_env,
target_arch = std::env::consts::ARCH,
"Profiling shutdown skipped"
);
}
pub async fn dump_cpu_pprof_for(_duration: Duration) -> Result<PathBuf, String> {
Err(unsupported_message("CPU profiling"))
}
pub async fn dump_memory_pprof_now() -> Result<PathBuf, String> {
Err(unsupported_message("Memory profiling"))
}
fn unsupported_message(feature: &str) -> String {
let target_env = option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown");
format!(
"{feature} is only supported on linux x86_64 gnu. target_os={}, target_env={target_env}, target_arch={}",
std::env::consts::OS,
std::env::consts::ARCH
)
}
pub async fn init_from_env() {
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_RUNTIME,
event = "profiling_runtime_skipped",
reason = LOCAL_CPU_PPROF_UNSUPPORTED_REASON,
target_os = std::env::consts::OS,
target_env = target_env(),
target_arch = std::env::consts::ARCH,
"Local pprof profiling runtime skipped"
);
}
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
pub use unsupported_impl::{dump_cpu_pprof_for, dump_memory_pprof_now, init_from_env, shutdown_profiling};
#[cfg(any(target_os = "linux", target_os = "macos"))]
mod linux_impl {
use pprof::protos::Message;
use rustfs_config::{
DEFAULT_CPU_DURATION_SECS, DEFAULT_CPU_FREQ, DEFAULT_CPU_INTERVAL_SECS, DEFAULT_CPU_MODE, DEFAULT_ENABLE_PROFILING,
DEFAULT_MEM_INTERVAL_SECS, DEFAULT_MEM_PERIODIC, DEFAULT_OUTPUT_DIR, ENV_CPU_DURATION_SECS, ENV_CPU_FREQ,
ENV_CPU_INTERVAL_SECS, ENV_CPU_MODE, ENV_ENABLE_PROFILING, ENV_MEM_INTERVAL_SECS, ENV_MEM_PERIODIC, ENV_OUTPUT_DIR,
};
use rustfs_utils::{get_env_bool, get_env_str, get_env_u64, get_env_usize};
use std::fs::{File, create_dir_all};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use tokio::sync::Mutex;
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};
const LOG_COMPONENT_PROFILING: &str = "profiling";
const LOG_SUBSYSTEM_CPU: &str = "cpu";
const LOG_SUBSYSTEM_MEMORY: &str = "memory";
const LOG_SUBSYSTEM_JEMALLOC: &str = "jemalloc";
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
static CPU_CONT_GUARD: OnceLock<Arc<Mutex<Option<pprof::ProfilerGuard<'static>>>>> = OnceLock::new();
static PROFILING_CANCEL_TOKEN: OnceLock<CancellationToken> = OnceLock::new();
/// CPU profiling mode
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum CpuMode {
Off,
Continuous,
Periodic,
}
/// Get or create output directory
fn output_dir() -> PathBuf {
let dir = get_env_str(ENV_OUTPUT_DIR, DEFAULT_OUTPUT_DIR);
let p = PathBuf::from(dir);
if let Err(e) = create_dir_all(&p) {
warn!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_RUNTIME,
event = "profiling_output_dir_fallback",
path = %p.display(),
error = %e,
fallback = ".",
"Profiling output directory fallback applied"
);
return PathBuf::from(".");
}
p
}
/// Read CPU profiling mode from env
fn read_cpu_mode() -> CpuMode {
match get_env_str(ENV_CPU_MODE, DEFAULT_CPU_MODE).to_lowercase().as_str() {
"continuous" => CpuMode::Continuous,
"periodic" => CpuMode::Periodic,
_ => CpuMode::Off,
}
}
/// Generate timestamp string for filenames
fn ts() -> String {
jiff::Zoned::now().strftime("%Y%m%dT%H%M%S").to_string()
}
/// Write pprof report to file in protobuf format
fn write_pprof_report_pb(report: &pprof::Report, path: &Path) -> Result<(), String> {
let profile = report.pprof().map_err(|e| format!("pprof() failed: {e}"))?;
let mut buf = Vec::with_capacity(512 * 1024);
profile.write_to_vec(&mut buf).map_err(|e| format!("encode failed: {e}"))?;
let mut f = File::create(path).map_err(|e| format!("create file failed: {e}"))?;
f.write_all(&buf).map_err(|e| format!("write file failed: {e}"))?;
Ok(())
}
/// Internal: dump CPU pprof from existing guard
async fn dump_cpu_with_guard(guard: &pprof::ProfilerGuard<'_>) -> Result<PathBuf, String> {
let report = guard.report().build().map_err(|e| format!("build report failed: {e}"))?;
let out = output_dir().join(format!("cpu_profile_{}.pb", ts()));
write_pprof_report_pb(&report, &out)?;
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_dump_exported",
profile_type = "cpu",
path = %out.display(),
"Profiling dump exported"
);
Ok(out)
}
// Public API: dump CPU for a duration; if continuous guard exists, snapshot immediately.
pub async fn dump_cpu_pprof_for(duration: Duration) -> Result<PathBuf, String> {
if let Some(cell) = CPU_CONT_GUARD.get() {
let guard_slot = cell.lock().await;
if let Some(ref guard) = *guard_slot {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_dump_source",
profile_type = "cpu",
source = "continuous_guard",
"Using continuous CPU profiling guard for dump"
);
return dump_cpu_with_guard(guard).await;
}
}
let freq = get_env_usize(ENV_CPU_FREQ, DEFAULT_CPU_FREQ) as i32;
let guard = pprof::ProfilerGuard::new(freq).map_err(|e| format!("create profiler failed: {e}"))?;
sleep(duration).await;
dump_cpu_with_guard(&guard).await
}
// Public API: dump memory pprof now (jemalloc)
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
pub async fn dump_memory_pprof_now() -> Result<PathBuf, String> {
let out = output_dir().join(format!("mem_profile_{}.pb", ts()));
let mut f = File::create(&out).map_err(|e| format!("create file failed: {e}"))?;
let prof_ctl_cell = jemalloc_pprof::PROF_CTL
.as_ref()
.ok_or_else(|| "jemalloc profiling control not available".to_string())?;
let mut prof_ctl = prof_ctl_cell.lock().await;
if !prof_ctl.activated() {
return Err("jemalloc profiling is not active".to_string());
}
let bytes = prof_ctl.dump_pprof().map_err(|e| format!("dump pprof failed: {e}"))?;
f.write_all(&bytes).map_err(|e| format!("write file failed: {e}"))?;
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_exported",
profile_type = "memory",
path = %out.display(),
"Profiling dump exported"
);
Ok(out)
}
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
pub async fn dump_memory_pprof_now() -> Result<PathBuf, String> {
Err(memory_profiling_unsupported_message())
}
// Jemalloc status check (No forced placement, only status observation)
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
pub async fn check_jemalloc_profiling() {
use tikv_jemalloc_ctl::{config, epoch, stats};
if let Err(e) = epoch::advance() {
warn!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_epoch_advance_failed",
error = %e,
"Jemalloc profiling state changed"
);
}
match config::malloc_conf::read() {
Ok(conf) => debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_malloc_conf",
result = "ok",
malloc_conf = %conf,
"Jemalloc profiling state checked"
),
Err(e) => debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_malloc_conf",
result = "read_failed",
error = %e,
"Jemalloc profiling state checked"
),
}
match std::env::var("MALLOC_CONF") {
Ok(v) => debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_malloc_conf_env",
state = "set",
malloc_conf = %v,
"Jemalloc profiling state checked"
),
Err(_) => debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_malloc_conf_env",
state = "unset",
"Jemalloc profiling state checked"
),
}
if let Some(lock) = jemalloc_pprof::PROF_CTL.as_ref() {
let ctl = lock.lock().await;
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_profiling_status",
activated = ctl.activated(),
"Jemalloc profiling status checked"
);
} else {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_profiling_status",
state = "unavailable",
"Jemalloc profiling status checked"
);
}
let _ = epoch::advance();
macro_rules! show {
($name:literal, $reader:expr) => {
match $reader {
Ok(v) => debug!(concat!($name, "={}"), v),
Err(e) => debug!(concat!($name, " read failed: {}"), e),
}
};
}
show!("allocated", stats::allocated::read());
show!("resident", stats::resident::read());
show!("mapped", stats::mapped::read());
show!("metadata", stats::metadata::read());
show!("active", stats::active::read());
}
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
pub async fn check_jemalloc_profiling() {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_JEMALLOC,
event = "jemalloc_profiling_status",
result = "skipped",
reason = "unsupported_target",
"Jemalloc profiling status checked"
);
}
// Internal: start continuous CPU profiling
async fn start_cpu_continuous(freq_hz: i32) {
let cell = CPU_CONT_GUARD.get_or_init(|| Arc::new(Mutex::new(None))).clone();
let mut slot = cell.lock().await;
if slot.is_some() {
warn!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_state",
profile_type = "cpu_continuous",
state = "already_running",
"CPU profiling already running"
);
return;
}
match pprof::ProfilerGuardBuilder::default()
.frequency(freq_hz)
.blocklist(&["libc", "libgcc", "pthread", "vdso"])
.build()
{
Ok(guard) => {
*slot = Some(guard);
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_state",
profile_type = "cpu_continuous",
state = "started",
freq_hz,
"CPU profiling started"
);
}
Err(e) => warn!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_state",
profile_type = "cpu_continuous",
state = "start_failed",
error = %e,
"CPU profiling failed to start"
),
}
}
// Internal: start periodic CPU sampling loop
async fn start_cpu_periodic(freq_hz: i32, interval: Duration, duration: Duration, token: CancellationToken) {
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_state",
profile_type = "cpu_periodic",
state = "started",
freq_hz,
?interval,
?duration,
"Periodic CPU profiling started"
);
tokio::spawn(async move {
loop {
tokio::select! {
_ = token.cancelled() => {
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_state",
profile_type = "cpu_periodic",
state = "cancelled",
"Periodic CPU profiling cancelled"
);
break;
}
_ = sleep(interval) => {}
}
if token.is_cancelled() {
break;
}
let guard = match pprof::ProfilerGuard::new(freq_hz) {
Ok(g) => g,
Err(e) => {
warn!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_capture_failed",
profile_type = "cpu_periodic",
stage = "create_guard",
error = %e,
"Profiling capture failed"
);
continue;
}
};
tokio::select! {
_ = token.cancelled() => {
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_state",
profile_type = "cpu_periodic",
state = "cancelled_during_capture",
"Periodic CPU profiling cancelled during capture"
);
break;
}
_ = sleep(duration) => {}
}
match guard.report().build() {
Ok(report) => {
let out = output_dir().join(format!("cpu_profile_{}.pb", ts()));
if let Err(e) = write_pprof_report_pb(&report, &out) {
warn!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_dump_failed",
profile_type = "cpu_periodic",
stage = "write_dump",
path = %out.display(),
error = %e,
"Periodic CPU dump write failed"
);
} else {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_dump_exported",
profile_type = "cpu_periodic",
path = %out.display(),
"Profiling dump exported"
);
}
}
Err(e) => warn!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_capture_failed",
profile_type = "cpu_periodic",
stage = "build_report",
error = %e,
"Profiling capture failed"
),
}
}
});
}
// Internal: start periodic memory dump when jemalloc profiling is active
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
async fn start_memory_periodic(interval: Duration, token: CancellationToken) {
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_state",
profile_type = "memory_periodic",
state = "started",
?interval,
"Periodic memory profiling started"
);
tokio::spawn(async move {
loop {
tokio::select! {
_ = token.cancelled() => {
info!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_state",
profile_type = "memory_periodic",
state = "cancelled",
"Periodic memory profiling cancelled"
);
break;
}
_ = sleep(interval) => {}
}
let Some(lock) = jemalloc_pprof::PROF_CTL.as_ref() else {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_skipped",
profile_type = "memory_periodic",
reason = "prof_ctl_unavailable",
"Profiling dump skipped"
);
continue;
};
let mut ctl = lock.lock().await;
if !ctl.activated() {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_skipped",
profile_type = "memory_periodic",
reason = "jemalloc_inactive",
"Profiling dump skipped"
);
continue;
}
let out = output_dir().join(format!("mem_profile_periodic_{}.pb", ts()));
match File::create(&out) {
Err(e) => {
tracing::error!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_failed",
profile_type = "memory_periodic",
stage = "create_file",
path = %out.display(),
error = %e,
"Periodic memory dump file creation failed"
);
continue;
}
Ok(mut f) => match ctl.dump_pprof() {
Ok(bytes) => {
if let Err(e) = f.write_all(&bytes) {
tracing::error!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_failed",
profile_type = "memory_periodic",
stage = "write_dump",
path = %out.display(),
error = %e,
"Periodic memory dump write failed"
);
} else {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_exported",
profile_type = "memory_periodic",
path = %out.display(),
"Profiling dump exported"
);
}
}
Err(e) => tracing::error!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_failed",
profile_type = "memory_periodic",
stage = "dump_pprof",
error = %e,
"Periodic memory dump export failed"
),
},
}
}
});
}
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
async fn start_memory_periodic(_interval: Duration, _token: CancellationToken) {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_runtime_skipped",
profile_type = "memory_periodic",
reason = "unsupported_target",
"Profiling runtime skipped"
);
}
// Public: unified init entry, avoid duplication/conflict
pub async fn init_from_env() {
let enabled = get_env_bool(ENV_ENABLE_PROFILING, DEFAULT_ENABLE_PROFILING);
if !enabled {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_RUNTIME,
event = "profiling_runtime_disabled",
reason = "env_flag",
"Profiling runtime disabled"
);
return;
}
// Jemalloc state check once (no dump)
check_jemalloc_profiling().await;
// Initialize cancellation token
let token = PROFILING_CANCEL_TOKEN.get_or_init(CancellationToken::new).clone();
// CPU
let cpu_mode = read_cpu_mode();
let cpu_freq = get_env_usize(ENV_CPU_FREQ, DEFAULT_CPU_FREQ) as i32;
let cpu_interval = Duration::from_secs(get_env_u64(ENV_CPU_INTERVAL_SECS, DEFAULT_CPU_INTERVAL_SECS));
let cpu_duration = Duration::from_secs(get_env_u64(ENV_CPU_DURATION_SECS, DEFAULT_CPU_DURATION_SECS));
match cpu_mode {
CpuMode::Off => debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_mode_selected",
profile_type = "cpu",
state = "off",
"Profiling mode selected"
),
CpuMode::Continuous => start_cpu_continuous(cpu_freq).await,
CpuMode::Periodic => start_cpu_periodic(cpu_freq, cpu_interval, cpu_duration, token.clone()).await,
}
// Memory
let mem_periodic = get_env_bool(ENV_MEM_PERIODIC, DEFAULT_MEM_PERIODIC);
let mem_interval = Duration::from_secs(get_env_u64(ENV_MEM_INTERVAL_SECS, DEFAULT_MEM_INTERVAL_SECS));
if mem_periodic {
start_memory_periodic(mem_interval, token).await;
}
}
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
fn memory_profiling_unsupported_message() -> String {
let target_env = option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown");
format!(
"Memory profiling is only supported on linux x86_64 gnu. target_os={}, target_env={target_env}, target_arch={}",
std::env::consts::OS,
std::env::consts::ARCH
)
}
/// Stop all background profiling tasks
pub fn shutdown_profiling() {
if let Some(token) = PROFILING_CANCEL_TOKEN.get() {
token.cancel();
}
}
pub fn shutdown_profiling() {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_RUNTIME,
event = "profiling_shutdown_skipped",
reason = LOCAL_CPU_PPROF_UNSUPPORTED_REASON,
target_os = std::env::consts::OS,
target_env = target_env(),
target_arch = std::env::consts::ARCH,
"Local pprof profiling shutdown skipped"
);
}
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub use linux_impl::{dump_cpu_pprof_for, dump_memory_pprof_now, init_from_env, shutdown_profiling};
pub fn log_cpu_pprof_dump_skipped() {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_CPU,
event = "profiling_dump_skipped",
profile_type = "cpu",
reason = LOCAL_CPU_PPROF_UNSUPPORTED_REASON,
"Local CPU pprof dump skipped"
);
}
pub fn log_memory_pprof_dump_skipped() {
debug!(
component = LOG_COMPONENT_PROFILING,
subsystem = LOG_SUBSYSTEM_MEMORY,
event = "profiling_dump_skipped",
profile_type = "memory",
reason = MEMORY_PPROF_UNSUPPORTED_REASON,
"Memory pprof dump skipped"
);
}
pub fn local_cpu_pprof_unsupported_message() -> String {
unsupported_message(LOCAL_CPU_PPROF_UNSUPPORTED_SUMMARY)
}
pub fn memory_pprof_unsupported_message() -> String {
unsupported_message(MEMORY_PPROF_UNSUPPORTED_SUMMARY)
}
fn unsupported_message(summary: &str) -> String {
format!(
"{summary}. target_os={}, target_env={}, target_arch={}",
std::env::consts::OS,
target_env(),
std::env::consts::ARCH
)
}
fn target_env() -> &'static str {
option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown")
}
rustfs/src/runtime_capabilities.rs
+2 -10
View File
@@ -143,11 +143,7 @@ fn target_env_name() -> Option<&'static str> {
}
fn cpu_profiling_status() -> CapabilityStatus {
if cfg!(any(target_os = "linux", target_os = "macos")) {
CapabilityStatus::supported()
} else {
CapabilityStatus::unsupported().with_reason("userspace CPU profiling supports linux and macos targets")
}
CapabilityStatus::unsupported().with_reason(crate::profiling::LOCAL_CPU_PPROF_UNSUPPORTED_SUMMARY)
}
fn ebpf_status() -> CapabilityStatus {
@@ -167,11 +163,7 @@ fn numa_status() -> CapabilityStatus {
}
fn memory_profiling_status() -> CapabilityStatus {
if cfg!(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")) {
CapabilityStatus::supported().with_reason("jemalloc memory profiling target")
} else {
CapabilityStatus::unsupported().with_reason("memory profiling supports linux gnu x86_64 targets")
}
CapabilityStatus::unsupported().with_reason(crate::profiling::MEMORY_PPROF_UNSUPPORTED_SUMMARY)
}
fn cgroup_memory_status() -> CapabilityStatus {