Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1e93a97caa | ||
|
|
e4cf0189fb |
Generated
+21
@@ -9219,6 +9219,7 @@ dependencies = [
|
||||
"temp-env",
|
||||
"tempfile",
|
||||
"thiserror 2.0.20",
|
||||
"tikv-jemallocator",
|
||||
"time",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
@@ -11943,6 +11944,26 @@ dependencies = [
|
||||
"cfg-if",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tikv-jemalloc-sys"
|
||||
version = "0.6.1+5.3.0-1-ge13ca993e8ccb9ba9847cc330696e02839f328f7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cd8aa5b2ab86a2cefa406d889139c162cbb230092f7d1d7cbc1716405d852a3b"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tikv-jemallocator"
|
||||
version = "0.6.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0359b4327f954e0567e69fb191cf1436617748813819c94b8cd4a431422d053a"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"tikv-jemalloc-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "time"
|
||||
version = "0.3.55"
|
||||
|
||||
@@ -594,3 +594,80 @@ mod tests {
|
||||
assert!(!same_file(&metadata1, &metadata2));
|
||||
}
|
||||
}
|
||||
|
||||
// Bucket existence cache - reduces statx syscalls for repeated bucket checks
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Mutex;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// Cache for bucket directory existence checks.
|
||||
struct BucketExistenceCache {
|
||||
cache: Mutex<HashMap<PathBuf, (Instant, bool)>>,
|
||||
ttl: Duration,
|
||||
}
|
||||
|
||||
impl BucketExistenceCache {
|
||||
fn new(ttl: Duration) -> Self {
|
||||
Self {
|
||||
cache: Mutex::new(HashMap::new()),
|
||||
ttl,
|
||||
}
|
||||
}
|
||||
|
||||
fn check_exists(&self, path: &PathBuf) -> Option<bool> {
|
||||
let mut cache = self.cache.lock().ok()?;
|
||||
if let Some((timestamp, exists)) = cache.get(path) {
|
||||
if timestamp.elapsed() < self.ttl {
|
||||
return Some(*exists);
|
||||
}
|
||||
cache.remove(path);
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn record(&self, path: PathBuf, exists: bool) {
|
||||
if let Ok(mut cache) = self.cache.lock() {
|
||||
cache.insert(path, (Instant::now(), exists));
|
||||
}
|
||||
}
|
||||
|
||||
fn invalidate(&self, path: &PathBuf) {
|
||||
if let Ok(mut cache) = self.cache.lock() {
|
||||
cache.remove(path);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static BUCKET_EXISTENCE_CACHE: std::sync::LazyLock<BucketExistenceCache> =
|
||||
std::sync::LazyLock::new(|| BucketExistenceCache::new(Duration::from_secs(60)));
|
||||
|
||||
/// Cached access check - reduces statx syscalls
|
||||
pub async fn cached_access(path: impl AsRef<Path>) -> io::Result<()> {
|
||||
let path_buf = path.as_ref().to_path_buf();
|
||||
|
||||
if let Some(exists) = BUCKET_EXISTENCE_CACHE.check_exists(&path_buf) {
|
||||
if exists {
|
||||
return Ok(());
|
||||
} else {
|
||||
return Err(io::Error::new(io::ErrorKind::NotFound, "bucket not found (cached)"));
|
||||
}
|
||||
}
|
||||
|
||||
let result = fs::metadata(&path_buf).await;
|
||||
|
||||
match &result {
|
||||
Ok(_) => BUCKET_EXISTENCE_CACHE.record(path_buf, true),
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => {
|
||||
BUCKET_EXISTENCE_CACHE.record(path_buf, false);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
|
||||
result?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn invalidate_bucket_cache(path: impl AsRef<Path>) {
|
||||
BUCKET_EXISTENCE_CACHE.invalidate(&path.as_ref().to_path_buf());
|
||||
}
|
||||
|
||||
@@ -35,7 +35,7 @@ use crate::disk::{
|
||||
error::{DiskError, Error, FileAccessDeniedWithContext, Result},
|
||||
error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error},
|
||||
format::FormatV3,
|
||||
fs::{O_APPEND, O_CREATE, O_RDONLY, O_TRUNC, O_WRONLY, access, lstat, lstat_std, remove, remove_all_std, remove_std, rename},
|
||||
fs::{O_APPEND, O_CREATE, O_RDONLY, O_TRUNC, O_WRONLY, access, cached_access, invalidate_bucket_cache, lstat, lstat_std, remove, remove_all_std, remove_std, rename},
|
||||
is_quota_mutation_fence_path, os,
|
||||
os::{check_path_length, is_dir_not_empty_error, is_empty_dir, is_root_disk, rename_all, rename_all_ignore_missing_source},
|
||||
quota_mutation_fence_path,
|
||||
@@ -3553,7 +3553,7 @@ impl LocalIoBackend for StdBackend {
|
||||
let access_check_start = metrics_enabled.then(std::time::Instant::now);
|
||||
let volume_dir = local_disk_bucket_path(self.io_root(), volume)?;
|
||||
if !skip_access_checks(volume) {
|
||||
access(&volume_dir)
|
||||
cached_access(&volume_dir)
|
||||
.await
|
||||
.map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
|
||||
}
|
||||
@@ -3618,7 +3618,7 @@ impl LocalIoBackend for StdBackend {
|
||||
async fn open_read_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader> {
|
||||
let volume_dir = local_disk_bucket_path(self.io_root(), volume)?;
|
||||
if !skip_access_checks(volume) {
|
||||
access(&volume_dir)
|
||||
cached_access(&volume_dir)
|
||||
.await
|
||||
.map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
|
||||
}
|
||||
@@ -3658,7 +3658,7 @@ impl LocalIoBackend for StdBackend {
|
||||
async fn open_full_read(&self, volume: &str, path: &str) -> Result<FileReader> {
|
||||
let volume_dir = local_disk_bucket_path(self.io_root(), volume)?;
|
||||
if !skip_access_checks(volume) {
|
||||
access(&volume_dir)
|
||||
cached_access(&volume_dir)
|
||||
.await
|
||||
.map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
|
||||
}
|
||||
@@ -3712,7 +3712,7 @@ impl LocalIoBackend for StdBackend {
|
||||
WriteMode::Append => {
|
||||
let volume_dir = local_disk_bucket_path(self.io_root(), volume)?;
|
||||
if !skip_access_checks(volume) {
|
||||
access(&volume_dir)
|
||||
cached_access(&volume_dir)
|
||||
.await
|
||||
.map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
|
||||
}
|
||||
@@ -5817,7 +5817,7 @@ impl LocalDisk {
|
||||
async fn delete_unleased(&self, volume: &str, path: &str, opt: &DeleteOptions) -> Result<()> {
|
||||
let volume_dir = self.io_get_bucket_path(volume)?;
|
||||
if !skip_access_checks(volume)
|
||||
&& let Err(e) = access(&volume_dir).await
|
||||
&& let Err(e) = cached_access(&volume_dir).await
|
||||
{
|
||||
return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
|
||||
}
|
||||
@@ -6739,7 +6739,7 @@ impl LocalDisk {
|
||||
let read_dir_result = match read_dir_entries_with_walk_stall(&dir_path_abs, -1, stall).await {
|
||||
Err(err) if err == Error::FileNotFound && !skip_access_checks(&opts.bucket) => {
|
||||
let volume_dir = self.io_get_bucket_path(&opts.bucket)?;
|
||||
if let Err(access_err) = access(&volume_dir).await {
|
||||
if let Err(access_err) = cached_access(&volume_dir).await {
|
||||
Err(to_access_error(access_err, DiskError::VolumeAccessDenied).into())
|
||||
} else {
|
||||
Err(err)
|
||||
@@ -8101,7 +8101,7 @@ impl DiskAPI for LocalDisk {
|
||||
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
|
||||
let volume_dir = self.io_get_bucket_path(volume)?;
|
||||
if !skip_access_checks(volume)
|
||||
&& let Err(e) = access(&volume_dir).await
|
||||
&& let Err(e) = cached_access(&volume_dir).await
|
||||
{
|
||||
return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
|
||||
}
|
||||
@@ -8309,7 +8309,7 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
if e == DiskError::FileNotFound {
|
||||
if !skip_access_checks(volume)
|
||||
&& let Err(err) = access(&volume_dir).await
|
||||
&& let Err(err) = cached_access(&volume_dir).await
|
||||
&& err.kind() == ErrorKind::NotFound
|
||||
{
|
||||
resp.results[i] = CHECK_PART_VOLUME_NOT_FOUND;
|
||||
@@ -8835,7 +8835,7 @@ impl DiskAPI for LocalDisk {
|
||||
Err(e) => {
|
||||
if e.kind() == ErrorKind::NotFound
|
||||
&& !skip_access_checks(volume)
|
||||
&& let Err(e) = access(&volume_dir).await
|
||||
&& let Err(e) = cached_access(&volume_dir).await
|
||||
{
|
||||
return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
|
||||
}
|
||||
@@ -8864,7 +8864,7 @@ impl DiskAPI for LocalDisk {
|
||||
let volume_dir = self.io_get_bucket_path(&opts.bucket)?;
|
||||
|
||||
if !skip_access_checks(&opts.bucket)
|
||||
&& let Err(e) = with_walk_stall_deadline(stall, access(&volume_dir)).await?
|
||||
&& let Err(e) = with_walk_stall_deadline(stall, cached_access(&volume_dir)).await?
|
||||
{
|
||||
return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
|
||||
}
|
||||
@@ -9944,9 +9944,10 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
let volume_dir = self.io_get_bucket_path(volume)?;
|
||||
|
||||
if let Err(e) = access(&volume_dir).await {
|
||||
if let Err(e) = cached_access(&volume_dir).await {
|
||||
if e.kind() == ErrorKind::NotFound {
|
||||
os::make_dir_all(&volume_dir, self.io_root()).await?;
|
||||
invalidate_bucket_cache(&volume_dir);
|
||||
return Ok(());
|
||||
}
|
||||
error!(
|
||||
@@ -10004,7 +10005,7 @@ impl DiskAPI for LocalDisk {
|
||||
async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> {
|
||||
let volume_dir = self.io_get_bucket_path(volume)?;
|
||||
if !skip_access_checks(volume) {
|
||||
access(&volume_dir)
|
||||
cached_access(&volume_dir)
|
||||
.await
|
||||
.map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?;
|
||||
}
|
||||
@@ -10837,6 +10838,7 @@ impl DiskAPI for LocalDisk {
|
||||
// hit path skips the volume-access check, so nothing else would notice)
|
||||
// (rustfs/backlog#1177).
|
||||
self.io_backend.invalidate_cached_fds_for_volume(volume);
|
||||
invalidate_bucket_cache(&p);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
+7
-3
@@ -40,7 +40,7 @@ name = "rustfs-cli"
|
||||
path = "src/bin/rustfs-cli.rs"
|
||||
|
||||
[features]
|
||||
default = ["ftps", "webdav"]
|
||||
default = ["ftps", "webdav", "mimalloc"]
|
||||
metrics-gpu = ["rustfs-obs/gpu"]
|
||||
ftps = ["rustfs-protocols/ftps"]
|
||||
swift = ["rustfs-protocols/swift"]
|
||||
@@ -56,6 +56,9 @@ rio-v2 = ["rustfs-ecstore/rio-v2"]
|
||||
pyroscope = ["rustfs-obs/pyroscope"]
|
||||
# Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`.
|
||||
dial9 = ["rustfs-obs/dial9"]
|
||||
# Allocator features
|
||||
mimalloc = ["dep:rustfs-mimalloc", "dep:rustfs-mimalloc-sys"]
|
||||
jemalloc = ["dep:tikv-jemallocator"]
|
||||
hotpath = [
|
||||
"hotpath/hotpath",
|
||||
"hotpath/tokio",
|
||||
@@ -336,13 +339,14 @@ opentelemetry = { workspace = true }
|
||||
tracing-opentelemetry = { workspace = true }
|
||||
# Data structures
|
||||
hashbrown = { workspace = true, features = ["serde", "rayon"] }
|
||||
rustfs-mimalloc = { workspace = true }
|
||||
rustfs-mimalloc = { workspace = true, optional = true }
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
libsystemd.workspace = true
|
||||
|
||||
[target.'cfg(not(target_os = "windows"))'.dependencies]
|
||||
rustfs-mimalloc-sys.workspace = true
|
||||
rustfs-mimalloc-sys = { workspace = true, optional = true }
|
||||
tikv-jemallocator = { version = "0.6", optional = true }
|
||||
|
||||
[dev-dependencies]
|
||||
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
|
||||
|
||||
Reference in New Issue
Block a user