mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-27 15:37:02 +00:00
feat(io-core): add truthful buffered io aliases (#4032)
This commit is contained in:
@@ -82,7 +82,7 @@ impl From<io::Error> for DirectIoError {
|
|||||||
/// # Platform Support
|
/// # Platform Support
|
||||||
///
|
///
|
||||||
/// Only available on Linux (uses `FileExt::read_at`). On other platforms,
|
/// Only available on Linux (uses `FileExt::read_at`). On other platforms,
|
||||||
/// use `ZeroCopyObjectReader` with memory mapping instead.
|
/// use `BytesBufferedReader` instead.
|
||||||
///
|
///
|
||||||
/// # Alignment Requirements
|
/// # Alignment Requirements
|
||||||
///
|
///
|
||||||
@@ -94,11 +94,11 @@ impl From<io::Error> for DirectIoError {
|
|||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// use rustfs_io_core::DirectIoReader;
|
/// use rustfs_io_core::AlignedPreadReader;
|
||||||
///
|
///
|
||||||
/// // Linux only
|
/// // Linux only
|
||||||
/// #[cfg(target_os = "linux")]
|
/// #[cfg(target_os = "linux")]
|
||||||
/// let reader = DirectIoReader::new(file, offset, size)?;
|
/// let reader = AlignedPreadReader::new(file, offset, size)?;
|
||||||
/// ```
|
/// ```
|
||||||
#[cfg(target_os = "linux")]
|
#[cfg(target_os = "linux")]
|
||||||
pub struct DirectIoReader {
|
pub struct DirectIoReader {
|
||||||
@@ -263,6 +263,12 @@ impl std::fmt::Debug for DirectIoReader {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Preferred name for aligned pread errors.
|
||||||
|
pub type AlignedPreadError = DirectIoError;
|
||||||
|
|
||||||
|
/// Preferred name for the aligned pread-based reader.
|
||||||
|
pub type AlignedPreadReader = DirectIoReader;
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
@@ -275,6 +281,12 @@ mod tests {
|
|||||||
let file = std::fs::File::open("/dev/zero").unwrap();
|
let file = std::fs::File::open("/dev/zero").unwrap();
|
||||||
assert!(DirectIoReader::new(file, 0, 512).is_ok(), "Should succeed with aligned offset and size");
|
assert!(DirectIoReader::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
|
// Invalid offset
|
||||||
let file = std::fs::File::open("/dev/zero").unwrap();
|
let file = std::fs::File::open("/dev/zero").unwrap();
|
||||||
assert!(DirectIoReader::new(file, 1, 512).is_err(), "Should fail with unaligned offset");
|
assert!(DirectIoReader::new(file, 1, 512).is_err(), "Should fail with unaligned offset");
|
||||||
|
|||||||
+10
-11
@@ -15,9 +15,9 @@
|
|||||||
//! Buffered I/O reader and writer implementations for RustFS.
|
//! Buffered I/O reader and writer implementations for RustFS.
|
||||||
//!
|
//!
|
||||||
//! This crate provides buffered readers and writers for I/O operations.
|
//! This crate provides buffered readers and writers for I/O operations.
|
||||||
//! Note: despite "ZeroCopy" naming in type names (kept for backward compatibility),
|
//! Prefer `BytesBufferedReader`, `BytesMutWriter`, and `AlignedPreadReader`
|
||||||
//! the actual implementations perform mmap-then-copy or aligned pread, not true
|
//! for new code. Historical `ZeroCopy*` and `DirectIo*` names remain exported
|
||||||
//! zero-copy file I/O or true Direct I/O.
|
//! for backward compatibility.
|
||||||
//!
|
//!
|
||||||
//! # Features
|
//! # Features
|
||||||
//!
|
//!
|
||||||
@@ -30,16 +30,15 @@
|
|||||||
//! # Example
|
//! # Example
|
||||||
//!
|
//!
|
||||||
//! ```ignore
|
//! ```ignore
|
||||||
//! use rustfs_io_core::{ZeroCopyObjectReader, BytesPool};
|
//! use rustfs_io_core::{BytesBufferedReader, BytesPool};
|
||||||
//! use bytes::Bytes;
|
//! use bytes::Bytes;
|
||||||
//!
|
//!
|
||||||
//! // Create from existing bytes (zero-copy)
|
//! // Create from existing bytes (zero-copy)
|
||||||
//! let data = Bytes::from("hello world");
|
//! let data = Bytes::from("hello world");
|
||||||
//! let reader = ZeroCopyObjectReader::from_bytes(data);
|
//! let reader = BytesBufferedReader::from_bytes(data);
|
||||||
//!
|
//!
|
||||||
//! // Create from file using mmap-then-copy (Unix only)
|
//! // Create from file using buffered reads
|
||||||
//! #[cfg(unix)]
|
//! let reader = BytesBufferedReader::from_file_read(&file, 0, 1024).await?;
|
||||||
//! let reader = ZeroCopyObjectReader::from_file_mmap(&file, 0, 1024).await?;
|
|
||||||
//!
|
//!
|
||||||
//! // Use BytesPool
|
//! // Use BytesPool
|
||||||
//! let pool = BytesPool::new_tiered();
|
//! let pool = BytesPool::new_tiered();
|
||||||
@@ -62,10 +61,10 @@ pub mod timeout_wrapper;
|
|||||||
pub mod writer;
|
pub mod writer;
|
||||||
|
|
||||||
#[cfg(target_os = "linux")]
|
#[cfg(target_os = "linux")]
|
||||||
pub use direct_io::{DirectIoError, DirectIoReader};
|
pub use direct_io::{AlignedPreadError, AlignedPreadReader, DirectIoError, DirectIoReader};
|
||||||
pub use pool::{BytesPool, BytesPoolConfig, BytesPoolMetrics, PooledBuffer};
|
pub use pool::{BytesPool, BytesPoolConfig, BytesPoolMetrics, PooledBuffer};
|
||||||
pub use reader::{ZeroCopyObjectReader, ZeroCopyReadError};
|
pub use reader::{BytesBufferedReader, ZeroCopyObjectReader, ZeroCopyReadError};
|
||||||
pub use writer::{ZeroCopyObjectWriter, ZeroCopyWriteError};
|
pub use writer::{BytesMutWriter, ZeroCopyObjectWriter, ZeroCopyWriteError};
|
||||||
|
|
||||||
// BufReader optimizer exports
|
// BufReader optimizer exports
|
||||||
pub use bufreader_optimizer::{BufReaderConfig, BufReaderOptimizer, BufReaderStats, BufferedSource};
|
pub use bufreader_optimizer::{BufReaderConfig, BufReaderOptimizer, BufReaderStats, BufferedSource};
|
||||||
|
|||||||
@@ -58,9 +58,12 @@ impl From<io::Error> for ZeroCopyReadError {
|
|||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// // Create from bytes (zero-copy)
|
/// use bytes::Bytes;
|
||||||
|
/// use rustfs_io_core::BytesBufferedReader;
|
||||||
|
///
|
||||||
|
/// // Create from bytes without copying the `Bytes` buffer
|
||||||
/// let data = Bytes::from("hello world");
|
/// let data = Bytes::from("hello world");
|
||||||
/// let reader = ZeroCopyObjectReader::from_bytes(data);
|
/// let reader = BytesBufferedReader::from_bytes(data);
|
||||||
///
|
///
|
||||||
/// // Read using AsyncRead trait
|
/// // Read using AsyncRead trait
|
||||||
/// let mut buf = vec![0u8; 1024];
|
/// let mut buf = vec![0u8; 1024];
|
||||||
@@ -73,8 +76,11 @@ pub struct ZeroCopyObjectReader {
|
|||||||
pos: usize,
|
pos: usize,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Preferred name for the bytes-backed object reader.
|
||||||
|
pub type BytesBufferedReader = ZeroCopyObjectReader;
|
||||||
|
|
||||||
impl ZeroCopyObjectReader {
|
impl ZeroCopyObjectReader {
|
||||||
/// Create a zero-copy reader from existing bytes.
|
/// Create a reader from existing bytes.
|
||||||
///
|
///
|
||||||
/// This is a true zero-copy operation - the Bytes are wrapped
|
/// This is a true zero-copy operation - the Bytes are wrapped
|
||||||
/// without any allocation or copying.
|
/// without any allocation or copying.
|
||||||
@@ -87,7 +93,7 @@ impl ZeroCopyObjectReader {
|
|||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// let data = Bytes::from("hello world");
|
/// let data = Bytes::from("hello world");
|
||||||
/// let reader = ZeroCopyObjectReader::from_bytes(data);
|
/// let reader = BytesBufferedReader::from_bytes(data);
|
||||||
/// ```
|
/// ```
|
||||||
pub fn from_bytes(data: Bytes) -> Self {
|
pub fn from_bytes(data: Bytes) -> Self {
|
||||||
Self { data, pos: 0 }
|
Self { data, pos: 0 }
|
||||||
@@ -102,7 +108,7 @@ impl ZeroCopyObjectReader {
|
|||||||
///
|
///
|
||||||
/// * `path` - Path to the file to memory map
|
/// * `path` - Path to the file to memory map
|
||||||
/// * `offset` - Offset within the file to start reading
|
/// * `offset` - Offset within the file to start reading
|
||||||
/// * `size` - Number of bytes to map
|
/// * `size` - Number of bytes to read
|
||||||
///
|
///
|
||||||
/// # Returns
|
/// # Returns
|
||||||
///
|
///
|
||||||
@@ -115,7 +121,7 @@ impl ZeroCopyObjectReader {
|
|||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// let reader = ZeroCopyObjectReader::from_file_mmap_path("large_file.bin", 0, 1024).await?;
|
/// let reader = BytesBufferedReader::from_file_mmap_path("large_file.bin", 0, 1024).await?;
|
||||||
/// ```
|
/// ```
|
||||||
#[cfg(unix)]
|
#[cfg(unix)]
|
||||||
// SAFETY: The mmap is created from a read-only file handle for the
|
// SAFETY: The mmap is created from a read-only file handle for the
|
||||||
@@ -154,7 +160,7 @@ impl ZeroCopyObjectReader {
|
|||||||
///
|
///
|
||||||
/// # Arguments
|
/// # Arguments
|
||||||
///
|
///
|
||||||
/// * `file` - File to memory map
|
/// * `file` - File to read from
|
||||||
/// * `offset` - Offset within the file to start reading
|
/// * `offset` - Offset within the file to start reading
|
||||||
/// * `size` - Number of bytes to map
|
/// * `size` - Number of bytes to map
|
||||||
///
|
///
|
||||||
@@ -164,13 +170,13 @@ impl ZeroCopyObjectReader {
|
|||||||
///
|
///
|
||||||
/// # Errors
|
/// # Errors
|
||||||
///
|
///
|
||||||
/// Returns an error if the file cannot be memory mapped.
|
/// Returns an error if the file cannot be read.
|
||||||
///
|
///
|
||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// let file = tokio::fs::File::open("large_file.bin").await?;
|
/// let file = tokio::fs::File::open("large_file.bin").await?;
|
||||||
/// let reader = ZeroCopyObjectReader::from_file_mmap(&file, 0, 1024).await?;
|
/// let reader = BytesBufferedReader::from_file_read(&file, 0, 1024).await?;
|
||||||
/// ```
|
/// ```
|
||||||
#[cfg(unix)]
|
#[cfg(unix)]
|
||||||
pub async fn from_file_mmap(file: &tokio::fs::File, offset: u64, size: usize) -> Result<Self, ZeroCopyReadError> {
|
pub async fn from_file_mmap(file: &tokio::fs::File, offset: u64, size: usize) -> Result<Self, ZeroCopyReadError> {
|
||||||
@@ -209,6 +215,14 @@ impl ZeroCopyObjectReader {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Create a Bytes-backed reader from a file using normal buffered reads.
|
||||||
|
///
|
||||||
|
/// This is the preferred name for new code. It is currently equivalent to
|
||||||
|
/// the historical `from_file_mmap` compatibility method.
|
||||||
|
pub async fn from_file_read(file: &tokio::fs::File, offset: u64, size: usize) -> Result<Self, ZeroCopyReadError> {
|
||||||
|
Self::from_file_mmap(file, offset, size).await
|
||||||
|
}
|
||||||
|
|
||||||
/// Get the remaining data as Bytes (zero-copy).
|
/// Get the remaining data as Bytes (zero-copy).
|
||||||
///
|
///
|
||||||
/// This returns a slice of the remaining data without copying.
|
/// This returns a slice of the remaining data without copying.
|
||||||
@@ -283,6 +297,18 @@ mod tests {
|
|||||||
assert_eq!(&buf[..n], b"hello world");
|
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]
|
#[tokio::test]
|
||||||
async fn test_remaining_bytes() {
|
async fn test_remaining_bytes() {
|
||||||
let data = Bytes::from("hello world");
|
let data = Bytes::from("hello world");
|
||||||
|
|||||||
@@ -33,16 +33,16 @@ use tokio::io::AsyncWrite;
|
|||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// use rustfs_io_core::ZeroCopyObjectWriter;
|
/// use rustfs_io_core::BytesMutWriter;
|
||||||
/// use bytes::Bytes;
|
/// use bytes::Bytes;
|
||||||
///
|
///
|
||||||
/// #[tokio::main]
|
/// #[tokio::main]
|
||||||
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||||
/// let mut writer = ZeroCopyObjectWriter::new();
|
/// let mut writer = BytesMutWriter::new();
|
||||||
///
|
///
|
||||||
/// // Write with zero-copy
|
/// // Write into the internal BytesMut buffer
|
||||||
/// let data = Bytes::from("hello world");
|
/// let data = Bytes::from("hello world");
|
||||||
/// writer.write_zero_copy(data).await?;
|
/// writer.write_buffered(data).await?;
|
||||||
///
|
///
|
||||||
/// // Get the result as Bytes (zero-copy conversion)
|
/// // Get the result as Bytes (zero-copy conversion)
|
||||||
/// let result = writer.into_bytes();
|
/// let result = writer.into_bytes();
|
||||||
@@ -59,19 +59,22 @@ pub struct ZeroCopyObjectWriter {
|
|||||||
finalized: bool,
|
finalized: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Preferred name for the BytesMut-backed object writer.
|
||||||
|
pub type BytesMutWriter = ZeroCopyObjectWriter;
|
||||||
|
|
||||||
impl ZeroCopyObjectWriter {
|
impl ZeroCopyObjectWriter {
|
||||||
/// Create a new zero-copy object writer with default capacity (8KB).
|
/// Create a new bytes-backed object writer with default capacity (8KB).
|
||||||
///
|
///
|
||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// let writer = ZeroCopyObjectWriter::new();
|
/// let writer = BytesMutWriter::new();
|
||||||
/// ```
|
/// ```
|
||||||
pub fn new() -> Self {
|
pub fn new() -> Self {
|
||||||
Self::with_capacity(8 * 1024)
|
Self::with_capacity(8 * 1024)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Create a new zero-copy object writer with specified capacity.
|
/// Create a new bytes-backed object writer with specified capacity.
|
||||||
///
|
///
|
||||||
/// # Arguments
|
/// # Arguments
|
||||||
///
|
///
|
||||||
@@ -80,7 +83,7 @@ impl ZeroCopyObjectWriter {
|
|||||||
/// # Example
|
/// # Example
|
||||||
///
|
///
|
||||||
/// ```ignore
|
/// ```ignore
|
||||||
/// let writer = ZeroCopyObjectWriter::with_capacity(64 * 1024);
|
/// let writer = BytesMutWriter::with_capacity(64 * 1024);
|
||||||
/// ```
|
/// ```
|
||||||
pub fn with_capacity(capacity: usize) -> Self {
|
pub fn with_capacity(capacity: usize) -> Self {
|
||||||
Self {
|
Self {
|
||||||
@@ -122,6 +125,14 @@ impl ZeroCopyObjectWriter {
|
|||||||
Ok(len)
|
Ok(len)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Write data into the internal buffer.
|
||||||
|
///
|
||||||
|
/// This is the preferred name for new code. It is currently equivalent to
|
||||||
|
/// the historical `write_zero_copy` compatibility method.
|
||||||
|
pub async fn write_buffered(&mut self, data: Bytes) -> Result<usize, ZeroCopyWriteError> {
|
||||||
|
self.write_zero_copy(data).await
|
||||||
|
}
|
||||||
|
|
||||||
/// Write a slice of data.
|
/// Write a slice of data.
|
||||||
///
|
///
|
||||||
/// # Arguments
|
/// # Arguments
|
||||||
@@ -311,6 +322,18 @@ mod tests {
|
|||||||
assert_eq!(writer.as_slice(), b"hello world");
|
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]
|
#[tokio::test]
|
||||||
async fn test_write_slice() {
|
async fn test_write_slice() {
|
||||||
let mut writer = ZeroCopyObjectWriter::new();
|
let mut writer = ZeroCopyObjectWriter::new();
|
||||||
|
|||||||
Reference in New Issue
Block a user