Performance: improve (#514)

* Performance: improve

Signed-off-by: junxiang Mu <1948535941@qq.com>

* remove dirty

Signed-off-by: junxiang Mu <1948535941@qq.com>

* fix some err

Signed-off-by: junxiang Mu <1948535941@qq.com>

---------

Signed-off-by: junxiang Mu <1948535941@qq.com>

crates/ecstore/Cargo.toml

@@ -101,6 +101,8 @@ rustfs-signer.workspace = true
 rustfs-checksums.workspace = true
 futures-util.workspace = true
 async-recursion.workspace = true
+parking_lot = "0.12"
+moka = { version = "0.12", features = ["future"] }
 
 [target.'cfg(not(windows))'.dependencies]
 nix = { workspace = true }

crates/ecstore/src/batch_processor.rs

@@ -0,0 +1,231 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! High-performance batch processor using JoinSet
+//!
+//! This module provides optimized batching utilities to reduce async runtime overhead
+//! and improve concurrent operation performance.
+
+use crate::disk::error::{Error, Result};
+use std::future::Future;
+use std::sync::Arc;
+use tokio::task::JoinSet;
+
+/// Batch processor that executes tasks concurrently with a semaphore
+pub struct AsyncBatchProcessor {
+    max_concurrent: usize,
+}
+
+impl AsyncBatchProcessor {
+    pub fn new(max_concurrent: usize) -> Self {
+        Self { max_concurrent }
+    }
+
+    /// Execute a batch of tasks concurrently with concurrency control
+    pub async fn execute_batch<T, F>(&self, tasks: Vec<F>) -> Vec<Result<T>>
+    where
+        T: Send + 'static,
+        F: Future<Output = Result<T>> + Send + 'static,
+    {
+        if tasks.is_empty() {
+            return Vec::new();
+        }
+
+        let semaphore = Arc::new(tokio::sync::Semaphore::new(self.max_concurrent));
+        let mut join_set = JoinSet::new();
+        let mut results = Vec::with_capacity(tasks.len());
+        for _ in 0..tasks.len() {
+            results.push(Err(Error::other("Not completed")));
+        }
+
+        // Spawn all tasks with semaphore control
+        for (i, task) in tasks.into_iter().enumerate() {
+            let sem = semaphore.clone();
+            join_set.spawn(async move {
+                let _permit = sem.acquire().await.map_err(|_| Error::other("Semaphore error"))?;
+                let result = task.await;
+                Ok::<(usize, Result<T>), Error>((i, result))
+            });
+        }
+
+        // Collect results
+        while let Some(join_result) = join_set.join_next().await {
+            match join_result {
+                Ok(Ok((index, task_result))) => {
+                    if index < results.len() {
+                        results[index] = task_result;
+                    }
+                }
+                Ok(Err(e)) => {
+                    // Semaphore or other system error - this is rare
+                    tracing::warn!("Batch processor system error: {:?}", e);
+                }
+                Err(join_error) => {
+                    // Task panicked - log but continue
+                    tracing::warn!("Task panicked in batch processor: {:?}", join_error);
+                }
+            }
+        }
+
+        results
+    }
+
+    /// Execute batch with early termination when sufficient successful results are obtained
+    pub async fn execute_batch_with_quorum<T, F>(&self, tasks: Vec<F>, required_successes: usize) -> Result<Vec<T>>
+    where
+        T: Send + 'static,
+        F: Future<Output = Result<T>> + Send + 'static,
+    {
+        let results = self.execute_batch(tasks).await;
+        let mut successes = Vec::new();
+
+        for value in results.into_iter().flatten() {
+            successes.push(value);
+            if successes.len() >= required_successes {
+                return Ok(successes);
+            }
+        }
+
+        if successes.len() >= required_successes {
+            Ok(successes)
+        } else {
+            Err(Error::other(format!(
+                "Insufficient successful results: got {}, needed {}",
+                successes.len(),
+                required_successes
+            )))
+        }
+    }
+}
+
+/// Global batch processor instances
+pub struct GlobalBatchProcessors {
+    read_processor: AsyncBatchProcessor,
+    write_processor: AsyncBatchProcessor,
+    metadata_processor: AsyncBatchProcessor,
+}
+
+impl GlobalBatchProcessors {
+    pub fn new() -> Self {
+        Self {
+            read_processor: AsyncBatchProcessor::new(16),     // Higher concurrency for reads
+            write_processor: AsyncBatchProcessor::new(8),     // Lower concurrency for writes
+            metadata_processor: AsyncBatchProcessor::new(12), // Medium concurrency for metadata
+        }
+    }
+
+    pub fn read_processor(&self) -> &AsyncBatchProcessor {
+        &self.read_processor
+    }
+
+    pub fn write_processor(&self) -> &AsyncBatchProcessor {
+        &self.write_processor
+    }
+
+    pub fn metadata_processor(&self) -> &AsyncBatchProcessor {
+        &self.metadata_processor
+    }
+}
+
+impl Default for GlobalBatchProcessors {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// Global instance
+use std::sync::OnceLock;
+
+static GLOBAL_PROCESSORS: OnceLock<GlobalBatchProcessors> = OnceLock::new();
+
+pub fn get_global_processors() -> &'static GlobalBatchProcessors {
+    GLOBAL_PROCESSORS.get_or_init(GlobalBatchProcessors::new)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[tokio::test]
+    async fn test_batch_processor_basic() {
+        let processor = AsyncBatchProcessor::new(4);
+
+        let tasks: Vec<_> = (0..10)
+            .map(|i| async move {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+                Ok::<i32, Error>(i)
+            })
+            .collect();
+
+        let results = processor.execute_batch(tasks).await;
+        assert_eq!(results.len(), 10);
+
+        // All tasks should succeed
+        for (i, result) in results.iter().enumerate() {
+            assert!(result.is_ok());
+            assert_eq!(result.as_ref().unwrap(), &(i as i32));
+        }
+    }
+
+    #[tokio::test]
+    async fn test_batch_processor_with_errors() {
+        let processor = AsyncBatchProcessor::new(2);
+
+        let tasks: Vec<_> = (0..5)
+            .map(|i| async move {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+                if i % 2 == 0 {
+                    Ok::<i32, Error>(i)
+                } else {
+                    Err(Error::other("Test error"))
+                }
+            })
+            .collect();
+
+        let results = processor.execute_batch(tasks).await;
+        assert_eq!(results.len(), 5);
+
+        // Check results pattern
+        for (i, result) in results.iter().enumerate() {
+            if i % 2 == 0 {
+                assert!(result.is_ok());
+                assert_eq!(result.as_ref().unwrap(), &(i as i32));
+            } else {
+                assert!(result.is_err());
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_batch_processor_quorum() {
+        let processor = AsyncBatchProcessor::new(4);
+
+        let tasks: Vec<_> = (0..10)
+            .map(|i| async move {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+                if i < 3 {
+                    Ok::<i32, Error>(i)
+                } else {
+                    Err(Error::other("Test error"))
+                }
+            })
+            .collect();
+
+        let results = processor.execute_batch_with_quorum(tasks, 2).await;
+        assert!(results.is_ok());
+        let successes = results.unwrap();
+        assert!(successes.len() >= 2);
+    }
+}

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -321,7 +321,7 @@ impl ExpiryState {
         let mut state = GLOBAL_ExpiryState.write().await;
 
         while state.tasks_tx.len() < n {
-            let (tx, rx) = mpsc::channel(10000);
+            let (tx, rx) = mpsc::channel(1000);
             let api = api.clone();
             let rx = Arc::new(tokio::sync::Mutex::new(rx));
             state.tasks_tx.push(tx);
@@ -432,7 +432,7 @@ pub struct TransitionState {
 impl TransitionState {
     #[allow(clippy::new_ret_no_self)]
     pub fn new() -> Arc<Self> {
-        let (tx1, rx1) = bounded(100000);
+        let (tx1, rx1) = bounded(1000);
         let (tx2, rx2) = bounded(1);
         Arc::new(Self {
             transition_tx: tx1,
@@ -467,8 +467,12 @@ impl TransitionState {
     }
 
     pub async fn init(api: Arc<ECStore>) {
-        let mut n = 10; //globalAPIConfig.getTransitionWorkers();
-        let tw = 10; //globalILMConfig.getTransitionWorkers(); 
+        let max_workers = std::env::var("RUSTFS_MAX_TRANSITION_WORKERS")
+            .ok()
+            .and_then(|s| s.parse::<i64>().ok())
+            .unwrap_or_else(|| std::cmp::min(num_cpus::get() as i64, 16));
+        let mut n = max_workers;
+        let tw = 8; //globalILMConfig.getTransitionWorkers(); 
         if tw > 0 {
             n = tw;
         }
@@ -561,8 +565,18 @@ impl TransitionState {
     pub async fn update_workers_inner(api: Arc<ECStore>, n: i64) {
         let mut n = n;
         if n == 0 {
-            n = 100;
+            let max_workers = std::env::var("RUSTFS_MAX_TRANSITION_WORKERS")
+                .ok()
+                .and_then(|s| s.parse::<i64>().ok())
+                .unwrap_or_else(|| std::cmp::min(num_cpus::get() as i64, 16));
+            n = max_workers;
         }
+        // Allow environment override of maximum workers
+        let absolute_max = std::env::var("RUSTFS_ABSOLUTE_MAX_WORKERS")
+            .ok()
+            .and_then(|s| s.parse::<i64>().ok())
+            .unwrap_or(32);
+        n = std::cmp::min(n, absolute_max);
 
         let mut num_workers = GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst);
         while num_workers < n {
@@ -585,7 +599,10 @@ impl TransitionState {
 }
 
 pub async fn init_background_expiry(api: Arc<ECStore>) {
-    let mut workers = num_cpus::get() / 2;
+    let mut workers = std::env::var("RUSTFS_MAX_EXPIRY_WORKERS")
+        .ok()
+        .and_then(|s| s.parse::<usize>().ok())
+        .unwrap_or_else(|| std::cmp::min(num_cpus::get(), 16));
     //globalILMConfig.getExpirationWorkers()
     if let Ok(env_expiration_workers) = env::var("_RUSTFS_ILM_EXPIRATION_WORKERS") {
         if let Ok(num_expirations) = env_expiration_workers.parse::<usize>() {
@@ -594,7 +611,10 @@ pub async fn init_background_expiry(api: Arc<ECStore>) {
     }
 
     if workers == 0 {
-        workers = 100;
+        workers = std::env::var("RUSTFS_DEFAULT_EXPIRY_WORKERS")
+            .ok()
+            .and_then(|s| s.parse::<usize>().ok())
+            .unwrap_or(8);
     }
 
     //let expiry_state = GLOBAL_ExpiryStSate.write().await;

crates/ecstore/src/disk/fs.rs

@@ -12,13 +12,45 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{fs::Metadata, path::Path};
+use std::{
+    fs::Metadata,
+    path::Path,
+    sync::{Arc, OnceLock},
+};
 
 use tokio::{
     fs::{self, File},
     io,
 };
 
+static READONLY_OPTIONS: OnceLock<Arc<fs::OpenOptions>> = OnceLock::new();
+static WRITEONLY_OPTIONS: OnceLock<Arc<fs::OpenOptions>> = OnceLock::new();
+static READWRITE_OPTIONS: OnceLock<Arc<fs::OpenOptions>> = OnceLock::new();
+
+fn get_readonly_options() -> &'static Arc<fs::OpenOptions> {
+    READONLY_OPTIONS.get_or_init(|| {
+        let mut opts = fs::OpenOptions::new();
+        opts.read(true);
+        Arc::new(opts)
+    })
+}
+
+fn get_writeonly_options() -> &'static Arc<fs::OpenOptions> {
+    WRITEONLY_OPTIONS.get_or_init(|| {
+        let mut opts = fs::OpenOptions::new();
+        opts.write(true);
+        Arc::new(opts)
+    })
+}
+
+fn get_readwrite_options() -> &'static Arc<fs::OpenOptions> {
+    READWRITE_OPTIONS.get_or_init(|| {
+        let mut opts = fs::OpenOptions::new();
+        opts.read(true).write(true);
+        Arc::new(opts)
+    })
+}
+
 #[cfg(not(windows))]
 pub fn same_file(f1: &Metadata, f2: &Metadata) -> bool {
     use std::os::unix::fs::MetadataExt;
@@ -84,35 +116,28 @@ pub const O_APPEND: FileMode = 0x00400;
 //     create_new: bool,
 
 pub async fn open_file(path: impl AsRef<Path>, mode: FileMode) -> io::Result<File> {
-    let mut opts = fs::OpenOptions::new();
-
-    match mode & (O_RDONLY | O_WRONLY | O_RDWR) {
-        O_RDONLY => {
-            opts.read(true);
-        }
-        O_WRONLY => {
-            opts.write(true);
-        }
-        O_RDWR => {
-            opts.read(true);
-            opts.write(true);
-        }
-        _ => (),
+    let base_opts = match mode & (O_RDONLY | O_WRONLY | O_RDWR) {
+        O_RDONLY => get_readonly_options(),
+        O_WRONLY => get_writeonly_options(),
+        O_RDWR => get_readwrite_options(),
+        _ => get_readonly_options(),
     };
 
-    if mode & O_CREATE != 0 {
-        opts.create(true);
+    if (mode & (O_CREATE | O_APPEND | O_TRUNC)) != 0 {
+        let mut opts = (**base_opts).clone();
+        if mode & O_CREATE != 0 {
+            opts.create(true);
+        }
+        if mode & O_APPEND != 0 {
+            opts.append(true);
+        }
+        if mode & O_TRUNC != 0 {
+            opts.truncate(true);
+        }
+        opts.open(path.as_ref()).await
+    } else {
+        base_opts.open(path.as_ref()).await
     }
-
-    if mode & O_APPEND != 0 {
-        opts.append(true);
-    }
-
-    if mode & O_TRUNC != 0 {
-        opts.truncate(true);
-    }
-
-    opts.open(path.as_ref()).await
 }
 
 pub async fn access(path: impl AsRef<Path>) -> io::Result<()> {
@@ -121,7 +146,7 @@ pub async fn access(path: impl AsRef<Path>) -> io::Result<()> {
 }
 
 pub fn access_std(path: impl AsRef<Path>) -> io::Result<()> {
-    tokio::task::block_in_place(|| std::fs::metadata(path))?;
+    std::fs::metadata(path)?;
     Ok(())
 }
 
@@ -130,7 +155,7 @@ pub async fn lstat(path: impl AsRef<Path>) -> io::Result<Metadata> {
 }
 
 pub fn lstat_std(path: impl AsRef<Path>) -> io::Result<Metadata> {
-    tokio::task::block_in_place(|| std::fs::metadata(path))
+    std::fs::metadata(path)
 }
 
 pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
@@ -159,26 +184,22 @@ pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
 #[tracing::instrument(level = "debug", skip_all)]
 pub fn remove_std(path: impl AsRef<Path>) -> io::Result<()> {
     let path = path.as_ref();
-    tokio::task::block_in_place(|| {
-        let meta = std::fs::metadata(path)?;
-        if meta.is_dir() {
-            std::fs::remove_dir(path)
-        } else {
-            std::fs::remove_file(path)
-        }
-    })
+    let meta = std::fs::metadata(path)?;
+    if meta.is_dir() {
+        std::fs::remove_dir(path)
+    } else {
+        std::fs::remove_file(path)
+    }
 }
 
 pub fn remove_all_std(path: impl AsRef<Path>) -> io::Result<()> {
     let path = path.as_ref();
-    tokio::task::block_in_place(|| {
-        let meta = std::fs::metadata(path)?;
-        if meta.is_dir() {
-            std::fs::remove_dir_all(path)
-        } else {
-            std::fs::remove_file(path)
-        }
-    })
+    let meta = std::fs::metadata(path)?;
+    if meta.is_dir() {
+        std::fs::remove_dir_all(path)
+    } else {
+        std::fs::remove_file(path)
+    }
 }
 
 pub async fn mkdir(path: impl AsRef<Path>) -> io::Result<()> {
@@ -190,7 +211,7 @@ pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<
 }
 
 pub fn rename_std(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
-    tokio::task::block_in_place(|| std::fs::rename(from, to))
+    std::fs::rename(from, to)
 }
 
 #[tracing::instrument(level = "debug", skip_all)]

crates/ecstore/src/disk/local.rs

@@ -41,18 +41,21 @@ use tokio::time::interval;
 
 use crate::erasure_coding::bitrot_verify;
 use bytes::Bytes;
-use path_absolutize::Absolutize;
+// use path_absolutize::Absolutize;  // Replaced with direct path operations for better performance
+use crate::file_cache::{get_global_file_cache, prefetch_metadata_patterns, read_metadata_cached};
+use parking_lot::RwLock as ParkingLotRwLock;
 use rustfs_filemeta::{
     Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, ObjectPartInfo, Opts, RawFileInfo, UpdateFn,
     get_file_info, read_xl_meta_no_data,
 };
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::os::get_info;
+use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fmt::Debug;
 use std::io::SeekFrom;
-use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 use std::{
     fs::Metadata,
@@ -101,6 +104,9 @@ pub struct LocalDisk {
     pub major: u64,
     pub minor: u64,
     pub nrrequests: u64,
+    // Performance optimization fields
+    path_cache: Arc<ParkingLotRwLock<HashMap<String, PathBuf>>>,
+    current_dir: Arc<OnceLock<PathBuf>>,
     // pub id: Mutex<Option<Uuid>>,
     // pub format_data: Mutex<Vec<u8>>,
     // pub format_file_info: Mutex<Option<Metadata>>,
@@ -130,8 +136,9 @@ impl Debug for LocalDisk {
 impl LocalDisk {
     pub async fn new(ep: &Endpoint, cleanup: bool) -> Result<Self> {
         debug!("Creating local disk");
-        let root = match PathBuf::from(ep.get_file_path()).absolutize() {
-            Ok(path) => path.into_owned(),
+        // Use optimized path resolution instead of absolutize() for better performance
+        let root = match std::fs::canonicalize(ep.get_file_path()) {
+            Ok(path) => path,
             Err(e) => {
                 if e.kind() == ErrorKind::NotFound {
                     return Err(DiskError::VolumeNotFound);
@@ -144,10 +151,8 @@ impl LocalDisk {
             // TODO: 删除 tmp 数据
         }
 
-        let format_path = Path::new(RUSTFS_META_BUCKET)
-            .join(Path::new(super::FORMAT_CONFIG_FILE))
-            .absolutize_virtually(&root)?
-            .into_owned();
+        // Use optimized path resolution instead of absolutize_virtually
+        let format_path = root.join(RUSTFS_META_BUCKET).join(super::FORMAT_CONFIG_FILE);
         debug!("format_path: {:?}", format_path);
         let (format_data, format_meta) = read_file_exists(&format_path).await?;
 
@@ -227,6 +232,8 @@ impl LocalDisk {
             // format_file_info: Mutex::new(format_meta),
             // format_data: Mutex::new(format_data),
             // format_last_check: Mutex::new(format_last_check),
+            path_cache: Arc::new(ParkingLotRwLock::new(HashMap::with_capacity(2048))),
+            current_dir: Arc::new(OnceLock::new()),
             exit_signal: None,
         };
         let (info, _root) = get_disk_info(root).await?;
@@ -351,19 +358,178 @@ impl LocalDisk {
         self.make_volumes(defaults).await
     }
 
+    // Optimized path resolution with caching
     pub fn resolve_abs_path(&self, path: impl AsRef<Path>) -> Result<PathBuf> {
-        Ok(path.as_ref().absolutize_virtually(&self.root)?.into_owned())
+        let path_ref = path.as_ref();
+        let path_str = path_ref.to_string_lossy();
+
+        // Fast cache read
+        {
+            let cache = self.path_cache.read();
+            if let Some(cached_path) = cache.get(path_str.as_ref()) {
+                return Ok(cached_path.clone());
+            }
+        }
+
+        // Calculate absolute path without using path_absolutize for better performance
+        let abs_path = if path_ref.is_absolute() {
+            path_ref.to_path_buf()
+        } else {
+            self.root.join(path_ref)
+        };
+
+        // Normalize path components to avoid filesystem calls
+        let normalized = self.normalize_path_components(&abs_path);
+
+        // Cache the result
+        {
+            let mut cache = self.path_cache.write();
+
+            // Simple cache size control
+            if cache.len() >= 4096 {
+                // Clear half the cache - simple eviction strategy
+                let keys_to_remove: Vec<_> = cache.keys().take(cache.len() / 2).cloned().collect();
+                for key in keys_to_remove {
+                    cache.remove(&key);
+                }
+            }
+
+            cache.insert(path_str.into_owned(), normalized.clone());
+        }
+
+        Ok(normalized)
     }
 
+    // Lightweight path normalization without filesystem calls
+    fn normalize_path_components(&self, path: &Path) -> PathBuf {
+        let mut result = PathBuf::new();
+
+        for component in path.components() {
+            match component {
+                std::path::Component::Normal(name) => {
+                    result.push(name);
+                }
+                std::path::Component::ParentDir => {
+                    result.pop();
+                }
+                std::path::Component::CurDir => {
+                    // Ignore current directory components
+                }
+                std::path::Component::RootDir => {
+                    result.push(component);
+                }
+                std::path::Component::Prefix(_prefix) => {
+                    result.push(component);
+                }
+            }
+        }
+
+        result
+    }
+
+    // Highly optimized object path generation
     pub fn get_object_path(&self, bucket: &str, key: &str) -> Result<PathBuf> {
-        let dir = Path::new(&bucket);
-        let file_path = Path::new(&key);
-        self.resolve_abs_path(dir.join(file_path))
+        // For high-frequency paths, use faster string concatenation
+        let cache_key = if key.is_empty() {
+            bucket.to_string()
+        } else {
+            // Use with_capacity to pre-allocate, reducing memory reallocations
+            let mut path_str = String::with_capacity(bucket.len() + key.len() + 1);
+            path_str.push_str(bucket);
+            path_str.push('/');
+            path_str.push_str(key);
+            path_str
+        };
+
+        // Fast path: directly calculate based on root, avoiding cache lookup overhead for simple cases
+        Ok(self.root.join(&cache_key))
     }
 
     pub fn get_bucket_path(&self, bucket: &str) -> Result<PathBuf> {
-        let dir = Path::new(&bucket);
-        self.resolve_abs_path(dir)
+        Ok(self.root.join(bucket))
+    }
+
+    // Batch path generation with single lock acquisition
+    pub fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
+        let mut results = Vec::with_capacity(requests.len());
+        let mut cache_misses = Vec::new();
+
+        // First attempt to get all paths from cache
+        {
+            let cache = self.path_cache.read();
+            for (i, (bucket, key)) in requests.iter().enumerate() {
+                let cache_key = format!("{}/{}", bucket, key);
+                if let Some(cached_path) = cache.get(&cache_key) {
+                    results.push((i, cached_path.clone()));
+                } else {
+                    cache_misses.push((i, bucket, key, cache_key));
+                }
+            }
+        }
+
+        // Handle cache misses
+        if !cache_misses.is_empty() {
+            let mut new_entries = Vec::new();
+            for (i, _bucket, _key, cache_key) in cache_misses {
+                let path = self.root.join(&cache_key);
+                results.push((i, path.clone()));
+                new_entries.push((cache_key, path));
+            }
+
+            // Batch update cache
+            {
+                let mut cache = self.path_cache.write();
+                for (key, path) in new_entries {
+                    cache.insert(key, path);
+                }
+            }
+        }
+
+        // Sort results back to original order
+        results.sort_by_key(|(i, _)| *i);
+        Ok(results.into_iter().map(|(_, path)| path).collect())
+    }
+
+    // Optimized metadata reading with caching
+    pub async fn read_metadata_cached(&self, path: PathBuf) -> Result<Arc<FileMeta>> {
+        read_metadata_cached(path).await
+    }
+
+    // Smart prefetching for related files
+    pub async fn read_version_with_prefetch(
+        &self,
+        volume: &str,
+        path: &str,
+        version_id: &str,
+        opts: &ReadOptions,
+    ) -> Result<FileInfo> {
+        let file_path = self.get_object_path(volume, path)?;
+
+        // Async prefetch related files, don't block current read
+        if let Some(parent) = file_path.parent() {
+            prefetch_metadata_patterns(parent, &[super::STORAGE_FORMAT_FILE, "part.1", "part.2", "part.meta"]).await;
+        }
+
+        // Main read logic
+        let file_dir = self.get_bucket_path(volume)?;
+        let (data, _) = self.read_raw(volume, file_dir, file_path, opts.read_data).await?;
+
+        get_file_info(&data, volume, path, version_id, FileInfoOpts { data: opts.read_data })
+            .await
+            .map_err(|_e| DiskError::Unexpected)
+    }
+
+    // Batch metadata reading for multiple objects
+    pub async fn read_metadata_batch(&self, requests: Vec<(String, String)>) -> Result<Vec<Option<Arc<FileMeta>>>> {
+        let paths: Vec<PathBuf> = requests
+            .iter()
+            .map(|(bucket, key)| self.get_object_path(bucket, &format!("{}/{}", key, super::STORAGE_FORMAT_FILE)))
+            .collect::<Result<Vec<_>>>()?;
+
+        let cache = get_global_file_cache();
+        let results = cache.get_metadata_batch(paths).await;
+
+        Ok(results.into_iter().map(|r| r.ok()).collect())
     }
 
     // /// Write to the filesystem atomically.
@@ -549,7 +715,15 @@ impl LocalDisk {
     }
 
     async fn read_metadata(&self, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
-        // TODO: support timeout
+        // Try to use cached file content reading for better performance, with safe fallback
+        let path = file_path.as_ref().to_path_buf();
+
+        // First, try the cache
+        if let Ok(bytes) = get_global_file_cache().get_file_content(path.clone()).await {
+            return Ok(bytes.to_vec());
+        }
+
+        // Fallback to direct read if cache fails
         let (data, _) = self.read_metadata_with_dmtime(file_path.as_ref()).await?;
         Ok(data)
     }

crates/ecstore/src/disk/mod.rs

@@ -668,7 +668,7 @@ pub struct VolumeInfo {
     pub created: Option<OffsetDateTime>,
 }
 
-#[derive(Deserialize, Serialize, Debug, Default)]
+#[derive(Deserialize, Serialize, Debug, Default, Clone)]
 pub struct ReadOptions {
     pub incl_free_versions: bool,
     pub read_data: bool,

crates/ecstore/src/endpoints.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use rustfs_utils::{XHost, check_local_server_addr, get_host_ip, get_host_ip_async, is_local_host};
+use rustfs_utils::{XHost, check_local_server_addr, get_host_ip, is_local_host};
 use tracing::{error, instrument, warn};
 
 use crate::{
@@ -248,8 +248,7 @@ impl PoolEndpointList {
                         Ok(ips) => ips,
                         Err(e) => {
                             error!("host {} not found, error:{}", host, e);
-                            get_host_ip_async(host.clone())
-                                .map_err(|e| Error::other(format!("host '{host}' cannot resolve: {e}")))?
+                            return Err(Error::other(format!("host '{host}' cannot resolve: {e}")));
                         }
                     };
                     host_ip_cache.insert(host.clone(), ips);

crates/ecstore/src/file_cache.rs

@@ -0,0 +1,332 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! High-performance file content and metadata caching using moka
+//!
+//! This module provides optimized caching for file operations to reduce
+//! redundant I/O and improve overall system performance.
+
+use super::disk::error::{Error, Result};
+use bytes::Bytes;
+use moka::future::Cache;
+use rustfs_filemeta::FileMeta;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+pub struct OptimizedFileCache {
+    // Use moka as high-performance async cache
+    metadata_cache: Cache<PathBuf, Arc<FileMeta>>,
+    file_content_cache: Cache<PathBuf, Bytes>,
+    // Performance monitoring
+    cache_hits: std::sync::atomic::AtomicU64,
+    cache_misses: std::sync::atomic::AtomicU64,
+}
+
+impl OptimizedFileCache {
+    pub fn new() -> Self {
+        Self {
+            metadata_cache: Cache::builder()
+                .max_capacity(2048)
+                .time_to_live(Duration::from_secs(300)) // 5 minutes TTL
+                .time_to_idle(Duration::from_secs(60)) // 1 minute idle
+                .build(),
+
+            file_content_cache: Cache::builder()
+                .max_capacity(512) // Smaller file content cache
+                .time_to_live(Duration::from_secs(120))
+                .weigher(|_key: &PathBuf, value: &Bytes| value.len() as u32)
+                .build(),
+
+            cache_hits: std::sync::atomic::AtomicU64::new(0),
+            cache_misses: std::sync::atomic::AtomicU64::new(0),
+        }
+    }
+
+    pub async fn get_metadata(&self, path: PathBuf) -> Result<Arc<FileMeta>> {
+        if let Some(cached) = self.metadata_cache.get(&path).await {
+            self.cache_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            return Ok(cached);
+        }
+
+        self.cache_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+
+        // Cache miss, read file
+        let data = tokio::fs::read(&path)
+            .await
+            .map_err(|e| Error::other(format!("Read metadata failed: {}", e)))?;
+
+        let mut meta = FileMeta::default();
+        meta.unmarshal_msg(&data)?;
+
+        let arc_meta = Arc::new(meta);
+        self.metadata_cache.insert(path, arc_meta.clone()).await;
+
+        Ok(arc_meta)
+    }
+
+    pub async fn get_file_content(&self, path: PathBuf) -> Result<Bytes> {
+        if let Some(cached) = self.file_content_cache.get(&path).await {
+            self.cache_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            return Ok(cached);
+        }
+
+        self.cache_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+
+        let data = tokio::fs::read(&path)
+            .await
+            .map_err(|e| Error::other(format!("Read file failed: {}", e)))?;
+
+        let bytes = Bytes::from(data);
+        self.file_content_cache.insert(path, bytes.clone()).await;
+
+        Ok(bytes)
+    }
+
+    // Prefetch related files
+    pub async fn prefetch_related(&self, base_path: &Path, patterns: &[&str]) {
+        let mut prefetch_tasks = Vec::new();
+
+        for pattern in patterns {
+            let path = base_path.join(pattern);
+            if tokio::fs::metadata(&path).await.is_ok() {
+                let cache = self.clone();
+                let path_clone = path.clone();
+                prefetch_tasks.push(async move {
+                    let _ = cache.get_metadata(path_clone).await;
+                });
+            }
+        }
+
+        // Parallel prefetch, don't wait for completion
+        if !prefetch_tasks.is_empty() {
+            tokio::spawn(async move {
+                futures::future::join_all(prefetch_tasks).await;
+            });
+        }
+    }
+
+    // Batch metadata reading with deduplication
+    pub async fn get_metadata_batch(
+        &self,
+        paths: Vec<PathBuf>,
+    ) -> Vec<std::result::Result<Arc<FileMeta>, rustfs_filemeta::Error>> {
+        let mut results = Vec::with_capacity(paths.len());
+        let mut cache_futures = Vec::new();
+
+        // First, attempt to get from cache
+        for (i, path) in paths.iter().enumerate() {
+            if let Some(cached) = self.metadata_cache.get(path).await {
+                results.push((i, Ok(cached)));
+                self.cache_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            } else {
+                cache_futures.push((i, path.clone()));
+            }
+        }
+
+        // For cache misses, read from filesystem
+        if !cache_futures.is_empty() {
+            let mut fs_results = Vec::new();
+
+            for (i, path) in cache_futures {
+                self.cache_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+
+                match tokio::fs::read(&path).await {
+                    Ok(data) => {
+                        let mut meta = FileMeta::default();
+                        match meta.unmarshal_msg(&data) {
+                            Ok(_) => {
+                                let arc_meta = Arc::new(meta);
+                                self.metadata_cache.insert(path, arc_meta.clone()).await;
+                                fs_results.push((i, Ok(arc_meta)));
+                            }
+                            Err(e) => {
+                                fs_results.push((i, Err(e)));
+                            }
+                        }
+                    }
+                    Err(_e) => {
+                        fs_results.push((i, Err(rustfs_filemeta::Error::Unexpected)));
+                    }
+                }
+            }
+
+            results.extend(fs_results);
+        }
+
+        // Sort results back to original order
+        results.sort_by_key(|(i, _)| *i);
+        results.into_iter().map(|(_, result)| result).collect()
+    }
+
+    // Invalidate cache entries for a path
+    pub async fn invalidate(&self, path: &Path) {
+        self.metadata_cache.remove(path).await;
+        self.file_content_cache.remove(path).await;
+    }
+
+    // Get cache statistics
+    pub fn get_stats(&self) -> FileCacheStats {
+        let hits = self.cache_hits.load(std::sync::atomic::Ordering::Relaxed);
+        let misses = self.cache_misses.load(std::sync::atomic::Ordering::Relaxed);
+        let hit_rate = if hits + misses > 0 {
+            (hits as f64 / (hits + misses) as f64) * 100.0
+        } else {
+            0.0
+        };
+
+        FileCacheStats {
+            metadata_cache_size: self.metadata_cache.entry_count(),
+            content_cache_size: self.file_content_cache.entry_count(),
+            cache_hits: hits,
+            cache_misses: misses,
+            hit_rate,
+            total_weight: 0, // Simplified for compatibility
+        }
+    }
+
+    // Clear all caches
+    pub async fn clear(&self) {
+        self.metadata_cache.invalidate_all();
+        self.file_content_cache.invalidate_all();
+
+        // Wait for invalidation to complete
+        self.metadata_cache.run_pending_tasks().await;
+        self.file_content_cache.run_pending_tasks().await;
+    }
+}
+
+impl Clone for OptimizedFileCache {
+    fn clone(&self) -> Self {
+        Self {
+            metadata_cache: self.metadata_cache.clone(),
+            file_content_cache: self.file_content_cache.clone(),
+            cache_hits: std::sync::atomic::AtomicU64::new(self.cache_hits.load(std::sync::atomic::Ordering::Relaxed)),
+            cache_misses: std::sync::atomic::AtomicU64::new(self.cache_misses.load(std::sync::atomic::Ordering::Relaxed)),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct FileCacheStats {
+    pub metadata_cache_size: u64,
+    pub content_cache_size: u64,
+    pub cache_hits: u64,
+    pub cache_misses: u64,
+    pub hit_rate: f64,
+    pub total_weight: u64,
+}
+
+impl Default for OptimizedFileCache {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+// Global cache instance
+use std::sync::OnceLock;
+
+static GLOBAL_FILE_CACHE: OnceLock<OptimizedFileCache> = OnceLock::new();
+
+pub fn get_global_file_cache() -> &'static OptimizedFileCache {
+    GLOBAL_FILE_CACHE.get_or_init(OptimizedFileCache::new)
+}
+
+// Utility functions for common operations
+pub async fn read_metadata_cached(path: PathBuf) -> Result<Arc<FileMeta>> {
+    get_global_file_cache().get_metadata(path).await
+}
+
+pub async fn read_file_content_cached(path: PathBuf) -> Result<Bytes> {
+    get_global_file_cache().get_file_content(path).await
+}
+
+pub async fn prefetch_metadata_patterns(base_path: &Path, patterns: &[&str]) {
+    get_global_file_cache().prefetch_related(base_path, patterns).await;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+    use tempfile::tempdir;
+
+    #[tokio::test]
+    async fn test_file_cache_basic() {
+        let cache = OptimizedFileCache::new();
+
+        // Create a temporary file
+        let dir = tempdir().unwrap();
+        let file_path = dir.path().join("test.txt");
+        let mut file = std::fs::File::create(&file_path).unwrap();
+        writeln!(file, "test content").unwrap();
+        drop(file);
+
+        // First read should be cache miss
+        let content1 = cache.get_file_content(file_path.clone()).await.unwrap();
+        assert_eq!(content1, Bytes::from("test content\n"));
+
+        // Second read should be cache hit
+        let content2 = cache.get_file_content(file_path.clone()).await.unwrap();
+        assert_eq!(content2, content1);
+
+        let stats = cache.get_stats();
+        assert!(stats.cache_hits > 0);
+        assert!(stats.cache_misses > 0);
+    }
+
+    #[tokio::test]
+    async fn test_metadata_batch_read() {
+        let cache = OptimizedFileCache::new();
+
+        // Create test files
+        let dir = tempdir().unwrap();
+        let mut paths = Vec::new();
+
+        for i in 0..5 {
+            let file_path = dir.path().join(format!("test_{}.txt", i));
+            let mut file = std::fs::File::create(&file_path).unwrap();
+            writeln!(file, "content {}", i).unwrap();
+            paths.push(file_path);
+        }
+
+        // Note: This test would need actual FileMeta files to work properly
+        // For now, we just test that the function runs without errors
+        let results = cache.get_metadata_batch(paths).await;
+        assert_eq!(results.len(), 5);
+    }
+
+    #[tokio::test]
+    async fn test_cache_invalidation() {
+        let cache = OptimizedFileCache::new();
+
+        let dir = tempdir().unwrap();
+        let file_path = dir.path().join("test.txt");
+        let mut file = std::fs::File::create(&file_path).unwrap();
+        writeln!(file, "test content").unwrap();
+        drop(file);
+
+        // Read file to populate cache
+        let _ = cache.get_file_content(file_path.clone()).await.unwrap();
+
+        // Invalidate cache
+        cache.invalidate(&file_path).await;
+
+        // Next read should be cache miss again
+        let _ = cache.get_file_content(file_path.clone()).await.unwrap();
+
+        let stats = cache.get_stats();
+        assert!(stats.cache_misses >= 2);
+    }
+}

crates/ecstore/src/lib.rs

@@ -16,6 +16,7 @@
 extern crate core;
 
 pub mod admin_server_info;
+pub mod batch_processor;
 pub mod bitrot;
 pub mod bucket;
 pub mod cache_value;
@@ -29,6 +30,7 @@ pub mod disks_layout;
 pub mod endpoints;
 pub mod erasure_coding;
 pub mod error;
+pub mod file_cache;
 pub mod global;
 pub mod lock_utils;
 pub mod metrics_realtime;

crates/ecstore/src/set_disk.rs

@@ -15,6 +15,7 @@
 #![allow(unused_imports)]
 #![allow(unused_variables)]
 
+use crate::batch_processor::{AsyncBatchProcessor, get_global_processors};
 use crate::bitrot::{create_bitrot_reader, create_bitrot_writer};
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::versioning::VersioningApi;
@@ -232,7 +233,10 @@ impl SetDisks {
             });
         }
 
-        let results = join_all(futures).await;
+        // Use optimized batch processor for disk info retrieval
+        let processor = get_global_processors().metadata_processor();
+        let results = processor.execute_batch(futures).await;
+
         for result in results {
             match result {
                 Ok(res) => {
@@ -507,21 +511,28 @@ impl SetDisks {
 
     #[tracing::instrument(skip(disks))]
     async fn cleanup_multipart_path(disks: &[Option<DiskStore>], paths: &[String]) {
-        let mut futures = Vec::with_capacity(disks.len());
-
         let mut errs = Vec::with_capacity(disks.len());
 
-        for disk in disks.iter() {
-            futures.push(async move {
-                if let Some(disk) = disk {
-                    disk.delete_paths(RUSTFS_META_MULTIPART_BUCKET, paths).await
-                } else {
-                    Err(DiskError::DiskNotFound)
+        // Use improved simple batch processor instead of join_all for better performance
+        let processor = get_global_processors().write_processor();
+
+        let tasks: Vec<_> = disks
+            .iter()
+            .map(|disk| {
+                let disk = disk.clone();
+                let paths = paths.to_vec();
+
+                async move {
+                    if let Some(disk) = disk {
+                        disk.delete_paths(RUSTFS_META_MULTIPART_BUCKET, &paths).await
+                    } else {
+                        Err(DiskError::DiskNotFound)
+                    }
                 }
             })
-        }
+            .collect();
 
-        let results = join_all(futures).await;
+        let results = processor.execute_batch(tasks).await;
         for result in results {
             match result {
                 Ok(_) => {
@@ -545,21 +556,32 @@ impl SetDisks {
         part_numbers: &[usize],
         read_quorum: usize,
     ) -> disk::error::Result<Vec<ObjectPartInfo>> {
-        let mut futures = Vec::with_capacity(disks.len());
-        for (i, disk) in disks.iter().enumerate() {
-            futures.push(async move {
-                if let Some(disk) = disk {
-                    disk.read_parts(bucket, part_meta_paths).await
-                } else {
-                    Err(DiskError::DiskNotFound)
-                }
-            });
-        }
-
         let mut errs = Vec::with_capacity(disks.len());
         let mut object_parts = Vec::with_capacity(disks.len());
 
-        let results = join_all(futures).await;
+        // Use batch processor for better performance
+        let processor = get_global_processors().read_processor();
+        let bucket = bucket.to_string();
+        let part_meta_paths = part_meta_paths.to_vec();
+
+        let tasks: Vec<_> = disks
+            .iter()
+            .map(|disk| {
+                let disk = disk.clone();
+                let bucket = bucket.clone();
+                let part_meta_paths = part_meta_paths.clone();
+
+                async move {
+                    if let Some(disk) = disk {
+                        disk.read_parts(&bucket, &part_meta_paths).await
+                    } else {
+                        Err(DiskError::DiskNotFound)
+                    }
+                }
+            })
+            .collect();
+
+        let results = processor.execute_batch(tasks).await;
         for result in results {
             match result {
                 Ok(res) => {
@@ -1369,22 +1391,71 @@ impl SetDisks {
             })
         });
 
+        // Wait for all tasks to complete
         let results = join_all(futures).await;
+
         for result in results {
-            match result? {
-                Ok(res) => {
-                    ress.push(res);
-                    errors.push(None);
-                }
-                Err(e) => {
+            match result {
+                Ok(res) => match res {
+                    Ok(file_info) => {
+                        ress.push(file_info);
+                        errors.push(None);
+                    }
+                    Err(e) => {
+                        ress.push(FileInfo::default());
+                        errors.push(Some(e));
+                    }
+                },
+                Err(_) => {
                     ress.push(FileInfo::default());
-                    errors.push(Some(e));
+                    errors.push(Some(DiskError::Unexpected));
                 }
             }
         }
         Ok((ress, errors))
     }
 
+    // Optimized version using batch processor with quorum support
+    pub async fn read_version_optimized(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: &str,
+        opts: &ReadOptions,
+    ) -> Result<Vec<rustfs_filemeta::FileInfo>> {
+        // Use existing disk selection logic
+        let disks = self.disks.read().await;
+        let required_reads = self.format.erasure.sets.len();
+
+        // Clone parameters outside the closure to avoid lifetime issues
+        let bucket = bucket.to_string();
+        let object = object.to_string();
+        let version_id = version_id.to_string();
+        let opts = opts.clone();
+
+        let processor = get_global_processors().read_processor();
+        let tasks: Vec<_> = disks
+            .iter()
+            .take(required_reads + 2) // Read a few extra for reliability
+            .filter_map(|disk| {
+                disk.as_ref().map(|d| {
+                    let disk = d.clone();
+                    let bucket = bucket.clone();
+                    let object = object.clone();
+                    let version_id = version_id.clone();
+                    let opts = opts.clone();
+
+                    async move { disk.read_version(&bucket, &bucket, &object, &version_id, &opts).await }
+                })
+            })
+            .collect();
+
+        match processor.execute_batch_with_quorum(tasks, required_reads).await {
+            Ok(results) => Ok(results),
+            Err(_) => Err(DiskError::FileNotFound.into()), // Use existing error type
+        }
+    }
+
     async fn read_all_xl(
         disks: &[Option<DiskStore>],
         bucket: &str,
@@ -1403,10 +1474,11 @@ impl SetDisks {
         object: &str,
         read_data: bool,
     ) -> (Vec<Option<RawFileInfo>>, Vec<Option<DiskError>>) {
-        let mut futures = Vec::with_capacity(disks.len());
         let mut ress = Vec::with_capacity(disks.len());
         let mut errors = Vec::with_capacity(disks.len());
 
+        let mut futures = Vec::with_capacity(disks.len());
+
         for disk in disks.iter() {
             futures.push(async move {
                 if let Some(disk) = disk {

crates/ecstore/src/tier/tier.rs

@@ -302,17 +302,19 @@ impl TierConfigMgr {
     }
 
     pub async fn get_driver<'a>(&'a mut self, tier_name: &str) -> std::result::Result<&'a WarmBackendImpl, AdminError> {
-        Ok(match self.driver_cache.entry(tier_name.to_string()) {
-            Entry::Occupied(e) => e.into_mut(),
-            Entry::Vacant(e) => {
-                let t = self.tiers.get(tier_name);
-                if t.is_none() {
-                    return Err(ERR_TIER_NOT_FOUND.clone());
-                }
-                let d = new_warm_backend(t.expect("err"), false).await?;
-                e.insert(d)
-            }
-        })
+        // Return cached driver if present
+        if self.driver_cache.contains_key(tier_name) {
+            return Ok(self.driver_cache.get(tier_name).unwrap());
+        }
+
+        // Get tier configuration and create new driver
+        let tier_config = self.tiers.get(tier_name).ok_or_else(|| ERR_TIER_NOT_FOUND.clone())?;
+
+        let driver = new_warm_backend(tier_config, false).await?;
+
+        // Insert and return reference
+        self.driver_cache.insert(tier_name.to_string(), driver);
+        Ok(self.driver_cache.get(tier_name).unwrap())
     }
 
     pub async fn reload(&mut self, api: Arc<ECStore>) -> std::result::Result<(), std::io::Error> {

crates/lock/Cargo.toml

@@ -47,3 +47,4 @@ smallvec = "1.11"
 smartstring = "1.0"
 crossbeam-queue = "0.3"
 heapless = "0.8"
+pprof.workspace = true

crates/policy/src/policy/doc.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize, de::Error};
 use time::OffsetDateTime;
 
 use super::Policy;
@@ -59,15 +59,17 @@ impl TryFrom<Vec<u8>> for PolicyDoc {
     type Error = serde_json::Error;
 
     fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> {
-        match serde_json::from_slice::<PolicyDoc>(&value) {
-            Ok(res) => Ok(res),
-            Err(err) => match serde_json::from_slice::<Policy>(&value) {
-                Ok(res2) => Ok(Self {
-                    policy: res2,
-                    ..Default::default()
-                }),
-                Err(_) => Err(err),
-            },
+        // Try to parse as PolicyDoc first
+        if let Ok(policy_doc) = serde_json::from_slice::<PolicyDoc>(&value) {
+            return Ok(policy_doc);
         }
+
+        // Fall back to parsing as Policy and wrap in PolicyDoc
+        serde_json::from_slice::<Policy>(&value)
+            .map(|policy| Self {
+                policy,
+                ..Default::default()
+            })
+            .map_err(|_| serde_json::Error::custom("Failed to parse as PolicyDoc or Policy".to_string()))
     }
 }

crates/utils/src/net.rs

@@ -16,17 +16,40 @@ use bytes::Bytes;
 use futures::pin_mut;
 use futures::{Stream, StreamExt};
 use std::net::Ipv6Addr;
-use std::sync::LazyLock;
+use std::sync::{LazyLock, Mutex};
 use std::{
-    collections::HashSet,
+    collections::{HashMap, HashSet},
     fmt::Display,
     net::{IpAddr, SocketAddr, TcpListener, ToSocketAddrs},
+    time::{Duration, Instant},
 };
 use transform_stream::AsyncTryStream;
 use url::{Host, Url};
 
 static LOCAL_IPS: LazyLock<Vec<IpAddr>> = LazyLock::new(|| must_get_local_ips().unwrap());
 
+#[derive(Debug, Clone)]
+struct DnsCacheEntry {
+    ips: HashSet<IpAddr>,
+    cached_at: Instant,
+}
+
+impl DnsCacheEntry {
+    fn new(ips: HashSet<IpAddr>) -> Self {
+        Self {
+            ips,
+            cached_at: Instant::now(),
+        }
+    }
+
+    fn is_expired(&self, ttl: Duration) -> bool {
+        self.cached_at.elapsed() > ttl
+    }
+}
+
+static DNS_CACHE: LazyLock<Mutex<HashMap<String, DnsCacheEntry>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+const DNS_CACHE_TTL: Duration = Duration::from_secs(300); // 5 minutes
+
 /// helper for validating if the provided arg is an ip address.
 pub fn is_socket_addr(addr: &str) -> bool {
     // TODO IPv6 zone information?
@@ -87,19 +110,19 @@ pub fn is_local_host(host: Host<&str>, port: u16, local_port: u16) -> std::io::R
 }
 
 /// returns IP address of given host using layered DNS resolution.
-pub fn get_host_ip_async(host: Host<&str>) -> std::io::Result<HashSet<IpAddr>> {
+///
+/// This is the async version of `get_host_ip()` that provides enhanced DNS resolution
+/// with Kubernetes support when the "net" feature is enabled.
+pub async fn get_host_ip_async(host: Host<&str>) -> std::io::Result<HashSet<IpAddr>> {
     match host {
         Host::Domain(domain) => {
             #[cfg(feature = "net")]
             {
                 use crate::dns_resolver::resolve_domain;
-                let handle = tokio::runtime::Handle::current();
-                handle.block_on(async {
-                    match resolve_domain(domain).await {
-                        Ok(ips) => Ok(ips.into_iter().collect()),
-                        Err(e) => Err(std::io::Error::other(format!("DNS resolution failed: {}", e))),
-                    }
-                })
+                match resolve_domain(domain).await {
+                    Ok(ips) => Ok(ips.into_iter().collect()),
+                    Err(e) => Err(std::io::Error::other(format!("DNS resolution failed: {}", e))),
+                }
             }
             #[cfg(not(feature = "net"))]
             {
@@ -128,17 +151,41 @@ pub fn get_host_ip_async(host: Host<&str>) -> std::io::Result<HashSet<IpAddr>> {
 
 /// returns IP address of given host using standard resolution.
 ///
-/// **Note**: This function uses standard library DNS resolution.
+/// **Note**: This function uses standard library DNS resolution with caching.
 /// For enhanced DNS resolution with Kubernetes support, use `get_host_ip_async()`.
 pub fn get_host_ip(host: Host<&str>) -> std::io::Result<HashSet<IpAddr>> {
     match host {
-        Host::Domain(domain) => match (domain, 0)
-            .to_socket_addrs()
-            .map(|v| v.map(|v| v.ip()).collect::<HashSet<_>>())
-        {
-            Ok(ips) => Ok(ips),
-            Err(err) => Err(std::io::Error::other(err)),
-        },
+        Host::Domain(domain) => {
+            // Check cache first
+            if let Ok(mut cache) = DNS_CACHE.lock() {
+                if let Some(entry) = cache.get(domain) {
+                    if !entry.is_expired(DNS_CACHE_TTL) {
+                        return Ok(entry.ips.clone());
+                    }
+                    // Remove expired entry
+                    cache.remove(domain);
+                }
+            }
+
+            // Perform DNS resolution
+            match (domain, 0)
+                .to_socket_addrs()
+                .map(|v| v.map(|v| v.ip()).collect::<HashSet<_>>())
+            {
+                Ok(ips) => {
+                    // Cache the result
+                    if let Ok(mut cache) = DNS_CACHE.lock() {
+                        cache.insert(domain.to_string(), DnsCacheEntry::new(ips.clone()));
+                        // Limit cache size to prevent memory bloat
+                        if cache.len() > 1000 {
+                            cache.retain(|_, v| !v.is_expired(DNS_CACHE_TTL));
+                        }
+                    }
+                    Ok(ips)
+                }
+                Err(err) => Err(std::io::Error::other(err)),
+            }
+        }
         Host::Ipv4(ip) => {
             let mut set = HashSet::with_capacity(1);
             set.insert(IpAddr::V4(ip));