mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-26 05:56:50 +00:00
refactor(ecstore): add mmap copy disk read name (#4061)
This commit is contained in:
@@ -171,7 +171,7 @@ Classification:
|
|||||||
| Remote shard read stream | `crates/ecstore/src/rpc/remote_disk.rs::RemoteDisk::read_file_stream`; `crates/ecstore/src/rpc/internode_data_transport.rs::InternodeDataTransport::open_read`; `crates/ecstore/src/bitrot.rs::create_bitrot_reader` | `rustfs/src/storage/rpc/http_service.rs::handle_read_file` | Covered by `InternodeDataTransport` | Object GET, repair reads, and erasure decode use this path for remote shard bytes. |
|
| Remote shard read stream | `crates/ecstore/src/rpc/remote_disk.rs::RemoteDisk::read_file_stream`; `crates/ecstore/src/rpc/internode_data_transport.rs::InternodeDataTransport::open_read`; `crates/ecstore/src/bitrot.rs::create_bitrot_reader` | `rustfs/src/storage/rpc/http_service.rs::handle_read_file` | Covered by `InternodeDataTransport` | Object GET, repair reads, and erasure decode use this path for remote shard bytes. |
|
||||||
| Remote shard write stream | `RemoteDisk::create_file`; `RemoteDisk::append_file`; `InternodeDataTransport::open_write`; `crates/ecstore/src/bitrot.rs::create_bitrot_writer` | `rustfs/src/storage/rpc/http_service.rs::handle_put_file` | Covered by `InternodeDataTransport` | Object PUT and multipart part upload use this path for remote shard bytes. |
|
| Remote shard write stream | `RemoteDisk::create_file`; `RemoteDisk::append_file`; `InternodeDataTransport::open_write`; `crates/ecstore/src/bitrot.rs::create_bitrot_writer` | `rustfs/src/storage/rpc/http_service.rs::handle_put_file` | Covered by `InternodeDataTransport` | Object PUT and multipart part upload use this path for remote shard bytes. |
|
||||||
| Remote namespace walk stream | `RemoteDisk::walk_dir`; `InternodeDataTransport::open_walk_dir`; `crates/ecstore/src/cache_value/metacache_set.rs` walk producers | `rustfs/src/storage/rpc/http_service.rs::handle_walk_dir` | Covered by `InternodeDataTransport` | High-volume listing/scanner/heal metadata stream. It is not object byte data, but it is a large internode stream. |
|
| Remote namespace walk stream | `RemoteDisk::walk_dir`; `InternodeDataTransport::open_walk_dir`; `crates/ecstore/src/cache_value/metacache_set.rs` walk producers | `rustfs/src/storage/rpc/http_service.rs::handle_walk_dir` | Covered by `InternodeDataTransport` | High-volume listing/scanner/heal metadata stream. It is not object byte data, but it is a large internode stream. |
|
||||||
| Remote zero-copy read fallback | `RemoteDisk::read_file_zero_copy` | same as remote shard read stream | Covered by `InternodeDataTransport` through `read_file_stream` | The remote path buffers the stream into `Bytes`; true zero-copy is not guaranteed for remote disks. |
|
| Remote mmap-copy read fallback | `RemoteDisk::read_file_mmap_copy` | same as remote shard read stream | Covered by `InternodeDataTransport` through `read_file_stream` | The remote path buffers the stream into `Bytes`; true zero-copy is not guaranteed for remote disks. |
|
||||||
|
|
||||||
### Still Direct TCP/HTTP/gRPC
|
### Still Direct TCP/HTTP/gRPC
|
||||||
|
|
||||||
@@ -210,7 +210,7 @@ Classification:
|
|||||||
| Medium | `ReadAll` and `WriteAll` still carry unary `bytes` over gRPC. They appear metadata-oriented today, but there is no size threshold or routing policy. |
|
| Medium | `ReadAll` and `WriteAll` still carry unary `bytes` over gRPC. They appear metadata-oriented today, but there is no size threshold or routing policy. |
|
||||||
| Medium | `ReadMultiple` can aggregate many metadata files into one gRPC response. |
|
| Medium | `ReadMultiple` can aggregate many metadata files into one gRPC response. |
|
||||||
| Low | Legacy gRPC `WalkDir` remains implemented while `RemoteDisk::walk_dir` uses HTTP through the transport. |
|
| Low | Legacy gRPC `WalkDir` remains implemented while `RemoteDisk::walk_dir` uses HTTP through the transport. |
|
||||||
| Medium | Remote `read_file_zero_copy` is a buffered read over the transport, not a remote zero-copy contract. |
|
| Medium | Remote `read_file_mmap_copy` is a buffered read over the transport, not a remote zero-copy contract. |
|
||||||
| Medium | Server-side TCP HTTP route handling is outside the client-side trait. |
|
| Medium | Server-side TCP HTTP route handling is outside the client-side trait. |
|
||||||
|
|
||||||
## Current Object Write Path
|
## Current Object Write Path
|
||||||
@@ -235,7 +235,7 @@ This is the primary write data-plane candidate.
|
|||||||
For object GETs and repair reads in distributed erasure mode, the relevant flow is:
|
For object GETs and repair reads in distributed erasure mode, the relevant flow is:
|
||||||
|
|
||||||
1. `SetDisks` prepares shard readers for the selected disks.
|
1. `SetDisks` prepares shard readers for the selected disks.
|
||||||
2. `create_bitrot_reader` uses local zero-copy only when `disk.is_local()`.
|
2. `create_bitrot_reader` uses local mmap-copy reads only when `disk.is_local()`.
|
||||||
3. For a remote disk, it calls `disk.read_file_stream(...)`.
|
3. For a remote disk, it calls `disk.read_file_stream(...)`.
|
||||||
4. `RemoteDisk::read_file_stream` delegates to
|
4. `RemoteDisk::read_file_stream` delegates to
|
||||||
`InternodeDataTransport::open_read`.
|
`InternodeDataTransport::open_read`.
|
||||||
@@ -385,7 +385,7 @@ The initial TCP backend can keep the current signed HTTP URLs internally.
|
|||||||
`RemoteDisk` delegates only these methods to the data transport:
|
`RemoteDisk` delegates only these methods to the data transport:
|
||||||
|
|
||||||
- `read_file_stream`
|
- `read_file_stream`
|
||||||
- `read_file_zero_copy` as the current wrapper over `read_file_stream`
|
- `read_file_mmap_copy` as the current wrapper over `read_file_stream`
|
||||||
- `append_file`
|
- `append_file`
|
||||||
- `create_file`
|
- `create_file`
|
||||||
- `walk_dir`
|
- `walk_dir`
|
||||||
|
|||||||
@@ -54,7 +54,7 @@ through `ParallelReader` in `crates/ecstore/src/erasure_coding/decode.rs` and
|
|||||||
| Bitrot verification | `BitrotReader::read` | caller `&mut [u8]`, `hash_buf: Vec<u8>` | No additional payload copy | The bitrot reader reads hash bytes into `hash_buf` and payload bytes directly into the supplied output slice. Hash calculation reads the slice. |
|
| Bitrot verification | `BitrotReader::read` | caller `&mut [u8]`, `hash_buf: Vec<u8>` | No additional payload copy | The bitrot reader reads hash bytes into `hash_buf` and payload bytes directly into the supplied output slice. Hash calculation reads the slice. |
|
||||||
| Erasure shard read | `ParallelReader::read` | `Vec<u8>` per shard | Yes | Each shard read allocates `vec![0u8; shard_size]`; data is filled there before decode/reconstruction. |
|
| Erasure shard read | `ParallelReader::read` | `Vec<u8>` per shard | Yes | Each shard read allocates `vec![0u8; shard_size]`; data is filled there before decode/reconstruction. |
|
||||||
| Object response write | `write_data_blocks` | slices of shard `Vec<u8>` | No extra staging copy | Decoded data block slices are written to the target writer with `write_all`; the target may copy internally. |
|
| Object response write | `write_data_blocks` | slices of shard `Vec<u8>` | No extra staging copy | Decoded data block slices are written to the target writer with `write_all`; the target may copy internally. |
|
||||||
| Remote zero-copy helper | `RemoteDisk::read_file_zero_copy` | `Vec<u8>` then `Bytes` | Yes | The remote implementation reads the full stream into a `Vec` and converts it into `Bytes`. It is a convenience fallback, not network zero-copy. |
|
| Remote mmap-copy helper | `RemoteDisk::read_file_mmap_copy` | `Vec<u8>` then `Bytes` | Yes | The remote implementation reads the full stream into a `Vec` and converts it into `Bytes`. It is a convenience fallback, not network zero-copy. |
|
||||||
|
|
||||||
## Write Stream
|
## Write Stream
|
||||||
|
|
||||||
@@ -90,7 +90,7 @@ through `ParallelReader` in `crates/ecstore/src/erasure_coding/decode.rs` and
|
|||||||
| 3 | `ParallelReader::read` shard buffers | High on read path | Each shard read allocates and fills a `Vec<u8>` before decode can proceed. This is also where degraded reads wait on quorum. |
|
| 3 | `ParallelReader::read` shard buffers | High on read path | Each shard read allocates and fills a `Vec<u8>` before decode can proceed. This is also where degraded reads wait on quorum. |
|
||||||
| 4 | `ReaderStream::with_capacity` plus `StreamReader` | Medium on read path | Server file reads create `Bytes` chunks, then client `AsyncRead` copies those chunks into the caller's `ReadBuf`. |
|
| 4 | `ReaderStream::with_capacity` plus `StreamReader` | Medium on read path | Server file reads create `Bytes` chunks, then client `AsyncRead` copies those chunks into the caller's `ReadBuf`. |
|
||||||
| 5 | `Erasure::encode` block and shard materialization | Medium on write path | Source data is first read into a block `Vec<u8>`, then encoded into per-shard `Bytes`. This is necessary for the current erasure API. |
|
| 5 | `Erasure::encode` block and shard materialization | Medium on write path | Source data is first read into a block `Vec<u8>`, then encoded into per-shard `Bytes`. This is necessary for the current erasure API. |
|
||||||
| 6 | `RemoteDisk::read_file_zero_copy` | Medium when used | Remote zero-copy reads buffer the whole stream into memory. The name does not mean zero-copy over the network. |
|
| 6 | `RemoteDisk::read_file_mmap_copy` | Medium when used | Remote mmap-copy reads buffer the whole stream into memory. The name does not mean zero-copy over the network. |
|
||||||
| 7 | URL/query/header/JSON serialization | Low | Metadata copies are small and not on the large payload hot path. |
|
| 7 | URL/query/header/JSON serialization | Low | Metadata copies are small and not on the large payload hot path. |
|
||||||
|
|
||||||
## Adapter Ownership Gaps
|
## Adapter Ownership Gaps
|
||||||
@@ -117,6 +117,6 @@ through `ParallelReader` in `crates/ecstore/src/erasure_coding/decode.rs` and
|
|||||||
7. The HTTP auth and URL construction boundary is part of the current TCP/HTTP
|
7. The HTTP auth and URL construction boundary is part of the current TCP/HTTP
|
||||||
backend. A non-HTTP backend would need equivalent peer authentication and
|
backend. A non-HTTP backend would need equivalent peer authentication and
|
||||||
disk addressing without assuming URL query parameters.
|
disk addressing without assuming URL query parameters.
|
||||||
8. Local disk zero-copy exists only for local reads via `read_file_zero_copy`.
|
8. Local disk mmap-copy reads exist only for local reads via `read_file_mmap_copy`.
|
||||||
Remote disks deliberately fall back to network streaming and full-buffer
|
Remote disks deliberately fall back to network streaming and full-buffer
|
||||||
collection for the zero-copy helper.
|
collection for the legacy zero-copy helper.
|
||||||
|
|||||||
@@ -1708,11 +1708,10 @@ impl DiskAPI for RemoteDisk {
|
|||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Zero-copy read for remote disks falls back to efficient network read.
|
/// Buffered read for remote disks.
|
||||||
/// Note: True zero-copy is not possible over network, but we avoid extra copies
|
/// The transport stream is collected into owned Bytes for caller sharing.
|
||||||
/// by reading directly into Bytes.
|
|
||||||
#[tracing::instrument(level = "debug", skip(self))]
|
#[tracing::instrument(level = "debug", skip(self))]
|
||||||
async fn read_file_zero_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
||||||
// For remote disks, use the regular reader and read into Bytes
|
// For remote disks, use the regular reader and read into Bytes
|
||||||
let reader = self.read_file_stream(volume, path, offset, length).await?;
|
let reader = self.read_file_stream(volume, path, offset, length).await?;
|
||||||
|
|
||||||
|
|||||||
@@ -1361,9 +1361,9 @@ impl DiskAPI for LocalDiskWrapper {
|
|||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn read_file_zero_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<bytes::Bytes> {
|
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<bytes::Bytes> {
|
||||||
self.track_disk_health(
|
self.track_disk_health(
|
||||||
|| async { self.disk.read_file_zero_copy(volume, path, offset, length).await },
|
|| async { self.disk.read_file_mmap_copy(volume, path, offset, length).await },
|
||||||
get_max_timeout_duration(),
|
get_max_timeout_duration(),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|||||||
@@ -2750,13 +2750,12 @@ impl DiskAPI for LocalDisk {
|
|||||||
Ok(Box::new(StallTimeoutReader::new(reader, get_object_disk_read_timeout())))
|
Ok(Box::new(StallTimeoutReader::new(reader, get_object_disk_read_timeout())))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Zero-copy file read using memory mapping (Unix) or efficient read (non-Unix).
|
/// File read using mmap-then-copy on Unix or efficient read on non-Unix.
|
||||||
/// Returns Bytes that can be shared without copying.
|
|
||||||
// SAFETY: Unix unsafe calls in this function only query page size and mmap
|
// SAFETY: Unix unsafe calls in this function only query page size and mmap
|
||||||
// a read-only file region after bounds and alignment are validated.
|
// a read-only file region after bounds and alignment are validated.
|
||||||
#[allow(unsafe_code)]
|
#[allow(unsafe_code)]
|
||||||
#[tracing::instrument(level = "debug", skip(self))]
|
#[tracing::instrument(level = "debug", skip(self))]
|
||||||
async fn read_file_zero_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
||||||
let volume_dir = self.get_bucket_path(volume)?;
|
let volume_dir = self.get_bucket_path(volume)?;
|
||||||
if !skip_access_checks(volume) {
|
if !skip_access_checks(volume) {
|
||||||
access(&volume_dir)
|
access(&volume_dir)
|
||||||
@@ -2784,7 +2783,7 @@ impl DiskAPI for LocalDisk {
|
|||||||
offset,
|
offset,
|
||||||
length,
|
length,
|
||||||
actual_size = meta.len(),
|
actual_size = meta.len(),
|
||||||
reason = "read_file_zero_copy_out_of_bounds",
|
reason = "read_file_mmap_copy_out_of_bounds",
|
||||||
"Disk local read fallback failed"
|
"Disk local read fallback failed"
|
||||||
);
|
);
|
||||||
return Err(DiskError::FileCorrupt);
|
return Err(DiskError::FileCorrupt);
|
||||||
@@ -4871,7 +4870,26 @@ mod test {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn test_read_file_zero_copy_rejects_offset_length_overflow() {
|
async fn test_read_file_mmap_copy_rejects_offset_length_overflow() {
|
||||||
|
use tempfile::tempdir;
|
||||||
|
|
||||||
|
let dir = tempdir().expect("operation should succeed");
|
||||||
|
let endpoint =
|
||||||
|
Endpoint::try_from(dir.path().to_str().expect("operation should succeed")).expect("operation should succeed");
|
||||||
|
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
|
||||||
|
|
||||||
|
disk.make_volume("test-volume").await.expect("operation should succeed");
|
||||||
|
disk.write_all("test-volume", "test-file.txt", Bytes::from_static(b"test"))
|
||||||
|
.await
|
||||||
|
.expect("operation should succeed");
|
||||||
|
|
||||||
|
let result = disk.read_file_mmap_copy("test-volume", "test-file.txt", usize::MAX, 1).await;
|
||||||
|
assert!(matches!(result, Err(DiskError::FileCorrupt)));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
#[allow(deprecated)]
|
||||||
|
async fn test_read_file_zero_copy_legacy_alias_rejects_offset_length_overflow() {
|
||||||
use tempfile::tempdir;
|
use tempfile::tempdir;
|
||||||
|
|
||||||
let dir = tempdir().expect("operation should succeed");
|
let dir = tempdir().expect("operation should succeed");
|
||||||
|
|||||||
@@ -291,10 +291,10 @@ impl DiskAPI for Disk {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tracing::instrument(skip(self))]
|
#[tracing::instrument(skip(self))]
|
||||||
async fn read_file_zero_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
||||||
match self {
|
match self {
|
||||||
Disk::Local(local_disk) => local_disk.read_file_zero_copy(volume, path, offset, length).await,
|
Disk::Local(local_disk) => local_disk.read_file_mmap_copy(volume, path, offset, length).await,
|
||||||
Disk::Remote(remote_disk) => remote_disk.read_file_zero_copy(volume, path, offset, length).await,
|
Disk::Remote(remote_disk) => remote_disk.read_file_mmap_copy(volume, path, offset, length).await,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -558,11 +558,17 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
|
|||||||
async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader>;
|
async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader>;
|
||||||
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader>;
|
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<FileReader>;
|
||||||
|
|
||||||
/// Zero-copy file read using memory mapping (Unix) or efficient read (non-Unix).
|
/// File read using mmap-then-copy on Unix or an efficient read on non-Unix.
|
||||||
/// Returns Bytes that can be shared without copying.
|
async fn read_file_mmap_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes>;
|
||||||
/// On Unix, this uses mmap for true zero-copy access.
|
|
||||||
/// On other platforms, falls back to efficient read operations.
|
/// Historical name for `read_file_mmap_copy`.
|
||||||
async fn read_file_zero_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes>;
|
#[deprecated(
|
||||||
|
since = "1.0.0-beta.8",
|
||||||
|
note = "use read_file_mmap_copy; this path copies mmap data into owned Bytes"
|
||||||
|
)]
|
||||||
|
async fn read_file_zero_copy(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result<Bytes> {
|
||||||
|
self.read_file_mmap_copy(volume, path, offset, length).await
|
||||||
|
}
|
||||||
|
|
||||||
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter>;
|
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter>;
|
||||||
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result<FileWriter>;
|
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> Result<FileWriter>;
|
||||||
|
|||||||
@@ -149,7 +149,7 @@ async fn open_disk_reader(
|
|||||||
) -> disk::error::Result<FileReader> {
|
) -> disk::error::Result<FileReader> {
|
||||||
if use_mmap_read && disk.is_local() {
|
if use_mmap_read && disk.is_local() {
|
||||||
let start = Instant::now();
|
let start = Instant::now();
|
||||||
match disk.read_file_zero_copy(bucket, path, offset, length).await {
|
match disk.read_file_mmap_copy(bucket, path, offset, length).await {
|
||||||
Ok(bytes) => {
|
Ok(bytes) => {
|
||||||
let duration_ms = start.elapsed().as_secs_f64() * 1000.0;
|
let duration_ms = start.elapsed().as_secs_f64() * 1000.0;
|
||||||
|
|
||||||
@@ -201,7 +201,7 @@ fn bitrot_encoded_range(offset: usize, length: usize, shard_size: usize, checksu
|
|||||||
/// * `shard_size` - Shard size for erasure coding
|
/// * `shard_size` - Shard size for erasure coding
|
||||||
/// * `checksum_algo` - Hash algorithm for bitrot verification
|
/// * `checksum_algo` - Hash algorithm for bitrot verification
|
||||||
/// * `skip_verify` - If true, skip checksum verification
|
/// * `skip_verify` - If true, skip checksum verification
|
||||||
/// * `use_mmap_read` - If true, use zero-copy read (mmap on Unix)
|
/// * `use_mmap_read` - If true, use mmap-copy read (mmap on Unix)
|
||||||
#[allow(clippy::too_many_arguments)]
|
#[allow(clippy::too_many_arguments)]
|
||||||
pub async fn create_bitrot_reader(
|
pub async fn create_bitrot_reader(
|
||||||
inline_data: Option<&[u8]>,
|
inline_data: Option<&[u8]>,
|
||||||
@@ -360,7 +360,7 @@ mod tests {
|
|||||||
let shard_size = 16;
|
let shard_size = 16;
|
||||||
let checksum_algo = HashAlgorithm::HighwayHash256S;
|
let checksum_algo = HashAlgorithm::HighwayHash256S;
|
||||||
|
|
||||||
// Test with zero-copy enabled (should work the same for inline data)
|
// Test with mmap-copy enabled (should work the same for inline data)
|
||||||
let result = create_bitrot_reader(
|
let result = create_bitrot_reader(
|
||||||
Some(test_data),
|
Some(test_data),
|
||||||
None,
|
None,
|
||||||
|
|||||||
Reference in New Issue
Block a user