rustfs-io-core

Overview

rustfs-io-core holds the shared I/O primitives for RustFS, a distributed object storage system. It provides:

• 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
• 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

Backpressure Control

System overload protection:

use rustfs_io_core::{BackpressureMonitor, BackpressureState, BackpressureConfig};

let config = BackpressureConfig {
    high_watermark: 0.8,  // 80% triggers backpressure
    low_watermark: 0.5,   // 50% releases backpressure
    ..Default::default()
};
let monitor = BackpressureMonitor::new(config);

// Check state
match monitor.state() {
    BackpressureState::Normal => println!("System normal"),
    BackpressureState::Warning => println!("System warning"),
    BackpressureState::Critical => println!("System overloaded"),
}

Deadlock Detection

Wait-for graph based deadlock detection:

use rustfs_io_core::{DeadlockDetector, LockType};

let detector = DeadlockDetector::with_defaults();

// Register locks
let lock1 = detector.register_lock(LockType::Mutex);
let lock2 = detector.register_lock(LockType::RwLockWrite);

// Record lock acquisition
detector.record_acquire(lock1, 1);  // Thread 1 acquires lock1
detector.record_wait(lock2, 1);     // Thread 1 waits for lock2

// Detect deadlock
if let Some(deadlock) = detector.detect_deadlock() {
    println!("Deadlock detected: {:?}", deadlock);
}

Lock Optimizer

Adaptive spin lock optimization:

use rustfs_io_core::{LockOptimizer, LockOptimizeConfig};

let optimizer = LockOptimizer::with_defaults();

// Record lock operations
optimizer.on_acquire();
// ... do work ...
optimizer.on_release(std::time::Duration::from_millis(10));

// View statistics
let stats = optimizer.stats();
println!("Locks acquired: {}", stats.total_acquired());

Progress Tracking

Byte progress and staleness for long-running operations:

use rustfs_io_core::OperationProgress;
use std::time::Duration;

let progress = OperationProgress::new(Some(1000), Duration::from_secs(5));

progress.update(500);
assert_eq!(progress.progress_percent(), Some(50.0));
assert!(!progress.is_stale());

Configuration

Code Configuration

use rustfs_io_core::IoSchedulerConfig;

let config = IoSchedulerConfig {
    max_concurrent_reads: 128,
    base_buffer_size: 128 * 1024,
    max_buffer_size: 4 * 1024 * 1024,
    high_priority_threshold: 64 * 1024,
    low_priority_threshold: 4 * 1024 * 1024,
    ..Default::default()
};

// Validate configuration
if let Err(e) = config.validate() {
    panic!("Invalid configuration: {}", e);
}

Module Structure

rustfs-io-core/
├── src/
│   ├── lib.rs              # Module entry
│   ├── config.rs           # Configuration types
│   ├── pool.rs             # Tiered buffer pool
│   ├── backpressure.rs     # Backpressure control
│   ├── deadlock_detector.rs # Deadlock detection
│   ├── lock_optimizer.rs   # Lock optimization
│   ├── progress.rs         # Operation progress tracking
│   └── io_profile.rs       # I/O profile
└── Cargo.toml

Testing

# Run all tests
cargo nextest run --package rustfs-io-core

# Run specific tests
cargo nextest run --package rustfs-io-core -E 'test(backpressure)'

Documentation

• API Documentation

Related Modules

• rustfs-io-metrics: Metrics collection and configuration
• rustfs: Main storage service

License

Apache License 2.0