diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 809d0a3f5..4475c2997 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -31,7 +31,7 @@ HTTP request → storage/ecfs (erasure coding, encryption, checksums) → ecstore (disk pool selection, data distribution) → rio (reader pipeline: encrypt → compress → hash → write) - → io-core (zero-copy I/O, buffer pool, direct I/O) + → io-core (buffer pool, storage profiling, admission control) → local disk / remote disk via RPC ``` @@ -55,7 +55,7 @@ rustfs/ # Workspace root (virtual manifest) ├── crates/ # library crates (authoritative list: Cargo.toml [workspace].members) │ ├── ecstore/ # Erasure-coded storage engine │ ├── rio/ # Reader I/O pipeline (encrypt, compress, hash) -│ ├── io-core/ # Zero-copy I/O, scheduling, buffer pool +│ ├── io-core/ # Buffer pool, storage profiling, admission control │ ├── io-metrics/ # I/O metrics collection │ ├── common/ # Shared runtime state, globals, data usage types │ ├── config/ # Configuration types and parsing @@ -302,7 +302,7 @@ The binary (`main.rs`) boots in this order: │ │ │ ┌─────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ │ ecstore │ │ rio │ │ io-core │ - │ (core) │ │ (readers) │ │ (zero-copy) │ + │ (core) │ │ (readers) │ │ (buffers) │ └─────┬──────┘ └─────────────┘ └─────────────┘ │ ┌─────┬──┼──┬─────┬──────┐ @@ -314,7 +314,7 @@ The binary (`main.rs`) boots in this order: - **"Where does S3 PutObject go?"** `server/` routes → `app/object_usecase` validates → `storage/ecfs` encodes → - `ecstore` distributes → `rio` encrypts/compresses → `io-core` writes + `ecstore` distributes → `rio` encrypts/compresses → `io-core` supplies buffers - **"Where are bucket policies enforced?"** `app/bucket_usecase` calls into `crates/policy/` diff --git a/Cargo.lock b/Cargo.lock index 83749d574..0fb09e78e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9613,7 +9613,6 @@ version = "1.0.0-rc.2" dependencies = [ "bytes", "hotpath", - "memmap2", "rustfs-io-metrics", "thiserror 2.0.20", "tokio", diff --git a/crates/io-core/CHANGELOG.md b/crates/io-core/CHANGELOG.md index 2ed1235c4..b60a494a1 100644 --- a/crates/io-core/CHANGELOG.md +++ b/crates/io-core/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Removed +#### rustfs-io-core +- **Zero-consumer modules** (added in 0.0.5): `reader`, `writer`, `bufreader_optimizer`, `shared_memory`, `direct_io`, `timeout_wrapper`, `io_priority_queue`, and `scheduler` had no caller in the workspace and were removed (rustfs/backlog#1824). The scheduling algorithm and the request timeout wrapper that RustFS actually runs live in `rustfs/src/storage/`; this crate keeps the config shapes they project into. `OperationProgress` moved to the new `progress` module and is still exported as `rustfs_io_core::OperationProgress`. + #### rustfs-io-metrics - **Unified configuration** (added in 0.0.5): the zero-consumer `IoConfig`, `CacheSettings`, `IoSchedulerSettings`, `BackpressureSettings`, `TimeoutSettings`, `DeadlockDetectionSettings` types and their `DEFAULT_*` constants were removed (rustfs/rustfs#6008); rustfs-io-core's `IoSchedulerConfig`/`BackpressureConfig` remain the canonical configuration types. diff --git a/crates/io-core/Cargo.toml b/crates/io-core/Cargo.toml index 9c309633b..f3118c19a 100644 --- a/crates/io-core/Cargo.toml +++ b/crates/io-core/Cargo.toml @@ -20,8 +20,8 @@ license.workspace = true repository.workspace = true rust-version.workspace = true homepage.workspace = true -description = "Buffered I/O reader and writer implementations for RustFS (mmap-then-copy, aligned pread)" -keywords = ["io", "reader", "writer", "rustfs", "mmap"] +description = "Shared I/O primitives for RustFS (buffer pool, storage profiling, backpressure, deadlock detection)" +keywords = ["io", "buffer", "pool", "rustfs", "backpressure"] categories = ["development-tools", "filesystem"] [lints] @@ -38,7 +38,6 @@ hotpath.workspace = true bytes = { workspace = true, features = ["serde"] } thiserror = { workspace = true } tokio = { workspace = true, features = ["io-util", "fs", "sync", "rt-multi-thread"] } -memmap2 = { workspace = true } rustfs-io-metrics = { workspace = true } tracing = { workspace = true } diff --git a/crates/io-core/README.md b/crates/io-core/README.md index ee5f7dd50..a372dab1b 100644 --- a/crates/io-core/README.md +++ b/crates/io-core/README.md @@ -23,67 +23,20 @@ ## Overview -**rustfs-io-core** is the core I/O scheduling module for [RustFS](https://rustfs.com), a distributed object storage system. It provides: +**rustfs-io-core** holds the shared I/O primitives for [RustFS](https://rustfs.com), a distributed object storage system. It provides: -- **I/O Scheduler**: Adaptive buffer size calculation and load management -- **Priority Queue**: Request priority scheduling with starvation prevention +- **Buffer Pool**: Tiered `BytesPool` for buffer reuse +- **Storage Profiling**: Storage-media and access-pattern model (`io_profile`) +- **Scheduler Configuration**: The `IoSchedulerConfig` / `IoPriorityQueueConfig` shapes the storage layer projects into - **Backpressure Control**: System overload protection with graceful degradation - **Deadlock Detection**: Wait-for graph based deadlock detection algorithm - **Lock Optimizer**: Adaptive spin lock optimization -- **Timeout Wrapper**: Dynamic timeout calculation and operation progress tracking +- **Progress Tracking**: Byte progress and staleness for long-running operations + +The scheduling algorithm itself lives in `rustfs/src/storage/concurrency/io_schedule.rs`; this crate carries the configuration shapes it projects into, not a second implementation. ## Features -### I/O Scheduler - -Adaptive I/O scheduling with dynamic buffer size calculation based on file size, access pattern, and system load: - -```rust -use rustfs_io_core::{IoScheduler, IoSchedulerConfig, IoLoadLevel}; -use rustfs_io_core::io_profile::{StorageMedia, AccessPattern}; - -// Create scheduler -let config = IoSchedulerConfig { - max_concurrent_reads: 64, - base_buffer_size: 64 * 1024, // 64 KB - max_buffer_size: 1024 * 1024, // 1 MB - ..Default::default() -}; -let scheduler = IoScheduler::new(config); - -// Calculate optimal buffer size -let buffer_size = calculate_optimal_buffer_size( - 10 * 1024 * 1024, // 10 MB file - 64 * 1024, // base buffer - true, // sequential access - 4, // concurrent requests - StorageMedia::Ssd, - IoLoadLevel::Low, -); -``` - -### Priority Queue - -Priority queue with starvation prevention: - -```rust -use rustfs_io_core::{IoPriorityQueue, IoPriority, IoQueueStatus}; - -let queue = IoPriorityQueue::<()>::new(100); - -// Enqueue request -let request_id = queue.enqueue(IoPriority::High, (), 1024); - -// Dequeue request -if let Some((priority, data)) = queue.dequeue() { - println!("Processing priority {:?} request", priority); -} - -// Check queue status -let status = queue.status(); -println!("High priority waiting: {}", status.high_priority_waiting); -``` - ### Backpressure Control System overload protection: @@ -148,71 +101,23 @@ let stats = optimizer.stats(); println!("Locks acquired: {}", stats.total_acquired()); ``` -### Timeout Wrapper +### Progress Tracking -Dynamic timeout calculation: +Byte progress and staleness for long-running operations: ```rust -use rustfs_io_core::{RequestTimeoutWrapper, TimeoutConfig}; +use rustfs_io_core::OperationProgress; use std::time::Duration; -let config = TimeoutConfig { - base_timeout: Duration::from_secs(5), - timeout_per_mb: Duration::from_millis(100), - max_timeout: Duration::from_secs(300), - ..Default::default() -}; -let wrapper = RequestTimeoutWrapper::new(config); +let progress = OperationProgress::new(Some(1000), Duration::from_secs(5)); -// Calculate operation timeout -let timeout = wrapper.calculate_timeout(10 * 1024 * 1024); // 10 MB -``` - -## Buffer Size Calculation - -Multiple buffer size calculation functions are provided: - -```rust -use rustfs_io_core::{ - get_concurrency_aware_buffer_size, - get_advanced_buffer_size, - get_buffer_size_for_media, - calculate_optimal_buffer_size, - KI_B, MI_B, -}; -use rustfs_io_core::io_profile::StorageMedia; - -// Basic calculation -let size1 = get_concurrency_aware_buffer_size(1024 * 1024, 64 * 1024); - -// Advanced calculation (considering access pattern) -let size2 = get_advanced_buffer_size(10 * 1024 * 1024, 64 * 1024, true); - -// Media type optimization -let size3 = get_buffer_size_for_media(64 * 1024, StorageMedia::Ssd); - -// Comprehensive calculation -let size4 = calculate_optimal_buffer_size( - 100 * 1024 * 1024, // 100 MB file - 64 * 1024, // base buffer - true, // sequential access - 4, // concurrent requests - StorageMedia::Nvme, - IoLoadLevel::Low, -); +progress.update(500); +assert_eq!(progress.progress_percent(), Some(50.0)); +assert!(!progress.is_stale()); ``` ## Configuration -### Environment Variables - -| Variable | Description | Default | -|----------|-------------|---------| -| `RUSTFS_MAX_CONCURRENT_READS` | Max concurrent reads | 64 | -| `RUSTFS_BASE_BUFFER_SIZE` | Base buffer size | 65536 | -| `RUSTFS_MAX_BUFFER_SIZE` | Max buffer size | 1048576 | -| `RUSTFS_IO_TIMEOUT_SECS` | I/O timeout seconds | 30 | - ### Code Configuration ```rust @@ -240,12 +145,11 @@ rustfs-io-core/ ├── src/ │ ├── lib.rs # Module entry │ ├── config.rs # Configuration types -│ ├── scheduler.rs # I/O scheduler -│ ├── io_priority_queue.rs # Priority queue +│ ├── pool.rs # Tiered buffer pool │ ├── backpressure.rs # Backpressure control │ ├── deadlock_detector.rs # Deadlock detection │ ├── lock_optimizer.rs # Lock optimization -│ ├── timeout_wrapper.rs # Timeout wrapper +│ ├── progress.rs # Operation progress tracking │ └── io_profile.rs # I/O profile └── Cargo.toml ``` @@ -254,21 +158,15 @@ rustfs-io-core/ ```bash # Run all tests -cargo test --package rustfs-io-core +cargo nextest run --package rustfs-io-core # Run specific tests -cargo test --package rustfs-io-core --lib scheduler - -# Run benchmarks -cargo bench --package rustfs-io-core +cargo nextest run --package rustfs-io-core -E 'test(backpressure)' ``` ## Documentation - [API Documentation](https://docs.rs/rustfs-io-core) -- [I/O Scheduler Design](./docs/scheduler-design.md) -- [Backpressure Control Design](./docs/backpressure-design.md) -- [Deadlock Detection Algorithm](./docs/deadlock-detection.md) ## Related Modules diff --git a/crates/io-core/README_zh.md b/crates/io-core/README_zh.md index 7471d86c5..9840c1ecd 100644 --- a/crates/io-core/README_zh.md +++ b/crates/io-core/README_zh.md @@ -23,71 +23,20 @@ ## 📖 概述 -**rustfs-io-core** 是 [RustFS](https://rustfs.com) 分布式对象存储系统的核心 I/O 调度模块。它提供了： +**rustfs-io-core** 是 [RustFS](https://rustfs.com) 分布式对象存储系统的共享 I/O 基础组件。它提供了： -- **I/O 调度器**：自适应缓冲区大小计算和负载管理 -- **优先级队列**：支持饥饿预防的请求优先级调度 +- **缓冲池**：分级复用的 `BytesPool` +- **存储画像**：存储介质与访问模式模型（`io_profile`） +- **调度配置**：存储层投影使用的 `IoSchedulerConfig` / `IoPriorityQueueConfig` - **背压控制**：系统过载保护和优雅降级 - **死锁检测**：基于等待图的死锁检测算法 - **锁优化**：自适应自旋锁优化 -- **超时包装器**：动态超时计算和操作进度追踪 +- **进度追踪**：长耗时操作的字节进度与停滞判定 + +调度算法本身位于 `rustfs/src/storage/concurrency/io_schedule.rs`；本 crate 只承载它投影使用的配置形状，不是第二套实现。 ## ✨ 核心功能 -### I/O 调度器 (IoScheduler) - -自适应 I/O 调度，根据文件大小、访问模式和系统负载动态调整缓冲区大小： - -```rust -use rustfs_io_core::{IoScheduler, IoSchedulerConfig, IoLoadLevel}; -use rustfs_io_core::io_profile::{StorageMedia, AccessPattern}; - -// 创建调度器 -let config = IoSchedulerConfig { - max_concurrent_reads: 64, - base_buffer_size: 64 * 1024, // 64 KB - max_buffer_size: 1024 * 1024, // 1 MB - ..Default::default() -}; -let scheduler = IoScheduler::new(config); - -// 计算最优缓冲区大小 -let buffer_size = scheduler.calculate_buffer_size( - 10 * 1024 * 1024, // 10 MB 文件 - true, // 顺序访问 - StorageMedia::Ssd, - IoLoadLevel::Low, -); -println!("缓冲区大小: {} bytes", buffer_size); -``` - -### 优先级队列 (IoPriorityQueue) - -支持饥饿预防的优先级队列： - -```rust -use rustfs_io_core::{IoPriorityQueue, IoPriority, IoQueueStatus}; - -let queue = IoPriorityQueue::<()>::new(100); - -// 入队请求 -let request_id = queue.enqueue( - IoPriority::High, - (), // 请求数据 - 1024, // 请求大小 -); - -// 出队请求 -if let Some((priority, data)) = queue.dequeue() { - println!("处理优先级 {:?} 的请求", priority); -} - -// 检查队列状态 -let status = queue.status(); -println!("高优先级等待: {}", status.high_priority_waiting); -println!("低优先级等待: {}", status.low_priority_waiting); -``` - ### 背压控制 (BackpressureMonitor) 系统过载保护： @@ -165,78 +114,23 @@ let stats = optimizer.stats(); println!("获取锁次数: {}", stats.locks_acquired.load(std::sync::atomic::Ordering::Relaxed)); ``` -### 超时包装器 (RequestTimeoutWrapper) +### 进度追踪 (OperationProgress) -动态超时计算： +长耗时操作的字节进度与停滞判定： ```rust -use rustfs_io_core::{RequestTimeoutWrapper, TimeoutConfig}; +use rustfs_io_core::OperationProgress; use std::time::Duration; -let config = TimeoutConfig { - base_timeout: Duration::from_secs(5), - timeout_per_mb: Duration::from_millis(100), - max_timeout: Duration::from_secs(300), - ..Default::default() -}; -let wrapper = RequestTimeoutWrapper::new(config); +let progress = OperationProgress::new(Some(1000), Duration::from_secs(5)); -// 计算操作超时 -let timeout = wrapper.calculate_timeout(10 * 1024 * 1024); // 10 MB -println!("超时时间: {:?}", timeout); - -// 执行带超时的操作 -let result = wrapper.execute_with_timeout(async { - // 异步操作 - Ok::<_, std::io::Error>(()) -}, timeout).await; -``` - -## 📊 缓冲区大小计算 - -模块提供了多种缓冲区大小计算函数： - -```rust -use rustfs_io_core::{ - get_concurrency_aware_buffer_size, - get_advanced_buffer_size, - get_buffer_size_for_media, - calculate_optimal_buffer_size, - KI_B, MI_B, -}; -use rustfs_io_core::io_profile::StorageMedia; - -// 基础计算 -let size1 = get_concurrency_aware_buffer_size(1024 * 1024, 64 * 1024); - -// 高级计算（考虑访问模式） -let size2 = get_advanced_buffer_size(10 * 1024 * 1024, 64 * 1024, true); - -// 媒体类型优化 -let size3 = get_buffer_size_for_media(64 * 1024, StorageMedia::Ssd); - -// 综合计算 -let size4 = calculate_optimal_buffer_size( - 100 * 1024 * 1024, // 100 MB 文件 - 64 * 1024, // 基础缓冲区 - true, // 顺序访问 - 4, // 并发请求数 - StorageMedia::Nvme, - IoLoadLevel::Low, -); +progress.update(500); +assert_eq!(progress.progress_percent(), Some(50.0)); +assert!(!progress.is_stale()); ``` ## 🔧 配置 -### 环境变量 - -| 变量名 | 描述 | 默认值 | -|--------|------|--------| -| `RUSTFS_MAX_CONCURRENT_READS` | 最大并发读数 | 64 | -| `RUSTFS_BASE_BUFFER_SIZE` | 基础缓冲区大小 | 65536 | -| `RUSTFS_MAX_BUFFER_SIZE` | 最大缓冲区大小 | 1048576 | -| `RUSTFS_IO_TIMEOUT_SECS` | I/O 超时秒数 | 30 | - ### 代码配置 ```rust @@ -264,12 +158,11 @@ rustfs-io-core/ ├── src/ │ ├── lib.rs # 模块入口 │ ├── config.rs # 配置类型 -│ ├── scheduler.rs # I/O 调度器 -│ ├── io_priority_queue.rs # 优先级队列 +│ ├── pool.rs # 分级缓冲池 │ ├── backpressure.rs # 背压控制 │ ├── deadlock_detector.rs # 死锁检测 │ ├── lock_optimizer.rs # 锁优化 -│ ├── timeout_wrapper.rs # 超时包装器 +│ ├── progress.rs # 操作进度追踪 │ └── io_profile.rs # I/O 配置文件 └── Cargo.toml ``` @@ -278,21 +171,15 @@ rustfs-io-core/ ```bash # 运行所有测试 -cargo test --package rustfs-io-core +cargo nextest run --package rustfs-io-core # 运行特定测试 -cargo test --package rustfs-io-core --lib scheduler - -# 运行基准测试 -cargo bench --package rustfs-io-core +cargo nextest run --package rustfs-io-core -E 'test(backpressure)' ``` ## 📚 文档 - [API 文档](https://docs.rs/rustfs-io-core) -- [I/O 调度器设计](./docs/scheduler-design.md) -- [背压控制原理](./docs/backpressure-design.md) -- [死锁检测算法](./docs/deadlock-detection.md) ## 🔗 相关模块 diff --git a/crates/io-core/examples/scheduler_example.rs b/crates/io-core/examples/scheduler_example.rs deleted file mode 100644 index 00fa560ae..000000000 --- a/crates/io-core/examples/scheduler_example.rs +++ /dev/null @@ -1,190 +0,0 @@ -// 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. - -//! Example demonstrating I/O scheduler usage. - -use rustfs_io_core::io_profile::StorageMedia; -use rustfs_io_core::{ - BackpressureMonitor, BackpressureState, DeadlockDetector, IoLoadLevel, IoScheduler, IoSchedulerConfig, KI_B, LockOptimizer, - LockType, MI_B, calculate_optimal_buffer_size, get_buffer_size_for_media, -}; -use std::time::Duration; - -fn main() { - println!("=== rustfs-io-core Example ===\n"); - - // 1. I/O scheduler example - io_scheduler_example(); - - // 2. Buffer size calculation example - buffer_size_example(); - - // 3. Backpressure control example - backpressure_example(); - - // 4. Deadlock detection example - deadlock_detection_example(); - - // 5. Lock optimizer example - lock_optimizer_example(); -} - -fn io_scheduler_example() { - println!("--- I/O Scheduler ---"); - - // Create scheduler with configuration - let config = IoSchedulerConfig { - max_concurrent_reads: 64, - base_buffer_size: 64 * KI_B, - max_buffer_size: MI_B, - ..Default::default() - }; - let scheduler = IoScheduler::new(config); - - println!(" Max concurrent reads: {}", scheduler.config().max_concurrent_reads); - println!(" Base buffer size: {} KB", scheduler.config().base_buffer_size / KI_B); - println!(" Max buffer size: {} KB", scheduler.config().max_buffer_size / KI_B); - - // Calculate buffer sizes for different scenarios - let scenarios = [ - ("Small file", 10 * KI_B as i64, true, StorageMedia::Ssd), - ("Medium file", MI_B as i64, true, StorageMedia::Ssd), - ("Large sequential", 100 * MI_B as i64, true, StorageMedia::Ssd), - ("Large random", 100 * MI_B as i64, false, StorageMedia::Ssd), - ("NVMe large", 100 * MI_B as i64, true, StorageMedia::Nvme), - ("HDD large", 100 * MI_B as i64, true, StorageMedia::Hdd), - ]; - - for (name, size, sequential, media) in scenarios { - let buffer = calculate_optimal_buffer_size(size, 64 * KI_B, sequential, 4, media, IoLoadLevel::Low); - println!(" {}: {} bytes ({} KB)", name, buffer, buffer / KI_B); - } - - println!(); -} - -fn buffer_size_example() { - println!("--- Buffer Size Calculation ---"); - - // Comprehensive calculation - let size1 = calculate_optimal_buffer_size(10 * MI_B as i64, 64 * KI_B, true, 4, StorageMedia::Ssd, IoLoadLevel::Low); - println!(" Comprehensive (10MB, sequential, SSD): {} KB", size1 / KI_B); - - // Media type optimization - let media_types = [ - StorageMedia::Nvme, - StorageMedia::Ssd, - StorageMedia::Hdd, - StorageMedia::Unknown, - ]; - for media in media_types { - let size = get_buffer_size_for_media(64 * KI_B, media); - println!(" {} optimized: {} KB", media.as_str(), size / KI_B); - } - - println!(); -} - -fn backpressure_example() { - println!("--- Backpressure Control ---"); - - let monitor = BackpressureMonitor::with_defaults(); - - // Check initial state - let state = monitor.state(); - let state_str = match state { - BackpressureState::Normal => "Normal", - BackpressureState::Warning => "Warning", - BackpressureState::Critical => "Critical", - }; - println!(" Initial state: {}", state_str); - - // Check if active - let is_active = monitor.is_active(); - println!(" Backpressure active: {}", is_active); - - // Try to acquire permit - if monitor.try_acquire() { - println!(" Successfully acquired permit"); - monitor.release(); - println!(" Released permit"); - } - - // View statistics - println!(" Total processed: {}", monitor.total_processed()); - println!(" Total rejected: {}", monitor.total_rejected()); - - println!(); -} - -fn deadlock_detection_example() { - println!("--- Deadlock Detection ---"); - - let detector = DeadlockDetector::with_defaults(); - - // Register locks - let mutex1 = detector.register_lock(LockType::Mutex); - let mutex2 = detector.register_lock(LockType::Mutex); - println!(" Registered locks: mutex1={}, mutex2={}", mutex1, mutex2); - - // Simulate normal operation - detector.record_acquire(mutex1, 1); // Thread 1 acquires mutex1 - detector.record_acquire(mutex2, 2); // Thread 2 acquires mutex2 - println!(" Normal operation: no deadlock"); - - // Detect deadlock - if detector.detect_deadlock().is_none() { - println!(" Detection result: no deadlock"); - } - - // Simulate deadlock scenario - detector.record_wait(mutex2, 1); // Thread 1 waits for mutex2 - detector.record_wait(mutex1, 2); // Thread 2 waits for mutex1 - - // Detect deadlock - if let Some(deadlock) = detector.detect_deadlock() { - println!(" Detection result: deadlock found {:?}", deadlock); - } - - // Cleanup - detector.unregister_lock(mutex1); - detector.unregister_lock(mutex2); - println!(); -} - -fn lock_optimizer_example() { - println!("--- Lock Optimizer ---"); - - let optimizer = LockOptimizer::with_defaults(); - - // Simulate lock operations - for _i in 0..5 { - optimizer.on_acquire(); - // Simulate work - std::thread::sleep(Duration::from_millis(10)); - optimizer.on_release(Duration::from_millis(10)); - } - - // View statistics - let stats = optimizer.stats(); - let acquired = stats.total_acquired(); - let avg_hold = stats.avg_hold_time(); - let contention = stats.contention_rate(); - - println!(" Locks acquired: {}", acquired); - println!(" Average hold time: {:?}", avg_hold); - println!(" Contention rate: {:.2}%", contention * 100.0); - - println!(); -} diff --git a/crates/io-core/src/bufreader_optimizer.rs b/crates/io-core/src/bufreader_optimizer.rs deleted file mode 100644 index 26ec968d5..000000000 --- a/crates/io-core/src/bufreader_optimizer.rs +++ /dev/null @@ -1,227 +0,0 @@ -// 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. - -//! BufReader layer optimizer for minimizing redundant buffering layers. -//! -//! This module provides optimization for BufReader usage in data paths, -//! including layer count limiting and dynamic buffer size adjustment. - -use std::sync::atomic::{AtomicU64, Ordering}; - -/// BufReader optimization configuration. -#[derive(Debug, Clone)] -pub struct BufReaderConfig { - /// Maximum number of nested BufReader layers (default: 2) - pub max_layers: u32, - - /// Buffer size for small files (default: 8KB) - pub small_file_buffer: usize, - - /// Buffer size for large files (default: 64KB) - pub large_file_buffer: usize, - - /// Threshold for large file classification (default: 1MB) - pub large_file_threshold: usize, -} - -impl Default for BufReaderConfig { - fn default() -> Self { - Self { - max_layers: 2, - small_file_buffer: 8 * 1024, // 8KB - large_file_buffer: 64 * 1024, // 64KB - large_file_threshold: 1024 * 1024, // 1MB - } - } -} - -/// BufReader optimization statistics. -#[derive(Debug, Default)] -pub struct BufReaderStats { - /// Total number of readers created - pub total_readers: AtomicU64, - - /// Number of redundant layers eliminated - pub eliminated_layers: AtomicU64, - - /// Number of buffer size adjustments - pub buffer_size_adjustments: AtomicU64, -} - -/// BufReader layer optimizer. -/// -/// Analyzes and optimizes BufReader nesting in data paths, -/// dynamically adjusting buffer sizes based on data characteristics. -pub struct BufReaderOptimizer { - config: BufReaderConfig, - stats: BufReaderStats, -} - -impl BufReaderOptimizer { - /// Create a new BufReader optimizer with the given configuration. - pub fn new(config: BufReaderConfig) -> Self { - Self { - config, - stats: BufReaderStats::default(), - } - } - - /// Create a new BufReader optimizer with default configuration. - pub fn with_defaults() -> Self { - Self::new(BufReaderConfig::default()) - } - - /// Calculate the optimal buffer size based on data size. - /// - /// Returns the appropriate buffer size based on whether the data - /// is classified as a small or large file. - pub fn optimal_buffer_size(&self, data_size: Option) -> usize { - match data_size { - Some(size) if size >= self.config.large_file_threshold => self.config.large_file_buffer, - Some(_) => self.config.small_file_buffer, - None => self.config.small_file_buffer, - } - } - - /// Optimize a reader by wrapping it with an appropriately sized BufReader. - /// - /// This method applies the optimal buffer size based on the expected - /// data size and tracks statistics. - pub fn optimize(&self, reader: R, data_size: Option) -> tokio::io::BufReader { - let buffer_size = self.optimal_buffer_size(data_size); - self.stats.total_readers.fetch_add(1, Ordering::Relaxed); - tokio::io::BufReader::with_capacity(buffer_size, reader) - } - - /// Get the statistics for this optimizer. - pub fn stats(&self) -> &BufReaderStats { - &self.stats - } - - /// Get the configuration for this optimizer. - pub fn config(&self) -> &BufReaderConfig { - &self.config - } -} - -/// Marker trait for buffered sources. -/// -/// Types implementing this trait are considered already buffered -/// and should not be wrapped with additional BufReader layers. -pub trait BufferedSource: tokio::io::AsyncRead {} - -impl BufReaderOptimizer { - /// Check if a reader is already a buffered source. - /// - /// Returns true if the reader implements `BufferedSource`, - /// indicating it should not be wrapped with BufReader. - pub fn is_buffered_source(&self, _reader: &R) -> bool { - true - } - - /// Eliminate redundant BufReader layers if possible. - /// - /// This method attempts to reduce the nesting depth of BufReader - /// layers to improve performance. - pub fn eliminate_redundant_layers(&self, reader: R) -> R { - // For now, just return the reader as-is - // Future implementation could detect and unwrap nested BufReaders - self.stats.eliminated_layers.fetch_add(0, Ordering::Relaxed); - reader - } -} - -#[cfg(test)] -mod tests { - use super::*; - use tokio::io::AsyncReadExt; - - #[test] - fn test_default_config() { - let config = BufReaderConfig::default(); - assert_eq!(config.max_layers, 2); - assert_eq!(config.small_file_buffer, 8 * 1024); - assert_eq!(config.large_file_buffer, 64 * 1024); - assert_eq!(config.large_file_threshold, 1024 * 1024); - } - - #[test] - fn test_optimal_buffer_size_small_file() { - let optimizer = BufReaderOptimizer::with_defaults(); - - // Small file (< 1MB) - assert_eq!(optimizer.optimal_buffer_size(Some(100)), 8 * 1024); - assert_eq!(optimizer.optimal_buffer_size(Some(1024)), 8 * 1024); - assert_eq!(optimizer.optimal_buffer_size(Some(512 * 1024)), 8 * 1024); - } - - #[test] - fn test_optimal_buffer_size_large_file() { - let optimizer = BufReaderOptimizer::with_defaults(); - - // Large file (>= 1MB) - assert_eq!(optimizer.optimal_buffer_size(Some(1024 * 1024)), 64 * 1024); - assert_eq!(optimizer.optimal_buffer_size(Some(10 * 1024 * 1024)), 64 * 1024); - } - - #[test] - fn test_optimal_buffer_size_unknown() { - let optimizer = BufReaderOptimizer::with_defaults(); - - // Unknown size - assert_eq!(optimizer.optimal_buffer_size(None), 8 * 1024); - } - - #[tokio::test] - async fn test_optimize_creates_bufreader() { - let optimizer = BufReaderOptimizer::with_defaults(); - let data = vec![1u8, 2, 3, 4, 5]; - let cursor = std::io::Cursor::new(data.clone()); - - let mut reader = optimizer.optimize(cursor, Some(5)); - - let mut buf = vec![0u8; 5]; - let n = reader.read(&mut buf).await.unwrap(); - - assert_eq!(n, 5); - assert_eq!(buf, data); - } - - #[test] - fn test_stats_tracking() { - let optimizer = BufReaderOptimizer::with_defaults(); - - assert_eq!(optimizer.stats().total_readers.load(Ordering::Relaxed), 0); - - let cursor = std::io::Cursor::new(vec![1u8, 2, 3]); - let _reader = optimizer.optimize(cursor, Some(3)); - - assert_eq!(optimizer.stats().total_readers.load(Ordering::Relaxed), 1); - } - - #[test] - fn test_custom_config() { - let config = BufReaderConfig { - max_layers: 3, - small_file_buffer: 4 * 1024, - large_file_buffer: 128 * 1024, - large_file_threshold: 2 * 1024 * 1024, - }; - - let optimizer = BufReaderOptimizer::new(config); - - assert_eq!(optimizer.optimal_buffer_size(Some(1024 * 1024)), 4 * 1024); - assert_eq!(optimizer.optimal_buffer_size(Some(3 * 1024 * 1024)), 128 * 1024); - } -} diff --git a/crates/io-core/src/direct_io.rs b/crates/io-core/src/direct_io.rs deleted file mode 100644 index 62e9834c8..000000000 --- a/crates/io-core/src/direct_io.rs +++ /dev/null @@ -1,332 +0,0 @@ -// 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. - -//! Aligned pread-based file reader. -//! -//! This module provides an aligned, position-based file reader that uses -//! `pread`/`FileExt::read_at` for I/O operations. It performs reads at -//! 512-byte-aligned offsets and sizes, making it suitable as a foundation -//! for workloads where alignment matters. -//! -//! Note: This reader does **not** set the `O_DIRECT` flag and therefore does -//! not bypass the OS page cache. It is an aligned `pread`-based reader, not -//! true Direct I/O. To implement true O_DIRECT on Linux, the file must be -//! opened with `O_DIRECT` via `libc::open`. -//! -//! # Platform Support -//! -//! The `read_at` implementation is only available on Unix-like platforms. -//! On other platforms, this reader will return an error. - -use std::io::{self}; -use std::pin::Pin; -use std::task::{Context, Poll}; -use tokio::io::{AsyncRead, ReadBuf}; - -/// Errors that can occur during aligned pread operations. -#[derive(Debug, Clone)] -pub enum AlignedPreadError { - /// Platform doesn't support `read_at`-based I/O - UnsupportedPlatform, - /// File descriptor doesn't support this reader - UnsupportedFile, - /// I/O error occurred - Io(String), - /// Invalid alignment (reads require 512-byte-aligned offset and size) - AlignmentError { offset: u64, size: usize }, -} - -impl std::fmt::Display for AlignedPreadError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::UnsupportedPlatform => write!(f, "Aligned pread not supported on this platform"), - Self::UnsupportedFile => write!(f, "File doesn't support this reader"), - Self::Io(msg) => write!(f, "I/O error: {}", msg), - Self::AlignmentError { offset, size } => { - write!(f, "Alignment error: offset={}, size={}", offset, size) - } - } - } -} - -impl std::error::Error for AlignedPreadError {} - -impl From for AlignedPreadError { - fn from(err: io::Error) -> Self { - Self::Io(err.to_string()) - } -} - -/// Aligned pread-based file reader for Unix platforms. -/// -/// This reader performs I/O using `pread`/`FileExt::read_at` at -/// 512-byte-aligned offsets and sizes, without modifying the file's -/// current position. -/// -/// **Note:** This reader does **not** set the `O_DIRECT` flag and therefore -/// does **not** bypass the OS page cache. It is an aligned `pread`-based -/// reader. To implement true O_DIRECT, the file must be opened with -/// `O_DIRECT` via `libc::open`. -/// -/// # Platform Support -/// -/// Only available on Linux (uses `FileExt::read_at`). On other platforms, -/// use `BytesBufferedReader` instead. -/// -/// # Alignment Requirements -/// -/// Reads have strict alignment requirements: -/// - File offset must be aligned to 512 bytes -/// - Buffer size must be a multiple of 512 bytes -/// - Buffer address must be aligned (handled internally) -/// -/// # Example -/// -/// ```ignore -/// use rustfs_io_core::AlignedPreadReader; -/// -/// // Linux only -/// #[cfg(target_os = "linux")] -/// let reader = AlignedPreadReader::new(file, offset, size)?; -/// ``` -#[cfg(target_os = "linux")] -pub struct AlignedPreadReader { - /// Underlying file handle used for aligned pread I/O - file: std::fs::File, - /// Current read position - pos: u64, - /// Remaining bytes to read - remaining: usize, - /// Buffer for aligned reads - buffer: Vec, - /// Current position in the buffer - buffer_pos: usize, - /// Amount of data in the buffer - buffer_len: usize, -} - -#[cfg(target_os = "linux")] -impl AlignedPreadReader { - /// Alignment requirement for reads (512 bytes for most systems) - pub const ALIGNMENT: usize = 512; - - /// Create a new aligned pread-based reader. - /// - /// # Arguments - /// - /// * `file` - File to read from - /// * `offset` - Starting offset in the file (must be 512-byte aligned) - /// * `size` - Number of bytes to read (must be 512-byte aligned) - /// - /// # Returns - /// - /// An `AlignedPreadReader` that reads the file at the given offset. - /// - /// # Errors - /// - /// Returns an error if offset or size are not 512-byte aligned. - pub fn new(file: std::fs::File, offset: u64, size: usize) -> Result { - // Check alignment - if !offset.is_multiple_of(Self::ALIGNMENT as u64) { - return Err(AlignedPreadError::AlignmentError { offset, size }); - } - if !size.is_multiple_of(Self::ALIGNMENT) { - return Err(AlignedPreadError::AlignmentError { offset, size }); - } - - Ok(Self { - file, - pos: offset, - remaining: size, - buffer: Vec::new(), - buffer_pos: 0, - buffer_len: 0, - }) - } - - /// Read a chunk of data using aligned pread. - /// - /// This method performs aligned reads and handles the buffering required - /// by this aligned pread implementation. It does not use `O_DIRECT`. - fn read_chunk(&mut self, buf: &mut [u8]) -> io::Result { - // If buffer is exhausted, read more data - if self.buffer_pos >= self.buffer_len { - if self.remaining == 0 { - return Ok(0); - } - - // Allocate aligned buffer - let chunk_size = (self.remaining).min(64 * 1024); // 64KB chunks - let aligned_size = chunk_size.div_ceil(Self::ALIGNMENT) * Self::ALIGNMENT; - - self.buffer = vec![0u8; aligned_size]; - - // Use pread for atomic read at position (no file offset modification) - use std::os::unix::fs::FileExt; - let n = self.file.read_at(&mut self.buffer, self.pos)?; - - self.buffer_pos = 0; - self.buffer_len = n; - self.pos += n as u64; - self.remaining -= n; - - if n == 0 { - return Ok(0); - } - } - - // Copy from buffer to user buffer - let available = self.buffer_len - self.buffer_pos; - let to_copy = buf.len().min(available); - buf[..to_copy].copy_from_slice(&self.buffer[self.buffer_pos..self.buffer_pos + to_copy]); - self.buffer_pos += to_copy; - - Ok(to_copy) - } -} - -#[cfg(target_os = "linux")] -impl AsyncRead for AlignedPreadReader { - fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - let filled = buf.filled().len(); - let mut remaining = buf.initialize_unfilled(); - - while !remaining.is_empty() { - match self.read_chunk(remaining) { - Ok(0) => break, - Ok(n) => { - remaining = &mut remaining[n..]; - } - Err(e) => return Poll::Ready(Err(e)), - } - } - - let _n_read = buf.filled().len() - filled; - Poll::Ready(Ok(())) - } -} - -/// Aligned pread reader stub for non-Linux platforms. -/// -/// On non-Linux platforms, `read_at`-based I/O is not available through this -/// type. This stub exists to provide a consistent API across platforms. -#[cfg(not(target_os = "linux"))] -pub struct AlignedPreadReader { - _priv: (), -} - -#[cfg(not(target_os = "linux"))] -impl AlignedPreadReader { - /// Create a new aligned pread reader (not supported on this platform). - /// - /// Always returns an error on non-Linux platforms. - pub fn new(_file: std::fs::File, _offset: u64, _size: usize) -> Result { - Err(AlignedPreadError::UnsupportedPlatform) - } -} - -#[cfg(not(target_os = "linux"))] -impl AsyncRead for AlignedPreadReader { - fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll> { - Poll::Ready(Err(io::Error::new( - io::ErrorKind::Unsupported, - "Aligned pread-based I/O not supported on this platform", - ))) - } -} - -impl std::fmt::Debug for AlignedPreadReader { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - #[cfg(target_os = "linux")] - { - f.debug_struct("AlignedPreadReader") - .field("pos", &self.pos) - .field("remaining", &self.remaining) - .field("buffer_len", &self.buffer_len) - .finish() - } - #[cfg(not(target_os = "linux"))] - { - f.debug_struct("AlignedPreadReader") - .field("platform", &"unsupported") - .finish() - } - } -} - -/// Historical name for aligned pread errors. -#[deprecated(since = "1.0.0-beta.8", note = "use AlignedPreadError; this reader does not set O_DIRECT")] -pub type DirectIoError = AlignedPreadError; - -/// Historical name for the aligned pread-based reader. -#[deprecated(since = "1.0.0-beta.8", note = "use AlignedPreadReader; this reader does not set O_DIRECT")] -pub type DirectIoReader = AlignedPreadReader; - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_alignment_check() { - #[cfg(target_os = "linux")] - { - // Valid alignment - let file = std::fs::File::open("/dev/zero").unwrap(); - assert!( - AlignedPreadReader::new(file, 0, 512).is_ok(), - "Should succeed with aligned offset and size" - ); - - let file = std::fs::File::open("/dev/zero").expect("open /dev/zero for alias"); - assert!( - AlignedPreadReader::new(file, 0, 512).is_ok(), - "Should succeed through aligned pread alias" - ); - - // Invalid offset - let file = std::fs::File::open("/dev/zero").unwrap(); - assert!(AlignedPreadReader::new(file, 1, 512).is_err(), "Should fail with unaligned offset"); - - // Invalid size - let file = std::fs::File::open("/dev/zero").unwrap(); - assert!(AlignedPreadReader::new(file, 0, 511).is_err(), "Should fail with unaligned size"); - } - - #[cfg(not(target_os = "linux"))] - { - // Non-Linux should return UnsupportedPlatform - let file = std::fs::File::open(std::env::current_exe().unwrap()).unwrap(); - assert!(matches!( - AlignedPreadReader::new(file, 0, 512), - Err(AlignedPreadError::UnsupportedPlatform) - )); - } - } - - #[test] - #[allow(deprecated)] - fn test_legacy_direct_io_alias() { - #[cfg(target_os = "linux")] - { - let file = std::fs::File::open("/dev/zero").unwrap(); - assert!(DirectIoReader::new(file, 0, 512).is_ok()); - } - - #[cfg(not(target_os = "linux"))] - { - let file = std::fs::File::open(std::env::current_exe().unwrap()).unwrap(); - assert!(matches!(DirectIoReader::new(file, 0, 512), Err(AlignedPreadError::UnsupportedPlatform))); - } - } -} diff --git a/crates/io-core/src/io_priority_queue.rs b/crates/io-core/src/io_priority_queue.rs deleted file mode 100644 index bb7b4cf45..000000000 --- a/crates/io-core/src/io_priority_queue.rs +++ /dev/null @@ -1,381 +0,0 @@ -// 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. - -//! I/O priority queue for scheduling I/O operations. -//! -//! This module provides a priority queue implementation for I/O operations -//! with support for starvation prevention and fair scheduling. - -use crate::config::IoPriorityQueueConfig; -use crate::scheduler::IoPriority; -use std::collections::VecDeque; -use std::time::{Duration, Instant}; - -/// A queued I/O request. -#[derive(Debug, Clone)] -pub struct IoRequest { - /// Request ID. - pub id: u64, - /// Request priority. - pub priority: IoPriority, - /// Request size in bytes. - pub size: usize, - /// Queue time. - pub queued_at: Instant, - /// Whether this is a sequential read. - pub is_sequential: bool, -} - -impl IoRequest { - /// Create a new I/O request. - pub fn new(id: u64, priority: IoPriority, size: usize, is_sequential: bool) -> Self { - Self { - id, - priority, - size, - queued_at: Instant::now(), - is_sequential, - } - } - - /// Get the wait time in the queue. - pub fn wait_time(&self) -> Duration { - self.queued_at.elapsed() - } -} - -/// Queue status for a priority level. -#[derive(Debug, Clone, Default)] -pub struct IoQueueStatus { - /// Number of requests in the queue. - pub count: usize, - /// Total size of all requests. - pub total_size: usize, - /// Oldest request wait time. - pub oldest_wait: Option, - /// Number of requests processed. - pub processed: u64, -} - -impl IoQueueStatus { - /// Create new queue status. - pub fn new() -> Self { - Self::default() - } -} - -/// I/O priority queue. -pub struct IoPriorityQueue { - /// Queue configuration. - config: IoPriorityQueueConfig, - /// High priority queue. - high: VecDeque, - /// Normal priority queue. - normal: VecDeque, - /// Low priority queue. - low: VecDeque, - /// Next request ID. - next_id: u64, - /// Last dequeue time for each priority (for starvation prevention). - last_dequeue: [Option; 3], - /// Statistics for each queue. - stats: [IoQueueStatus; 3], -} - -impl IoPriorityQueue { - /// Create a new priority queue with the given configuration. - pub fn new(config: IoPriorityQueueConfig) -> Self { - Self { - config, - high: VecDeque::with_capacity(100), - normal: VecDeque::with_capacity(500), - low: VecDeque::with_capacity(200), - next_id: 0, - last_dequeue: [None, None, None], - stats: [IoQueueStatus::new(), IoQueueStatus::new(), IoQueueStatus::new()], - } - } - - /// Create with default configuration. - pub fn with_defaults() -> Self { - Self::new(IoPriorityQueueConfig::default()) - } - - /// Get the configuration. - pub fn config(&self) -> &IoPriorityQueueConfig { - &self.config - } - - /// Enqueue a request. - pub fn enqueue(&mut self, priority: IoPriority, size: usize, is_sequential: bool) -> u64 { - let id = self.next_id; - self.next_id += 1; - - let request = IoRequest::new(id, priority, size, is_sequential); - - match priority { - IoPriority::High => { - if self.high.len() < self.config.high_capacity { - self.high.push_back(request); - } - } - IoPriority::Normal => { - if self.normal.len() < self.config.normal_capacity { - self.normal.push_back(request); - } - } - IoPriority::Low => { - if self.low.len() < self.config.low_capacity { - self.low.push_back(request); - } - } - } - - id - } - - /// Dequeue the next request. - /// - /// Uses weighted fair queuing with starvation prevention. - pub fn dequeue(&mut self) -> Option { - let now = Instant::now(); - - // Check for starvation: if a lower priority queue hasn't been served in a while, - // give it priority - let normal_starved = self.is_starved(IoPriority::Normal, now); - let low_starved = self.is_starved(IoPriority::Low, now); - - // Priority order with starvation consideration - // Check conditions first, then dequeue - let dequeue_high = !self.high.is_empty() && !low_starved && !normal_starved; - let dequeue_normal = !self.normal.is_empty() && !low_starved; - let dequeue_low = !self.low.is_empty(); - let dequeue_high_fallback = !self.high.is_empty(); - let dequeue_normal_fallback = !self.normal.is_empty(); - - if dequeue_high { - let request = self.high.pop_front(); - if request.is_some() { - self.last_dequeue[0] = Some(Instant::now()); - self.stats[0].processed += 1; - } - request - } else if dequeue_normal { - let request = self.normal.pop_front(); - if request.is_some() { - self.last_dequeue[1] = Some(Instant::now()); - self.stats[1].processed += 1; - } - request - } else if dequeue_low { - let request = self.low.pop_front(); - if request.is_some() { - self.last_dequeue[2] = Some(Instant::now()); - self.stats[2].processed += 1; - } - request - } else if dequeue_high_fallback { - let request = self.high.pop_front(); - if request.is_some() { - self.last_dequeue[0] = Some(Instant::now()); - self.stats[0].processed += 1; - } - request - } else if dequeue_normal_fallback { - let request = self.normal.pop_front(); - if request.is_some() { - self.last_dequeue[1] = Some(Instant::now()); - self.stats[1].processed += 1; - } - request - } else { - None - } - } - - /// Check if a priority level is starved. - fn is_starved(&self, priority: IoPriority, now: Instant) -> bool { - let idx = match priority { - IoPriority::High => 0, - IoPriority::Normal => 1, - IoPriority::Low => 2, - }; - - if let Some(last) = self.last_dequeue[idx] { - now.duration_since(last) > self.config.starvation_threshold - } else { - false - } - } - - /// Get the total number of queued requests. - pub fn len(&self) -> usize { - self.high.len() + self.normal.len() + self.low.len() - } - - /// Check if the queue is empty. - pub fn is_empty(&self) -> bool { - self.high.is_empty() && self.normal.is_empty() && self.low.is_empty() - } - - /// Get queue status for a priority level. - pub fn status(&self, priority: IoPriority) -> IoQueueStatus { - let (queue, idx) = match priority { - IoPriority::High => (&self.high, 0), - IoPriority::Normal => (&self.normal, 1), - IoPriority::Low => (&self.low, 2), - }; - - let mut status = self.stats[idx].clone(); - status.count = queue.len(); - status.total_size = queue.iter().map(|r| r.size).sum(); - status.oldest_wait = queue.front().map(|r| r.wait_time()); - status - } - - /// Get the total queue status. - pub fn total_status(&self) -> IoQueueStatus { - let mut total = IoQueueStatus::new(); - total.count = self.len(); - total.total_size = self - .high - .iter() - .chain(self.normal.iter()) - .chain(self.low.iter()) - .map(|r| r.size) - .sum(); - total.processed = self.stats.iter().map(|s| s.processed).sum(); - total.oldest_wait = self - .high - .front() - .map(|r| r.wait_time()) - .or_else(|| self.normal.front().map(|r| r.wait_time())) - .or_else(|| self.low.front().map(|r| r.wait_time())); - total - } - - /// Clear all queues. - pub fn clear(&mut self) { - self.high.clear(); - self.normal.clear(); - self.low.clear(); - } - - /// Peek at the next request without removing it. - pub fn peek(&self) -> Option<&IoRequest> { - if !self.high.is_empty() { - self.high.front() - } else if !self.normal.is_empty() { - self.normal.front() - } else { - self.low.front() - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_enqueue_dequeue() { - let mut queue = IoPriorityQueue::with_defaults(); - - let id1 = queue.enqueue(IoPriority::High, 1024, true); - let id2 = queue.enqueue(IoPriority::Normal, 2048, false); - let id3 = queue.enqueue(IoPriority::Low, 4096, true); - - assert_eq!(queue.len(), 3); - - // High priority should be dequeued first - let req1 = queue.dequeue().unwrap(); - assert_eq!(req1.id, id1); - assert_eq!(req1.priority, IoPriority::High); - - let req2 = queue.dequeue().unwrap(); - assert_eq!(req2.id, id2); - assert_eq!(req2.priority, IoPriority::Normal); - - let req3 = queue.dequeue().unwrap(); - assert_eq!(req3.id, id3); - assert_eq!(req3.priority, IoPriority::Low); - - assert!(queue.is_empty()); - } - - #[test] - fn test_queue_status() { - let mut queue = IoPriorityQueue::with_defaults(); - - queue.enqueue(IoPriority::High, 1024, true); - queue.enqueue(IoPriority::High, 2048, true); - queue.enqueue(IoPriority::Normal, 4096, false); - - let high_status = queue.status(IoPriority::High); - assert_eq!(high_status.count, 2); - assert_eq!(high_status.total_size, 3072); - - let normal_status = queue.status(IoPriority::Normal); - assert_eq!(normal_status.count, 1); - assert_eq!(normal_status.total_size, 4096); - - let total = queue.total_status(); - assert_eq!(total.count, 3); - assert_eq!(total.total_size, 7168); - } - - #[test] - fn test_queue_capacity() { - let config = IoPriorityQueueConfig { - high_capacity: 2, - normal_capacity: 2, - low_capacity: 2, - ..Default::default() - }; - let mut queue = IoPriorityQueue::new(config); - - queue.enqueue(IoPriority::High, 1024, true); - queue.enqueue(IoPriority::High, 1024, true); - queue.enqueue(IoPriority::High, 1024, true); // Should be dropped - - assert_eq!(queue.status(IoPriority::High).count, 2); - } - - #[test] - fn test_clear() { - let mut queue = IoPriorityQueue::with_defaults(); - - queue.enqueue(IoPriority::High, 1024, true); - queue.enqueue(IoPriority::Normal, 2048, false); - queue.enqueue(IoPriority::Low, 4096, true); - - assert_eq!(queue.len(), 3); - queue.clear(); - assert!(queue.is_empty()); - } - - #[test] - fn test_peek() { - let mut queue = IoPriorityQueue::with_defaults(); - - queue.enqueue(IoPriority::Normal, 2048, false); - queue.enqueue(IoPriority::High, 1024, true); - - let peeked = queue.peek().unwrap(); - assert_eq!(peeked.priority, IoPriority::High); - - // Peek shouldn't remove the item - assert_eq!(queue.len(), 2); - } -} diff --git a/crates/io-core/src/lib.rs b/crates/io-core/src/lib.rs index 67a374ea7..ca5f20c53 100644 --- a/crates/io-core/src/lib.rs +++ b/crates/io-core/src/lib.rs @@ -12,85 +12,39 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Buffered I/O reader and writer implementations for RustFS. +//! Shared I/O primitives for RustFS. //! -//! This crate provides buffered readers and writers for I/O operations. -//! Prefer `BytesBufferedReader`, `BytesMutWriter`, and `AlignedPreadReader` -//! for new code. Historical `ZeroCopy*` and `DirectIo*` names remain exported -//! for backward compatibility. +//! This crate holds the buffer pool and the concurrency-control primitives +//! that the storage layer builds on: //! -//! # Features -//! -//! - Memory-mapped file reading (mmap-then-copy) on Unix platforms -//! - Bytes-based buffered wrapping -//! - AsyncRead trait implementations -//! - Tiered BytesPool for buffer management -//! - Aligned pread-based reader (NOT true Direct I/O / O_DIRECT) +//! - Tiered `BytesPool` for buffer management +//! - Storage-media and access-pattern profiling (`io_profile`) +//! - Scheduler and priority-queue configuration shapes +//! - Backpressure admission, deadlock detection, lock optimization +//! - Progress tracking for long-running operations //! //! # Example //! //! ```ignore -//! use rustfs_io_core::{BytesBufferedReader, BytesPool}; -//! use bytes::Bytes; +//! use rustfs_io_core::BytesPool; //! -//! // Create from existing bytes (zero-copy) -//! let data = Bytes::from("hello world"); -//! let reader = BytesBufferedReader::from_bytes(data); -//! -//! // Create from file using buffered reads -//! let reader = BytesBufferedReader::from_file_read(&file, 0, 1024).await?; -//! -//! // Use BytesPool //! let pool = BytesPool::new_tiered(); //! let mut buffer = pool.acquire_buffer(8192).await; //! ``` pub mod backpressure; -pub mod bufreader_optimizer; pub mod config; pub mod deadlock_detector; -pub mod direct_io; -pub mod io_priority_queue; pub mod io_profile; pub mod lock_optimizer; pub mod pool; -pub mod reader; -pub mod scheduler; -pub mod shared_memory; -pub mod timeout_wrapper; -pub mod writer; +pub mod progress; -#[cfg(target_os = "linux")] -pub use direct_io::{AlignedPreadError, AlignedPreadReader}; -#[cfg(target_os = "linux")] -#[allow(deprecated)] -pub use direct_io::{DirectIoError, DirectIoReader}; pub use pool::{BytesPool, BytesPoolConfig, BytesPoolMetrics, PooledBuffer}; -#[allow(deprecated)] -pub use reader::ZeroCopyObjectReader; -pub use reader::{BytesBufferedReader, ZeroCopyReadError}; -#[allow(deprecated)] -pub use writer::ZeroCopyObjectWriter; -pub use writer::{BytesMutWriter, ZeroCopyWriteError}; - -// BufReader optimizer exports -pub use bufreader_optimizer::{BufReaderConfig, BufReaderOptimizer, BufReaderStats, BufferedSource}; - -// Shared memory exports -pub use shared_memory::{ArcData, ArcMetadata, SharedMemoryConfig, SharedMemoryPool, SharedMemoryStats}; // Config exports pub use config::{ConfigError, IoPriorityQueueConfig, IoSchedulerConfig}; -// Scheduler exports -pub use scheduler::{ - BandwidthTier, IoLoadLevel, IoLoadMetrics, IoPriority, IoScheduler, IoSchedulingContext, IoStrategy, KI_B, MI_B, - calculate_optimal_buffer_size, get_advanced_buffer_size, get_buffer_size_for_media, get_concurrency_aware_buffer_size, -}; - -// Priority queue exports -pub use io_priority_queue::{IoPriorityQueue, IoQueueStatus, IoRequest}; - // Backpressure exports pub use backpressure::{BackpressureConfig, BackpressureError, BackpressureMonitor, BackpressureState}; @@ -100,8 +54,5 @@ pub use deadlock_detector::{DeadlockDetector, DeadlockDetectorConfig, LockInfo, // Lock optimizer exports pub use lock_optimizer::{LockGuard, LockOptimizeConfig, LockOptimizer, LockStats}; -// Timeout wrapper exports -pub use timeout_wrapper::{ - OperationProgress, RequestTimeoutWrapper, TimeoutConfig, TimeoutError, TimeoutStats, calculate_adaptive_timeout, - estimate_bytes_per_second, -}; +// Progress tracking exports +pub use progress::OperationProgress; diff --git a/crates/io-core/src/progress.rs b/crates/io-core/src/progress.rs new file mode 100644 index 000000000..44e4460dd --- /dev/null +++ b/crates/io-core/src/progress.rs @@ -0,0 +1,138 @@ +// 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. + +//! Progress tracking for long-running I/O operations. +//! +//! Re-exported as `rustfs_concurrency::OperationProgress` for the storage +//! timeout implementation, which uses `is_stale` to tell a slow transfer +//! apart from a stalled one. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +/// Operation progress tracker. +#[derive(Debug)] +pub struct OperationProgress { + /// Total size (if known). + pub total_size: Option, + /// Bytes processed. + bytes_processed: AtomicU64, + /// Last update time. + last_update: std::sync::Mutex, + /// Stale timeout. + stale_timeout: Duration, + /// Start time for transfer rate calculation. + start_time: Instant, +} + +impl OperationProgress { + /// Create new operation progress. + pub fn new(total_size: Option, stale_timeout: Duration) -> Self { + Self { + total_size, + bytes_processed: AtomicU64::new(0), + last_update: std::sync::Mutex::new(Instant::now()), + stale_timeout, + start_time: Instant::now(), + } + } + + /// Update progress. + pub fn update(&self, bytes: u64) { + self.bytes_processed.store(bytes, Ordering::Relaxed); + if let Ok(mut last) = self.last_update.lock() { + *last = Instant::now(); + } + } + + /// Add to progress. + pub fn add(&self, bytes: u64) { + self.bytes_processed.fetch_add(bytes, Ordering::Relaxed); + if let Ok(mut last) = self.last_update.lock() { + *last = Instant::now(); + } + } + + /// Get current progress. + pub fn current(&self) -> u64 { + self.bytes_processed.load(Ordering::Relaxed) + } + + /// Check if progress is stale. + pub fn is_stale(&self) -> bool { + if let Ok(last) = self.last_update.lock() { + last.elapsed() > self.stale_timeout + } else { + false + } + } + + /// Get progress percentage. + pub fn progress_percent(&self) -> Option { + self.total_size.map(|total| { + if total == 0 { + 100.0 + } else { + let processed = self.bytes_processed.load(Ordering::Relaxed); + (processed as f64 / total as f64 * 100.0).min(100.0) + } + }) + } + + /// Get remaining bytes. + pub fn remaining(&self) -> Option { + self.total_size.map(|total| { + let processed = self.bytes_processed.load(Ordering::Relaxed); + total.saturating_sub(processed) + }) + } + + /// Calculate transfer rate in bytes per second. + /// + /// Returns 0 if no time has elapsed or no data transferred. + pub fn transfer_rate(&self) -> u64 { + let processed = self.bytes_processed.load(Ordering::Relaxed); + if processed == 0 { + return 0; + } + + let elapsed = self.start_time.elapsed().as_secs_f64(); + if elapsed > 0.0 { + (processed as f64 / elapsed) as u64 + } else { + 0 + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_operation_progress() { + let progress = OperationProgress::new(Some(1000), Duration::from_secs(5)); + + assert_eq!(progress.current(), 0); + assert_eq!(progress.progress_percent(), Some(0.0)); + + progress.update(500); + assert_eq!(progress.current(), 500); + assert_eq!(progress.progress_percent(), Some(50.0)); + + progress.add(300); + assert_eq!(progress.current(), 800); + assert_eq!(progress.remaining(), Some(200)); + } +} diff --git a/crates/io-core/src/reader.rs b/crates/io-core/src/reader.rs deleted file mode 100644 index 49433af73..000000000 --- a/crates/io-core/src/reader.rs +++ /dev/null @@ -1,412 +0,0 @@ -// 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. - -//! Bytes-backed object reader implementation. - -use bytes::Bytes; -use std::io; -use std::pin::Pin; -use std::task::{Context, Poll}; -use tokio::io::{AsyncRead, ReadBuf}; - -/// Errors that can occur during Bytes-backed read operations. -#[derive(Debug, Clone)] -pub enum ZeroCopyReadError { - /// I/O error occurred. - Io(String), - /// Memory mapping error. - Mmap(String), - /// Invalid offset or size. - InvalidRange, -} - -impl std::fmt::Display for ZeroCopyReadError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Io(msg) => write!(f, "I/O error: {}", msg), - Self::Mmap(msg) => write!(f, "Mmap error: {}", msg), - Self::InvalidRange => write!(f, "Invalid offset or size"), - } - } -} - -impl std::error::Error for ZeroCopyReadError {} - -impl From for ZeroCopyReadError { - fn from(err: io::Error) -> Self { - Self::Io(err.to_string()) - } -} - -/// Bytes-backed object reader. -/// -/// `from_bytes` wraps existing `Bytes` without copying, but file constructors -/// copy file data into owned `Bytes` after mmap or normal reads. -/// -/// # Example -/// -/// ```ignore -/// use bytes::Bytes; -/// use rustfs_io_core::BytesBufferedReader; -/// -/// // Create from bytes without copying the `Bytes` buffer -/// let data = Bytes::from("hello world"); -/// let reader = BytesBufferedReader::from_bytes(data); -/// -/// // Read using AsyncRead trait -/// let mut buf = vec![0u8; 1024]; -/// let n = reader.read(&mut buf[..]).await?; -/// ``` -pub struct BytesBufferedReader { - /// Internal data source (could be mmap or owned bytes) - data: Bytes, - /// Current read position - pos: usize, -} - -/// Historical name for the bytes-backed object reader. -#[deprecated( - since = "1.0.0-beta.8", - note = "use BytesBufferedReader; file constructors copy into owned Bytes" -)] -pub type ZeroCopyObjectReader = BytesBufferedReader; - -impl BytesBufferedReader { - /// Create a reader from existing bytes. - /// - /// This is a true zero-copy operation - the Bytes are wrapped - /// without any allocation or copying. - /// - /// # Arguments - /// - /// * `data` - Bytes to wrap - /// - /// # Example - /// - /// ```ignore - /// let data = Bytes::from("hello world"); - /// let reader = BytesBufferedReader::from_bytes(data); - /// ``` - pub fn from_bytes(data: Bytes) -> Self { - Self { data, pos: 0 } - } - - /// Create a Bytes-backed reader from a file using mmap-then-copy. - /// - /// This maps the requested file range and copies it into owned `Bytes` - /// before returning. It does not expose the mmap as a zero-copy buffer. - /// - /// # Arguments - /// - /// * `path` - Path to the file to memory map - /// * `offset` - Offset within the file to start reading - /// * `size` - Number of bytes to read - /// - /// # Returns - /// - /// A reader backed by copied file data. - /// - /// # Errors - /// - /// Returns an error if the file cannot be memory mapped. - /// - /// # Example - /// - /// ```ignore - /// let reader = BytesBufferedReader::from_file_mmap_path("large_file.bin", 0, 1024).await?; - /// ``` - #[cfg(unix)] - // SAFETY: The mmap is created from a read-only file handle for the - // caller-provided range, then copied into owned `Bytes` before the file and - // mapping are dropped. - #[allow(unsafe_code)] - pub async fn from_file_mmap_path(path: &std::path::Path, offset: u64, size: usize) -> Result { - use memmap2::MmapOptions; - - let path = path.to_path_buf(); - let (offset, size) = (offset, size); - - tokio::task::spawn_blocking(move || { - // Open the file in sync context - let std_file = std::fs::File::open(&path).map_err(|e| ZeroCopyReadError::Io(e.to_string()))?; - - // SAFETY: `std_file` remains open while the mapping is created and - // copied, and the mapped bytes are not exposed beyond this closure. - let mmap = unsafe { MmapOptions::new().offset(offset).len(size).map(&std_file) } - .map_err(|e| ZeroCopyReadError::Mmap(e.to_string()))?; - - // Convert to Bytes (this is a copy, but only done once) - Ok(Self { - data: Bytes::copy_from_slice(&mmap), - pos: 0, - }) - }) - .await - .map_err(|e| ZeroCopyReadError::Io(e.to_string()))? - } - - /// Create a Bytes-backed reader from a file using normal reads. - /// - /// This path reads the requested range into an owned buffer and wraps it in - /// `Bytes`. It does not perform mmap or zero-copy file I/O. - /// - /// # Arguments - /// - /// * `file` - File to read from - /// * `offset` - Offset within the file to start reading - /// * `size` - Number of bytes to map - /// - /// # Returns - /// - /// A reader backed by copied file data. - /// - /// # Errors - /// - /// Returns an error if the file cannot be read. - /// - /// # Example - /// - /// ```ignore - /// let file = tokio::fs::File::open("large_file.bin").await?; - /// let reader = BytesBufferedReader::from_file_read(&file, 0, 1024).await?; - /// ``` - #[cfg(unix)] - pub async fn from_file_read(file: &tokio::fs::File, offset: u64, size: usize) -> Result { - use tokio::io::{AsyncReadExt, AsyncSeekExt, SeekFrom}; - - let mut cloned = file.try_clone().await?; - cloned.seek(SeekFrom::Start(offset)).await?; - - let mut buffer = vec![0u8; size]; - cloned.read_exact(&mut buffer).await?; - - Ok(Self { - data: Bytes::from(buffer), - pos: 0, - }) - } - - /// Create a Bytes-backed reader from a file (non-Unix fallback). - /// - /// On platforms that don't support mmap, this falls back to regular file I/O. - #[cfg(not(unix))] - pub async fn from_file_read(file: &tokio::fs::File, offset: u64, size: usize) -> Result { - use tokio::io::{AsyncReadExt, AsyncSeekExt, SeekFrom}; - - let mut cloned = file.try_clone().await?; - cloned.seek(SeekFrom::Start(offset)).await?; - - let mut buffer = vec![0u8; size]; - cloned.read_exact(&mut buffer).await?; - - Ok(Self { - data: Bytes::from(buffer), - pos: 0, - }) - } - - /// Historical name for `from_file_read`. - #[deprecated( - since = "1.0.0-beta.8", - note = "use from_file_read; this method performs normal reads into owned Bytes" - )] - pub async fn from_file_mmap(file: &tokio::fs::File, offset: u64, size: usize) -> Result { - Self::from_file_read(file, offset, size).await - } - - /// Get the remaining data as Bytes (zero-copy). - /// - /// This returns a slice of the remaining data without copying. - /// The returned Bytes shares the underlying memory with this reader. - /// - /// # Example - /// - /// ```ignore - /// let remaining = reader.remaining_bytes(); - /// println!("Remaining: {} bytes", remaining.len()); - /// ``` - pub fn remaining_bytes(&self) -> Bytes { - self.data.slice(self.pos..) - } - - /// Get the total length of the data. - pub fn len(&self) -> usize { - self.data.len() - } - - /// Check if the reader has reached the end. - pub fn is_empty(&self) -> bool { - self.pos >= self.data.len() - } - - /// Get the current read position. - pub fn position(&self) -> usize { - self.pos - } -} - -impl AsyncRead for BytesBufferedReader { - fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - let remaining = self.data.len() - self.pos; - if remaining == 0 { - return Poll::Ready(Ok(())); - } - - let to_read = std::cmp::min(remaining, buf.remaining()); - let slice = &self.data[self.pos..self.pos + to_read]; - buf.put_slice(slice); - self.pos += to_read; - - Poll::Ready(Ok(())) - } -} - -impl std::fmt::Debug for BytesBufferedReader { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("BytesBufferedReader") - .field("data_len", &self.data.len()) - .field("pos", &self.pos) - .field("remaining", &(self.data.len() - self.pos)) - .finish() - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::path::PathBuf; - use tokio::io::AsyncReadExt; - - fn temp_file_path(test_name: &str) -> PathBuf { - let nonce = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("system time should be after unix epoch") - .as_nanos(); - std::env::temp_dir().join(format!("rustfs-io-core-{test_name}-{}-{nonce}", std::process::id())) - } - - #[tokio::test] - async fn test_from_bytes() { - let data = Bytes::from("hello world"); - let mut reader = BytesBufferedReader::from_bytes(data.clone()); - - let mut buf = [0u8; 11]; - let n = reader.read(&mut buf[..]).await.unwrap(); - - assert_eq!(n, 11); - assert_eq!(&buf[..n], b"hello world"); - } - - #[tokio::test] - async fn test_preferred_reader_alias() { - let data = Bytes::from("hello world"); - let mut reader = BytesBufferedReader::from_bytes(data); - - let mut buf = [0u8; 5]; - let n = reader.read(&mut buf[..]).await.expect("read bytes from alias"); - - assert_eq!(n, 5); - assert_eq!(&buf[..n], b"hello"); - } - - #[tokio::test] - async fn test_from_file_read_reads_requested_range() { - let path = temp_file_path("from-file-read"); - tokio::fs::write(&path, b"hello world") - .await - .expect("write temp file for reader test"); - - let file = tokio::fs::File::open(&path).await.expect("open temp file for reader test"); - let mut reader = BytesBufferedReader::from_file_read(&file, 6, 5) - .await - .expect("read requested range into Bytes"); - - let mut output = Vec::new(); - reader.read_to_end(&mut output).await.expect("drain reader output"); - - assert_eq!(output, b"world"); - - let _ = tokio::fs::remove_file(path).await; - } - - #[tokio::test] - #[allow(deprecated)] - async fn test_from_file_mmap_legacy_alias_reads_requested_range() { - let path = temp_file_path("from-file-mmap"); - tokio::fs::write(&path, b"hello world") - .await - .expect("write temp file for legacy reader test"); - - let file = tokio::fs::File::open(&path) - .await - .expect("open temp file for legacy reader test"); - let mut reader = BytesBufferedReader::from_file_mmap(&file, 0, 5) - .await - .expect("read requested range through legacy alias"); - - let mut output = Vec::new(); - reader.read_to_end(&mut output).await.expect("drain legacy reader output"); - - assert_eq!(output, b"hello"); - - let _ = tokio::fs::remove_file(path).await; - } - - #[tokio::test] - async fn test_remaining_bytes() { - let data = Bytes::from("hello world"); - let reader = BytesBufferedReader::from_bytes(data); - - let remaining = reader.remaining_bytes(); - assert_eq!(remaining.len(), 11); - assert_eq!(&remaining[..], b"hello world"); - } - - #[tokio::test] - async fn test_position() { - let data = Bytes::from("hello world"); - let mut reader = BytesBufferedReader::from_bytes(data); - - assert_eq!(reader.position(), 0); - - let mut buf = [0u8; 5]; - reader.read_exact(&mut buf[..]).await.unwrap(); - - assert_eq!(reader.position(), 5); - } - - #[tokio::test] - async fn test_is_empty() { - let data = Bytes::from(""); - let reader = BytesBufferedReader::from_bytes(data); - assert!(reader.is_empty()); - - let data = Bytes::from("hello"); - let reader = BytesBufferedReader::from_bytes(data); - assert!(!reader.is_empty()); - } - - #[tokio::test] - #[allow(deprecated)] - async fn test_legacy_reader_alias() { - let data = Bytes::from("hello world"); - let mut reader = ZeroCopyObjectReader::from_bytes(data); - - let mut buf = [0u8; 5]; - let n = reader.read(&mut buf[..]).await.expect("read bytes through legacy alias"); - - assert_eq!(n, 5); - assert_eq!(&buf[..n], b"hello"); - } -} diff --git a/crates/io-core/src/scheduler.rs b/crates/io-core/src/scheduler.rs deleted file mode 100644 index 76afa58cf..000000000 --- a/crates/io-core/src/scheduler.rs +++ /dev/null @@ -1,882 +0,0 @@ -// 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. - -//! I/O scheduler for adaptive buffer sizing and load management. -//! -//! This module provides the core I/O scheduling logic that determines -//! optimal buffer sizes, I/O strategies, and load management decisions. - -use crate::config::IoSchedulerConfig; -use crate::io_profile::{AccessPattern, StorageMedia, StorageProfile}; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::time::Duration; - -/// I/O priority levels. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] -pub enum IoPriority { - /// High priority for small, latency-sensitive operations. - High, - /// Normal priority for standard operations. - #[default] - Normal, - /// Low priority for large, throughput-oriented operations. - Low, -} - -impl IoPriority { - /// Determine priority based on request size. - /// - /// A negative `size` means the size is unknown (-1 by convention) and maps - /// to `Normal`; casting it to `usize` would wrap to a huge value and - /// misclassify the request as `Low`. - pub fn from_size(size: i64, high_threshold: usize, low_threshold: usize) -> Self { - if size < 0 { - return IoPriority::Normal; - } - let size = size as usize; - if size < high_threshold { - IoPriority::High - } else if size > low_threshold { - IoPriority::Low - } else { - IoPriority::Normal - } - } - - /// Get the priority as a string for metrics labels. - pub fn as_str(&self) -> &'static str { - match self { - IoPriority::High => "high", - IoPriority::Normal => "normal", - IoPriority::Low => "low", - } - } - - /// Check if this is high priority. - pub fn is_high(&self) -> bool { - matches!(self, IoPriority::High) - } - - /// Check if this is normal priority. - pub fn is_normal(&self) -> bool { - matches!(self, IoPriority::Normal) - } - - /// Check if this is low priority. - pub fn is_low(&self) -> bool { - matches!(self, IoPriority::Low) - } -} - -impl std::fmt::Display for IoPriority { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -/// I/O load level. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Default)] -pub enum IoLoadLevel { - /// Low load - system is underutilized. - Low, - /// Medium load - system is moderately utilized. - #[default] - Medium, - /// High load - system is heavily utilized. - High, - /// Critical load - system is overloaded. - Critical, -} - -impl IoLoadLevel { - /// Get the load level as a string for metrics labels. - pub fn as_str(&self) -> &'static str { - match self { - IoLoadLevel::Low => "low", - IoLoadLevel::Medium => "medium", - IoLoadLevel::High => "high", - IoLoadLevel::Critical => "critical", - } - } - - /// Determine load level from wait time. - pub fn from_wait_time(wait_time: Duration, low_threshold: Duration, high_threshold: Duration) -> Self { - if wait_time <= low_threshold { - IoLoadLevel::Low - } else if wait_time <= high_threshold { - IoLoadLevel::Medium - } else if wait_time <= high_threshold * 2 { - IoLoadLevel::High - } else { - IoLoadLevel::Critical - } - } -} - -impl std::fmt::Display for IoLoadLevel { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", self.as_str()) - } -} - -/// Bandwidth tier classification. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)] -pub enum BandwidthTier { - /// Low bandwidth (< 100 MB/s). - Low, - /// Medium bandwidth (100-500 MB/s). - #[default] - Medium, - /// High bandwidth (> 500 MB/s). - High, - /// Unknown bandwidth. - Unknown, -} - -impl BandwidthTier { - /// Determine bandwidth tier from bytes per second. - pub fn from_bps(bps: u64) -> Self { - const MB: u64 = 1024 * 1024; - if bps < 100 * MB { - BandwidthTier::Low - } else if bps < 500 * MB { - BandwidthTier::Medium - } else { - BandwidthTier::High - } - } - - /// Get the tier as a string for metrics labels. - pub fn as_str(&self) -> &'static str { - match self { - BandwidthTier::Low => "low", - BandwidthTier::Medium => "medium", - BandwidthTier::High => "high", - BandwidthTier::Unknown => "unknown", - } - } -} - -/// I/O strategy decision. -#[derive(Debug, Clone)] -pub struct IoStrategy { - /// Buffer size to use for I/O operations. - pub buffer_size: usize, - /// Buffer multiplier based on storage media. - pub buffer_multiplier: f64, - /// Whether to enable readahead. - pub enable_readahead: bool, - /// Whether to use buffered I/O. - pub use_buffered_io: bool, - - // Performance state - /// Current number of concurrent requests. - pub concurrent_requests: usize, - /// Observed bandwidth in bytes per second. - pub observed_bandwidth_bps: Option, - /// Bandwidth tier classification. - pub bandwidth_tier: BandwidthTier, - /// Current load level. - pub load_level: IoLoadLevel, - - // Priority - /// I/O priority for this operation. - pub priority: IoPriority, - - // Decision flags - /// Whether to throttle random I/O. - pub should_throttle_random_io: bool, - /// Whether to expand buffer for sequential access. - pub should_expand_for_sequential: bool, - /// Whether to reduce buffer due to concurrency. - pub should_reduce_for_concurrency: bool, - /// Whether to reduce buffer due to low bandwidth. - pub should_reduce_for_bandwidth: bool, -} - -impl Default for IoStrategy { - fn default() -> Self { - Self { - buffer_size: 128 * 1024, - buffer_multiplier: 1.0, - enable_readahead: true, - use_buffered_io: true, - concurrent_requests: 0, - observed_bandwidth_bps: None, - bandwidth_tier: BandwidthTier::Medium, - load_level: IoLoadLevel::Low, - priority: IoPriority::Normal, - should_throttle_random_io: false, - should_expand_for_sequential: false, - should_reduce_for_concurrency: false, - should_reduce_for_bandwidth: false, - } - } -} - -impl IoStrategy { - /// Create a new strategy with default values. - pub fn new() -> Self { - Self::default() - } - - /// Create a strategy for sequential access. - pub fn sequential(buffer_size: usize) -> Self { - Self { - buffer_size, - enable_readahead: true, - should_expand_for_sequential: true, - ..Self::default() - } - } - - /// Create a strategy for random access. - pub fn random(buffer_size: usize) -> Self { - Self { - buffer_size, - enable_readahead: false, - should_throttle_random_io: true, - ..Self::default() - } - } -} - -/// I/O load metrics. -#[derive(Debug, Clone, Default)] -pub struct IoLoadMetrics { - /// Number of samples in the current window. - pub sample_count: usize, - /// Total wait time in the window. - pub total_wait_time: Duration, - /// Maximum wait time in the window. - pub max_wait_time: Duration, - /// Average wait time. - pub avg_wait_time: Duration, - /// Current load level. - pub load_level: IoLoadLevel, -} - -impl IoLoadMetrics { - /// Create new load metrics. - pub fn new() -> Self { - Self::default() - } - - /// Add a wait time sample. - pub fn add_sample(&mut self, wait_time: Duration) { - self.sample_count += 1; - self.total_wait_time += wait_time; - if wait_time > self.max_wait_time { - self.max_wait_time = wait_time; - } - self.avg_wait_time = if self.sample_count > 0 { - self.total_wait_time / self.sample_count as u32 - } else { - Duration::ZERO - }; - } - - /// Update load level based on thresholds. - pub fn update_load_level(&mut self, low_threshold: Duration, high_threshold: Duration) { - self.load_level = IoLoadLevel::from_wait_time(self.avg_wait_time, low_threshold, high_threshold); - } - - /// Reset the metrics. - pub fn reset(&mut self) { - *self = Self::default(); - } -} - -/// I/O scheduler. -pub struct IoScheduler { - /// Scheduler configuration. - config: IoSchedulerConfig, - /// Active request counter. - active_requests: AtomicUsize, - /// Load metrics. - load_metrics: std::sync::Mutex, -} - -impl IoScheduler { - /// Create a new I/O scheduler with the given configuration. - pub fn new(config: IoSchedulerConfig) -> Self { - Self { - config, - active_requests: AtomicUsize::new(0), - load_metrics: std::sync::Mutex::new(IoLoadMetrics::new()), - } - } - - /// Create a new I/O scheduler with default configuration. - pub fn with_defaults() -> Self { - Self::new(IoSchedulerConfig::default()) - } - - /// Get the scheduler configuration. - pub fn config(&self) -> &IoSchedulerConfig { - &self.config - } - - /// Get the current number of active requests. - pub fn active_requests(&self) -> usize { - self.active_requests.load(Ordering::Relaxed) - } - - /// Increment the active request count. - pub fn increment_requests(&self) { - self.active_requests.fetch_add(1, Ordering::Relaxed); - } - - /// Decrement the active request count. - pub fn decrement_requests(&self) { - self.active_requests.fetch_sub(1, Ordering::Relaxed); - } - - /// Calculate I/O strategy for a request. - pub fn calculate_strategy(&self, file_size: i64, permit_wait_time: Duration, is_sequential: bool) -> IoStrategy { - let concurrent_requests = self.active_requests.load(Ordering::Relaxed); - - // Determine priority based on file size - let priority = IoPriority::from_size( - file_size, - self.config.high_priority_size_threshold, - self.config.low_priority_size_threshold, - ); - - // Determine load level - let load_level = - IoLoadLevel::from_wait_time(permit_wait_time, self.config.load_low_threshold(), self.config.load_high_threshold()); - - // Calculate base buffer size - let base_buffer = self.config.base_buffer_size; - - // Adjust for concurrency - let concurrency_factor = match concurrent_requests { - 0..=2 => 1.0, - 3..=4 => 0.75, - 5..=8 => 0.5, - _ => 0.4, - }; - - // Adjust for load level - let load_factor = match load_level { - IoLoadLevel::Low => 1.2, - IoLoadLevel::Medium => 1.0, - IoLoadLevel::High => 0.7, - IoLoadLevel::Critical => 0.5, - }; - - // Adjust for access pattern - let sequential_factor = if is_sequential { 1.5 } else { 1.0 }; - - // Calculate final buffer size - let buffer_size = (base_buffer as f64 * concurrency_factor * load_factor * sequential_factor) as usize; - let buffer_size = buffer_size.clamp(self.config.min_buffer_size, self.config.max_buffer_size); - - IoStrategy { - buffer_size, - buffer_multiplier: concurrency_factor * load_factor * sequential_factor, - enable_readahead: is_sequential && load_level != IoLoadLevel::Critical, - use_buffered_io: true, - concurrent_requests, - observed_bandwidth_bps: None, - bandwidth_tier: BandwidthTier::Unknown, - load_level, - priority, - should_throttle_random_io: !is_sequential && load_level >= IoLoadLevel::High, - should_expand_for_sequential: is_sequential && load_level <= IoLoadLevel::Medium, - should_reduce_for_concurrency: concurrent_requests > 4, - should_reduce_for_bandwidth: false, - } - } - - /// Calculate multi-factor I/O strategy. - pub fn calculate_multi_factor_strategy( - &self, - file_size: i64, - permit_wait_time: Duration, - is_sequential: bool, - storage_profile: Option<&StorageProfile>, - ) -> IoStrategy { - let mut strategy = self.calculate_strategy(file_size, permit_wait_time, is_sequential); - - // Apply storage profile adjustments - if let Some(profile) = storage_profile { - // Adjust buffer size based on storage media - let media_factor = match profile.media { - StorageMedia::Nvme => 1.5, - StorageMedia::Ssd => 1.2, - StorageMedia::Hdd => 0.8, - StorageMedia::Unknown => 1.0, - }; - - strategy.buffer_size = (strategy.buffer_size as f64 * media_factor).min(self.config.max_buffer_size as f64) as usize; - - // Apply sequential boost if applicable - if is_sequential { - strategy.buffer_size = (strategy.buffer_size as f64 * profile.sequential_boost_multiplier) - .min(self.config.max_buffer_size as f64) as usize; - } - - // Apply random penalty if applicable - if !is_sequential { - strategy.buffer_size = (strategy.buffer_size as f64 * profile.random_penalty_multiplier) - .max(self.config.min_buffer_size as f64) as usize; - } - - // Update readahead preference - strategy.enable_readahead = strategy.enable_readahead && profile.prefers_readahead; - } - - strategy - } - - /// Record a wait time sample for load tracking. - pub fn record_wait_time(&self, wait_time: Duration) { - if let Ok(mut metrics) = self.load_metrics.lock() { - metrics.add_sample(wait_time); - metrics.update_load_level(self.config.load_low_threshold(), self.config.load_high_threshold()); - } - } - - /// Get current load metrics. - pub fn load_metrics(&self) -> IoLoadMetrics { - if let Ok(metrics) = self.load_metrics.lock() { - metrics.clone() - } else { - IoLoadMetrics::default() - } - } -} - -impl Default for IoScheduler { - fn default() -> Self { - Self::with_defaults() - } -} - -// ============================================================================ -// Buffer Size Calculation Functions -// ============================================================================ - -/// Constants for buffer size calculations. -pub const KI_B: usize = 1024; -pub const MI_B: usize = 1024 * 1024; - -/// Get concurrency-aware buffer size. -/// -/// Adjusts buffer size based on the current level of concurrent requests. -/// Higher concurrency leads to smaller buffers to reduce memory pressure. -/// -/// # Arguments -/// -/// * `file_size` - Size of the file being read (-1 if unknown) -/// * `base_buffer_size` - Base buffer size from workload profile -/// -/// # Returns -/// -/// Adjusted buffer size in bytes -pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize) -> usize { - // Get current concurrency level from global counter - let concurrent_requests = 1; // Default to 1 if no global counter available - - // Define concurrency thresholds - let medium_threshold = 4; - let high_threshold = 8; - - // Calculate adaptive multiplier based on concurrency - let adaptive_multiplier = if concurrent_requests <= 2 { - // Low concurrency (1-2): use full buffer size - 1.0 - } else if concurrent_requests <= medium_threshold { - // Medium concurrency (3-4): slightly reduce buffer size (75% of base) - 0.75 - } else if concurrent_requests <= high_threshold { - // Higher concurrency (5-8): more aggressive reduction (50% of base) - 0.5 - } else { - // Very high concurrency (>8): minimize memory per request (40% of base) - 0.4 - }; - - // Calculate the adjusted buffer size - let adjusted_size = (base_buffer_size as f64 * adaptive_multiplier) as usize; - - // Ensure we stay within reasonable bounds - let min_buffer = if file_size > 0 && file_size < 100 * KI_B as i64 { - 32 * KI_B // For very small files, use minimum buffer - } else { - 64 * KI_B // Standard minimum buffer size - }; - - let max_buffer = if concurrent_requests > high_threshold { - 256 * KI_B // Cap at 256KB for high concurrency - } else { - MI_B // Cap at 1MB for lower concurrency - }; - - adjusted_size.clamp(min_buffer, max_buffer) -} - -/// Advanced concurrency-aware buffer sizing with file size optimization. -/// -/// This enhanced version considers both concurrency level and file size patterns -/// to provide even better performance characteristics. -/// -/// # Arguments -/// -/// * `file_size` - Size of the file being read (-1 if unknown) -/// * `base_buffer_size` - Baseline buffer size from workload profile -/// * `is_sequential` - Whether this is a sequential read (hint for optimization) -/// * `concurrent_requests` - Current number of concurrent requests -/// -/// # Returns -/// -/// Optimized buffer size in bytes -pub fn get_advanced_buffer_size( - file_size: i64, - base_buffer_size: usize, - is_sequential: bool, - concurrent_requests: usize, -) -> usize { - // For very small files, use smaller buffers regardless of concurrency - if file_size > 0 && file_size < 256 * KI_B as i64 { - return (file_size as usize / 4).clamp(16 * KI_B, 64 * KI_B); - } - - // Base calculation from standard function - let standard_size = get_concurrency_aware_buffer_size(file_size, base_buffer_size); - - let medium_threshold = 4; - let high_threshold = 8; - - // For sequential reads, we can be more aggressive with buffer sizes - if is_sequential && concurrent_requests <= medium_threshold { - // Boost buffer size for sequential reads under low concurrency - let boosted = (standard_size as f64 * 1.5) as usize; - return boosted.min(MI_B); - } - - // For random reads under high concurrency, reduce buffer size - if !is_sequential && concurrent_requests > high_threshold { - let reduced = (standard_size as f64 * 0.7) as usize; - return reduced.max(32 * KI_B); - } - - standard_size -} - -/// Get buffer size with storage media optimization. -/// -/// Adjusts buffer size based on storage media characteristics. -/// -/// # Arguments -/// -/// * `base_size` - Base buffer size -/// * `media` - Storage media type -/// -/// # Returns -/// -/// Optimized buffer size for the storage media -pub fn get_buffer_size_for_media(base_size: usize, media: StorageMedia) -> usize { - let multiplier = match media { - StorageMedia::Nvme => 1.5, // NVMe can handle larger buffers - StorageMedia::Ssd => 1.2, // SSD benefits from moderate buffers - StorageMedia::Hdd => 0.8, // HDD prefers smaller buffers to reduce seek overhead - StorageMedia::Unknown => 1.0, - }; - - (base_size as f64 * multiplier).min(MI_B as f64) as usize -} - -/// Calculate optimal buffer size using multi-factor analysis. -/// -/// This is the main entry point for buffer size calculation, considering -/// all factors: concurrency, storage media, access pattern, and load. -/// -/// # Arguments -/// -/// * `file_size` - Size of the file being read -/// * `base_buffer_size` - Base buffer size -/// * `is_sequential` - Whether access is sequential -/// * `concurrent_requests` - Current concurrency level -/// * `media` - Storage media type -/// * `load_level` - Current I/O load level -/// -/// # Returns -/// -/// Optimally calculated buffer size -pub fn calculate_optimal_buffer_size( - file_size: i64, - base_buffer_size: usize, - is_sequential: bool, - concurrent_requests: usize, - media: StorageMedia, - load_level: IoLoadLevel, -) -> usize { - // Start with advanced buffer size calculation - let mut buffer_size = get_advanced_buffer_size(file_size, base_buffer_size, is_sequential, concurrent_requests); - - // Apply storage media optimization - buffer_size = get_buffer_size_for_media(buffer_size, media); - - // Apply load-based adjustment - let load_multiplier = match load_level { - IoLoadLevel::Low => 1.2, - IoLoadLevel::Medium => 1.0, - IoLoadLevel::High => 0.7, - IoLoadLevel::Critical => 0.5, - }; - - buffer_size = (buffer_size as f64 * load_multiplier) as usize; - - // Final bounds check - buffer_size.clamp(32 * KI_B, MI_B) -} - -/// I/O scheduling context for multi-factor strategy calculation. -#[derive(Debug, Clone)] -pub struct IoSchedulingContext { - /// File size in bytes (-1 if unknown). - pub file_size: i64, - /// Base buffer size from configuration. - pub base_buffer_size: usize, - /// Time spent waiting for permit. - pub permit_wait_duration: Duration, - /// Whether access is sequential. - pub is_sequential_hint: bool, - /// Detected access pattern. - pub access_pattern: AccessPattern, - /// Detected storage media. - pub storage_media: StorageMedia, - /// Observed bandwidth in bytes per second. - pub observed_bandwidth_bps: Option, - /// Current concurrent request count. - pub concurrent_requests: usize, -} - -impl Default for IoSchedulingContext { - fn default() -> Self { - Self { - file_size: -1, - base_buffer_size: 128 * KI_B, - permit_wait_duration: Duration::ZERO, - is_sequential_hint: true, - access_pattern: AccessPattern::Unknown, - storage_media: StorageMedia::Unknown, - observed_bandwidth_bps: None, - concurrent_requests: 1, - } - } -} - -impl IoSchedulingContext { - /// Create a new scheduling context. - pub fn new(file_size: i64, base_buffer_size: usize) -> Self { - Self { - file_size, - base_buffer_size, - ..Self::default() - } - } - - /// Builder pattern: set sequential hint. - pub fn with_sequential(mut self, is_sequential: bool) -> Self { - self.is_sequential_hint = is_sequential; - self.access_pattern = if is_sequential { - AccessPattern::Sequential - } else { - AccessPattern::Random - }; - self - } - - /// Builder pattern: set storage media. - pub fn with_media(mut self, media: StorageMedia) -> Self { - self.storage_media = media; - self - } - - /// Builder pattern: set bandwidth. - pub fn with_bandwidth(mut self, bps: u64) -> Self { - self.observed_bandwidth_bps = Some(bps); - self - } - - /// Builder pattern: set concurrency. - pub fn with_concurrency(mut self, count: usize) -> Self { - self.concurrent_requests = count; - self - } - - /// Builder pattern: set wait duration. - pub fn with_wait_duration(mut self, duration: Duration) -> Self { - self.permit_wait_duration = duration; - self - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_io_priority() { - assert_eq!(IoPriority::from_size(1024, 64 * 1024, 4 * 1024 * 1024), IoPriority::High); - assert_eq!(IoPriority::from_size(1024 * 1024, 64 * 1024, 4 * 1024 * 1024), IoPriority::Normal); - assert_eq!(IoPriority::from_size(10 * 1024 * 1024, 64 * 1024, 4 * 1024 * 1024), IoPriority::Low); - } - - #[test] - fn test_io_priority_unknown_size_is_normal() { - // -1 means "size unknown" and must not wrap to usize::MAX (=> Low). - assert_eq!(IoPriority::from_size(-1, 64 * 1024, 4 * 1024 * 1024), IoPriority::Normal); - assert_eq!(IoPriority::from_size(i64::MIN, 64 * 1024, 4 * 1024 * 1024), IoPriority::Normal); - } - - #[test] - fn test_io_load_level() { - let low = Duration::from_millis(5); - let high = Duration::from_millis(50); - - assert_eq!(IoLoadLevel::from_wait_time(Duration::from_millis(1), low, high), IoLoadLevel::Low); - assert_eq!(IoLoadLevel::from_wait_time(Duration::from_millis(20), low, high), IoLoadLevel::Medium); - assert_eq!(IoLoadLevel::from_wait_time(Duration::from_millis(60), low, high), IoLoadLevel::High); - assert_eq!(IoLoadLevel::from_wait_time(Duration::from_millis(150), low, high), IoLoadLevel::Critical); - } - - #[test] - fn test_bandwidth_tier() { - assert_eq!(BandwidthTier::from_bps(50 * 1024 * 1024), BandwidthTier::Low); - assert_eq!(BandwidthTier::from_bps(200 * 1024 * 1024), BandwidthTier::Medium); - assert_eq!(BandwidthTier::from_bps(600 * 1024 * 1024), BandwidthTier::High); - } - - #[test] - fn test_io_strategy_default() { - let strategy = IoStrategy::default(); - assert!(strategy.buffer_size > 0); - assert!(strategy.enable_readahead); - } - - #[test] - fn test_io_scheduler() { - let scheduler = IoScheduler::with_defaults(); - - let strategy = scheduler.calculate_strategy(1024 * 1024, Duration::from_millis(5), true); - assert!(strategy.buffer_size > 0); - assert!(strategy.enable_readahead); - assert_eq!(strategy.load_level, IoLoadLevel::Low); - } - - #[test] - fn test_io_scheduler_with_concurrency() { - let scheduler = IoScheduler::with_defaults(); - - // Simulate concurrent requests - scheduler.increment_requests(); - scheduler.increment_requests(); - scheduler.increment_requests(); - - let strategy = scheduler.calculate_strategy(1024 * 1024, Duration::from_millis(5), true); - assert_eq!(strategy.concurrent_requests, 3); - } - - #[test] - fn test_load_metrics() { - let mut metrics = IoLoadMetrics::new(); - - metrics.add_sample(Duration::from_millis(10)); - metrics.add_sample(Duration::from_millis(20)); - metrics.add_sample(Duration::from_millis(30)); - - assert_eq!(metrics.sample_count, 3); - assert_eq!(metrics.avg_wait_time, Duration::from_millis(20)); - assert_eq!(metrics.max_wait_time, Duration::from_millis(30)); - } - - #[test] - fn test_get_concurrency_aware_buffer_size() { - // Test with default concurrency (1) - let size = get_concurrency_aware_buffer_size(1024 * 1024, 128 * KI_B); - assert!(size >= 64 * KI_B); - assert!(size <= MI_B); - - // Test with small file - let size = get_concurrency_aware_buffer_size(50 * KI_B as i64, 128 * KI_B); - assert!(size >= 32 * KI_B); - } - - #[test] - fn test_get_advanced_buffer_size() { - // Sequential read with low concurrency - let size = get_advanced_buffer_size(10 * MI_B as i64, 128 * KI_B, true, 2); - assert!(size >= 128 * KI_B); - - // Random read with high concurrency - let size = get_advanced_buffer_size(10 * MI_B as i64, 128 * KI_B, false, 10); - assert!(size >= 32 * KI_B); - - // Very small file - let size = get_advanced_buffer_size(100 * KI_B as i64, 128 * KI_B, true, 1); - assert!(size <= 64 * KI_B); - } - - #[test] - fn test_get_buffer_size_for_media() { - let base = 128 * KI_B; - - // NVMe should get larger buffers - let nvme_size = get_buffer_size_for_media(base, StorageMedia::Nvme); - assert!(nvme_size > base); - - // SSD should get slightly larger buffers - let ssd_size = get_buffer_size_for_media(base, StorageMedia::Ssd); - assert!(ssd_size > base); - - // HDD should get smaller buffers - let hdd_size = get_buffer_size_for_media(base, StorageMedia::Hdd); - assert!(hdd_size < base); - } - - #[test] - fn test_calculate_optimal_buffer_size() { - // Low load, sequential, NVMe - let size = calculate_optimal_buffer_size(10 * MI_B as i64, 128 * KI_B, true, 2, StorageMedia::Nvme, IoLoadLevel::Low); - assert!(size >= 32 * KI_B); - assert!(size <= MI_B); - - // Critical load, random, HDD - let size = - calculate_optimal_buffer_size(10 * MI_B as i64, 128 * KI_B, false, 10, StorageMedia::Hdd, IoLoadLevel::Critical); - assert!(size >= 32 * KI_B); - assert!(size <= MI_B); - } - - #[test] - fn test_io_scheduling_context() { - let ctx = IoSchedulingContext::new(10 * MI_B as i64, 256 * KI_B) - .with_sequential(true) - .with_media(StorageMedia::Nvme) - .with_bandwidth(500 * MI_B as u64) - .with_concurrency(4); - - assert_eq!(ctx.file_size, 10 * MI_B as i64); - assert_eq!(ctx.base_buffer_size, 256 * KI_B); - assert!(ctx.is_sequential_hint); - assert_eq!(ctx.storage_media, StorageMedia::Nvme); - assert_eq!(ctx.observed_bandwidth_bps, Some(500 * MI_B as u64)); - assert_eq!(ctx.concurrent_requests, 4); - } -} diff --git a/crates/io-core/src/shared_memory.rs b/crates/io-core/src/shared_memory.rs deleted file mode 100644 index e21781beb..000000000 --- a/crates/io-core/src/shared_memory.rs +++ /dev/null @@ -1,320 +0,0 @@ -// 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. - -//! Shared memory pool for zero-copy data sharing. -//! -//! This module provides Arc-based shared memory management for -//! efficient cross-task data passing without serialization. - -use std::convert::AsRef; -use std::ops::Deref; -use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::Instant; - -/// Shared memory pool configuration. -#[derive(Debug, Clone)] -pub struct SharedMemoryConfig { - /// Whether shared memory is enabled - pub enabled: bool, - - /// Maximum pool size in bytes - pub max_pool_size: usize, - - /// Maximum object size in bytes - pub max_object_size: usize, -} - -impl Default for SharedMemoryConfig { - fn default() -> Self { - Self { - enabled: true, - max_pool_size: 100 * 1024 * 1024, // 100MB - max_object_size: 10 * 1024 * 1024, // 10MB - } - } -} - -/// Shared memory pool statistics. -#[derive(Debug, Default)] -pub struct SharedMemoryStats { - /// Total number of objects created - pub total_objects: AtomicU64, - - /// Total number of shared references - pub total_shared_refs: AtomicU64, - - /// Current memory usage in bytes - pub current_memory: AtomicU64, - - /// Peak memory usage in bytes - pub peak_memory: AtomicU64, -} - -/// Arc data metadata. -#[derive(Clone, Debug)] -pub struct ArcMetadata { - /// Size of the data (if measurable) - pub size: Option, - - /// Creation timestamp - pub created_at: Instant, -} - -/// Arc-based data wrapper for zero-copy sharing. -/// -/// This wrapper uses Arc to enable shared ownership of data -/// across multiple tasks without copying. -pub struct ArcData { - /// The wrapped data - inner: Arc, - - /// Metadata about the data - metadata: ArcMetadata, -} - -impl Clone for ArcData { - fn clone(&self) -> Self { - Self { - inner: Arc::clone(&self.inner), - metadata: self.metadata.clone(), - } - } -} - -impl ArcData { - /// Create a new ArcData wrapper. - pub fn new(data: T) -> Self { - ArcData { - inner: Arc::new(data), - metadata: ArcMetadata { - size: None, - created_at: Instant::now(), - }, - } - } - - /// Create a new ArcData wrapper with known size. - pub fn with_size(data: T, size: usize) -> Self { - ArcData { - inner: Arc::new(data), - metadata: ArcMetadata { - size: Some(size), - created_at: Instant::now(), - }, - } - } - - /// Get the reference count. - pub fn ref_count(&self) -> usize { - Arc::strong_count(&self.inner) - } - - /// Convert into the underlying Arc. - pub fn into_arc(self) -> Arc { - self.inner - } - - /// Get the metadata. - pub fn metadata(&self) -> &ArcMetadata { - &self.metadata - } - - /// Get the size if known. - pub fn size(&self) -> Option { - self.metadata.size - } -} - -impl AsRef for ArcData { - fn as_ref(&self) -> &T { - &self.inner - } -} - -impl Deref for ArcData { - type Target = T; - - fn deref(&self) -> &Self::Target { - &self.inner - } -} - -impl std::fmt::Debug for ArcData { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("ArcData") - .field("ref_count", &self.ref_count()) - .field("metadata", &self.metadata) - .finish() - } -} - -/// Shared memory pool for managing Arc-based shared data. -pub struct SharedMemoryPool { - config: SharedMemoryConfig, - stats: SharedMemoryStats, -} - -impl SharedMemoryPool { - /// Create a new shared memory pool with the given configuration. - pub fn new(config: SharedMemoryConfig) -> Self { - Self { - config, - stats: SharedMemoryStats::default(), - } - } - - /// Create a new shared memory pool with default configuration. - pub fn with_defaults() -> Self { - Self::new(SharedMemoryConfig::default()) - } - - /// Create shared data. - /// - /// This method wraps the data in an ArcData for zero-copy sharing. - pub fn create(&self, data: T) -> ArcData { - self.stats.total_objects.fetch_add(1, Ordering::Relaxed); - ArcData::new(data) - } - - /// Create shared data with known size. - /// - /// This method tracks memory usage for statistics. - pub fn create_with_size(&self, data: T, size: usize) -> ArcData { - self.stats.total_objects.fetch_add(1, Ordering::Relaxed); - - // Update memory statistics - self.stats.current_memory.fetch_add(size as u64, Ordering::Relaxed); - - // Update peak memory - let current = self.stats.current_memory.load(Ordering::Relaxed); - let mut peak = self.stats.peak_memory.load(Ordering::Relaxed); - if current > peak { - peak = current; - self.stats.peak_memory.store(peak, Ordering::Relaxed); - } - - ArcData::with_size(data, size) - } - - /// Share data by increasing reference count. - /// - /// This method creates a new ArcData that shares the underlying data - /// without copying. - pub fn share(&self, data: &ArcData) -> ArcData { - self.stats.total_shared_refs.fetch_add(1, Ordering::Relaxed); - data.clone() - } - - /// Get the statistics for this pool. - pub fn stats(&self) -> &SharedMemoryStats { - &self.stats - } - - /// Get the configuration for this pool. - pub fn config(&self) -> &SharedMemoryConfig { - &self.config - } - - /// Check if the pool is enabled. - pub fn is_enabled(&self) -> bool { - self.config.enabled - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_arc_data_new() { - let data = vec![1u8, 2, 3, 4, 5]; - let arc_data = ArcData::new(data.clone()); - - assert_eq!(arc_data.as_ref(), &data); - assert_eq!(arc_data.ref_count(), 1); - } - - #[test] - fn test_arc_data_clone() { - let data = vec![1u8, 2, 3, 4, 5]; - let arc_data = ArcData::new(data); - - assert_eq!(arc_data.ref_count(), 1); - - let arc_data2 = arc_data.clone(); - assert_eq!(arc_data.ref_count(), 2); - assert_eq!(arc_data2.ref_count(), 2); - - let arc_data3 = arc_data.clone(); - assert_eq!(arc_data.ref_count(), 3); - assert_eq!(arc_data2.ref_count(), 3); - assert_eq!(arc_data3.ref_count(), 3); - } - - #[test] - fn test_arc_data_deref() { - let data = vec![1u8, 2, 3, 4, 5]; - let arc_data = ArcData::new(data); - - // Test Deref trait - assert_eq!(arc_data.len(), 5); - assert_eq!(arc_data[0], 1); - } - - #[test] - fn test_shared_memory_pool_create() { - let pool = SharedMemoryPool::with_defaults(); - let data = vec![1u8, 2, 3, 4, 5]; - - let arc_data = pool.create(data.clone()); - - assert_eq!(arc_data.as_ref(), &data); - assert_eq!(pool.stats().total_objects.load(Ordering::Relaxed), 1); - } - - #[test] - fn test_shared_memory_pool_share() { - let pool = SharedMemoryPool::with_defaults(); - let data = vec![1u8, 2, 3, 4, 5]; - - let arc_data = pool.create(data); - assert_eq!(arc_data.ref_count(), 1); - - let shared = pool.share(&arc_data); - assert_eq!(arc_data.ref_count(), 2); - assert_eq!(shared.ref_count(), 2); - assert_eq!(pool.stats().total_shared_refs.load(Ordering::Relaxed), 1); - } - - #[test] - fn test_shared_memory_pool_with_size() { - let pool = SharedMemoryPool::with_defaults(); - let data = vec![1u8; 1024]; - - let arc_data = pool.create_with_size(data, 1024); - - assert_eq!(arc_data.size(), Some(1024)); - assert_eq!(pool.stats().current_memory.load(Ordering::Relaxed), 1024); - } - - #[test] - fn test_default_config() { - let config = SharedMemoryConfig::default(); - - assert!(config.enabled); - assert_eq!(config.max_pool_size, 100 * 1024 * 1024); - assert_eq!(config.max_object_size, 10 * 1024 * 1024); - } -} diff --git a/crates/io-core/src/timeout_wrapper.rs b/crates/io-core/src/timeout_wrapper.rs deleted file mode 100644 index 9d9ce4b22..000000000 --- a/crates/io-core/src/timeout_wrapper.rs +++ /dev/null @@ -1,501 +0,0 @@ -// 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. - -//! Timeout wrapper for I/O operations. -//! -//! This module provides timeout management for I/O operations with -//! dynamic timeout calculation based on operation size. - -use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::{Duration, Instant}; - -/// Timeout configuration. -#[derive(Debug, Clone)] -pub struct TimeoutConfig { - /// Base timeout for small operations. - pub base_timeout: Duration, - /// Timeout per MB of data. - pub timeout_per_mb: Duration, - /// Maximum timeout. - pub max_timeout: Duration, - /// Minimum timeout. - pub min_timeout: Duration, - /// GetObject operation timeout. - pub get_object_timeout: Duration, - /// PutObject operation timeout. - pub put_object_timeout: Duration, - /// ListObjects operation timeout. - pub list_objects_timeout: Duration, - /// Whether dynamic timeout is enabled. - pub enable_dynamic_timeout: bool, -} - -impl Default for TimeoutConfig { - fn default() -> Self { - Self { - base_timeout: Duration::from_secs(5), - timeout_per_mb: Duration::from_millis(100), - max_timeout: Duration::from_secs(300), - min_timeout: Duration::from_secs(1), - get_object_timeout: Duration::from_secs(30), - put_object_timeout: Duration::from_secs(60), - list_objects_timeout: Duration::from_secs(10), - enable_dynamic_timeout: true, - } - } -} - -impl TimeoutConfig { - /// Create new timeout configuration. - pub fn new() -> Self { - Self::default() - } - - /// Calculate dynamic timeout based on size. - pub fn calculate_timeout(&self, size_bytes: u64) -> Duration { - if !self.enable_dynamic_timeout { - return self.base_timeout; - } - - let mb = size_bytes as f64 / (1024.0 * 1024.0); - let timeout = self.base_timeout + self.timeout_per_mb.mul_f64(mb); - timeout.clamp(self.min_timeout, self.max_timeout) - } - - /// Validate the configuration. - pub fn validate(&self) -> Result<(), TimeoutError> { - if self.min_timeout > self.max_timeout { - return Err(TimeoutError::InvalidConfig("min_timeout must be <= max_timeout".to_string())); - } - if self.base_timeout < self.min_timeout || self.base_timeout > self.max_timeout { - return Err(TimeoutError::InvalidConfig( - "base_timeout must be between min_timeout and max_timeout".to_string(), - )); - } - Ok(()) - } -} - -/// Timeout error. -#[derive(Debug, Clone, thiserror::Error)] -pub enum TimeoutError { - /// Operation timed out. - #[error("Operation timed out after {0:?}")] - TimedOut(Duration), - /// Invalid configuration. - #[error("Invalid timeout config: {0}")] - InvalidConfig(String), -} - -/// Operation progress tracker. -#[derive(Debug)] -pub struct OperationProgress { - /// Total size (if known). - pub total_size: Option, - /// Bytes processed. - bytes_processed: AtomicU64, - /// Last update time. - last_update: std::sync::Mutex, - /// Stale timeout. - stale_timeout: Duration, - /// Start time for transfer rate calculation. - start_time: Instant, -} - -impl OperationProgress { - /// Create new operation progress. - pub fn new(total_size: Option, stale_timeout: Duration) -> Self { - Self { - total_size, - bytes_processed: AtomicU64::new(0), - last_update: std::sync::Mutex::new(Instant::now()), - stale_timeout, - start_time: Instant::now(), - } - } - - /// Update progress. - pub fn update(&self, bytes: u64) { - self.bytes_processed.store(bytes, Ordering::Relaxed); - if let Ok(mut last) = self.last_update.lock() { - *last = Instant::now(); - } - } - - /// Add to progress. - pub fn add(&self, bytes: u64) { - self.bytes_processed.fetch_add(bytes, Ordering::Relaxed); - if let Ok(mut last) = self.last_update.lock() { - *last = Instant::now(); - } - } - - /// Get current progress. - pub fn current(&self) -> u64 { - self.bytes_processed.load(Ordering::Relaxed) - } - - /// Check if progress is stale. - pub fn is_stale(&self) -> bool { - if let Ok(last) = self.last_update.lock() { - last.elapsed() > self.stale_timeout - } else { - false - } - } - - /// Get progress percentage. - pub fn progress_percent(&self) -> Option { - self.total_size.map(|total| { - if total == 0 { - 100.0 - } else { - let processed = self.bytes_processed.load(Ordering::Relaxed); - (processed as f64 / total as f64 * 100.0).min(100.0) - } - }) - } - - /// Get remaining bytes. - pub fn remaining(&self) -> Option { - self.total_size.map(|total| { - let processed = self.bytes_processed.load(Ordering::Relaxed); - total.saturating_sub(processed) - }) - } - - /// Calculate transfer rate in bytes per second. - /// - /// Returns 0 if no time has elapsed or no data transferred. - pub fn transfer_rate(&self) -> u64 { - let processed = self.bytes_processed.load(Ordering::Relaxed); - if processed == 0 { - return 0; - } - - let elapsed = self.start_time.elapsed().as_secs_f64(); - if elapsed > 0.0 { - (processed as f64 / elapsed) as u64 - } else { - 0 - } - } -} - -/// Request timeout wrapper. -pub struct RequestTimeoutWrapper { - /// Configuration. - config: TimeoutConfig, - /// Start time. - start_time: Instant, - /// Operation progress. - progress: Option, -} - -impl RequestTimeoutWrapper { - /// Create a new timeout wrapper. - pub fn new(config: TimeoutConfig) -> Self { - Self { - config, - start_time: Instant::now(), - progress: None, - } - } - - /// Create with progress tracking. - pub fn with_progress(config: TimeoutConfig, total_size: Option, stale_timeout: Duration) -> Self { - Self { - config, - start_time: Instant::now(), - progress: Some(OperationProgress::new(total_size, stale_timeout)), - } - } - - /// Get the configuration. - pub fn config(&self) -> &TimeoutConfig { - &self.config - } - - /// Get elapsed time. - pub fn elapsed(&self) -> Duration { - self.start_time.elapsed() - } - - /// Get remaining time. - pub fn remaining(&self, timeout: Duration) -> Option { - let elapsed = self.elapsed(); - if elapsed >= timeout { None } else { Some(timeout - elapsed) } - } - - /// Check if timed out. - pub fn is_timed_out(&self, size: Option) -> bool { - let timeout = self.get_timeout(size); - self.elapsed() > timeout - } - - /// Get the timeout for a given size. - pub fn get_timeout(&self, size: Option) -> Duration { - if self.config.enable_dynamic_timeout { - if let Some(s) = size { - self.config.calculate_timeout(s) - } else { - self.config.base_timeout - } - } else { - self.config.base_timeout - } - } - - /// Check if timed out and return error if so. - pub fn check_timeout(&self, size: Option) -> Result<(), TimeoutError> { - if self.is_timed_out(size) { - Err(TimeoutError::TimedOut(self.get_timeout(size))) - } else { - Ok(()) - } - } - - /// Get progress. - pub fn progress(&self) -> Option<&OperationProgress> { - self.progress.as_ref() - } - - /// Update progress. - pub fn update_progress(&self, bytes: u64) { - if let Some(ref progress) = self.progress { - progress.update(bytes); - } - } - - /// Check if operation is stalled (no progress for a while). - pub fn is_stalled(&self) -> bool { - self.progress.as_ref().is_some_and(|p| p.is_stale()) - } - - /// Get progress percentage. - pub fn progress_percent(&self) -> Option { - self.progress.as_ref().and_then(|p| p.progress_percent()) - } -} - -/// Timeout statistics. -#[derive(Debug, Default)] -pub struct TimeoutStats { - /// Total operations. - pub total_operations: AtomicU64, - /// Timed out operations. - pub timed_out: AtomicU64, - /// Total wait time in nanoseconds. - pub total_wait_time_ns: AtomicU64, - /// Maximum wait time in nanoseconds. - pub max_wait_time_ns: AtomicU64, -} - -impl TimeoutStats { - /// Create new timeout statistics. - pub fn new() -> Self { - Self::default() - } - - /// Record an operation. - pub fn record_operation(&self, wait_time: Duration) { - self.total_operations.fetch_add(1, Ordering::Relaxed); - let ns = wait_time.as_nanos() as u64; - self.total_wait_time_ns.fetch_add(ns, Ordering::Relaxed); - - let mut current = self.max_wait_time_ns.load(Ordering::Relaxed); - while ns > current { - match self - .max_wait_time_ns - .compare_exchange_weak(current, ns, Ordering::Relaxed, Ordering::Relaxed) - { - Ok(_) => break, - Err(actual) => current = actual, - } - } - } - - /// Record a timeout. - pub fn record_timeout(&self) { - self.timed_out.fetch_add(1, Ordering::Relaxed); - } - - /// Get timeout rate. - pub fn timeout_rate(&self) -> f64 { - let total = self.total_operations.load(Ordering::Relaxed); - let timed_out = self.timed_out.load(Ordering::Relaxed); - if total == 0 { 0.0 } else { timed_out as f64 / total as f64 } - } - - /// Get average wait time. - pub fn avg_wait_time(&self) -> Duration { - let total = self.total_wait_time_ns.load(Ordering::Relaxed); - let count = self.total_operations.load(Ordering::Relaxed); - total.checked_div(count).map(Duration::from_nanos).unwrap_or(Duration::ZERO) - } - - /// Reset statistics. - pub fn reset(&self) { - self.total_operations.store(0, Ordering::Relaxed); - self.timed_out.store(0, Ordering::Relaxed); - self.total_wait_time_ns.store(0, Ordering::Relaxed); - self.max_wait_time_ns.store(0, Ordering::Relaxed); - } -} - -/// Calculate adaptive timeout based on historical data and current conditions. -/// -/// This function adjusts the timeout based on: -/// - Historical transfer rate -/// - Recent timeout count -/// - Object size -pub fn calculate_adaptive_timeout( - base_timeout: Duration, - historical_rate_bps: Option, - recent_timeout_count: u32, - object_size: u64, -) -> Duration { - // If we have recent timeouts, increase timeout - let timeout_multiplier = if recent_timeout_count > 3 { - 2.0 // Double timeout if many recent timeouts - } else if recent_timeout_count > 1 { - 1.5 // 50% increase if some timeouts - } else { - 1.0 // No adjustment - }; - - // Adaptive timeout bounds: 5 seconds minimum, 10 minutes maximum. - const MIN_SECS: f64 = 5.0; - const MAX_SECS: f64 = 600.0; - - // If we have historical rate data, use it for estimation - let estimated_secs = match historical_rate_bps { - Some(rate) if rate > 0 => (object_size as f64 / rate as f64) * 1.2, // 20% buffer - _ => base_timeout.as_secs_f64(), - }; - - // Clamp BEFORE constructing the Duration: `from_secs_f64` panics when the - // estimate overflows Duration (huge object_size with a tiny historical rate). - Duration::from_secs_f64((estimated_secs * timeout_multiplier).clamp(MIN_SECS, MAX_SECS)) -} - -/// Estimate bytes per second transfer rate. -/// -/// This is used for adaptive timeout calculation. -pub fn estimate_bytes_per_second(object_size: u64, expected_duration: Duration) -> u64 { - let secs = expected_duration.as_secs_f64(); - if secs > 0.0 { - (object_size as f64 / secs) as u64 - } else { - // Return a reasonable default (1 MB/s) - 1024 * 1024 - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_timeout_config() { - let config = TimeoutConfig::default(); - assert!(config.validate().is_ok()); - - // Small file - let timeout = config.calculate_timeout(1024); - assert!(timeout >= config.min_timeout); - - // Large file - let timeout = config.calculate_timeout(100 * 1024 * 1024); - assert!(timeout <= config.max_timeout); - } - - #[test] - fn test_timeout_config_validation() { - let config = TimeoutConfig { - min_timeout: Duration::from_secs(10), - max_timeout: Duration::from_secs(5), - ..Default::default() - }; - assert!(config.validate().is_err()); - } - - #[test] - fn test_adaptive_timeout_extreme_estimate_does_not_panic() { - // A huge object with a tiny historical rate used to overflow - // Duration::from_secs_f64 and panic; it must clamp to the upper bound. - let timeout = calculate_adaptive_timeout(Duration::from_secs(30), Some(1), 0, u64::MAX); - assert_eq!(timeout, Duration::from_secs(600)); - - // Tiny estimates clamp to the lower bound. - let timeout = calculate_adaptive_timeout(Duration::from_secs(30), Some(u64::MAX), 0, 1); - assert_eq!(timeout, Duration::from_secs(5)); - } - - #[test] - fn test_operation_progress() { - let progress = OperationProgress::new(Some(1000), Duration::from_secs(5)); - - assert_eq!(progress.current(), 0); - assert_eq!(progress.progress_percent(), Some(0.0)); - - progress.update(500); - assert_eq!(progress.current(), 500); - assert_eq!(progress.progress_percent(), Some(50.0)); - - progress.add(300); - assert_eq!(progress.current(), 800); - assert_eq!(progress.remaining(), Some(200)); - } - - #[test] - fn test_request_timeout_wrapper() { - let config = TimeoutConfig { - base_timeout: Duration::from_millis(100), - enable_dynamic_timeout: false, - ..Default::default() - }; - let wrapper = RequestTimeoutWrapper::new(config); - - assert!(!wrapper.is_timed_out(None)); - - std::thread::sleep(Duration::from_millis(150)); - - assert!(wrapper.is_timed_out(None)); - assert!(wrapper.check_timeout(None).is_err()); - } - - #[test] - fn test_timeout_stats() { - let stats = TimeoutStats::new(); - - stats.record_operation(Duration::from_millis(10)); - stats.record_operation(Duration::from_millis(20)); - stats.record_timeout(); - - assert_eq!(stats.total_operations.load(Ordering::Relaxed), 2); - assert_eq!(stats.timed_out.load(Ordering::Relaxed), 1); - assert!((stats.timeout_rate() - 0.5).abs() < 0.01); - } - - #[test] - fn test_progress_tracking() { - let config = TimeoutConfig::default(); - let wrapper = RequestTimeoutWrapper::with_progress(config, Some(1000), Duration::from_secs(1)); - - wrapper.update_progress(500); - assert_eq!(wrapper.progress_percent(), Some(50.0)); - assert!(!wrapper.is_stalled()); - } -} diff --git a/crates/io-core/src/writer.rs b/crates/io-core/src/writer.rs deleted file mode 100644 index 2d1f1a754..000000000 --- a/crates/io-core/src/writer.rs +++ /dev/null @@ -1,443 +0,0 @@ -// 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. - -//! BytesMut-backed object writer for optimized write operations. -//! -//! It uses `BytesMut` for efficient buffering; writes into that buffer may -//! still copy input bytes. The historical `ZeroCopyObjectWriter` name remains -//! available as a deprecated compatibility alias. - -use bytes::{BufMut, Bytes, BytesMut}; -use std::pin::Pin; -use std::task::{Context, Poll}; -use tokio::io::AsyncWrite; - -/// BytesMut-backed object writer for optimized write operations. -/// -/// This writer minimizes memory allocations by: -/// - Using BytesMut for efficient buffer growth -/// - Accepting `Bytes` inputs for efficient buffer handling -/// - Optional integration with BytesPool for buffer reuse -/// -/// # Example -/// -/// ```ignore -/// use rustfs_io_core::BytesMutWriter; -/// use bytes::Bytes; -/// -/// #[tokio::main] -/// async fn main() -> Result<(), Box> { -/// let mut writer = BytesMutWriter::new(); -/// -/// // Write into the internal BytesMut buffer -/// let data = Bytes::from("hello world"); -/// writer.write_buffered(data).await?; -/// -/// // Get the result as Bytes (zero-copy conversion) -/// let result = writer.into_bytes(); -/// -/// Ok(()) -/// } -/// ``` -pub struct BytesMutWriter { - /// Internal buffer using BytesMut for efficient growth - buffer: BytesMut, - /// Total bytes written - bytes_written: usize, - /// Whether the writer has been finalized - finalized: bool, -} - -/// Historical name for the BytesMut-backed object writer. -#[deprecated(since = "1.0.0-beta.8", note = "use BytesMutWriter; writes append into a BytesMut buffer")] -pub type ZeroCopyObjectWriter = BytesMutWriter; - -impl BytesMutWriter { - /// Create a new bytes-backed object writer with default capacity (8KB). - /// - /// # Example - /// - /// ```ignore - /// let writer = BytesMutWriter::new(); - /// ``` - pub fn new() -> Self { - Self::with_capacity(8 * 1024) - } - - /// Create a new bytes-backed object writer with specified capacity. - /// - /// # Arguments - /// - /// * `capacity` - Initial buffer capacity in bytes - /// - /// # Example - /// - /// ```ignore - /// let writer = BytesMutWriter::with_capacity(64 * 1024); - /// ``` - pub fn with_capacity(capacity: usize) -> Self { - Self { - buffer: BytesMut::with_capacity(capacity), - bytes_written: 0, - finalized: false, - } - } - - /// Write data into the internal buffer. - /// - /// This method accepts `Bytes` for API compatibility, then appends the - /// bytes into the internal `BytesMut` buffer. - /// - /// # Arguments - /// - /// * `data` - Data to append to the internal buffer - /// - /// # Returns - /// - /// * `Ok(usize)` - Number of bytes written - /// * `Err(ZeroCopyWriteError)` - Write error - /// - /// # Example - /// - /// ```ignore - /// let data = Bytes::from("hello world"); - /// let written = writer.write_buffered(data).await?; - /// ``` - pub async fn write_buffered(&mut self, data: Bytes) -> Result { - if self.finalized { - return Err(ZeroCopyWriteError::Finalized("Cannot write to finalized writer".to_string())); - } - - let len = data.len(); - self.buffer.put(data); - - self.bytes_written += len; - Ok(len) - } - - /// Historical name for `write_buffered`. - #[deprecated( - since = "1.0.0-beta.8", - note = "use write_buffered; this method appends bytes into an internal buffer" - )] - pub async fn write_zero_copy(&mut self, data: Bytes) -> Result { - self.write_buffered(data).await - } - - /// Write a slice of data. - /// - /// # Arguments - /// - /// * `data` - Data slice to write - /// - /// # Returns - /// - /// * `Ok(usize)` - Number of bytes written - /// * `Err(ZeroCopyWriteError)` - Write error - pub async fn write_slice(&mut self, data: &[u8]) -> Result { - if self.finalized { - return Err(ZeroCopyWriteError::Finalized("Cannot write to finalized writer".to_string())); - } - - let len = data.len(); - self.buffer.put_slice(data); - self.bytes_written += len; - Ok(len) - } - - /// Finalize the writer and consume it, returning the written data as Bytes. - /// - /// This converts the internal BytesMut to Bytes, which is a zero-copy - /// operation that freezes the buffer. - /// - /// # Returns - /// - /// The written data as Bytes - /// - /// # Example - /// - /// ```ignore - /// let result = writer.into_bytes(); - /// ``` - pub fn into_bytes(mut self) -> Bytes { - self.finalized = true; - self.buffer.freeze() - } - - /// Get the current buffer as a slice (without consuming). - /// - /// # Returns - /// - /// Slice of the current buffer content - pub fn as_slice(&self) -> &[u8] { - &self.buffer[..] - } - - /// Get the total number of bytes written. - /// - /// # Returns - /// - /// Number of bytes written - pub fn bytes_written(&self) -> usize { - self.bytes_written - } - - /// Get the current buffer capacity. - /// - /// # Returns - /// - /// Current buffer capacity in bytes - pub fn capacity(&self) -> usize { - self.buffer.capacity() - } - - /// Get the current buffer length. - /// - /// # Returns - /// - /// Current buffer length in bytes - pub fn len(&self) -> usize { - self.buffer.len() - } - - /// Check if the buffer is empty. - /// - /// # Returns - /// - /// `true` if buffer is empty, `false` otherwise - pub fn is_empty(&self) -> bool { - self.buffer.is_empty() - } - - /// Clear the buffer, resetting it to empty. - /// - /// This does not change the capacity, just resets the length to 0. - pub fn clear(&mut self) { - self.buffer.clear(); - self.bytes_written = 0; - self.finalized = false; - } - - /// Reserve additional capacity in the buffer. - /// - /// # Arguments - /// - /// * `additional` - Additional capacity to reserve - pub fn reserve(&mut self, additional: usize) { - self.buffer.reserve(additional); - } -} - -impl Default for BytesMutWriter { - fn default() -> Self { - Self::new() - } -} - -impl std::fmt::Debug for BytesMutWriter { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("BytesMutWriter") - .field("buffer_len", &self.buffer.len()) - .field("buffer_capacity", &self.buffer.capacity()) - .field("bytes_written", &self.bytes_written) - .field("finalized", &self.finalized) - .finish() - } -} - -/// AsyncWrite implementation for BytesMutWriter. -/// -/// This allows the writer to be used with tokio's async I/O utilities. -impl AsyncWrite for BytesMutWriter { - fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { - if self.finalized { - return Poll::Ready(Err(tokio::io::Error::new( - tokio::io::ErrorKind::WriteZero, - "Cannot write to finalized writer", - ))); - } - - let len = buf.len(); - self.buffer.put_slice(buf); - self.bytes_written += len; - Poll::Ready(Ok(len)) - } - - fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - // Nothing to flush for in-memory buffer - Poll::Ready(Ok(())) - } - - fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - self.finalized = true; - Poll::Ready(Ok(())) - } -} - -/// Zero-copy write error types. -#[derive(Debug, thiserror::Error)] -pub enum ZeroCopyWriteError { - /// I/O error occurred - #[error("I/O error: {0}")] - Io(#[from] tokio::io::Error), - - /// Writer has been finalized and cannot accept more writes - #[error("Writer finalized: {0}")] - Finalized(String), - - /// Invalid input provided - #[error("Invalid input: {0}")] - InvalidInput(String), -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_new_writer() { - let writer = BytesMutWriter::new(); - assert!(writer.is_empty()); - assert_eq!(writer.bytes_written(), 0); - assert!(writer.capacity() >= 8 * 1024); - } - - #[tokio::test] - async fn test_write_buffered() { - let mut writer = BytesMutWriter::new(); - let data = Bytes::from("hello world"); - - let written = writer.write_buffered(data).await.unwrap(); - assert_eq!(written, 11); - assert_eq!(writer.bytes_written(), 11); - assert_eq!(writer.as_slice(), b"hello world"); - } - - #[tokio::test] - async fn test_preferred_writer_alias() { - let mut writer = BytesMutWriter::new(); - let written = writer - .write_buffered(Bytes::from("hello world")) - .await - .expect("write bytes through alias"); - - assert_eq!(written, 11); - assert_eq!(writer.as_slice(), b"hello world"); - } - - #[tokio::test] - async fn test_write_slice() { - let mut writer = BytesMutWriter::new(); - let data = b"hello world"; - - let written = writer.write_slice(data).await.unwrap(); - assert_eq!(written, 11); - assert_eq!(writer.bytes_written(), 11); - assert_eq!(writer.as_slice(), b"hello world"); - } - - #[tokio::test] - async fn test_into_bytes() { - let mut writer = BytesMutWriter::new(); - let data = Bytes::from("hello world"); - - writer.write_buffered(data).await.unwrap(); - let result = writer.into_bytes(); - - assert_eq!(result.as_ref(), b"hello world"); - } - - #[tokio::test] - async fn test_write_after_finalize() { - let mut writer = BytesMutWriter::new(); - let data = Bytes::from("hello"); - - writer.write_buffered(data).await.unwrap(); - let _result = writer.into_bytes(); - - // Create new writer and try to write after finalize - let mut writer2 = BytesMutWriter::new(); - writer2.write_buffered(Bytes::from("test")).await.unwrap(); - let _ = writer2.into_bytes(); - - // Writing to a consumed writer should work via new writer - let mut writer3 = BytesMutWriter::new(); - let result = writer3.write_buffered(Bytes::from("final")).await; - assert!(result.is_ok()); - } - - #[tokio::test] - async fn test_clear() { - let mut writer = BytesMutWriter::new(); - writer.write_slice(b"hello").await.unwrap(); - - writer.clear(); - assert!(writer.is_empty()); - assert_eq!(writer.bytes_written(), 0); - // Capacity should remain - assert!(writer.capacity() > 0); - } - - #[tokio::test] - async fn test_reserve() { - let mut writer = BytesMutWriter::with_capacity(10); - let initial_capacity = writer.capacity(); - - writer.reserve(1000); - // Reserve ensures at least the additional capacity can be added - // but may allocate more than requested - assert!(writer.capacity() >= initial_capacity); - } - - #[tokio::test] - async fn test_multiple_writes() { - let mut writer = BytesMutWriter::new(); - - writer.write_buffered(Bytes::from("hello ")).await.unwrap(); - writer.write_slice(b"world").await.unwrap(); - - assert_eq!(writer.as_slice(), b"hello world"); - assert_eq!(writer.bytes_written(), 11); - } - - #[tokio::test] - async fn test_async_write() { - use tokio::io::AsyncWriteExt; - - let mut writer = BytesMutWriter::new(); - let data = b"hello world"; - - let written = writer.write(data).await.unwrap(); - assert_eq!(written, 11); - assert_eq!(writer.as_slice(), b"hello world"); - } - - #[tokio::test] - async fn test_debug() { - let writer = BytesMutWriter::new(); - let debug_str = format!("{:?}", writer); - assert!(debug_str.contains("BytesMutWriter")); - assert!(debug_str.contains("buffer_len")); - } - - #[tokio::test] - #[allow(deprecated)] - async fn test_legacy_writer_alias() { - let mut writer = ZeroCopyObjectWriter::new(); - let written = writer.write_zero_copy(Bytes::from("hello")).await.unwrap(); - - assert_eq!(written, 5); - assert_eq!(writer.as_slice(), b"hello"); - } -}