gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 06:18:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore): bound xl.meta read workers and enable HEAD early-stop (#8846)

Browse Source 
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a
This commit is contained in:
Ramakrishna Chilaka
2026-10-23 20:02:48 +05:30
committed by GitHub
parent 623601d2f9
commit cfee2ffaba
5 changed files with 597 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/local.rs
+374 -39
View File
@@ -273,6 +273,169 @@ fn read_all_data_std(
Ok((bytes, modtime))
}
fn read_metadata_only_std(opened: std::io::Result<std::fs::File>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
let mut f = opened.map_err(to_file_error)?;
let meta = f.metadata().map_err(to_file_error)?;
if meta.is_dir() {
return Err(Error::FileNotFound);
}
let size = usize::try_from(meta.len()).map_err(Error::other)?;
let data = read_xl_meta_no_data_sync(&mut f, size)?;
let modtime = match meta.modified() {
Ok(md) => Some(OffsetDateTime::from(md)),
Err(_) => None,
};
Ok((data, modtime))
}
const MAX_METADATA_READ_WORKERS_PER_DISK: usize = 2;
type ReadAllDataResult = core::result::Result<(Vec<u8>, Option<OffsetDateTime>), ReadAllError>;
type ReadMetadataOnlyResult = Result<(Vec<u8>, Option<OffsetDateTime>)>;
enum MetadataReadOp {
ReadAll {
tx: tokio::sync::oneshot::Sender<ReadAllDataResult>,
},
ReadMetadataOnly {
tx: tokio::sync::oneshot::Sender<ReadMetadataOnlyResult>,
},
#[cfg(test)]
TestCustom {
is_closed: Box<dyn Fn() -> bool + Send>,
run: Box<dyn FnOnce() + Send>,
},
}
struct MetadataReadJob {
path: PathBuf,
io_permit: Option<DiskIoPermit>,
op: MetadataReadOp,
}
impl MetadataReadJob {
#[inline]
fn is_cancelled(&self) -> bool {
match &self.op {
MetadataReadOp::ReadAll { tx } => tx.is_closed(),
MetadataReadOp::ReadMetadataOnly { tx } => tx.is_closed(),
#[cfg(test)]
MetadataReadOp::TestCustom { is_closed, .. } => is_closed(),
}
}
#[inline]
fn execute(self, root: &os::PublicationRoot) {
if self.is_cancelled() {
return;
}
let _io_permit = self.io_permit;
match self.op {
MetadataReadOp::ReadAll { tx } => {
let _ = tx.send(read_all_data_std(root.open_readonly(&self.path)));
}
MetadataReadOp::ReadMetadataOnly { tx } => {
let _ = tx.send(read_metadata_only_std(root.open_readonly(&self.path)));
}
#[cfg(test)]
MetadataReadOp::TestCustom { run, .. } => run(),
}
}
}
#[derive(Default)]
struct MetadataReadQueueState {
jobs: std::collections::VecDeque<MetadataReadJob>,
active_workers: usize,
}
struct MetadataReadQueue {
root: os::PublicationRoot,
state: ParkingLotMutex<MetadataReadQueueState>,
}
struct MetadataReadWorkerGuard {
queue: Arc<MetadataReadQueue>,
disarmed: bool,
}
impl Drop for MetadataReadWorkerGuard {
fn drop(&mut self) {
if self.disarmed {
return;
}
let should_respawn = {
let mut state = self.queue.state.lock();
state.active_workers = state.active_workers.saturating_sub(1);
if !state.jobs.is_empty() && state.active_workers < MAX_METADATA_READ_WORKERS_PER_DISK {
state.active_workers += 1;
true
} else {
false
}
};
if should_respawn {
MetadataReadQueue::spawn_worker(Arc::clone(&self.queue));
}
}
}
impl MetadataReadQueue {
fn new(root: os::PublicationRoot) -> Arc<Self> {
Arc::new(Self {
root,
state: ParkingLotMutex::new(MetadataReadQueueState::default()),
})
}
fn enqueue(self: &Arc<Self>, job: MetadataReadJob) {
let should_spawn = {
let mut state = self.state.lock();
state.jobs.push_back(job);
if state.active_workers < MAX_METADATA_READ_WORKERS_PER_DISK {
state.active_workers += 1;
true
} else {
false
}
};
if should_spawn {
Self::spawn_worker(Arc::clone(self));
}
}
fn spawn_worker(queue: Arc<Self>) {
drop(tokio::task::spawn_blocking(move || {
let mut guard = MetadataReadWorkerGuard {
queue: Arc::clone(&queue),
disarmed: false,
};
loop {
let Some(job) = ({
let mut state = queue.state.lock();
match state.jobs.pop_front() {
Some(job) => Some(job),
None => {
state.active_workers = state.active_workers.saturating_sub(1);
guard.disarmed = true;
None
}
}
}) else {
return;
};
job.execute(&queue.root);
}
}));
}
}
/// Reads the `xl.meta` header/metadata envelope and 5-byte CRC for directory
/// walks (matching `read_xl_meta_no_data_sync`) without reading trailing
/// `InlineData` object bodies. If the header or metadata envelope is corrupt or
@@ -5586,6 +5749,7 @@ pub struct LocalDisk {
exit_signal: Option<tokio::sync::broadcast::Sender<()>>,
io_backend: Arc<dyn LocalIoBackend>,
file_sync_permits: Arc<Semaphore>,
metadata_read_queue: Arc<MetadataReadQueue>,
snapshot_leases: Arc<ParkingLotMutex<SnapshotLeaseRegistry>>,
}
@@ -6041,6 +6205,7 @@ impl LocalDisk {
// TODO(backlog): add O_DIRECT I/O support for performance-critical paths
// TODO(backlog): populate DiskInfo in constructor
let metadata_read_queue = MetadataReadQueue::new(publication_root.clone());
let mut disk = Self {
root: root.clone(),
publication_root,
@@ -6071,6 +6236,7 @@ impl LocalDisk {
exit_signal: None,
io_backend: build_local_io_backend(io_root.clone()).await,
file_sync_permits: os::disk_file_sync_limiter(&root),
metadata_read_queue,
snapshot_leases: Arc::new(ParkingLotMutex::new(SnapshotLeaseRegistry::default())),
};
let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| {
@@ -6563,6 +6729,26 @@ impl LocalDisk {
self.io_get_object_path(bucket, key)
}
/// Only for operations that open through PublicationRoot or a commit guard:
/// containment and symlink checks happen with the actual open, off the runtime.
fn io_get_bucket_and_object_open_path(&self, bucket: &str, key: &str) -> Result<(PathBuf, PathBuf)> {
#[cfg(target_os = "linux")]
{
let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key);
let normalized = check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?;
if matches!(normalized, std::borrow::Cow::Owned(_)) {
reject_local_disk_symlink_components_at(self.io_root(), &self.mount_lease, &normalized)?;
}
Ok((bucket_path, path))
}
#[cfg(not(target_os = "linux"))]
{
let path = self.io_get_object_path(bucket, key)?;
let bucket_path = self.io_get_bucket_path(bucket)?;
Ok((bucket_path, path))
}
}
fn io_get_bucket_path(&self, bucket: &str) -> Result<PathBuf> {
self.local_disk_bucket_path(self.io_root(), bucket)
}
@@ -7984,32 +8170,13 @@ impl LocalDisk {
// - metadata failure -> to_file_error
// - parse failure -> propagated verbatim from read_xl_meta_no_data_sync (`?`)
let path = file_path.as_ref().to_path_buf();
let root = self.publication_root.clone();
let (data, modtime) =
spawn_blocking_with_disk_io_permit(io_permit, move || -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
// Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)).
let mut f = root.open_readonly(&path).map_err(to_file_error)?;
let meta = f.metadata().map_err(to_file_error)?;
if meta.is_dir() {
// fix use io::Error
return Err(Error::FileNotFound);
}
let size = meta.len() as usize;
let data = read_xl_meta_no_data_sync(&mut f, size)?;
let modtime = match meta.modified() {
Ok(md) => Some(OffsetDateTime::from(md)),
Err(_) => None,
};
Ok((data, modtime))
})
.await
.map_err(DiskError::from)??;
let (tx, rx) = tokio::sync::oneshot::channel();
self.metadata_read_queue.enqueue(MetadataReadJob {
path,
io_permit,
op: MetadataReadOp::ReadMetadataOnly { tx },
});
let (data, modtime) = rx.await.map_err(|_| DiskError::other("metadata read worker cancelled"))??;
Ok((data, modtime))
}
@@ -8180,10 +8347,13 @@ impl LocalDisk {
// gating the volume fallback on the open error alone is equivalent to
// the original code, where the fallback lived solely in the open match arm.
let path = file_path.as_ref().to_path_buf();
let root = self.publication_root.clone();
let res = spawn_blocking_with_disk_io_permit(io_permit, move || read_all_data_std(root.open_readonly(&path)))
.await
.map_err(DiskError::from)?;
let (tx, rx) = tokio::sync::oneshot::channel();
self.metadata_read_queue.enqueue(MetadataReadJob {
path,
io_permit,
op: MetadataReadOp::ReadAll { tx },
});
let res = rx.await.map_err(|_| DiskError::other("metadata read worker cancelled"))?;
let (bytes, modtime) = self.resolve_read_all_result(volume, volume_dir.as_ref(), res).await?;
@@ -10127,19 +10297,23 @@ fn check_local_disk_valid_path(root: &Path, path: impl AsRef<Path>) -> Result<()
#[cfg(target_os = "linux")]
fn check_local_disk_valid_object_path_at(root: &Path, root_fd: &std::fs::File, bucket_path: &Path, path: &Path) -> Result<()> {
check_local_disk_object_path_components(root, bucket_path, path)?;
reject_local_disk_symlink_components_at(root, root_fd, &normalize_path_components(path))
let normalized_path = check_local_disk_object_path_components(root, bucket_path, path)?;
reject_local_disk_symlink_components_at(root, root_fd, &normalized_path)
}
#[cfg(target_os = "linux")]
fn check_local_disk_object_path_components(root: &Path, bucket_path: &Path, path: &Path) -> Result<()> {
let bucket_path = normalize_path_components(bucket_path);
let path = normalize_path_components(path);
fn check_local_disk_object_path_components<'a>(
root: &Path,
bucket_path: &Path,
path: &'a Path,
) -> Result<std::borrow::Cow<'a, Path>> {
let bucket_path = normalize_path_cow(bucket_path);
let path = normalize_path_cow(path);
if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) {
return Err(DiskError::InvalidPath);
}
Ok(())
Ok(path)
}
#[cfg(target_os = "linux")]
@@ -10226,6 +10400,18 @@ fn reject_local_disk_symlink_components(root: &Path, path: &Path) -> Result<()>
Ok(())
}
#[cfg(target_os = "linux")]
#[inline]
fn normalize_path_cow(path: &Path) -> std::borrow::Cow<'_, Path> {
if !path
.components()
.any(|c| matches!(c, std::path::Component::ParentDir | std::path::Component::CurDir))
{
return std::borrow::Cow::Borrowed(path);
}
std::borrow::Cow::Owned(normalize_path_components(path))
}
// Lightweight path normalization without filesystem calls
fn normalize_path_components(path: impl AsRef<Path>) -> PathBuf {
let path = path.as_ref();
@@ -12428,8 +12614,7 @@ impl DiskAPI for LocalDisk {
}
let path_resolve_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let file_path = self.io_get_object_path(volume, path)?;
let volume_dir = self.io_get_bucket_path(volume)?;
let (volume_dir, file_path) = self.io_get_bucket_and_object_open_path(volume, path)?;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_RESOLVE, path_resolve_start);
let path_check_start = get_stage_timer_if_enabled(stage_metrics_enabled);
@@ -12439,7 +12624,7 @@ impl DiskAPI for LocalDisk {
let read_data = opts.read_data;
let xlmeta_read_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let raw_read_result = self.read_raw(volume, volume_dir.clone(), file_path, read_data).await;
let raw_read_result = self.read_raw(volume, &volume_dir, &file_path, read_data).await;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_XLMETA_READ, xlmeta_read_start);
let (data, _) = raw_read_result.map_err(|e| {
if e == DiskError::FileNotFound && !version_id.is_empty() {
@@ -25523,6 +25708,156 @@ mod test {
assert!(matches!(disk.get_object_path_for_io("bucket", "object"), Err(DiskError::InvalidPath)));
}
#[cfg(unix)]
#[tokio::test]
async fn read_version_rejects_symlink_prefix_and_leaf() {
use std::os::unix::fs::symlink;
let root_dir = tempfile::tempdir().expect("temp dir should be created");
let outside_dir = tempfile::tempdir().expect("outside temp dir should be created");
let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
disk.make_volume("bucket").await.expect("bucket should be created");
let valid_meta = test_meta(test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline"))));
fs::create_dir_all(outside_dir.path().join("object"))
.await
.expect("outside object dir should be created");
fs::write(outside_dir.path().join("object").join(STORAGE_FORMAT_FILE), &valid_meta)
.await
.expect("outside prefix xl.meta should be written");
fs::write(outside_dir.path().join(STORAGE_FORMAT_FILE), &valid_meta)
.await
.expect("outside leaf xl.meta should be written");
symlink(outside_dir.path(), root_dir.path().join("bucket/symlink-prefix")).expect("prefix symlink should be created");
symlink(outside_dir.path(), root_dir.path().join("bucket/symlink-leaf")).expect("leaf symlink should be created");
for read_data in [false, true] {
let opts = ReadOptions {
read_data,
..Default::default()
};
assert!(matches!(
disk.read_version("", "bucket", "symlink-prefix/object", "", &opts).await,
Err(DiskError::InvalidPath)
));
assert!(matches!(
disk.read_version("", "bucket", "symlink-leaf", "", &opts).await,
Err(DiskError::InvalidPath)
));
}
}
#[tokio::test]
async fn metadata_read_queue_bounds_concurrency_skips_cancelled_and_recovers_from_panic() {
let dir = tempfile::tempdir().expect("tempdir should be created");
let root = os::PublicationRoot::new(dir.path()).expect("publication root should be created");
let queue = MetadataReadQueue::new(root);
// 1. Concurrency bound + cancelled job skip:
let active = Arc::new(AtomicUsize::new(0));
let max_active = Arc::new(AtomicUsize::new(0));
let executed_cancelled = Arc::new(AtomicBool::new(false));
let (entered_tx, mut entered_rx) = tokio::sync::mpsc::unbounded_channel();
let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
let release_rx = Arc::new(ParkingLotMutex::new(release_rx));
for _ in 0..MAX_METADATA_READ_WORKERS_PER_DISK {
let active = Arc::clone(&active);
let max_active = Arc::clone(&max_active);
let entered_tx = entered_tx.clone();
let release_rx = Arc::clone(&release_rx);
queue.enqueue(MetadataReadJob {
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| false),
run: Box::new(move || {
let cur = active.fetch_add(1, Ordering::AcqRel) + 1;
max_active.fetch_max(cur, Ordering::AcqRel);
let _ = entered_tx.send(());
let _ = release_rx.lock().recv();
active.fetch_sub(1, Ordering::AcqRel);
}),
},
});
}
for _ in 0..MAX_METADATA_READ_WORKERS_PER_DISK {
tokio::time::timeout(Duration::from_secs(5), entered_rx.recv())
.await
.expect("initial workers should enter")
.expect("channel should stay open");
}
assert_eq!(queue.state.lock().active_workers, MAX_METADATA_READ_WORKERS_PER_DISK);
// Enqueue a cancelled production ReadMetadataOnly job and a cancelled TestCustom job while workers are busy.
let (cancelled_tx, cancelled_rx) = tokio::sync::oneshot::channel();
drop(cancelled_rx);
queue.enqueue(MetadataReadJob {
path: dir.path().join("cancelled-never-opened"),
io_permit: None,
op: MetadataReadOp::ReadMetadataOnly { tx: cancelled_tx },
});
let executed_cancelled_clone = Arc::clone(&executed_cancelled);
queue.enqueue(MetadataReadJob {
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| true),
run: Box::new(move || {
executed_cancelled_clone.store(true, Ordering::Release);
}),
},
});
// Enqueue a panicking job followed by a normal completion job to verify panic respawn.
queue.enqueue(MetadataReadJob {
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| false),
run: Box::new(|| panic!("synthetic worker panic")),
},
});
let (done_tx, done_rx) = tokio::sync::oneshot::channel();
let active_trailing = Arc::clone(&active);
let max_active_trailing = Arc::clone(&max_active);
queue.enqueue(MetadataReadJob {
path: PathBuf::new(),
io_permit: None,
op: MetadataReadOp::TestCustom {
is_closed: Box::new(|| false),
run: Box::new(move || {
let cur = active_trailing.fetch_add(1, Ordering::AcqRel) + 1;
max_active_trailing.fetch_max(cur, Ordering::AcqRel);
active_trailing.fetch_sub(1, Ordering::AcqRel);
let _ = done_tx.send(());
}),
},
});
assert_eq!(
queue.state.lock().active_workers,
MAX_METADATA_READ_WORKERS_PER_DISK,
"extra queued jobs must not exceed per-disk worker cap"
);
// Release only one blocked worker so the panicking job must trigger a respawn
// to execute the trailing completion job while the other worker stays blocked.
release_tx.send(()).expect("release one blocked worker");
tokio::time::timeout(Duration::from_secs(5), done_rx)
.await
.expect("trailing job should finish after worker panic recovery")
.expect("trailing job sender should complete");
for _ in 1..MAX_METADATA_READ_WORKERS_PER_DISK {
release_tx.send(()).expect("release remaining blocked worker");
}
assert!(!executed_cancelled.load(Ordering::Acquire), "cancelled job must be skipped");
assert!(max_active.load(Ordering::Acquire) <= MAX_METADATA_READ_WORKERS_PER_DISK);
}
#[tokio::test]
async fn get_object_path_rejects_key_traversal_out_of_bucket() {
let root_dir = tempfile::tempdir().expect("temp dir should be created");
crates/ecstore/src/set_disk/core/metadata_read.rs
+23 -14
View File
@@ -67,6 +67,7 @@ pub(in crate::set_disk) struct MetadataReadResult {
pub(in crate::set_disk) slots: Vec<MetadataDiskObservation>,
pub(in crate::set_disk) diagnostics: MetadataFanoutDiagnostics,
pub(in crate::set_disk) condition_stopped: bool,
pub(in crate::set_disk) verified_inline_body: Option<bytes::Bytes>,
}
impl MetadataReadResult {
@@ -634,7 +635,7 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
parts_metadata: &[FileInfo],
disks: &[Option<DiskStore>],
) -> Option<&'static str> {
inline_body_miss_reason_from_observations(
verify_inline_body_from_observations(
bucket,
object,
candidate,
@@ -645,17 +646,18 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
disks,
)
.await
.err()
}
async fn inline_body_miss_reason_from_observations<'a>(
async fn verify_inline_body_from_observations<'a>(
bucket: &str,
object: &str,
candidate: &FileInfo,
observations: impl IntoIterator<Item = (usize, Option<&'a FileInfo>)>,
disks: &[Option<DiskStore>],
) -> Option<&'static str> {
) -> Result<bytes::Bytes, &'static str> {
if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
return Some(reason);
return Err(reason);
}
let Ok(erasure) = coding::Erasure::try_new_with_options(
@@ -664,21 +666,21 @@ async fn inline_body_miss_reason_from_observations<'a>(
candidate.erasure.block_size,
candidate.uses_legacy_checksum,
) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
};
let data_files =
match collect_inline_data_shard_fileinfos_from_observations(observations, candidate, erasure.data_shards, |index| {
disks.get(index).is_some_and(Option::is_some)
}) {
Ok(data_files) => data_files,
Err(reason) => return Some(reason),
Err(reason) => return Err(reason),
};
let Some(part) = candidate.parts.first() else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
};
let Ok(object_size) = usize::try_from(candidate.size) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
};
let checksum_info = candidate.erasure.get_checksum_info(part.number);
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -698,14 +700,14 @@ async fn inline_body_miss_reason_from_observations<'a>(
let Ok(mut readers) =
build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
};
#[cfg(test)]
crate::set_disk::disk_call_counters::record(object, crate::set_disk::disk_call_counters::KIND_BODY_DECODE, 0);
match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
Some(body) if body.len() == object_size => None,
_ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
Some(body) if body.len() == object_size => Ok(body),
_ => Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
}
}
@@ -1148,6 +1150,7 @@ impl SetDisks {
slots,
diagnostics,
condition_stopped: false,
verified_inline_body: None,
})
}
@@ -1368,6 +1371,7 @@ impl SetDisks {
),
)
});
let mut verified_inline_body = None;
let should_return_early = if condition_stopped {
true
} else if force_full_wait {
@@ -1377,7 +1381,7 @@ impl SetDisks {
Some(_candidate) if non_inline_candidate_eligible == Some(true) => {
accumulator.candidate_has_read_reserve()
}
Some(candidate) => match inline_body_miss_reason_from_observations(
Some(candidate) => match verify_inline_body_from_observations(
bucket.as_ref(),
object.as_ref(),
candidate,
@@ -1386,8 +1390,11 @@ impl SetDisks {
)
.await
{
None => true,
Some(reason) => {
Ok(body) => {
verified_inline_body = Some(body);
true
}
Err(reason) => {
final_miss_reason_override = Some(reason);
if bounded_fanout
&& reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD
@@ -1445,6 +1452,7 @@ impl SetDisks {
slots,
diagnostics,
condition_stopped,
verified_inline_body,
});
}
}
@@ -1527,6 +1535,7 @@ impl SetDisks {
slots,
diagnostics,
condition_stopped: false,
verified_inline_body: None,
})
}
crates/ecstore/src/set_disk/mod.rs
+152 -2
View File
@@ -939,6 +939,7 @@ struct OwnedGetObjectFileInfo {
fi: FileInfo,
parts_metadata: Vec<FileInfo>,
online_disks: Vec<Option<DiskStore>>,
verified_inline_body: Option<Bytes>,
}
enum LateMetadataFanout {
@@ -955,6 +956,7 @@ impl GetObjectFileInfo {
fi,
parts_metadata,
online_disks,
verified_inline_body: None,
}),
shared: None,
condition_stopped: false,
@@ -1017,6 +1019,27 @@ impl GetObjectFileInfo {
self.late_metadata_fanout.is_some()
}
fn set_verified_inline_body(&mut self, body: Option<Bytes>) {
if let Some(owned) = self.owned.as_mut() {
owned.verified_inline_body = body;
}
}
fn take_verified_inline_body(&mut self) -> Option<Bytes> {
self.owned.as_mut().and_then(|owned| owned.verified_inline_body.take())
}
fn into_object_info(self, bucket: &str, object: &str, versioned: bool) -> ObjectInfo {
match (self.owned, self.shared) {
(Some(snapshot), None) => ObjectInfo::from_owned_file_info(snapshot.fi, bucket, object, versioned),
(None, Some(entry)) => match Arc::try_unwrap(entry) {
Ok(entry) => ObjectInfo::from_owned_file_info(entry.fi, bucket, object, versioned),
Err(entry) => ObjectInfo::from_file_info(&entry.fi, bucket, object, versioned),
},
_ => unreachable!("GET metadata snapshot representation must be exclusive"),
}
}
fn into_owned(self) -> (FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>) {
let (fi, parts_metadata, online_disks, _) = self.into_owned_with_late_metadata_fanout();
(fi, parts_metadata, online_disks)
@@ -2206,7 +2229,7 @@ mod prepared_get_object_metadata_tests {
let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
rustfs_io_metrics::set_get_stage_metrics_enabled(true);
let (restored, object_size, calls_total) = metrics::with_local_recorder(&recorder, || {
let (restored, object_size, calls_total, body_decodes) = metrics::with_local_recorder(&recorder, || {
runtime.block_on(async {
let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
let opts = ObjectOptions {
@@ -2260,7 +2283,12 @@ mod prepared_get_object_metadata_tests {
.await
.expect("inline GET body should stream");
(restored, object_size, calls.total(disk_call_counters::KIND_READ_VERSION))
(
restored,
object_size,
calls.total(disk_call_counters::KIND_READ_VERSION),
calls.total(disk_call_counters::KIND_BODY_DECODE),
)
},
)
.await
@@ -2274,6 +2302,10 @@ mod prepared_get_object_metadata_tests {
calls_total, 4,
"default production inline GET should schedule the initial bounded quorum plus one hedge"
);
assert_eq!(
body_decodes, 1,
"cold inline early-stop GET must reuse the quorum-verified inline body instead of decoding shards twice"
);
assert_eq!(
recorder.histogram_values(
"rustfs_io_get_object_metadata_fanout_scheduled",
@@ -2300,6 +2332,124 @@ mod prepared_get_object_metadata_tests {
);
}
#[test]
#[serial_test::serial(body_cache_hook)]
fn get_object_info_uses_early_stop_for_locked_or_cache_safe_reads_and_full_wait_for_unlocked() {
use crate::storage_api_contracts::object::ObjectOperations;
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("current-thread runtime should build");
let bucket = "get-object-info-early-stop-bucket";
let object = object_with_initial_data_shards(bucket, "get-object-info-early-stop-object");
let payload = b"head object early-stop payload".repeat(128);
runtime.block_on(async {
let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
let write_opts = ObjectOptions {
no_lock: true,
..Default::default()
};
set_disks
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("bucket should be created");
let mut put_reader = PutObjReader::from_vec(payload.clone());
set_disks
.put_object(bucket, &object, &mut put_reader, &write_opts)
.await
.expect("inline object should be written");
temp_env::async_with_vars(
[
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>),
("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
],
async {
let slow_parity_disk = bounded_initial_parity_disk_index(bucket, &object);
let barrier =
rename_fanout_barrier::arm(&object, slow_parity_disk, rename_fanout_barrier::PHASE_READ_VERSION);
let calls = disk_call_counters::observe(&object);
let set_disks_for_head = Arc::clone(&set_disks);
let object_for_head = object.clone();
let head_opts = ObjectOptions {
no_lock: true,
metadata_cache_safe: true,
..Default::default()
};
let mut head_task =
tokio::spawn(
async move { set_disks_for_head.get_object_info(bucket, &object_for_head, &head_opts).await },
);
tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
.await
.expect("cache-safe HEAD should pause a slow parity metadata read");
let info = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut head_task)
.await
.expect("cache-safe HEAD should return before the paused parity metadata response")
.expect("HEAD task should not panic")
.expect("HEAD should succeed");
assert_eq!(info.size, payload.len() as i64);
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
4,
"cache-safe HEAD should schedule bounded quorum plus one hedge"
);
assert_eq!(
calls.total(disk_call_counters::KIND_BODY_DECODE),
1,
"early-stop HEAD verifies inline quorum once"
);
// Second cache-safe HEAD should hit the QuorumSelected metadata cache entry (0 new disk reads).
let cached_info = set_disks
.get_object_info(bucket, &object, &ObjectOptions::default())
.await
.expect("cached HEAD with default locked options should succeed");
assert_eq!(cached_info.size, payload.len() as i64);
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
4,
"repeat locked HEAD must hit the QuorumSelected metadata cache entry"
);
// Release the slow parity barrier so full-wait can complete all 4 disks.
barrier.release();
// Unlocked internal preflight (no_lock=true, metadata_cache_safe=false) must bypass cache and full-wait.
let unlocked_info = set_disks
.get_object_info(
bucket,
&object,
&ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
..Default::default()
},
)
.await
.expect("unlocked preflight get_object_info should succeed");
assert_eq!(unlocked_info.size, payload.len() as i64);
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
8,
"unlocked preflight get_object_info must query all 4 disks"
);
assert_eq!(
calls.total(disk_call_counters::KIND_BODY_DECODE),
1,
"unlocked preflight full-wait must not run early-stop inline body decode"
);
},
)
.await;
});
}
#[test]
#[serial_test::serial(body_cache_hook)]
fn prepared_metadata_uses_full_fanout_even_when_data_read_early_stop_is_enabled() {
crates/ecstore/src/set_disk/ops/object.rs
+44 -4
View File
@@ -2873,7 +2873,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
};
let metadata_stage_start = stage_metrics_enabled.then(Instant::now);
let (snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() {
let (mut snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() {
(prepared.snapshot, prepared.object_info)
} else {
match self
@@ -2903,6 +2903,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
}
}
};
let verified_inline_body = snapshot.take_verified_inline_body();
let fi = snapshot.fi();
let files = snapshot.parts_metadata();
let disks = snapshot.online_disks();
@@ -3010,6 +3011,42 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
disks.get(index).is_some_and(Option::is_some)
})
{
if let Some(body) = verified_inline_body.filter(|body| body.len() == object_size) {
if let Some(inline_prepare_stage_start) = inline_prepare_stage_start.take() {
rustfs_io_metrics::record_get_object_stage_duration_by_size(
GET_OBJECT_PATH_INLINE_DIRECT,
GET_STAGE_INLINE_PREPARE,
object_class.as_str(),
size_bucket,
inline_prepare_stage_start.elapsed().as_secs_f64(),
);
}
if rustfs_io_metrics::get_stage_metrics_enabled() {
rustfs_io_metrics::record_get_object_stage_duration_by_size(
GET_OBJECT_PATH_INLINE_DIRECT,
GET_STAGE_READER_SETUP,
object_class.as_str(),
size_bucket,
0.0,
);
rustfs_io_metrics::record_get_object_stage_duration_by_size(
GET_OBJECT_PATH_INLINE_DIRECT,
GET_STAGE_DECODE,
object_class.as_str(),
size_bucket,
0.0,
);
}
record_get_object_reader_path_observation(GET_OBJECT_PATH_INLINE_DIRECT, object_class, size_bucket);
let reader = GetObjectReader {
stream: Box::new(Cursor::new(body.clone())),
object_info,
buffered_body: Some(body),
body_source: GetObjectBodySource::Unprobed,
};
return Ok(reader);
}
let read_length = inline_erasure_shard_file_offset(
0,
object_size,
@@ -10978,8 +11015,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
// Use the same full xl.meta read path as GetObject metadata resolution.
// This avoids HEAD/GetObject metadata visibility skew immediately after
// PutObject/CompleteMultipartUpload.
let read = self.get_object_fileinfo(bucket, object, opts, true, false);
// PutObject/CompleteMultipartUpload. Read-only callers that hold the
// shared lock and permit metadata caching can early-stop once write
// quorum agrees on the object identity.
let allow_early_stop = !opts.no_lock || opts.metadata_cache_safe;
let read = self.get_object_fileinfo(bucket, object, opts, true, allow_early_stop);
#[cfg(feature = "diagnostic-read-version-caller")]
let read = crate::cluster::rpc::runtime_sources::with_read_version_entry(
rustfs_io_metrics::internode_metrics::ReadVersionEntrySource::ObjectInfoRead,
@@ -10992,7 +11032,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
);
let snapshot = read.await.map_err(|e| to_object_err(e, vec![bucket, object]))?;
let oi = ObjectInfo::from_file_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
let oi = snapshot.into_object_info(bucket, object, opts.versioned || opts.version_suspended);
Ok(oi)
}
crates/ecstore/src/set_disk/read.rs
+4 -2
View File
@@ -774,7 +774,7 @@ impl SetDisks {
.await;
#[cfg(not(feature = "diagnostic-read-version-caller"))]
let metadata_read_result = metadata_read.await;
let metadata_read = match metadata_read_result {
let mut metadata_read = match metadata_read_result {
Ok(metadata_read) => metadata_read,
Err(err) => {
let err = StorageError::from(err);
@@ -784,6 +784,7 @@ impl SetDisks {
};
let metadata_fanout_complete = metadata_read.is_complete();
let condition_stopped = metadata_read.condition_stopped;
let verified_inline_body = metadata_read.verified_inline_body.take();
let (mut parts_metadata, errs, metadata_fanout_diagnostics) = metadata_read.into_legacy();
metadata_fanout_diagnostics.record(metadata_metrics_path);
// warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
@@ -902,6 +903,7 @@ impl SetDisks {
GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks)
};
snapshot.condition_stopped = condition_stopped;
snapshot.set_verified_inline_body(verified_inline_body);
Ok(snapshot)
}
@@ -916,7 +918,7 @@ impl SetDisks {
Ok(snapshot) => snapshot,
Err(e) => return (ObjectInfo::default(), FileInfo::default(), 0, Some(e)),
};
let fi = snapshot.fi().clone();
let (fi, _, _) = snapshot.into_owned();
let write_quorum = fi.write_quorum(self.default_write_quorum());