mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
feat(observability): feature-gated hotpath instrumentation for the data path (#4394)
Merge the hotpath-rs wall-time instrumentation from the backlog#936 analysis worktree behind an opt-in 'hotpath' cargo feature, keeping the default build at zero overhead and zero dependency. - hotpath is an optional dependency everywhere (dep:hotpath feature syntax); the default dependency tree contains no hotpath crate at all - 40+ measurement points across S3 handlers, ECStore/SetDisks object and multipart ops, erasure encode/decode, bitrot, LocalDisk I/O, FileMeta codec, and HashReader - attribute sites use #[cfg_attr(feature = "hotpath", hotpath::measure)]; async_trait bodies use per-crate hp_guard! macros (ecstore + rustfs bin); rio gates measure_block! behind hp_measure_block! - feature chain: rustfs -> rustfs-ecstore -> rustfs-rio / rustfs-filemeta, each crate owning its own gate - hotpath-alloc is intentionally not wired up (hotpath 0.21.x TLS panic on cross-thread guard drop under tokio, see backlog#935); mimalloc stays the unconditional global allocator - docs/development/hotpath-profiling.md documents building, HOTPATH_* env vars, SIGTERM report flow, and how to reproduce the backlog#936 timing reports Refs: https://github.com/rustfs/backlog/issues/935 (HP-14, item 2), https://github.com/rustfs/backlog/issues/936 Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
Generated
+146
-2
@@ -560,6 +560,12 @@ dependencies = [
|
||||
"regex-syntax",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ascii"
|
||||
version = "1.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16"
|
||||
|
||||
[[package]]
|
||||
name = "asn1-rs"
|
||||
version = "0.7.2"
|
||||
@@ -1838,6 +1844,12 @@ dependencies = [
|
||||
"phf 0.12.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chunked_transfer"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901"
|
||||
|
||||
[[package]]
|
||||
name = "ciborium"
|
||||
version = "0.2.2"
|
||||
@@ -1992,7 +2004,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "958c5d6ecf1f214b4c2bbbbf6ab9523a864bd136dcf71a7e8904799acfe1ad47"
|
||||
dependencies = [
|
||||
"unicode-segmentation",
|
||||
"unicode-width",
|
||||
"unicode-width 0.2.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -3594,6 +3606,16 @@ dependencies = [
|
||||
"dirs-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dirs-next"
|
||||
version = "2.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"dirs-sys-next",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dirs-sys"
|
||||
version = "0.5.0"
|
||||
@@ -3602,10 +3624,21 @@ checksum = "e01a3366d27ee9890022452ee61b2b63a67e6f13f58900b651ff5665f0bb1fab"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"option-ext",
|
||||
"redox_users",
|
||||
"redox_users 0.5.2",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dirs-sys-next"
|
||||
version = "0.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"redox_users 0.4.6",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dispatch2"
|
||||
version = "0.3.1"
|
||||
@@ -4682,6 +4715,16 @@ dependencies = [
|
||||
"hashbrown 0.17.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hdrhistogram"
|
||||
version = "7.5.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "765c9198f173dd59ce26ff9f95ef0aafd0a0fe01fb9d72841bc5066a4c06511d"
|
||||
dependencies = [
|
||||
"byteorder",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "headers"
|
||||
version = "0.4.1"
|
||||
@@ -4873,6 +4916,39 @@ dependencies = [
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hotpath"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1ff6b552a6afa29d9e33f8d555bee9093c142dd449501ae128e6494a303f03dc"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"cfg-if",
|
||||
"crossbeam-channel",
|
||||
"futures-util",
|
||||
"hdrhistogram",
|
||||
"hotpath-macros",
|
||||
"libc",
|
||||
"pin-project-lite",
|
||||
"prettytable-rs",
|
||||
"quanta",
|
||||
"regex",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tiny_http",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hotpath-macros"
|
||||
version = "0.21.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4f15322569d3cfadf84c0de7ef72be435b8f4b4839ee4ace78a7eaca48a87ded"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.118",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "htmlescape"
|
||||
version = "0.3.1"
|
||||
@@ -5274,6 +5350,17 @@ dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "is-terminal"
|
||||
version = "0.4.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "is_debug"
|
||||
version = "1.1.0"
|
||||
@@ -7591,6 +7678,19 @@ dependencies = [
|
||||
"syn 2.0.118",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prettytable-rs"
|
||||
version = "0.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "eea25e07510aa6ab6547308ebe3c036016d162b8da920dbb079e3ba8acf3d95a"
|
||||
dependencies = [
|
||||
"encode_unicode",
|
||||
"is-terminal",
|
||||
"lazy_static",
|
||||
"term",
|
||||
"unicode-width 0.1.14",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "primefield"
|
||||
version = "0.14.0"
|
||||
@@ -8264,6 +8364,17 @@ dependencies = [
|
||||
"bitflags 2.13.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_users"
|
||||
version = "0.4.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43"
|
||||
dependencies = [
|
||||
"getrandom 0.2.17",
|
||||
"libredox",
|
||||
"thiserror 1.0.69",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_users"
|
||||
version = "0.5.2"
|
||||
@@ -8759,6 +8870,7 @@ dependencies = [
|
||||
"hashbrown 0.17.1",
|
||||
"hex-simd",
|
||||
"hmac 0.13.0",
|
||||
"hotpath",
|
||||
"http 1.4.2",
|
||||
"http-body 1.0.1",
|
||||
"http-body-util",
|
||||
@@ -9009,6 +9121,7 @@ dependencies = [
|
||||
"google-cloud-storage",
|
||||
"hex-simd",
|
||||
"hmac 0.13.0",
|
||||
"hotpath",
|
||||
"http 1.4.2",
|
||||
"http-body 1.0.1",
|
||||
"http-body-util",
|
||||
@@ -9126,6 +9239,7 @@ dependencies = [
|
||||
"bytes",
|
||||
"crc-fast",
|
||||
"criterion",
|
||||
"hotpath",
|
||||
"proptest",
|
||||
"regex",
|
||||
"rmp",
|
||||
@@ -9639,6 +9753,7 @@ dependencies = [
|
||||
"faster-hex",
|
||||
"futures",
|
||||
"hex-simd",
|
||||
"hotpath",
|
||||
"http 1.4.2",
|
||||
"http-body-util",
|
||||
"md-5 0.11.0",
|
||||
@@ -11262,6 +11377,17 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "term"
|
||||
version = "0.7.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c59df8ac95d96ff9bede18eb7300b0fda5e5d8d90960e76f8e14ae765eedbf1f"
|
||||
dependencies = [
|
||||
"dirs-next",
|
||||
"rustversion",
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "termcolor"
|
||||
version = "1.4.1"
|
||||
@@ -11395,6 +11521,18 @@ dependencies = [
|
||||
"crunchy",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tiny_http"
|
||||
version = "0.12.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "389915df6413a2e74fb181895f933386023c71110878cd0825588928e64cdc82"
|
||||
dependencies = [
|
||||
"ascii",
|
||||
"chunked_transfer",
|
||||
"httpdate",
|
||||
"log",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tinystr"
|
||||
version = "0.8.3"
|
||||
@@ -11999,6 +12137,12 @@ version = "1.13.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c6f5d3c3b1bf09027a88a6bc961fc00497d651009560b5463668dc81b0fa87a8"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-width"
|
||||
version = "0.1.14"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-width"
|
||||
version = "0.2.2"
|
||||
|
||||
@@ -338,6 +338,7 @@ dav-server = "0.11.0"
|
||||
|
||||
# Performance Analysis and Memory Profiling
|
||||
mimalloc = "0.1"
|
||||
hotpath = "0.21"
|
||||
# Snapshot testing for output format regression detection
|
||||
insta = { version = "1.48", features = ["yaml", "json"] }
|
||||
|
||||
|
||||
@@ -33,8 +33,10 @@ workspace = true
|
||||
[features]
|
||||
default = []
|
||||
rio-v2 = ["dep:rustfs-rio-v2"]
|
||||
hotpath = ["dep:hotpath", "hotpath/hotpath", "rustfs-filemeta/hotpath", "rustfs-rio/hotpath"]
|
||||
|
||||
[dependencies]
|
||||
hotpath = { workspace = true, optional = true }
|
||||
rustfs-filemeta.workspace = true
|
||||
rustfs-utils = { workspace = true, features = ["full"] }
|
||||
rustfs-rio.workspace = true
|
||||
|
||||
@@ -2338,6 +2338,7 @@ impl LocalDisk {
|
||||
Ok((buf, mtime))
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
|
||||
check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
|
||||
|
||||
@@ -2364,6 +2365,7 @@ impl LocalDisk {
|
||||
Ok((data, modtime))
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
|
||||
// TODO: timeout support
|
||||
let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
|
||||
@@ -3417,6 +3419,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
|
||||
crate::hp_guard!("LocalDisk::read_all");
|
||||
if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE {
|
||||
let format_info = self.format_info.read().await;
|
||||
if !format_info.data.is_empty() {
|
||||
@@ -3433,11 +3436,13 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
#[tracing::instrument(level = "debug", skip_all)]
|
||||
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::write_all");
|
||||
self.write_all_public(volume, path, data).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::delete");
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
if !skip_access_checks(volume)
|
||||
&& let Err(e) = access(&volume_dir).await
|
||||
@@ -3794,6 +3799,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::rename_file");
|
||||
let src_volume_dir = self.get_bucket_path(src_volume)?;
|
||||
let dst_volume_dir = self.get_bucket_path(dst_volume)?;
|
||||
if !skip_access_checks(src_volume) {
|
||||
@@ -3852,6 +3858,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: i64) -> Result<FileWriter> {
|
||||
crate::hp_guard!("LocalDisk::create_file");
|
||||
if !origvolume.is_empty() {
|
||||
let origvolume_dir = self.get_bucket_path(origvolume)?;
|
||||
if !skip_access_checks(origvolume) {
|
||||
@@ -3875,11 +3882,13 @@ impl DiskAPI for LocalDisk {
|
||||
// TODO: io verifier
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
|
||||
crate::hp_guard!("LocalDisk::read_file");
|
||||
self.io_backend.open_full_read(volume, path).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
|
||||
crate::hp_guard!("LocalDisk::read_file_stream");
|
||||
self.io_backend.open_read_stream(volume, path, offset, length).await
|
||||
}
|
||||
|
||||
@@ -4031,6 +4040,7 @@ impl DiskAPI for LocalDisk {
|
||||
dst_volume: &str,
|
||||
dst_path: &str,
|
||||
) -> Result<RenameDataResp> {
|
||||
crate::hp_guard!("LocalDisk::rename_data");
|
||||
let src_volume_dir = self.get_bucket_path(src_volume)?;
|
||||
if !skip_access_checks(src_volume)
|
||||
&& let Err(e) = super::fs::access_std(&src_volume_dir)
|
||||
@@ -4524,6 +4534,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
|
||||
crate::hp_guard!("LocalDisk::write_metadata");
|
||||
let p = self.get_object_path(volume, format!("{path}/{STORAGE_FORMAT_FILE}").as_str())?;
|
||||
|
||||
let mut meta = FileMeta::new();
|
||||
@@ -4557,6 +4568,7 @@ impl DiskAPI for LocalDisk {
|
||||
version_id: &str,
|
||||
opts: &ReadOptions,
|
||||
) -> Result<FileInfo> {
|
||||
crate::hp_guard!("LocalDisk::read_version");
|
||||
if !org_volume.is_empty() {
|
||||
let org_volume_path = self.get_bucket_path(org_volume)?;
|
||||
if !skip_access_checks(org_volume) {
|
||||
@@ -4652,6 +4664,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
|
||||
crate::hp_guard!("LocalDisk::read_xl");
|
||||
let file_path = self.get_object_path(volume, path)?;
|
||||
let file_dir = self.get_bucket_path(volume)?;
|
||||
|
||||
@@ -4881,6 +4894,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn read_metadata(&self, volume: &str, path: &str) -> Result<Bytes> {
|
||||
crate::hp_guard!("LocalDisk::read_metadata");
|
||||
let file_path = self.get_object_path(volume, path)?;
|
||||
let volume_dir = self.get_bucket_path(volume)?;
|
||||
let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
|
||||
|
||||
@@ -60,6 +60,7 @@ where
|
||||
|
||||
/// Read a single (hash+data) block, verify hash, and return the number of bytes read into `out`.
|
||||
/// Returns an error if hash verification fails or data exceeds shard_size.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
|
||||
self.last_verify_duration = Duration::ZERO;
|
||||
if out.len() > self.shard_size {
|
||||
@@ -154,6 +155,7 @@ where
|
||||
|
||||
/// Write a (hash+data) block. Returns the number of data bytes written.
|
||||
/// Returns an error if called after a short write or if data exceeds shard_size.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(label = "BitrotWriter::write"))]
|
||||
pub async fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
||||
if buf.is_empty() {
|
||||
return Ok(0);
|
||||
@@ -232,6 +234,7 @@ pub fn bitrot_shard_file_size(size: usize, shard_size: usize, algo: HashAlgorith
|
||||
size.div_ceil(shard_size) * algo.size() + size
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn bitrot_verify<R: AsyncRead + Unpin + Send>(
|
||||
mut r: R,
|
||||
want_size: usize,
|
||||
|
||||
@@ -654,6 +654,7 @@ impl<R> ParallelReader<R>
|
||||
where
|
||||
R: AsyncRead + Unpin + Send + Sync,
|
||||
{
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn read(&mut self) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>) {
|
||||
// On the reconstruction-verifying GET path, read every live shard reader
|
||||
// in lockstep so all readers advance one block per stripe and stay
|
||||
@@ -1432,6 +1433,7 @@ where
|
||||
}
|
||||
|
||||
impl Erasure {
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn decode<W, R>(
|
||||
&self,
|
||||
writer: &mut W,
|
||||
|
||||
@@ -368,6 +368,7 @@ impl Erasure {
|
||||
Ok((reader, total))
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn encode<R>(
|
||||
self: Arc<Self>,
|
||||
mut reader: R,
|
||||
@@ -510,6 +511,7 @@ impl Erasure {
|
||||
Ok((reader, total))
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn encode_batched<R>(
|
||||
self: Arc<Self>,
|
||||
mut reader: R,
|
||||
@@ -637,6 +639,7 @@ impl Erasure {
|
||||
|
||||
/// Fast path for small inline objects: skip tokio::spawn + mpsc channel.
|
||||
/// Reads all data, encodes directly, writes shards sequentially.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn encode_inline_small<R>(
|
||||
self: Arc<Self>,
|
||||
reader: R,
|
||||
@@ -651,6 +654,7 @@ impl Erasure {
|
||||
|
||||
/// Fast path for single-block non-inline objects: avoids the producer/consumer
|
||||
/// pipeline in `encode()` while keeping the same writer/quorum/shutdown semantics.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn encode_single_block_non_inline<R>(
|
||||
self: Arc<Self>,
|
||||
reader: R,
|
||||
|
||||
@@ -504,6 +504,7 @@ impl Erasure {
|
||||
/// # Returns
|
||||
/// A vector of encoded shards as `Bytes`.
|
||||
#[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub fn encode_data(&self, data: &[u8]) -> io::Result<Vec<Bytes>> {
|
||||
let shard_size_fn = if self.uses_legacy {
|
||||
calc_shard_size_legacy
|
||||
@@ -551,6 +552,7 @@ impl Erasure {
|
||||
|
||||
/// Encode owned data, avoiding a copy when the caller already has a heap buffer.
|
||||
/// Falls back to copying into a new buffer if zero-copy conversion fails.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub fn encode_data_owned(&self, data: Vec<u8>) -> io::Result<Vec<Bytes>> {
|
||||
let shard_size_fn = if self.uses_legacy {
|
||||
calc_shard_size_legacy
|
||||
@@ -608,6 +610,7 @@ impl Erasure {
|
||||
|
||||
/// Encode data from an owned `BytesMut` buffer, avoiding the initial copy
|
||||
/// from a borrowed slice into a fresh `BytesMut`.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub fn encode_data_bytes_mut(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
|
||||
let shard_size_fn = if self.uses_legacy {
|
||||
calc_shard_size_legacy
|
||||
@@ -660,6 +663,7 @@ impl Erasure {
|
||||
///
|
||||
/// # Returns
|
||||
/// Ok if reconstruction succeeds, error otherwise.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub fn decode_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
||||
if self.parity_shards > 0 {
|
||||
if self.uses_legacy {
|
||||
@@ -679,6 +683,7 @@ impl Erasure {
|
||||
}
|
||||
|
||||
/// Decode and reconstruct missing data shards, then regenerate parity shards.
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub fn decode_data_and_parity(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
||||
if self.parity_shards > 0 {
|
||||
if self.uses_legacy {
|
||||
|
||||
@@ -12,6 +12,24 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
/// Scope-based hotpath measurement for `#[async_trait]` methods, where
|
||||
/// `#[cfg_attr(feature = "hotpath", hotpath::measure)]` would only time the boxed-future construction.
|
||||
/// The guard records wall time from this statement until the enclosing
|
||||
/// (desugared) async block completes, including early returns via `?`.
|
||||
#[cfg(feature = "hotpath")]
|
||||
#[macro_export]
|
||||
macro_rules! hp_guard {
|
||||
($label:expr) => {
|
||||
let _hotpath_scope_guard = ::hotpath::functions::build_measurement_guard_sync($label, false);
|
||||
};
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
#[macro_export]
|
||||
macro_rules! hp_guard {
|
||||
($label:expr) => {};
|
||||
}
|
||||
|
||||
pub mod api;
|
||||
mod bucket;
|
||||
mod cache_value;
|
||||
|
||||
@@ -264,6 +264,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
data: &mut PutObjReader,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<PartInfo> {
|
||||
crate::hp_guard!("SetDisks::put_object_part");
|
||||
let upload_id_path = Self::get_upload_id_dir(bucket, object, upload_id);
|
||||
|
||||
let (fi, _) = self.check_upload_id_exists(bucket, object, upload_id, true).await?;
|
||||
@@ -786,6 +787,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<MultipartUploadResult> {
|
||||
crate::hp_guard!("SetDisks::new_multipart_upload");
|
||||
let mut _object_lock_guard = None;
|
||||
|
||||
if opts.http_preconditions.is_some() {
|
||||
@@ -955,6 +957,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
uploaded_parts: Vec<CompletePart>,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<ObjectInfo> {
|
||||
crate::hp_guard!("SetDisks::complete_multipart_upload");
|
||||
self.invalidate_get_object_metadata_cache(bucket, object).await;
|
||||
|
||||
let mut object_lock_guard = None;
|
||||
|
||||
@@ -40,6 +40,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
h: HeaderMap,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<GetObjectReader> {
|
||||
crate::hp_guard!("SetDisks::get_object_reader");
|
||||
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
|
||||
// Check if lock optimization is enabled for reads that are fully materialized in memory.
|
||||
let lock_optimization_enabled = is_lock_optimization_enabled();
|
||||
@@ -550,6 +551,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
|
||||
#[tracing::instrument(skip(self, data,))]
|
||||
async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
|
||||
crate::hp_guard!("SetDisks::put_object");
|
||||
self.invalidate_get_object_metadata_cache(bucket, object).await;
|
||||
|
||||
let disks = self.get_disks_internal().await;
|
||||
@@ -1705,6 +1707,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
|
||||
#[tracing::instrument(skip(self))]
|
||||
async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
|
||||
crate::hp_guard!("SetDisks::get_object_info");
|
||||
// Acquire a shared read-lock to protect consistency during info fetch
|
||||
let _read_lock_guard = if !opts.no_lock {
|
||||
Some(self.acquire_read_lock_diag("get_object_info", bucket, object).await?)
|
||||
|
||||
@@ -120,6 +120,7 @@ impl SetDisks {
|
||||
.await;
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub async fn read_version_optimized(
|
||||
&self,
|
||||
bucket: &str,
|
||||
@@ -161,6 +162,7 @@ impl SetDisks {
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub(super) async fn get_object_fileinfo(
|
||||
&self,
|
||||
bucket: &str,
|
||||
@@ -319,6 +321,7 @@ impl SetDisks {
|
||||
Ok((fi, parts_metadata, op_online_disks))
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub(super) async fn get_object_info_and_quorum(
|
||||
&self,
|
||||
bucket: &str,
|
||||
@@ -515,6 +518,7 @@ impl SetDisks {
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub(super) async fn get_object_with_fileinfo<W>(
|
||||
// &self,
|
||||
bucket: &str,
|
||||
@@ -1042,6 +1046,7 @@ impl SetDisks {
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub(super) async fn get_object_decode_reader_with_fileinfo(
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
@@ -1193,6 +1198,7 @@ impl SetDisks {
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
async fn build_codec_streaming_part_reader(
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
@@ -1330,6 +1336,7 @@ fn multipart_part_checksum_algo(fi: &FileInfo, part_number: usize) -> HashAlgori
|
||||
/// `get_object_with_fileinfo` (backlog#870) so both report the same
|
||||
/// stage-duration semantics.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
async fn setup_multipart_part_readers(
|
||||
files: &[FileInfo],
|
||||
disks: &[Option<DiskStore>],
|
||||
|
||||
@@ -186,6 +186,7 @@ impl ECStore {
|
||||
}
|
||||
|
||||
#[instrument(skip(self, data))]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub(super) async fn handle_put_object_part(
|
||||
&self,
|
||||
bucket: &str,
|
||||
|
||||
@@ -628,6 +628,7 @@ impl ECStore {
|
||||
}
|
||||
|
||||
#[instrument(level = "debug", skip(self))]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub(super) async fn handle_get_object_reader(
|
||||
&self,
|
||||
bucket: &str,
|
||||
@@ -661,6 +662,7 @@ impl ECStore {
|
||||
}
|
||||
|
||||
#[instrument(level = "debug", skip(self, data))]
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure)]
|
||||
pub(super) async fn handle_put_object(
|
||||
&self,
|
||||
bucket: &str,
|
||||
|
||||
@@ -25,7 +25,12 @@ keywords = ["file-metadata", "storage", "retrieval", "rustfs", "Minio"]
|
||||
categories = ["web-programming", "development-tools", "filesystem"]
|
||||
documentation = "https://docs.rs/rustfs-filemeta/latest/rustfs_filemeta/"
|
||||
|
||||
[features]
|
||||
default = []
|
||||
hotpath = ["dep:hotpath", "hotpath/hotpath"]
|
||||
|
||||
[dependencies]
|
||||
hotpath = { workspace = true, optional = true }
|
||||
crc-fast = { workspace = true }
|
||||
rmp.workspace = true
|
||||
rmp-serde.workspace = true
|
||||
|
||||
@@ -19,6 +19,7 @@ impl FileMeta {
|
||||
!matches!(Self::check_xl2_v1(buf), Err(_e))
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "FileMeta"))]
|
||||
pub fn load(buf: &[u8]) -> Result<FileMeta> {
|
||||
let mut xl = FileMeta::default();
|
||||
xl.unmarshal_msg(buf)?;
|
||||
@@ -111,6 +112,7 @@ impl FileMeta {
|
||||
Ok((bin_len, remaining))
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "FileMeta"))]
|
||||
pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result<u64> {
|
||||
let i = buf.len() as u64;
|
||||
|
||||
@@ -324,6 +326,7 @@ impl FileMeta {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg_attr(feature = "hotpath", hotpath::measure(impl_type = "FileMeta"))]
|
||||
pub fn marshal_msg(&self) -> Result<Vec<u8>> {
|
||||
let mut wr = Vec::new();
|
||||
|
||||
|
||||
@@ -28,7 +28,12 @@ documentation = "https://docs.rs/rustfs-rio/latest/rustfs_rio/"
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
[features]
|
||||
default = []
|
||||
hotpath = ["dep:hotpath", "hotpath/hotpath"]
|
||||
|
||||
[dependencies]
|
||||
hotpath = { workspace = true, optional = true }
|
||||
arc-swap.workspace = true
|
||||
tokio = { workspace = true, features = ["full"] }
|
||||
rand = { workspace = true }
|
||||
|
||||
@@ -531,7 +531,7 @@ impl AsyncRead for HashReader {
|
||||
let before = buf.filled().len();
|
||||
match this.inner.poll_read(cx, buf) {
|
||||
Poll::Pending => Poll::Pending,
|
||||
Poll::Ready(Ok(())) => {
|
||||
Poll::Ready(Ok(())) => crate::hp_measure_block!("HashReader::on_chunk", {
|
||||
let data = &buf.filled()[before..];
|
||||
let filled = data.len();
|
||||
|
||||
@@ -612,7 +612,7 @@ impl AsyncRead for HashReader {
|
||||
*this.checksum_on_finish = true;
|
||||
}
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}),
|
||||
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,26 @@
|
||||
// Default encryption block size - aligned with system default read buffer size (1MB)
|
||||
pub const DEFAULT_ENCRYPTION_BLOCK_SIZE: usize = 1024 * 1024;
|
||||
|
||||
/// Feature-gated wrapper around `hotpath::measure_block!` for expression
|
||||
/// positions (e.g. inside `poll_read`) where an attribute cannot be used.
|
||||
/// With the `hotpath` feature off it expands to the body unchanged, so the
|
||||
/// default build carries zero profiling overhead and no hotpath dependency.
|
||||
#[cfg(feature = "hotpath")]
|
||||
#[macro_export]
|
||||
macro_rules! hp_measure_block {
|
||||
($label:expr, $body:expr) => {
|
||||
::hotpath::measure_block!($label, $body)
|
||||
};
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
#[macro_export]
|
||||
macro_rules! hp_measure_block {
|
||||
($label:expr, $body:expr) => {
|
||||
$body
|
||||
};
|
||||
}
|
||||
|
||||
macro_rules! delegate_reader_capabilities_generic {
|
||||
($name:ident<$inner_ty:ident>, $inner:ident) => {
|
||||
impl<$inner_ty> crate::EtagResolvable for $name<$inner_ty>
|
||||
|
||||
@@ -60,11 +60,19 @@ full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp"]
|
||||
manual-test-runners = []
|
||||
rio-v2 = ["rustfs-ecstore/rio-v2"]
|
||||
pyroscope = ["rustfs-obs/pyroscope"]
|
||||
hotpath = [
|
||||
"dep:hotpath",
|
||||
"hotpath/hotpath",
|
||||
"rustfs-ecstore/hotpath",
|
||||
"rustfs-filemeta/hotpath",
|
||||
"rustfs-rio/hotpath",
|
||||
]
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
[dependencies]
|
||||
hotpath = { workspace = true, optional = true }
|
||||
# RustFS Internal Crates
|
||||
rustfs-heal = { workspace = true }
|
||||
rustfs-audit = { workspace = true }
|
||||
|
||||
@@ -50,6 +50,25 @@
|
||||
//! tests where you start one server in a background task, run all your
|
||||
//! tests, and then shut it down.
|
||||
|
||||
/// Scope-based hotpath measurement for `#[async_trait]` methods, where
|
||||
/// `#[hotpath::measure]` would only time the boxed-future construction.
|
||||
/// The guard records wall time from this statement until the enclosing
|
||||
/// (desugared) async block completes, including early returns via `?`.
|
||||
/// With the `hotpath` feature off it expands to nothing.
|
||||
#[cfg(feature = "hotpath")]
|
||||
#[macro_export]
|
||||
macro_rules! hp_guard {
|
||||
($label:expr) => {
|
||||
let _hotpath_scope_guard = ::hotpath::functions::build_measurement_guard_sync($label, false);
|
||||
};
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "hotpath"))]
|
||||
#[macro_export]
|
||||
macro_rules! hp_guard {
|
||||
($label:expr) => {};
|
||||
}
|
||||
|
||||
pub mod admin;
|
||||
pub mod allocator_reclaim;
|
||||
pub mod app;
|
||||
|
||||
@@ -16,5 +16,19 @@
|
||||
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
|
||||
|
||||
fn main() {
|
||||
// Wall-time profiling guard for `--features hotpath` builds. The guard
|
||||
// lives for the whole process; dropping it on normal exit (SIGINT/SIGTERM
|
||||
// unwind back through run_process) prints the timing report.
|
||||
//
|
||||
// hotpath's alloc mode is intentionally NOT wired up: hotpath 0.21.x
|
||||
// panics when a MeasurementGuardSync drops on a different thread than it
|
||||
// was created on (TLS index mismatch, alloc/core.rs:151), which tokio's
|
||||
// work-stealing runtime does constantly. See rustfs/backlog#935.
|
||||
#[cfg(feature = "hotpath")]
|
||||
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main")
|
||||
.percentiles(&[50.0, 95.0, 99.0])
|
||||
.functions_limit(0)
|
||||
.build();
|
||||
|
||||
rustfs::startup_entrypoint::run_process();
|
||||
}
|
||||
|
||||
@@ -257,6 +257,7 @@ impl S3 for FS {
|
||||
&self,
|
||||
req: S3Request<CompleteMultipartUploadInput>,
|
||||
) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
|
||||
crate::hp_guard!("S3::complete_multipart_upload");
|
||||
let usecase = s3_api::default_multipart_usecase();
|
||||
Box::pin(usecase.execute_complete_multipart_upload(req)).await
|
||||
}
|
||||
@@ -283,6 +284,7 @@ impl S3 for FS {
|
||||
&self,
|
||||
req: S3Request<CreateMultipartUploadInput>,
|
||||
) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
|
||||
crate::hp_guard!("S3::create_multipart_upload");
|
||||
let usecase = s3_api::default_multipart_usecase();
|
||||
usecase.execute_create_multipart_upload(req).await
|
||||
}
|
||||
@@ -374,6 +376,7 @@ impl S3 for FS {
|
||||
/// Delete an object
|
||||
#[instrument(level = "debug", skip(self, req))]
|
||||
async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
|
||||
crate::hp_guard!("S3::delete_object");
|
||||
let usecase = s3_api::default_object_usecase();
|
||||
Box::pin(usecase.execute_delete_object(req)).await
|
||||
}
|
||||
@@ -659,6 +662,7 @@ impl S3 for FS {
|
||||
fields(start_time=?time::OffsetDateTime::now_utc())
|
||||
)]
|
||||
async fn get_object(&self, req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
|
||||
crate::hp_guard!("S3::get_object");
|
||||
let usecase = s3_api::default_object_usecase();
|
||||
Box::pin(usecase.execute_get_object(req)).await
|
||||
}
|
||||
@@ -943,6 +947,7 @@ impl S3 for FS {
|
||||
|
||||
#[instrument(level = "debug", skip(self, req))]
|
||||
async fn head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
|
||||
crate::hp_guard!("S3::head_object");
|
||||
let usecase = s3_api::default_object_usecase();
|
||||
usecase.execute_head_object(req).await
|
||||
}
|
||||
@@ -982,6 +987,7 @@ impl S3 for FS {
|
||||
|
||||
#[instrument(level = "debug", skip(self, req))]
|
||||
async fn list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
|
||||
crate::hp_guard!("S3::list_objects_v2");
|
||||
record_s3_op(S3Operation::ListObjectsV2, &req.input.bucket);
|
||||
let usecase = s3_api::default_bucket_usecase();
|
||||
usecase.execute_list_objects_v2(req).await
|
||||
@@ -1189,6 +1195,7 @@ impl S3 for FS {
|
||||
|
||||
#[instrument(level = "debug", skip(self, req))]
|
||||
async fn put_object(&self, req: S3Request<PutObjectInput>) -> S3Result<S3Response<PutObjectOutput>> {
|
||||
crate::hp_guard!("S3::put_object");
|
||||
let usecase = s3_api::default_object_usecase();
|
||||
Box::pin(usecase.execute_put_object(self, req)).await
|
||||
}
|
||||
@@ -1549,6 +1556,7 @@ impl S3 for FS {
|
||||
|
||||
#[instrument(level = "debug", skip(self, req))]
|
||||
async fn upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
|
||||
crate::hp_guard!("S3::upload_part");
|
||||
record_s3_op(S3Operation::UploadPart, &req.input.bucket);
|
||||
let usecase = s3_api::default_multipart_usecase();
|
||||
usecase.execute_upload_part(req).await
|
||||
|
||||
Reference in New Issue
Block a user