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

refactor(io): make accurate reader names primary (#4060)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-29 20:20:22 +08:00
committed by GitHub
parent 14e9fa2c8d
commit 9567e7b3be
4 changed files with 114 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/io-core/src/direct_io.rs
+37 -15
View File
@@ -101,7 +101,7 @@ impl From<io::Error> for DirectIoError {
/// let reader = AlignedPreadReader::new(file, offset, size)?;
/// ```
#[cfg(target_os = "linux")]
pub struct DirectIoReader {
pub struct AlignedPreadReader {
/// Underlying file handle used for aligned pread I/O
file: std::fs::File,
/// Current read position
@@ -117,7 +117,7 @@ pub struct DirectIoReader {
}
#[cfg(target_os = "linux")]
impl DirectIoReader {
impl AlignedPreadReader {
/// Alignment requirement for reads (512 bytes for most systems)
pub const ALIGNMENT: usize = 512;
@@ -131,7 +131,7 @@ impl DirectIoReader {
///
/// # Returns
///
/// A `DirectIoReader` that reads the file at the given offset.
/// An `AlignedPreadReader` that reads the file at the given offset.
///
/// # Errors
///
@@ -197,7 +197,7 @@ impl DirectIoReader {
}
#[cfg(target_os = "linux")]
impl AsyncRead for DirectIoReader {
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();
@@ -222,12 +222,12 @@ impl AsyncRead for DirectIoReader {
/// 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 DirectIoReader {
pub struct AlignedPreadReader {
_priv: (),
}
#[cfg(not(target_os = "linux"))]
impl DirectIoReader {
impl AlignedPreadReader {
/// Create a new aligned pread reader (not supported on this platform).
///
/// Always returns an error on non-Linux platforms.
@@ -237,7 +237,7 @@ impl DirectIoReader {
}
#[cfg(not(target_os = "linux"))]
impl AsyncRead for DirectIoReader {
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,
@@ -246,11 +246,11 @@ impl AsyncRead for DirectIoReader {
}
}
impl std::fmt::Debug for DirectIoReader {
impl std::fmt::Debug for AlignedPreadReader {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
#[cfg(target_os = "linux")]
{
f.debug_struct("DirectIoReader")
f.debug_struct("AlignedPreadReader")
.field("pos", &self.pos)
.field("remaining", &self.remaining)
.field("buffer_len", &self.buffer_len)
@@ -258,7 +258,9 @@ impl std::fmt::Debug for DirectIoReader {
}
#[cfg(not(target_os = "linux"))]
{
f.debug_struct("DirectIoReader").field("platform", &"unsupported").finish()
f.debug_struct("AlignedPreadReader")
.field("platform", &"unsupported")
.finish()
}
}
}
@@ -266,8 +268,9 @@ 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;
/// 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 {
@@ -279,7 +282,10 @@ mod tests {
{
// Valid alignment
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!(
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!(
@@ -289,16 +295,32 @@ mod tests {
// Invalid offset
let file = std::fs::File::open("/dev/zero").unwrap();
assert!(DirectIoReader::new(file, 1, 512).is_err(), "Should fail with unaligned offset");
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!(DirectIoReader::new(file, 0, 511).is_err(), "Should fail with unaligned size");
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(DirectIoError::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(DirectIoError::UnsupportedPlatform)));
}
crates/io-core/src/lib.rs
+10 -3
View File
@@ -61,10 +61,17 @@ pub mod timeout_wrapper;
pub mod writer;
#[cfg(target_os = "linux")]
pub use direct_io::{AlignedPreadError, AlignedPreadReader, DirectIoError, DirectIoReader};
#[allow(deprecated)]
pub use direct_io::DirectIoReader;
#[cfg(target_os = "linux")]
pub use direct_io::{AlignedPreadError, AlignedPreadReader, DirectIoError};
pub use pool::{BytesPool, BytesPoolConfig, BytesPoolMetrics, PooledBuffer};
pub use reader::{BytesBufferedReader, ZeroCopyObjectReader, ZeroCopyReadError};
pub use writer::{BytesMutWriter, ZeroCopyObjectWriter, ZeroCopyWriteError};
#[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};
crates/io-core/src/reader.rs
+31 -15
View File
@@ -51,9 +51,8 @@ impl From<io::Error> for ZeroCopyReadError {
/// Bytes-backed object reader.
///
/// This reader keeps the historical `ZeroCopyObjectReader` name for public API
/// compatibility. `from_bytes` wraps existing `Bytes` without copying, but file
/// constructors copy file data into owned `Bytes` after mmap or normal reads.
/// `from_bytes` wraps existing `Bytes` without copying, but file constructors
/// copy file data into owned `Bytes` after mmap or normal reads.
///
/// # Example
///
@@ -69,17 +68,21 @@ impl From<io::Error> for ZeroCopyReadError {
/// let mut buf = vec![0u8; 1024];
/// let n = reader.read(&mut buf[..]).await?;
/// ```
pub struct ZeroCopyObjectReader {
pub struct BytesBufferedReader {
/// Internal data source (could be mmap or owned bytes)
data: Bytes,
/// Current read position
pos: usize,
}
/// Preferred name for the bytes-backed object reader.
pub type BytesBufferedReader = ZeroCopyObjectReader;
/// 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 ZeroCopyObjectReader {
impl BytesBufferedReader {
/// Create a reader from existing bytes.
///
/// This is a true zero-copy operation - the Bytes are wrapped
@@ -254,7 +257,7 @@ impl ZeroCopyObjectReader {
}
}
impl AsyncRead for ZeroCopyObjectReader {
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 {
@@ -270,9 +273,9 @@ impl AsyncRead for ZeroCopyObjectReader {
}
}
impl std::fmt::Debug for ZeroCopyObjectReader {
impl std::fmt::Debug for BytesBufferedReader {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ZeroCopyObjectReader")
f.debug_struct("BytesBufferedReader")
.field("data_len", &self.data.len())
.field("pos", &self.pos)
.field("remaining", &(self.data.len() - self.pos))
@@ -288,7 +291,7 @@ mod tests {
#[tokio::test]
async fn test_from_bytes() {
let data = Bytes::from("hello world");
let mut reader = ZeroCopyObjectReader::from_bytes(data.clone());
let mut reader = BytesBufferedReader::from_bytes(data.clone());
let mut buf = [0u8; 11];
let n = reader.read(&mut buf[..]).await.unwrap();
@@ -312,7 +315,7 @@ mod tests {
#[tokio::test]
async fn test_remaining_bytes() {
let data = Bytes::from("hello world");
let reader = ZeroCopyObjectReader::from_bytes(data);
let reader = BytesBufferedReader::from_bytes(data);
let remaining = reader.remaining_bytes();
assert_eq!(remaining.len(), 11);
@@ -322,7 +325,7 @@ mod tests {
#[tokio::test]
async fn test_position() {
let data = Bytes::from("hello world");
let mut reader = ZeroCopyObjectReader::from_bytes(data);
let mut reader = BytesBufferedReader::from_bytes(data);
assert_eq!(reader.position(), 0);
@@ -335,11 +338,24 @@ mod tests {
#[tokio::test]
async fn test_is_empty() {
let data = Bytes::from("");
let reader = ZeroCopyObjectReader::from_bytes(data);
let reader = BytesBufferedReader::from_bytes(data);
assert!(reader.is_empty());
let data = Bytes::from("hello");
let reader = ZeroCopyObjectReader::from_bytes(data);
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/writer.rs
+36 -25
View File
@@ -14,9 +14,9 @@
//! BytesMut-backed object writer for optimized write operations.
//!
//! This module keeps the historical `ZeroCopyObjectWriter` name for public API
//! compatibility. It uses `BytesMut` for efficient buffering; writes into that
//! buffer may still copy input bytes.
//! 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;
@@ -50,7 +50,7 @@ use tokio::io::AsyncWrite;
/// Ok(())
/// }
/// ```
pub struct ZeroCopyObjectWriter {
pub struct BytesMutWriter {
/// Internal buffer using BytesMut for efficient growth
buffer: BytesMut,
/// Total bytes written
@@ -59,10 +59,11 @@ pub struct ZeroCopyObjectWriter {
finalized: bool,
}
/// Preferred name for the BytesMut-backed object writer.
pub type BytesMutWriter = ZeroCopyObjectWriter;
/// 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 ZeroCopyObjectWriter {
impl BytesMutWriter {
/// Create a new bytes-backed object writer with default capacity (8KB).
///
/// # Example
@@ -237,15 +238,15 @@ impl ZeroCopyObjectWriter {
}
}
impl Default for ZeroCopyObjectWriter {
impl Default for BytesMutWriter {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Debug for ZeroCopyObjectWriter {
impl std::fmt::Debug for BytesMutWriter {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ZeroCopyObjectWriter")
f.debug_struct("BytesMutWriter")
.field("buffer_len", &self.buffer.len())
.field("buffer_capacity", &self.buffer.capacity())
.field("bytes_written", &self.bytes_written)
@@ -254,10 +255,10 @@ impl std::fmt::Debug for ZeroCopyObjectWriter {
}
}
/// AsyncWrite implementation for ZeroCopyObjectWriter.
/// AsyncWrite implementation for BytesMutWriter.
///
/// This allows the writer to be used with tokio's async I/O utilities.
impl AsyncWrite for ZeroCopyObjectWriter {
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(
@@ -305,7 +306,7 @@ mod tests {
#[tokio::test]
async fn test_new_writer() {
let writer = ZeroCopyObjectWriter::new();
let writer = BytesMutWriter::new();
assert!(writer.is_empty());
assert_eq!(writer.bytes_written(), 0);
assert!(writer.capacity() >= 8 * 1024);
@@ -313,7 +314,7 @@ mod tests {
#[tokio::test]
async fn test_write_zero_copy() {
let mut writer = ZeroCopyObjectWriter::new();
let mut writer = BytesMutWriter::new();
let data = Bytes::from("hello world");
let written = writer.write_zero_copy(data).await.unwrap();
@@ -336,7 +337,7 @@ mod tests {
#[tokio::test]
async fn test_write_slice() {
let mut writer = ZeroCopyObjectWriter::new();
let mut writer = BytesMutWriter::new();
let data = b"hello world";
let written = writer.write_slice(data).await.unwrap();
@@ -347,7 +348,7 @@ mod tests {
#[tokio::test]
async fn test_into_bytes() {
let mut writer = ZeroCopyObjectWriter::new();
let mut writer = BytesMutWriter::new();
let data = Bytes::from("hello world");
writer.write_zero_copy(data).await.unwrap();
@@ -358,26 +359,26 @@ mod tests {
#[tokio::test]
async fn test_write_after_finalize() {
let mut writer = ZeroCopyObjectWriter::new();
let mut writer = BytesMutWriter::new();
let data = Bytes::from("hello");
writer.write_zero_copy(data).await.unwrap();
let _result = writer.into_bytes();
// Create new writer and try to write after finalize
let mut writer2 = ZeroCopyObjectWriter::new();
let mut writer2 = BytesMutWriter::new();
writer2.write_zero_copy(Bytes::from("test")).await.unwrap();
let _ = writer2.into_bytes();
// Writing to a consumed writer should work via new writer
let mut writer3 = ZeroCopyObjectWriter::new();
let mut writer3 = BytesMutWriter::new();
let result = writer3.write_zero_copy(Bytes::from("final")).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_clear() {
let mut writer = ZeroCopyObjectWriter::new();
let mut writer = BytesMutWriter::new();
writer.write_slice(b"hello").await.unwrap();
writer.clear();
@@ -389,7 +390,7 @@ mod tests {
#[tokio::test]
async fn test_reserve() {
let mut writer = ZeroCopyObjectWriter::with_capacity(10);
let mut writer = BytesMutWriter::with_capacity(10);
let initial_capacity = writer.capacity();
writer.reserve(1000);
@@ -400,7 +401,7 @@ mod tests {
#[tokio::test]
async fn test_multiple_writes() {
let mut writer = ZeroCopyObjectWriter::new();
let mut writer = BytesMutWriter::new();
writer.write_zero_copy(Bytes::from("hello ")).await.unwrap();
writer.write_slice(b"world").await.unwrap();
@@ -413,7 +414,7 @@ mod tests {
async fn test_async_write() {
use tokio::io::AsyncWriteExt;
let mut writer = ZeroCopyObjectWriter::new();
let mut writer = BytesMutWriter::new();
let data = b"hello world";
let written = writer.write(data).await.unwrap();
@@ -423,9 +424,19 @@ mod tests {
#[tokio::test]
async fn test_debug() {
let writer = ZeroCopyObjectWriter::new();
let writer = BytesMutWriter::new();
let debug_str = format!("{:?}", writer);
assert!(debug_str.contains("ZeroCopyObjectWriter"));
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");
}
}