gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-01 09:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(ecstore): consolidate non-inline read planning (#6892)

Browse Source 
Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-31 01:15:48 +08:00
committed by GitHub
parent 8fd364a99c
commit d214c27583
3 changed files with 430 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/core/io_primitives.rs
+284 -48
View File
@@ -37,6 +37,8 @@ use super::super::ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS;
#[cfg(test)]
use super::super::ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX;
#[cfg(test)]
use super::super::ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE;
#[cfg(test)]
use super::super::get_metadata_slowtail_fault_delay;
use super::super::{
Bytes, CHECK_PART_DISK_NOT_FOUND, DeleteOptions, DiskError, DiskStore, EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
@@ -50,10 +52,10 @@ use super::super::{
disk, file_info_is_valid_for_metadata, get_metadata_slowtail_fault_request, info, inline_erasure_shard_file_offset,
inline_erasure_shard_size, is_err_object_not_found, is_err_version_not_found, is_get_metadata_data_read_early_stop_enabled,
is_get_metadata_early_stop_bounded_fanout_enabled, is_get_metadata_early_stop_enabled,
is_get_metadata_two_phase_read_plan_enabled, is_object_dangling, is_version_early_stop_enabled, issue3031_diag_enabled,
join_all, join_errs, log_multipart_write_quorum_failure, merge_file_meta_versions, path_join_buf, record_global_dirty_scope,
reduce_read_quorum_errs, reduce_write_quorum_errs, send_heal_request_with_admission, should_prevent_write, to_object_err,
try_read_inline_data_shards_direct, warn,
is_get_metadata_non_inline_data_read_early_stop_enabled, is_object_dangling, is_version_early_stop_enabled,
issue3031_diag_enabled, join_all, join_errs, log_multipart_write_quorum_failure, merge_file_meta_versions, path_join_buf,
record_global_dirty_scope, reduce_read_quorum_errs, reduce_write_quorum_errs, send_heal_request_with_admission,
should_prevent_write, to_object_err, try_read_inline_data_shards_direct, warn,
};
#[cfg(test)]
use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
@@ -688,6 +690,10 @@ pub(in crate::set_disk) struct MetadataQuorumAccumulator {
pub(in crate::set_disk) hard_errors: usize,
pub(in crate::set_disk) candidate: Option<FileInfo>,
pub(in crate::set_disk) candidate_votes: usize,
// Bitset of shard indexes whose metadata matches the candidate. Erasure
// layouts are capped at 16 shards, so this stays allocation-free on the
// GET metadata hot path.
candidate_shard_mask: u16,
pub(in crate::set_disk) conflicting_metadata: bool,
pub(in crate::set_disk) delete_marker_seen: bool,
pub(in crate::set_disk) delete_marker_candidates: Vec<(FileInfo, usize)>,
@@ -709,6 +715,7 @@ impl MetadataQuorumAccumulator {
hard_errors: 0,
candidate: None,
candidate_votes: 0,
candidate_shard_mask: 0,
conflicting_metadata: false,
delete_marker_seen: false,
delete_marker_candidates: Vec::new(),
@@ -724,6 +731,14 @@ impl MetadataQuorumAccumulator {
}
pub(in crate::set_disk) fn observe_file_info(&mut self, file_info: &FileInfo) {
self.observe_file_info_with_index(None, file_info);
}
pub(in crate::set_disk) fn observe_file_info_at(&mut self, disk_index: usize, file_info: &FileInfo) {
self.observe_file_info_with_index(Some(disk_index), file_info);
}
fn observe_file_info_with_index(&mut self, disk_index: Option<usize>, file_info: &FileInfo) {
if !file_info_is_valid_for_metadata(file_info) {
self.hard_errors = self.hard_errors.saturating_add(1);
return;
@@ -763,6 +778,11 @@ impl MetadataQuorumAccumulator {
match &self.candidate {
Some(candidate) if metadata_early_stop_candidate_matches(candidate, file_info) => {
self.candidate_votes = self.candidate_votes.saturating_add(1);
if let Some(disk_index) = disk_index
&& let Some(bit) = Self::candidate_shard_bit(candidate, file_info, disk_index)
{
self.candidate_shard_mask |= bit;
}
}
Some(_) => {
self.conflicting_metadata = true;
@@ -770,10 +790,38 @@ impl MetadataQuorumAccumulator {
None => {
self.candidate = Some(file_info.clone());
self.candidate_votes = 1;
if let Some(disk_index) = disk_index
&& let Some(bit) = Self::candidate_shard_bit(file_info, file_info, disk_index)
{
self.candidate_shard_mask |= bit;
}
}
}
}
fn candidate_shard_bit(candidate: &FileInfo, file_info: &FileInfo, disk_index: usize) -> Option<u16> {
let &erasure_index = candidate.erasure.distribution.get(disk_index)?;
if erasure_index == 0 || erasure_index > u16::BITS as usize || file_info.erasure.index != erasure_index {
return None;
}
Some(1u16 << (erasure_index - 1))
}
pub(in crate::set_disk) fn candidate_has_read_reserve(&self) -> bool {
self.candidate_read_reserve_target()
.is_some_and(|required| self.candidate_shard_mask.count_ones() as usize >= required)
}
pub(in crate::set_disk) fn candidate_read_reserve_target(&self) -> Option<usize> {
let candidate = self.candidate.as_ref()?;
Some(
candidate
.erasure
.data_blocks
.saturating_add(usize::from(candidate.erasure.parity_blocks > 0)),
)
}
pub(in crate::set_disk) fn observe_error(&mut self, err: &DiskError) {
match err {
DiskError::FileNotFound | DiskError::VolumeNotFound => {
@@ -1084,11 +1132,7 @@ fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Op
None
}
fn non_inline_data_read_candidate_is_safe(
candidate: &FileInfo,
parts_metadata: &[FileInfo],
disks: &[Option<DiskStore>],
) -> bool {
fn non_inline_data_read_candidate_is_safe(candidate: &FileInfo) -> bool {
if candidate.inline_data()
|| candidate.is_compressed()
|| candidate.is_remote()
@@ -1100,34 +1144,11 @@ fn non_inline_data_read_candidate_is_safe(
{
return false;
}
let Ok(erasure) = coding::Erasure::try_new_with_options(
candidate.erasure.data_blocks,
candidate.erasure.parity_blocks,
candidate.erasure.block_size,
candidate.uses_legacy_checksum,
) else {
return false;
};
// The regular reader setup can reconstruct missing data shards from any
// `data_shards` matching metadata entries. Requiring every data slot here
// would unnecessarily wait for one slow data disk even when parity and
// the remaining data shards already form a read quorum.
let mut available_shards = vec![false; erasure.data_shards + erasure.parity_shards];
for ((file_info, disk), &erasure_index) in parts_metadata
.iter()
.zip(disks.iter())
.zip(candidate.erasure.distribution.iter())
{
if erasure_index == 0 || erasure_index > available_shards.len() || disk.is_none() {
continue;
}
if metadata_early_stop_candidate_matches(file_info, candidate) && file_info.erasure.index == erasure_index {
available_shards[erasure_index - 1] = true;
}
}
available_shards.into_iter().filter(|present| *present).count() >= erasure.data_shards
candidate.has_valid_erasure_geometry()
}
const NON_INLINE_SINGLE_PENDING_HEDGE_DELAY: Duration = Duration::from_millis(100);
fn data_read_inline_missing_shards_are_pending(
candidate: &FileInfo,
parts_metadata: &[FileInfo],
@@ -2910,7 +2931,7 @@ impl SetDisks {
read_data,
healing,
incl_free_versions,
read_data && is_get_metadata_two_phase_read_plan_enabled(),
read_data && is_get_metadata_non_inline_data_read_early_stop_enabled(),
default_parity_count,
allow_coalescing,
)
@@ -3081,6 +3102,8 @@ impl SetDisks {
let mut scheduled_count = 0usize;
let mut force_full_wait = false;
let mut final_miss_reason_override = None;
let mut non_inline_candidate_eligible = None;
let mut single_pending_hedge_deadline = None;
let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
let spawn_read_version =
|join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
@@ -3128,18 +3151,54 @@ impl SetDisks {
}
}
while let Some(result) = join_set.join_next().await {
loop {
let mut defer_pending_inline_data_shard = false;
let result = if let Some(deadline) = single_pending_hedge_deadline.take() {
tokio::select! {
result = join_set.join_next() => result,
_ = tokio::time::sleep_until(deadline) => {
if bounded_fanout
&& !force_full_wait
&& join_set.len() == 1
&& non_inline_candidate_eligible == Some(true)
&& !accumulator.candidate_has_read_reserve()
&& next_fanout_index < disks.len()
{
while next_fanout_index < disks.len() {
let disk_index = fanout_order[next_fanout_index];
next_fanout_index = next_fanout_index.saturating_add(1);
if let Some(disk) = disks.get(disk_index).cloned() {
spawn_read_version(&mut join_set, disk_index, disk);
scheduled_count = scheduled_count.saturating_add(1);
break;
}
}
}
continue;
}
}
} else {
join_set.join_next().await
};
let Some(result) = result else { break };
match result {
Ok((index, res, elapsed)) => match res {
Ok(file_info) => {
observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
accumulator.observe_file_info(&file_info);
if allow_non_inline_data_read_early_stop {
accumulator.observe_file_info_at(index, &file_info);
} else {
accumulator.observe_file_info(&file_info);
}
if allow_non_inline_data_read_early_stop && non_inline_candidate_eligible.is_none() {
non_inline_candidate_eligible =
accumulator.candidate.as_ref().map(non_inline_data_read_candidate_is_safe);
}
if bounded_fanout
&& read_data
&& !force_full_wait
&& let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
&& !(allow_non_inline_data_read_early_stop
&& !(non_inline_candidate_eligible == Some(true)
&& reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE)
{
force_full_wait = true;
@@ -3171,11 +3230,8 @@ impl SetDisks {
{
let should_return_early = if read_data {
match accumulator.candidate.as_ref() {
Some(candidate)
if allow_non_inline_data_read_early_stop
&& non_inline_data_read_candidate_is_safe(candidate, &ress, disks) =>
{
true
Some(_candidate) if non_inline_candidate_eligible == Some(true) => {
accumulator.candidate_has_read_reserve()
}
Some(candidate) => match data_read_early_stop_inline_body_miss_reason(
bucket.as_ref(),
@@ -3247,12 +3303,37 @@ impl SetDisks {
}
let pending_responses = join_set.len();
let should_hedge_single_pending_data_read = read_data
// Inline verification can still depend on a missing data shard;
// issue one immediate spare when only that shard remains. The
// non-inline path keeps its delayed hedge below to avoid healthy
// reads paying speculative I/O before the candidate is classified.
let should_hedge_single_pending_inline_read = read_data
&& !force_full_wait
&& !defer_pending_inline_data_shard
&& pending_responses == 1
&& non_inline_candidate_eligible != Some(true)
&& accumulator.can_still_reach_early_stop_with_pending(pending_responses);
if bounded_fanout && force_full_wait {
// A non-inline plan must retain one extra matching shard as a
// reconstruction reserve. Schedule that reserve only after the
// candidate is known to be eligible, so inline GETs do not pay an
// extra fanout and the healthy path remains allocation-free.
let needs_non_inline_read_reserve = non_inline_candidate_eligible == Some(true)
&& !accumulator.candidate_has_read_reserve()
&& accumulator
.candidate_read_reserve_target()
.is_some_and(|reserve_target| scheduled_count < reserve_target || pending_responses == 0);
if bounded_fanout
&& !force_full_wait
&& (needs_non_inline_read_reserve || should_hedge_single_pending_inline_read)
&& next_fanout_index < disks.len()
{
let disk_index = fanout_order[next_fanout_index];
if let Some(disk) = disks.get(disk_index).cloned() {
spawn_read_version(&mut join_set, disk_index, disk);
scheduled_count = scheduled_count.saturating_add(1);
}
next_fanout_index = next_fanout_index.saturating_add(1);
} else if bounded_fanout && force_full_wait {
while next_fanout_index < disks.len() {
let disk_index = fanout_order[next_fanout_index];
if let Some(disk) = disks.get(disk_index).cloned() {
@@ -3264,8 +3345,7 @@ impl SetDisks {
} else if bounded_fanout
&& !defer_pending_inline_data_shard
&& next_fanout_index < disks.len()
&& (!accumulator.can_still_reach_early_stop_with_pending(pending_responses)
|| should_hedge_single_pending_data_read)
&& !accumulator.can_still_reach_early_stop_with_pending(pending_responses)
{
let disk_index = fanout_order[next_fanout_index];
if let Some(disk) = disks.get(disk_index).cloned() {
@@ -3274,6 +3354,17 @@ impl SetDisks {
}
next_fanout_index = next_fanout_index.saturating_add(1);
}
if bounded_fanout
&& !force_full_wait
&& !defer_pending_inline_data_shard
&& join_set.len() == 1
&& non_inline_candidate_eligible == Some(true)
&& !accumulator.candidate_has_read_reserve()
&& accumulator.can_still_reach_early_stop_with_pending(join_set.len())
&& next_fanout_index < disks.len()
{
single_pending_hedge_deadline = Some(tokio::time::Instant::now() + NON_INLINE_SINGLE_PENDING_HEDGE_DELAY);
}
}
let accumulator_miss_reason = accumulator.final_miss_reason();
@@ -7358,6 +7449,90 @@ mod tests {
drop(dirs);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn metadata_slowtail_fault_gate_stops_before_unneeded_tail() {
const DISKS: usize = 4;
let bucket = "metadata-slowtail-gated-bucket";
let object = "objects/metadata-slowtail-gated-object";
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
install_mapped_metadata_fanout_fileinfo(&disks, bucket, object).await;
let order = bounded_metadata_fanout_order(bucket, object, DISKS, 2);
let slow_disk = *order.get(3).expect("four-disk fanout should have a deferred tail disk");
let slow_disk_env = slow_disk.to_string();
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some(slow_disk_env.as_str())),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
async {
let calls = disk_call_counters::observe(object);
let read_with_data =
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2);
let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_millis(500), read_with_data)
.await
.expect("gated metadata read should stop before the deferred slow tail")
.expect("gated metadata fanout should resolve");
assert!(parts_metadata.iter().filter(|fi| fi.name == object).count() >= 3);
assert!(errs.iter().all(Option::is_none));
assert!(diagnostics.total_responses() < DISKS);
assert_eq!(calls.total(disk_call_counters::KIND_METADATA_SLOWTAIL_FAULT), 0);
},
)
.await;
drop(dirs);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn metadata_slowtail_fault_gate_hedges_an_initial_slow_data_shard() {
const DISKS: usize = 4;
let bucket = "metadata-slowtail-gated-initial-bucket";
let object = "objects/metadata-slowtail-gated-initial-object";
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
install_mapped_metadata_fanout_fileinfo(&disks, bucket, object).await;
let order = bounded_metadata_fanout_order(bucket, object, DISKS, 2);
let slow_disk = *order.get(1).expect("four-disk fanout should have an initial data disk");
let spare_disk = *order.get(3).expect("four-disk fanout should have a spare disk");
let slow_disk_env = slow_disk.to_string();
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("500")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some(slow_disk_env.as_str())),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
async {
let calls = disk_call_counters::observe(object);
let read_with_data =
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2);
let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_millis(300), read_with_data)
.await
.expect("gated metadata read should hedge the initial slow shard")
.expect("gated metadata fanout should resolve");
assert!(parts_metadata.iter().filter(|fi| fi.name == object).count() >= 3);
assert!(errs.iter().all(Option::is_none));
assert!(diagnostics.total_responses() < DISKS);
assert_eq!(calls.for_disk(disk_call_counters::KIND_METADATA_SLOWTAIL_FAULT, slow_disk), 1);
assert_eq!(calls.for_disk(disk_call_counters::KIND_READ_VERSION, spare_disk), 1);
},
)
.await;
drop(dirs);
}
/// Demo / regression guard for the backlog#1325 per-disk call counters.
///
/// The metadata fan-out issues each `read_version` inside its own
@@ -7771,6 +7946,32 @@ mod tests {
}
}
async fn install_mapped_metadata_fanout_fileinfo(disks: &[Option<DiskStore>], bucket: &str, object: &str) {
let version_id = Uuid::new_v4();
let data_dir = Uuid::new_v4();
let mod_time = OffsetDateTime::now_utc();
let distribution = FileInfo::new(&metadata_distribution_key(bucket, object), 2, 2)
.erasure
.distribution;
for (index, disk) in disks
.iter()
.enumerate()
.filter_map(|(index, disk)| disk.as_ref().map(|disk| (index, disk)))
{
disk.write_all(bucket, &format!("{object}/{data_dir}/part.1"), Bytes::from_static(b"x"))
.await
.expect("part data should be installed on every disk");
let mut file_info = valid_metadata_fanout_fileinfo(bucket, object, version_id, data_dir, mod_time);
file_info.erasure.distribution = distribution.clone();
file_info.erasure.index = *distribution
.get(index)
.expect("mapped metadata distribution should cover every disk");
disk.write_metadata(bucket, bucket, object, file_info)
.await
.expect("mapped metadata should be installed on every disk");
}
}
async fn inline_metadata_fanout_fileinfos_with_mode(
bucket: &str,
object: &str,
@@ -10393,6 +10594,41 @@ mod tests {
assert_eq!(accumulator.candidate_latest_quorum(&impossible_parity), None);
}
#[test]
fn metadata_quorum_accumulator_tracks_mapped_shards_and_requires_a_reserve() {
let version_id = Uuid::new_v4();
let data_dir = Uuid::new_v4();
let base = valid_metadata_fanout_fileinfo("bucket", "object", version_id, data_dir, OffsetDateTime::now_utc());
let distribution = base.erasure.distribution.clone();
let mut accumulator = MetadataQuorumAccumulator::new(4, 2, true);
for (disk_index, &erasure_index) in distribution.iter().take(2).enumerate() {
let mut file_info = base.clone();
file_info.erasure.index = erasure_index;
accumulator.observe_file_info_at(disk_index, &file_info);
}
assert!(
!accumulator.candidate_has_read_reserve(),
"data quorum without parity reserve must not early-stop"
);
let mut mismatched = base.clone();
mismatched.erasure.index = distribution[3];
accumulator.observe_file_info_at(2, &mismatched);
assert!(
!accumulator.candidate_has_read_reserve(),
"mapped index mismatch must not count as a reserve"
);
let mut reserve = base;
reserve.erasure.index = distribution[2];
accumulator.observe_file_info_at(2, &reserve);
assert!(
accumulator.candidate_has_read_reserve(),
"one matching reserve shard should complete the read reserve"
);
}
#[test]
fn metadata_quorum_accumulator_treats_invalid_default_parity_as_full_fanout() {
let accumulator = MetadataQuorumAccumulator::new(2, 2, true);
crates/ecstore/src/set_disk/mod.rs
+140 -20
View File
@@ -773,10 +773,11 @@ const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false;
const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE";
const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
// Two-phase metadata/read-plan rollout (backlog#1309). The first phase reads
// metadata without inline payloads and only fetches inline data from the
// selected data-shard slots. Keep this opt-in until the Linux multi-node
// slow-tail and small-inline cost gates are complete.
// Opt-in non-inline data-read quorum early-stop rollout (backlog#1309). The
// existing metadata fanout still reads data-bearing metadata; this gate only
// permits a safe plain single-part candidate to stop before the full fanout.
// Keep it opt-in until the Linux multi-node slow-tail and small-inline cost
// gates are complete. The environment name is retained for compatibility.
const ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE: &str = "RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE";
const DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE: bool = false;
@@ -1037,14 +1038,19 @@ mod prepared_get_object_metadata_tests {
const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {
object_with_initial_data_shards_for_geometry(bucket, prefix, 4, 2)
}
fn object_with_initial_data_shards_for_geometry(bucket: &str, prefix: &str, total_disks: usize, parity: usize) -> String {
(0..1000)
.map(|index| format!("{prefix}-{index}.bin"))
.find(|name| {
let order = bounded_metadata_fanout_order(bucket, name, 4, 2);
let distribution = FileInfo::new(&[bucket, name].join("/"), 2, 2).erasure.distribution;
let mut seen = [false; 2];
for disk_index in order.into_iter().take(3) {
if let Some(block_index @ 1..=2) = distribution.get(disk_index).copied() {
let order = bounded_metadata_fanout_order(bucket, name, total_disks, parity);
let data = total_disks.saturating_sub(parity);
let distribution = FileInfo::new(&[bucket, name].join("/"), data, parity).erasure.distribution;
let mut seen = vec![false; data];
for disk_index in order.into_iter().take(total_disks.saturating_sub(parity).saturating_add(1)) {
if let Some(block_index) = distribution.get(disk_index).copied().filter(|index| *index <= data) {
seen[block_index - 1] = true;
}
}
@@ -1177,9 +1183,9 @@ mod prepared_get_object_metadata_tests {
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn two_phase_read_plan_uses_metadata_only_for_non_inline_get() {
async fn non_inline_data_read_early_stop_uses_quorum_plan() {
let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
let bucket = "two-phase-read-plan";
let bucket = "non-inline-read-plan";
let object = object_with_initial_data_shards(bucket, "non-inline-object");
let payload = vec![0x5a; 2 * 1024 * 1024];
let opts = ObjectOptions {
@@ -1209,17 +1215,17 @@ mod prepared_get_object_metadata_tests {
let mut reader = set_disks
.get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
.await
.expect("two-phase GET reader should open");
.expect("quorum GET reader should open");
let mut restored = Vec::new();
reader
.stream
.read_to_end(&mut restored)
.await
.expect("two-phase GET body should stream");
.expect("quorum GET body should stream");
assert_eq!(restored, payload);
assert!(
calls.total(disk_call_counters::KIND_READ_VERSION) < 4,
"non-inline two-phase GET should stop metadata fanout at a quorum"
"non-inline quorum GET should retain a reconstruction reserve"
);
},
)
@@ -1228,11 +1234,62 @@ mod prepared_get_object_metadata_tests {
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn two_phase_read_plan_preserves_inline_early_stop_path() {
async fn non_inline_data_read_early_stop_keeps_reserve_on_unequal_layout() {
let (_dirs, set_disks) = make_local_set_disks(6, 2).await;
let bucket = "non-inline-read-reserve";
let object = object_with_initial_data_shards_for_geometry(bucket, "reserve-object", 6, 2);
let payload = vec![0x5a; 2 * 1024 * 1024];
let 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, &opts)
.await
.expect("object should be written");
temp_env::async_with_vars(
[
("RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
],
async {
let calls = disk_call_counters::observe(&object);
let mut reader = set_disks
.get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
.await
.expect("quorum GET reader should open");
let mut restored = Vec::new();
reader
.stream
.read_to_end(&mut restored)
.await
.expect("quorum GET body should stream");
assert_eq!(restored, payload);
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
5,
"the unequal layout should schedule exactly one reserve beyond its data quorum"
);
},
)
.await;
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn non_inline_data_read_early_stop_preserves_inline_path() {
let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
let bucket = "two-phase-read-plan-inline";
let bucket = "non-inline-read-plan-inline";
let object = object_with_initial_data_shards(bucket, "inline-object");
let payload = b"two-phase inline payload".repeat(256);
let payload = b"quorum inline payload".repeat(256);
let opts = ObjectOptions {
no_lock: true,
..Default::default()
@@ -1259,13 +1316,13 @@ mod prepared_get_object_metadata_tests {
let mut reader = set_disks
.get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
.await
.expect("two-phase inline GET reader should open");
.expect("inline GET reader should open");
let mut restored = Vec::new();
reader
.stream
.read_to_end(&mut restored)
.await
.expect("two-phase inline GET body should stream");
.expect("inline GET body should stream");
assert_eq!(restored, payload);
assert_eq!(
test_get_object_reader_path_id(),
@@ -1278,6 +1335,69 @@ mod prepared_get_object_metadata_tests {
.await;
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn non_inline_data_read_early_stop_does_not_add_inline_fanout_on_unequal_layout() {
let (_dirs, set_disks) = make_local_set_disks(6, 2).await;
let bucket = "inline-read-plan-unequal";
let object = object_with_initial_data_shards_for_geometry(bucket, "inline-object", 6, 2);
let payload = b"inline quorum payload".repeat(256);
let 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, &opts)
.await
.expect("inline object should be written");
let read_once = |enabled: bool| {
let set_disks = Arc::clone(&set_disks);
let bucket = bucket.to_string();
let object = object.clone();
let payload = payload.clone();
let opts = opts.clone();
async move {
temp_env::async_with_vars(
[
(
"RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE",
Some(if enabled { "true" } else { "false" }),
),
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
],
async {
let calls = disk_call_counters::observe(&object);
let mut reader = set_disks
.get_object_reader(&bucket, &object, None, HeaderMap::new(), &opts)
.await
.expect("inline GET reader should open");
let mut restored = Vec::new();
reader
.stream
.read_to_end(&mut restored)
.await
.expect("inline GET body should stream");
assert_eq!(restored, payload);
calls.total(disk_call_counters::KIND_READ_VERSION)
},
)
.await
}
};
let gate_off_calls = read_once(false).await;
let gate_on_calls = read_once(true).await;
assert_eq!(gate_on_calls, gate_off_calls, "inline gate must not add reserve fanout");
}
#[test]
#[serial_test::serial(body_cache_hook)]
fn inline_data_read_early_stop_defaults_return_exact_body() {
@@ -2024,7 +2144,7 @@ fn is_get_metadata_data_read_early_stop_enabled() -> bool {
}
}
fn is_get_metadata_two_phase_read_plan_enabled() -> bool {
fn is_get_metadata_non_inline_data_read_early_stop_enabled() -> bool {
#[cfg(test)]
{
rustfs_utils::get_env_bool(
crates/ecstore/src/set_disk/read.rs
+6 -5
View File
@@ -128,7 +128,7 @@ use super::is_get_metadata_early_stop_bounded_fanout_enabled;
#[cfg(test)]
use super::is_get_metadata_early_stop_enabled;
#[cfg(test)]
use super::is_get_metadata_two_phase_read_plan_enabled;
use super::is_get_metadata_non_inline_data_read_early_stop_enabled;
#[cfg(test)]
use super::is_version_early_stop_enabled;
#[cfg(test)]
@@ -4272,15 +4272,16 @@ mod tests {
}
#[test]
fn two_phase_read_plan_gate_defaults_off_and_honors_override() {
#[serial(body_cache_hook)]
fn non_inline_data_read_early_stop_gate_defaults_off_and_honors_override() {
temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, None::<&str>, || {
assert!(!is_get_metadata_two_phase_read_plan_enabled());
assert!(!is_get_metadata_non_inline_data_read_early_stop_enabled());
});
temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("true"), || {
assert!(is_get_metadata_two_phase_read_plan_enabled());
assert!(is_get_metadata_non_inline_data_read_early_stop_enabled());
});
temp_env::with_var(ENV_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE, Some("false"), || {
assert!(!is_get_metadata_two_phase_read_plan_enabled());
assert!(!is_get_metadata_non_inline_data_read_early_stop_enabled());
});
}