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

chore(io-core): drop eight zero-consumer modules (#6201)

Browse Source
This commit is contained in:
Zhengchao An
2026-08-18 16:14:05 +08:00
committed by GitHub
parent a5800033bd
commit bd296eff9e
17 changed files with 195 additions and 4008 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/io-core/CHANGELOG.md
+3
View File
@@ -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.
crates/io-core/Cargo.toml
+2 -3
View File
@@ -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 }
crates/io-core/README.md
+18 -120
View File
@@ -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
crates/io-core/README_zh.md
+18 -131
View File
@@ -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)
## 🔗 相关模块
crates/io-core/examples/scheduler_example.rs
-190
View File
@@ -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!();
}
crates/io-core/src/bufreader_optimizer.rs
-227
View File
@@ -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>) -> 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<R: tokio::io::AsyncRead + Unpin>(&self, reader: R, data_size: Option<usize>) -> tokio::io::BufReader<R> {
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<R: BufferedSource + ?Sized>(&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<R: tokio::io::AsyncRead + Unpin>(&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);
}
}
crates/io-core/src/direct_io.rs
-332
View File
@@ -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<io::Error> 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<u8>,
/// 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<Self, AlignedPreadError> {
// 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<usize> {
// 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<io::Result<()>> {
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<Self, AlignedPreadError> {
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<io::Result<()>> {
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)));
}
}
}
crates/io-core/src/io_priority_queue.rs
-381
View File
@@ -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<Duration>,
/// 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<IoRequest>,
/// Normal priority queue.
normal: VecDeque<IoRequest>,
/// Low priority queue.
low: VecDeque<IoRequest>,
/// Next request ID.
next_id: u64,
/// Last dequeue time for each priority (for starvation prevention).
last_dequeue: [Option<Instant>; 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<IoRequest> {
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);
}
}
crates/io-core/src/lib.rs
+12 -61
View File
@@ -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;
crates/io-core/src/progress.rs
+138
View File
@@ -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<u64>,
/// Bytes processed.
bytes_processed: AtomicU64,
/// Last update time.
last_update: std::sync::Mutex<Instant>,
/// 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<u64>, 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<f64> {
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<u64> {
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));
}
}
crates/io-core/src/reader.rs
-412
View File
@@ -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<io::Error> 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<Self, ZeroCopyReadError> {
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<Self, ZeroCopyReadError> {
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<Self, ZeroCopyReadError> {
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, ZeroCopyReadError> {
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<io::Result<()>> {
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");
}
}
crates/io-core/src/scheduler.rs
-882
View File
@@ -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<u64>,
/// 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<IoLoadMetrics>,
}
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<u64>,
/// 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);
}
}
crates/io-core/src/shared_memory.rs
-320
View File
@@ -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<usize>,
/// 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<T> {
/// The wrapped data
inner: Arc<T>,
/// Metadata about the data
metadata: ArcMetadata,
}
impl<T> Clone for ArcData<T> {
fn clone(&self) -> Self {
Self {
inner: Arc::clone(&self.inner),
metadata: self.metadata.clone(),
}
}
}
impl<T> ArcData<T> {
/// 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<T> {
self.inner
}
/// Get the metadata.
pub fn metadata(&self) -> &ArcMetadata {
&self.metadata
}
/// Get the size if known.
pub fn size(&self) -> Option<usize> {
self.metadata.size
}
}
impl<T> AsRef<T> for ArcData<T> {
fn as_ref(&self) -> &T {
&self.inner
}
}
impl<T> Deref for ArcData<T> {
type Target = T;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl<T> std::fmt::Debug for ArcData<T> {
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<T>(&self, data: T) -> ArcData<T> {
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<T>(&self, data: T, size: usize) -> ArcData<T> {
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<T>(&self, data: &ArcData<T>) -> ArcData<T> {
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);
}
}
crates/io-core/src/timeout_wrapper.rs
-501
View File
@@ -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<u64>,
/// Bytes processed.
bytes_processed: AtomicU64,
/// Last update time.
last_update: std::sync::Mutex<Instant>,
/// 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<u64>, 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<f64> {
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<u64> {
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<OperationProgress>,
}
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<u64>, 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<Duration> {
let elapsed = self.elapsed();
if elapsed >= timeout { None } else { Some(timeout - elapsed) }
}
/// Check if timed out.
pub fn is_timed_out(&self, size: Option<u64>) -> bool {
let timeout = self.get_timeout(size);
self.elapsed() > timeout
}
/// Get the timeout for a given size.
pub fn get_timeout(&self, size: Option<u64>) -> 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<u64>) -> 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<f64> {
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<u64>,
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());
}
}
crates/io-core/src/writer.rs
-443
View File
@@ -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<dyn std::error::Error>> {
/// 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<usize, ZeroCopyWriteError> {
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<usize, ZeroCopyWriteError> {
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<usize, ZeroCopyWriteError> {
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<Result<usize, tokio::io::Error>> {
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<Result<(), tokio::io::Error>> {
// Nothing to flush for in-memory buffer
Poll::Ready(Ok(()))
}
fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), tokio::io::Error>> {
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");
}
}