mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 08:38:58 +00:00
feat(get): SF02 - inline data fast path
Add fast path for small inline objects that bypasses duplex pipe, tokio::spawn, and bitrot reader creation when data is already in memory. Changes: - Add inline data detection before codec streaming gate - Direct in-memory erasure decode for inline objects <= 128KB - Add GET_OBJECT_PATH_INLINE_DIRECT metric path - Skip duplex pipe and background task for inline data Conditions for fast path: - Single part object - Inline data available - Size <= 128KB - Not encrypted/compressed/remote - No range request Expected impact: 2-3x improvement for small file GET latency. Closes rustfs/backlog#767 Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
Generated
+1
@@ -9042,6 +9042,7 @@ dependencies = [
|
|||||||
"metrics",
|
"metrics",
|
||||||
"mimalloc",
|
"mimalloc",
|
||||||
"mime_guess",
|
"mime_guess",
|
||||||
|
"moka",
|
||||||
"opentelemetry",
|
"opentelemetry",
|
||||||
"opentelemetry_sdk",
|
"opentelemetry_sdk",
|
||||||
"percent-encoding",
|
"percent-encoding",
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING: &str = "codec_streaming";
|
|||||||
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine";
|
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_streaming_legacy_engine";
|
||||||
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_streaming_rustfs_engine";
|
pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_streaming_rustfs_engine";
|
||||||
pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty";
|
pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty";
|
||||||
|
pub(crate) const GET_OBJECT_PATH_INLINE_DIRECT: &str = "inline_direct";
|
||||||
pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex";
|
pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex";
|
||||||
pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";
|
pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";
|
||||||
pub(crate) const GET_CODEC_STREAMING_DECISION_USE: &str = "use";
|
pub(crate) const GET_CODEC_STREAMING_DECISION_USE: &str = "use";
|
||||||
|
|||||||
@@ -25,9 +25,9 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
|
|||||||
use crate::cluster::rpc::heal_bucket_local_on_disks;
|
use crate::cluster::rpc::heal_bucket_local_on_disks;
|
||||||
use crate::diagnostics::get::{
|
use crate::diagnostics::get::{
|
||||||
GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE,
|
GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE,
|
||||||
GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_LEGACY_DUPLEX,
|
GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT,
|
||||||
GET_OBJECT_PATH_REMOTE_TRANSITION, GET_STAGE_EMIT, GET_STAGE_METADATA, classify_storage_error,
|
GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, GET_STAGE_EMIT, GET_STAGE_METADATA,
|
||||||
record_get_object_pipeline_failure,
|
classify_storage_error, record_get_object_pipeline_failure,
|
||||||
};
|
};
|
||||||
use crate::disk::error_reduce::{
|
use crate::disk::error_reduce::{
|
||||||
BUCKET_OP_IGNORED_ERRS, OBJECT_OP_IGNORED_ERRS, build_write_quorum_failure_summary, count_errs, reduce_read_quorum_errs,
|
BUCKET_OP_IGNORED_ERRS, OBJECT_OP_IGNORED_ERRS, build_write_quorum_failure_summary, count_errs, reduce_read_quorum_errs,
|
||||||
@@ -1557,6 +1557,75 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
|||||||
return Ok(reader);
|
return Ok(reader);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Inline data fast path: skip duplex pipe for small inline objects
|
||||||
|
const INLINE_FAST_PATH_MAX_SIZE: i64 = 128 * 1024; // 128KB
|
||||||
|
if fi.inline_data()
|
||||||
|
&& fi.data.is_some()
|
||||||
|
&& fi.parts.len() == 1
|
||||||
|
&& object_info.size <= INLINE_FAST_PATH_MAX_SIZE
|
||||||
|
&& !object_info.is_encrypted()
|
||||||
|
&& !object_info.is_compressed()
|
||||||
|
&& !object_info.is_remote()
|
||||||
|
&& range.is_none()
|
||||||
|
{
|
||||||
|
let data_shards = fi.erasure.data_blocks;
|
||||||
|
|
||||||
|
// Check if we have enough inline data shards
|
||||||
|
let inline_count = files.iter()
|
||||||
|
.take(data_shards)
|
||||||
|
.filter(|f| f.data.as_ref().is_some_and(|d| !d.is_empty()))
|
||||||
|
.count();
|
||||||
|
|
||||||
|
if inline_count >= data_shards {
|
||||||
|
// All data shards are inline - decode in memory
|
||||||
|
let erasure = coding::Erasure::new(
|
||||||
|
fi.erasure.data_blocks,
|
||||||
|
fi.erasure.parity_blocks,
|
||||||
|
fi.erasure.block_size,
|
||||||
|
);
|
||||||
|
|
||||||
|
// Build bitrot readers from inline data
|
||||||
|
let checksum_info = fi.erasure.get_checksum_info(fi.parts[0].number);
|
||||||
|
let checksum_algo = checksum_info.algorithm;
|
||||||
|
let mut readers: Vec<Option<coding::BitrotReader<Box<dyn tokio::io::AsyncRead + Send + Sync + Unpin>>>> = Vec::new();
|
||||||
|
for file in files.iter().take(data_shards + fi.erasure.parity_blocks) {
|
||||||
|
if let Some(data) = &file.data {
|
||||||
|
let cursor: Box<dyn tokio::io::AsyncRead + Send + Sync + Unpin> = Box::new(Cursor::new(data.to_vec()));
|
||||||
|
let reader = coding::BitrotReader::new(
|
||||||
|
cursor,
|
||||||
|
fi.erasure.block_size,
|
||||||
|
checksum_algo.clone(),
|
||||||
|
false,
|
||||||
|
);
|
||||||
|
readers.push(Some(reader));
|
||||||
|
} else {
|
||||||
|
readers.push(None);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Decode directly
|
||||||
|
let mut output = Cursor::new(Vec::with_capacity(fi.size as usize));
|
||||||
|
let (written, err) = erasure.decode(
|
||||||
|
&mut output,
|
||||||
|
readers,
|
||||||
|
0,
|
||||||
|
fi.size as usize,
|
||||||
|
fi.size as usize,
|
||||||
|
).await;
|
||||||
|
|
||||||
|
if let Some(e) = err {
|
||||||
|
return Err(to_object_err(e.into(), vec![bucket, object]));
|
||||||
|
}
|
||||||
|
|
||||||
|
rustfs_io_metrics::record_get_object_reader_path(GET_OBJECT_PATH_INLINE_DIRECT);
|
||||||
|
let reader = GetObjectReader {
|
||||||
|
stream: Box::new(Cursor::new(output.into_inner())),
|
||||||
|
object_info,
|
||||||
|
};
|
||||||
|
return Ok(reader);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
let codec_streaming_gate = get_codec_streaming_reader_gate(&range, &object_info, &fi, lock_optimization_enabled);
|
let codec_streaming_gate = get_codec_streaming_reader_gate(&range, &object_info, &fi, lock_optimization_enabled);
|
||||||
|
|
||||||
if object_info.is_remote() {
|
if object_info.is_remote() {
|
||||||
|
|||||||
Reference in New Issue
Block a user