mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 00:38:16 +00:00
perf(ecstore): k-way heap merge for ListObjects, drop clone-to-parse (#4347)
* perf(ecstore): replace linear merge scan with k-way heap and drop clone-to-parse (backlog#874 backlog#875) merge_entry_channels advanced the k-way merge with a linear scan over all channel heads (O(entries x channels)) and allocated two fresh Strings per pairwise comparison via path::clean. Every step also cloned MetaCacheEntry values, including entry.clone().xl_meta() clone-to-parse calls. - Introduce MergeHead with a cached cleaned name (allocated only when the raw name is not already clean) and drive the merge with a BinaryHeap of boxed heads: O(log channels) per entry, allocation-free comparisons. - Move entries through the merge instead of cloning; the winner is sent without an intermediate copy. - Remove the dead merge_file_meta_versions block: it only ran for prefix-dir groups whose entries have empty metadata, so xl_meta() always failed; cross-drive version merging happens in the resolve path. - Keep legacy same-name semantics (dir groups collapse, objects shadow prefix dirs, later object candidate wins) and add regression tests for interleaved ordering, dir/object precedence, uncleaned-name grouping, and prefix-dir collapse. Co-Authored-By: heihutu <heihutu@gmail.com> * fix(ecstore): honor ascending versions_sort in ListObjects walk (#4348) * fix(ecstore): honor ascending versions_sort in walk and document ordering invariant (backlog#876) The walk loop carried a bare `//TODO: SORT` inside the `WalkVersionsSortOrder::Ascending` branch, so the requested ascending order was silently ignored and versions streamed newest-first (the raw FileMeta order). WalkOptions defaults to Ascending, so every default walker -- notably replication resync, which replays versions and needs oldest-first to preserve the version-stack order -- received the exact opposite of the contract. FileMeta maintains versions newest-first (sort_by_mod_time is descending) and into_file_info_versions preserves that order, so ascending emission is the exact reverse of file_info_versions output. Reverse in place when ascending is requested and add a regression test locking the newest-first invariant plus the reversal contract. Key-ordering audit result (no gap found): per-disk walkers emit sorted streams, merge_entry_channels performs an ordered k-way merge, and gather_results only filters by marker/limit, so ListObjects key order is guaranteed upstream and needs no post-sort. Co-Authored-By: heihutu <heihutu@gmail.com> * perf(ecstore): enable GET metadata early-stop by default (#4349) * perf(ecstore): enable GET metadata early-stop by default with env opt-out (backlog#872) The metadata early-stop fanout (read_all_fileinfo_early_stop) has been implemented and instrumented for a while but stayed behind an opt-in flag, so default GETs always waited for every disk to answer the metadata read even after quorum agreement was reached. Flip RUSTFS_GET_METADATA_EARLY_STOP_ENABLE to default-on. The gate stays conservative: should_allow_metadata_early_stop only admits metadata-only reads (read_data=false) without version_id, healing, or free-version requirements, everything else falls back to the full-wait fanout, and setting the env var to false restores the old behavior entirely. The version-aware gate (RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE) remains opt-in because versioned reads carry a higher stale-selection risk profile. Also replace the stale "optimize concurrency" TODO in get_object_fileinfo with a pointer to the early-stop implementation and add regression tests for the new default plus the explicit opt-out path. Co-Authored-By: heihutu <heihutu@gmail.com> * perf(ecstore): lazily construct codec streaming multipart readers (#4350) * perf(ecstore): lazily construct codec streaming multipart part readers (backlog#871) get_object_decode_reader_with_fileinfo opened shard readers for every part of a multipart object before returning the streaming reader, so TTFB paid for parts x disks file opens up front and an early client disconnect wasted the setup work for every unread part. Replace the eager loop with LazyMultipartCodecStreamingReader: the first part is still built eagerly so the dominant fallback conditions (missing shards / read quorum) are detected before any byte is streamed and the whole request can fall back to the legacy duplex path exactly as before. Each subsequent part is built on demand -- when the previous part hits EOF -- via a spawned task handle owned by the reader; dropping the reader aborts an in-flight build so disconnects stop all further IO. If a later part hits a fallback condition mid-stream (a shard vanished after the request started), the reader surfaces an explicit read error with a pipeline-failure metric instead of silently degrading; the client's retry then detects the condition on the eager first-part setup and takes the legacy path cleanly. Adds unit tests for in-order streaming across lazy boundaries, deferred construction (no build when the client stops within part 1), and the mid-stream fallback error path. Co-Authored-By: heihutu <heihutu@gmail.com> * perf(ecstore): prefetch next multipart part reader setup during decode (#4351) * perf(ecstore): prefetch next multipart part reader setup during decode (backlog#870) get_object_with_fileinfo processed multipart parts strictly serially: the next part's bitrot reader setup (file opens + read-quorum wait across all disks) only started after the current part finished decoding, so large multipart reads paid full setup latency between every part. Overlap the two stages with a depth-one pipeline: right after the current part's readers are obtained, the next part's setup is spawned (shared inputs behind Arc) and joined when the loop reaches that part. The shared setup_multipart_part_readers helper keeps stage-duration metrics semantics identical for both paths; a failed or stale prefetch falls back to the synchronous setup, and the PrefetchedReaderSetup guard aborts the in-flight task on error returns, early breaks, or caller drop so disconnects stop background disk IO. Gate: RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH (default on, env opt-out). Adds a three-part end-to-end read test covering the prefetch hit path and cross-part content ordering. Co-Authored-By: heihutu <heihutu@gmail.com> * perf(ecstore): move FileInfo through GET shuffle instead of cloning (#4352) perf(ecstore): move FileInfo entries through the GET shuffle instead of cloning (backlog#873) shuffle_disks_and_parts_metadata_by_index deep-cloned every valid FileInfo (parts, erasure info, metadata map) once per disk on each GET. Add an ownership-taking variant that runs the same by-index consistency check as a read-only first pass and then moves entries into their shuffled slots with mem::take, and switch get_object_with_fileinfo to it -- that call site already owned the parts metadata vector. Disk handles are Arc clones and stay cheap. Scope notes from the backlog#873 audit: - get_object_fileinfo's disks.clone() stays: DiskStore is Arc<Disk>, so the clone is per-slot refcounting and correctly avoids holding the RwLock read guard across the metadata fanout awaits. - get_object_decode_reader_with_fileinfo keeps the borrowing shuffle: its caller must retain files/disks for the legacy fallback path, so an owned variant would just shift the same clone upstream. - The metadata-cache hit path still clones parts_metadata; sharing the cached entry via Arc changes the read-path return types and is left as a follow-up. Equivalence tests cover both the by-index placement and the mod-time fallback against the borrowing variant. Co-authored-by: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> * fix(ecstore): gate merge emission on cleaned key and clear clippy redundant_clone Address review + CI findings on the ListObjects/GET optimization PR: - merge_entry_channels gated emission on the raw entry name while the heap orders by the cleaned key, so entries whose cleaned order and raw byte order disagree (e.g. redundant slashes) could be dropped. Gate on the same cleaned sort key the heap uses; add a regression test (`a//c` after `a/b`). - Drop three redundant `.clone()` calls in test code flagged by clippy::redundant_clone (owned-shuffle equivalence tests and the walk ascending-versions contract test) that failed the CI clippy gate. - Document the known mid-stream fallback limitation of the opt-in multipart codec streaming reader (default off) and mark the in-place per-part legacy degradation as a follow-up. Co-Authored-By: heihutu <heihutu@gmail.com> * fix(ecstore): force full metadata fanout for object tagging writes (backlog#872) put_object_tags reads the object fileinfo with read_data=false and then writes the updated tags to the online-disk set that read returned. With metadata early-stop enabled by default, that read now returns as soon as read quorum is reached, so the online-disk set is only a read-quorum subset. Writing tags to that subset fails write quorum -> ErasureWriteQuorum -> S3 SlowDown, which is exactly the s3-tests tagging failures (PutObjectTagging/DeleteObjectTagging, reached max retries). Thread a caller-controlled `allow_early_stop` gate through read_all_fileinfo_observed/_inner and add get_object_fileinfo_gated; put_object_tags calls it with allow_early_stop=false so the metadata read does the full quorum fanout and returns the complete online-disk set as the write target. Pure-read callers (GET/HEAD/tag read) keep the early-stop fast path unchanged. Extract metadata_early_stop_permitted() as the single gate and add a unit test locking the invariant: caller opt-out (and observe=false, and data reads) never early-stop even with the env flags on. Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
@@ -1721,6 +1721,23 @@ pub(in crate::set_disk) fn should_allow_metadata_early_stop(
|
||||
|| (is_version_early_stop_enabled() && !version_id.is_empty() && !healing)
|
||||
}
|
||||
|
||||
/// Final gate for the metadata early-stop fast path.
|
||||
///
|
||||
/// `caller_allows_early_stop=false` unconditionally forces the full quorum
|
||||
/// fanout so read-before-write callers (object tagging) get the complete
|
||||
/// online-disk set as their write target; the early-stop subset would only
|
||||
/// carry read quorum and fail write quorum (backlog#872 regression).
|
||||
pub(in crate::set_disk) fn metadata_early_stop_permitted(
|
||||
caller_allows_early_stop: bool,
|
||||
observe: bool,
|
||||
read_data: bool,
|
||||
version_id: &str,
|
||||
healing: bool,
|
||||
incl_free_versions: bool,
|
||||
) -> bool {
|
||||
caller_allows_early_stop && observe && should_allow_metadata_early_stop(read_data, version_id, healing, incl_free_versions)
|
||||
}
|
||||
|
||||
impl SetDisks {
|
||||
pub(in crate::set_disk) async fn read_parts(
|
||||
disks: &[Option<DiskStore>],
|
||||
@@ -1797,6 +1814,7 @@ impl SetDisks {
|
||||
healing,
|
||||
incl_free_versions,
|
||||
false,
|
||||
true,
|
||||
0,
|
||||
)
|
||||
.await?;
|
||||
@@ -1813,6 +1831,7 @@ impl SetDisks {
|
||||
read_data: bool,
|
||||
healing: bool,
|
||||
incl_free_versions: bool,
|
||||
allow_early_stop: bool,
|
||||
default_parity_count: usize,
|
||||
) -> disk::error::Result<(Vec<FileInfo>, Vec<Option<DiskError>>, MetadataFanoutDiagnostics)> {
|
||||
Self::read_all_fileinfo_inner(
|
||||
@@ -1825,6 +1844,7 @@ impl SetDisks {
|
||||
healing,
|
||||
incl_free_versions,
|
||||
true,
|
||||
allow_early_stop,
|
||||
default_parity_count,
|
||||
)
|
||||
.await
|
||||
@@ -1841,10 +1861,18 @@ impl SetDisks {
|
||||
healing: bool,
|
||||
incl_free_versions: bool,
|
||||
observe: bool,
|
||||
// When false, the caller opts out of the early-stop fast path even for
|
||||
// otherwise-eligible reads. Read-before-write callers (e.g. object
|
||||
// tagging) must set this so the returned online-disk set reflects the
|
||||
// full quorum fanout rather than the early-stop subset — writing to the
|
||||
// subset would fail write quorum (backlog#872 regression).
|
||||
caller_allows_early_stop: bool,
|
||||
default_parity_count: usize,
|
||||
) -> disk::error::Result<(Vec<FileInfo>, Vec<Option<DiskError>>, MetadataFanoutDiagnostics)> {
|
||||
let early_stop_enabled = observe && (is_get_metadata_early_stop_enabled() || is_version_early_stop_enabled());
|
||||
let allow_early_stop = observe && should_allow_metadata_early_stop(read_data, version_id, healing, incl_free_versions);
|
||||
let early_stop_enabled =
|
||||
caller_allows_early_stop && observe && (is_get_metadata_early_stop_enabled() || is_version_early_stop_enabled());
|
||||
let allow_early_stop =
|
||||
metadata_early_stop_permitted(caller_allows_early_stop, observe, read_data, version_id, healing, incl_free_versions);
|
||||
if allow_early_stop {
|
||||
return Self::read_all_fileinfo_early_stop(
|
||||
disks,
|
||||
|
||||
@@ -839,6 +839,55 @@ impl SetDisks {
|
||||
Err(DiskError::ErasureReadQuorum)
|
||||
}
|
||||
|
||||
/// Ownership-taking variant of `shuffle_disks_and_parts_metadata_by_index`
|
||||
/// (backlog#873): callers that already own the vectors avoid one deep
|
||||
/// `FileInfo` clone per disk by moving entries into their shuffled slots.
|
||||
///
|
||||
/// Semantics match the borrowing variant, including the fallback to the
|
||||
/// mod-time based placement when `parity_blocks` or more sources are
|
||||
/// inconsistent; the consistency check runs as a read-only first pass so
|
||||
/// the fallback still sees the untouched inputs.
|
||||
pub(super) fn shuffle_disks_and_parts_metadata_by_index_owned(
|
||||
mut disks: Vec<Option<DiskStore>>,
|
||||
mut parts_metadata: Vec<FileInfo>,
|
||||
fi: &FileInfo,
|
||||
) -> (Vec<Option<DiskStore>>, Vec<FileInfo>) {
|
||||
let distribution = &fi.erasure.distribution;
|
||||
|
||||
let mut inconsistent = 0;
|
||||
for (k, v) in parts_metadata.iter().enumerate() {
|
||||
if disks[k].is_none() || !v.is_valid() || distribution[k] != v.erasure.index {
|
||||
inconsistent += 1;
|
||||
}
|
||||
}
|
||||
|
||||
let use_by_index = inconsistent < fi.erasure.parity_blocks;
|
||||
let init = fi.mod_time.is_none();
|
||||
|
||||
let mut shuffled_disks = vec![None; disks.len()];
|
||||
let mut shuffled_parts_metadata = vec![FileInfo::default(); parts_metadata.len()];
|
||||
|
||||
for k in 0..parts_metadata.len() {
|
||||
if disks[k].is_none() {
|
||||
continue;
|
||||
}
|
||||
let eligible = if use_by_index {
|
||||
parts_metadata[k].is_valid() && distribution[k] == parts_metadata[k].erasure.index
|
||||
} else {
|
||||
init || parts_metadata[k].is_valid()
|
||||
};
|
||||
if !eligible {
|
||||
continue;
|
||||
}
|
||||
|
||||
let block_idx = distribution[k];
|
||||
shuffled_parts_metadata[block_idx - 1] = std::mem::take(&mut parts_metadata[k]);
|
||||
shuffled_disks[block_idx - 1] = disks[k].take();
|
||||
}
|
||||
|
||||
(shuffled_disks, shuffled_parts_metadata)
|
||||
}
|
||||
|
||||
pub(super) fn shuffle_disks_and_parts_metadata_by_index(
|
||||
disks: &[Option<DiskStore>],
|
||||
parts_metadata: &[FileInfo],
|
||||
@@ -949,3 +998,78 @@ impl SetDisks {
|
||||
shuffled_parts_errs
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
async fn shuffle_test_disks(tempdir: &tempfile::TempDir, count: usize) -> Vec<Option<DiskStore>> {
|
||||
let endpoint =
|
||||
Endpoint::try_from(tempdir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
|
||||
let disk = new_disk(
|
||||
&endpoint,
|
||||
&DiskOption {
|
||||
cleanup: false,
|
||||
health_check: false,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("disk should be created");
|
||||
// The shuffle only inspects Some/None and clones the Arc handle, so
|
||||
// one shared disk handle per slot is sufficient.
|
||||
(0..count).map(|_| Some(disk.clone())).collect()
|
||||
}
|
||||
|
||||
fn shuffle_fixture(consistent: bool) -> (FileInfo, Vec<FileInfo>) {
|
||||
let mut fi = FileInfo::new("bucket/object", 2, 1);
|
||||
fi.mod_time = Some(time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"));
|
||||
fi.size = 1;
|
||||
fi.add_object_part(1, String::new(), 1, None, 1, None, None);
|
||||
|
||||
let slots = fi.erasure.distribution.len();
|
||||
let parts = (0..slots)
|
||||
.map(|k| {
|
||||
let mut part_fi = fi.clone();
|
||||
part_fi.erasure.index = if consistent {
|
||||
fi.erasure.distribution[k]
|
||||
} else {
|
||||
// Misplace every source so the by-index pass is rejected
|
||||
// and the mod-time fallback placement runs instead.
|
||||
fi.erasure.distribution[(k + 1) % slots]
|
||||
};
|
||||
part_fi
|
||||
})
|
||||
.collect();
|
||||
(fi, parts)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn owned_shuffle_matches_borrowing_variant_when_consistent() {
|
||||
let tempdir = tempfile::tempdir().expect("tempdir should be created");
|
||||
let (fi, parts) = shuffle_fixture(true);
|
||||
let disks = shuffle_test_disks(&tempdir, parts.len()).await;
|
||||
|
||||
let (expected_disks, expected_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index(&disks, &parts, &fi);
|
||||
let (owned_disks, owned_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index_owned(disks, parts, &fi);
|
||||
|
||||
assert_eq!(owned_parts, expected_parts, "owned shuffle must place identical metadata");
|
||||
let expected_slots: Vec<bool> = expected_disks.iter().map(Option::is_some).collect();
|
||||
let owned_slots: Vec<bool> = owned_disks.iter().map(Option::is_some).collect();
|
||||
assert_eq!(owned_slots, expected_slots, "owned shuffle must fill identical disk slots");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn owned_shuffle_matches_borrowing_variant_on_fallback() {
|
||||
let tempdir = tempfile::tempdir().expect("tempdir should be created");
|
||||
let (fi, parts) = shuffle_fixture(false);
|
||||
let disks = shuffle_test_disks(&tempdir, parts.len()).await;
|
||||
|
||||
let (expected_disks, expected_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index(&disks, &parts, &fi);
|
||||
let (owned_disks, owned_parts) = SetDisks::shuffle_disks_and_parts_metadata_by_index_owned(disks, parts, &fi);
|
||||
|
||||
assert_eq!(owned_parts, expected_parts, "fallback placement must match the borrowing variant");
|
||||
let expected_slots: Vec<bool> = expected_disks.iter().map(Option::is_some).collect();
|
||||
let owned_slots: Vec<bool> = owned_disks.iter().map(Option::is_some).collect();
|
||||
assert_eq!(owned_slots, expected_slots, "fallback disk slots must match the borrowing variant");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -459,7 +459,12 @@ const DEFAULT_RUSTFS_GET_SMALL_OBJECT_DIRECT_MEMORY_THRESHOLD: usize = 128 * 102
|
||||
// --- Metadata Early-Stop Configuration ---
|
||||
|
||||
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ENABLE";
|
||||
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = false;
|
||||
// Enabled by default (backlog#872): the early-stop path only engages for
|
||||
// requests `should_allow_metadata_early_stop` classifies as safe (metadata-only
|
||||
// reads without version_id / healing / free-version needs) and still requires
|
||||
// a full read-quorum agreement before stopping. Set the env var to `false` to
|
||||
// fall back to full-wait metadata fanout.
|
||||
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true;
|
||||
|
||||
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT";
|
||||
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100;
|
||||
@@ -467,6 +472,11 @@ const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100;
|
||||
const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE";
|
||||
const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false;
|
||||
|
||||
// --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
|
||||
|
||||
const ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH: &str = "RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH";
|
||||
const DEFAULT_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH: bool = true;
|
||||
|
||||
static OBJECT_LOCK_DIAG_ENABLED: OnceLock<bool> = OnceLock::new();
|
||||
|
||||
mod core;
|
||||
@@ -682,6 +692,31 @@ fn is_version_early_stop_enabled() -> bool {
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if multipart reads prefetch the next part's bitrot reader setup
|
||||
/// while the current part decodes (backlog#870).
|
||||
///
|
||||
/// **Note**: Cached via `OnceLock` in production. In test builds the env var
|
||||
/// is read directly so that `temp_env` overrides take effect.
|
||||
fn is_multipart_reader_setup_prefetch_enabled() -> bool {
|
||||
#[cfg(test)]
|
||||
{
|
||||
rustfs_utils::get_env_bool(
|
||||
ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH,
|
||||
DEFAULT_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH,
|
||||
)
|
||||
}
|
||||
#[cfg(not(test))]
|
||||
{
|
||||
static CACHED: OnceLock<bool> = OnceLock::new();
|
||||
*CACHED.get_or_init(|| {
|
||||
rustfs_utils::get_env_bool(
|
||||
ENV_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH,
|
||||
DEFAULT_RUSTFS_GET_MULTIPART_READER_SETUP_PREFETCH,
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// --- Rollout Percentage Functions ---
|
||||
|
||||
fn get_codec_streaming_rollout_pct() -> u32 {
|
||||
@@ -6899,6 +6934,119 @@ mod tests {
|
||||
assert_eq!(out, payload[range_offset..range_offset + range_length]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn multipart_reads_stream_all_parts_with_setup_prefetch() {
|
||||
use tokio::io::AsyncReadExt;
|
||||
use uuid::Uuid;
|
||||
|
||||
let tempdir = tempfile::tempdir().expect("tempdir should be created");
|
||||
let endpoint =
|
||||
Endpoint::try_from(tempdir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
|
||||
let disk = new_disk(
|
||||
&endpoint,
|
||||
&DiskOption {
|
||||
cleanup: false,
|
||||
health_check: false,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("disk should be created");
|
||||
|
||||
let bucket = "bucket";
|
||||
let object = "object";
|
||||
// Three parts with distinct fill bytes so cross-part ordering bugs and
|
||||
// prefetch boundary mistakes surface as content mismatches
|
||||
// (backlog#870 exercises the prefetch hit path for parts 2 and 3).
|
||||
let parts: Vec<Vec<u8>> = vec![
|
||||
vec![b'a'; 2 * 1024 * 1024 + 111],
|
||||
vec![b'b'; 1024 * 1024 + 17],
|
||||
vec![b'c'; 3 * 1024 * 1024 + 923],
|
||||
];
|
||||
let total_size: usize = parts.iter().map(|part| part.len()).sum();
|
||||
|
||||
disk.make_volume(bucket).await.expect("bucket should be created");
|
||||
|
||||
let mut fi = FileInfo::new(&format!("{bucket}/{object}"), 1, 0);
|
||||
let data_dir = Uuid::new_v4();
|
||||
fi.data_dir = Some(data_dir);
|
||||
fi.size = total_size as i64;
|
||||
for (index, part) in parts.iter().enumerate() {
|
||||
fi.add_object_part(index + 1, String::new(), part.len(), None, part.len() as i64, None, None);
|
||||
}
|
||||
|
||||
let erasure = coding::Erasure::new_with_options(
|
||||
fi.erasure.data_blocks,
|
||||
fi.erasure.parity_blocks,
|
||||
fi.erasure.block_size,
|
||||
fi.uses_legacy_checksum,
|
||||
);
|
||||
|
||||
for (index, payload) in parts.iter().enumerate() {
|
||||
let part_number = index + 1;
|
||||
let shard_path = format!("{object}/{data_dir}/part.{part_number}");
|
||||
let checksum_info = fi.erasure.get_checksum_info(part_number);
|
||||
|
||||
let mut bitrot_writer = create_bitrot_writer(
|
||||
true,
|
||||
None,
|
||||
bucket,
|
||||
&shard_path,
|
||||
payload.len() as i64,
|
||||
erasure.shard_size(),
|
||||
checksum_info.algorithm.clone(),
|
||||
)
|
||||
.await
|
||||
.expect("bitrot writer should be created");
|
||||
|
||||
for chunk in payload.chunks(erasure.shard_size()) {
|
||||
bitrot_writer.write(chunk).await.expect("payload chunk should be written");
|
||||
}
|
||||
|
||||
let encoded = bitrot_writer.into_inline_data().expect("bitrot encoded data should exist");
|
||||
disk.write_all(bucket, &shard_path, Bytes::from(encoded))
|
||||
.await
|
||||
.expect("encoded shard should be stored");
|
||||
}
|
||||
|
||||
let files = vec![fi.clone()];
|
||||
let disks = vec![Some(disk.clone())];
|
||||
let (mut reader, mut writer) = tokio::io::duplex(64 * 1024);
|
||||
let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size);
|
||||
|
||||
let read_task = tokio::spawn(async move {
|
||||
SetDisks::get_object_with_fileinfo(
|
||||
bucket,
|
||||
object,
|
||||
0,
|
||||
total_size as i64,
|
||||
&mut writer,
|
||||
fi,
|
||||
files,
|
||||
&disks,
|
||||
0,
|
||||
0,
|
||||
true,
|
||||
false,
|
||||
GET_OBJECT_PATH_LEGACY_DUPLEX,
|
||||
GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
|
||||
metrics_size_bucket,
|
||||
)
|
||||
.await
|
||||
});
|
||||
|
||||
let mut out = Vec::new();
|
||||
reader.read_to_end(&mut out).await.expect("all part bytes should be readable");
|
||||
|
||||
read_task
|
||||
.await
|
||||
.expect("read task should complete")
|
||||
.expect("multipart read should succeed");
|
||||
|
||||
let expected: Vec<u8> = parts.concat();
|
||||
assert_eq!(out.len(), expected.len(), "all parts should be streamed");
|
||||
assert_eq!(out, expected, "part contents and ordering must survive setup prefetch");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parts_after_marker_uses_marker_position() {
|
||||
let part_numbers = (1..=1002).collect::<Vec<_>>();
|
||||
|
||||
@@ -2171,7 +2171,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
// }
|
||||
// _lock_guard = guard_opt;
|
||||
// }
|
||||
let (mut fi, _, disks) = self.get_object_fileinfo(bucket, object, opts, false).await?;
|
||||
// Force the full quorum fanout (allow_early_stop=false): `disks` is the
|
||||
// write target below, and an early-stop subset would only carry read
|
||||
// quorum, failing write quorum on update_object_meta (backlog#872).
|
||||
let (mut fi, _, disks) = self.get_object_fileinfo_gated(bucket, object, opts, false, false).await?;
|
||||
|
||||
fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned());
|
||||
|
||||
|
||||
@@ -167,6 +167,24 @@ impl SetDisks {
|
||||
object: &str,
|
||||
opts: &ObjectOptions,
|
||||
read_data: bool,
|
||||
) -> Result<(FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>)> {
|
||||
// Read-only callers (GET/HEAD/tag read) may use the metadata early-stop
|
||||
// fast path.
|
||||
self.get_object_fileinfo_gated(bucket, object, opts, read_data, true).await
|
||||
}
|
||||
|
||||
/// Like `get_object_fileinfo`, but `allow_early_stop=false` forces the full
|
||||
/// quorum fanout. Read-before-write callers (object tagging) must use this:
|
||||
/// the returned online-disk set is the write target, and the early-stop
|
||||
/// subset would fail write quorum (backlog#872 regression fix).
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
pub(in crate::set_disk) async fn get_object_fileinfo_gated(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
opts: &ObjectOptions,
|
||||
read_data: bool,
|
||||
allow_early_stop: bool,
|
||||
) -> Result<(FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>)> {
|
||||
let vid = opts.version_id.clone().unwrap_or_default();
|
||||
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
|
||||
@@ -221,7 +239,10 @@ impl SetDisks {
|
||||
|
||||
let disks = disks.clone();
|
||||
|
||||
// TODO: optimize concurrency and break once enough slots are available
|
||||
// Early-stop for safe metadata reads is handled inside
|
||||
// read_all_fileinfo_observed (see read_all_fileinfo_early_stop in
|
||||
// core/io_primitives.rs); unsafe requests and callers that opt out
|
||||
// (allow_early_stop=false) fall back to full-wait.
|
||||
let (parts_metadata, errs, metadata_fanout_diagnostics) = Self::read_all_fileinfo_observed(
|
||||
&disks,
|
||||
"",
|
||||
@@ -231,6 +252,7 @@ impl SetDisks {
|
||||
read_data,
|
||||
false,
|
||||
opts.incl_free_versions,
|
||||
allow_early_stop,
|
||||
self.default_parity_count,
|
||||
)
|
||||
.await?;
|
||||
@@ -516,7 +538,10 @@ impl SetDisks {
|
||||
{
|
||||
let pipeline_started = Instant::now();
|
||||
debug!(bucket, object, requested_length = length, offset, "get_object_with_fileinfo start");
|
||||
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi);
|
||||
// Owned shuffle (backlog#873): `files` is consumed and its FileInfo
|
||||
// entries move into their shuffled slots, avoiding one deep clone per
|
||||
// disk; the disk handles are Arc clones and stay cheap.
|
||||
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index_owned(disks.to_vec(), files, &fi);
|
||||
|
||||
let total_size = fi.size as usize;
|
||||
|
||||
@@ -607,6 +632,17 @@ impl SetDisks {
|
||||
let part_indices: Vec<usize> = (part_index..=last_part_index).collect();
|
||||
debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
|
||||
|
||||
// Pipeline prefetch (backlog#870): while the current part decodes and
|
||||
// streams out, the next part's bitrot reader setup (file opens +
|
||||
// read-quorum wait across all disks) runs concurrently, so multipart
|
||||
// reads no longer serialize setup latency between parts. Depth is one
|
||||
// part; shared inputs move behind Arc so the prefetch task is 'static.
|
||||
let use_mmap_read = object_mmap_read_enabled();
|
||||
let files = Arc::new(files);
|
||||
let disks = Arc::new(disks);
|
||||
let prefetch_enabled = is_multipart_reader_setup_prefetch_enabled();
|
||||
let mut prefetched: Option<(usize, PrefetchedReaderSetup)> = None;
|
||||
|
||||
let mut total_read = 0;
|
||||
for current_part in part_indices {
|
||||
if total_read == length {
|
||||
@@ -647,43 +683,98 @@ impl SetDisks {
|
||||
"Streaming multipart part"
|
||||
);
|
||||
|
||||
let checksum_info = fi.erasure.get_checksum_info(part_number);
|
||||
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
|
||||
HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
let checksum_algo = multipart_part_checksum_algo(&fi, part_number);
|
||||
let read_length = till_offset.saturating_sub(read_offset);
|
||||
|
||||
let use_mmap_read = object_mmap_read_enabled();
|
||||
|
||||
let reader_setup_stage_start = Instant::now();
|
||||
let read_costs = coding::decode::should_collect_shard_read_costs().then(|| shard_read_costs_for_disks(&disks));
|
||||
let reader_setup = create_bitrot_readers_until_quorum_with_preference(
|
||||
&files,
|
||||
&disks,
|
||||
bucket,
|
||||
object,
|
||||
let sync_spec = PartReaderSetupSpec {
|
||||
part_number,
|
||||
read_offset,
|
||||
read_length,
|
||||
erasure.shard_size(),
|
||||
checksum_algo,
|
||||
skip_verify_bitrot,
|
||||
use_mmap_read,
|
||||
erasure.data_shards,
|
||||
erasure.parity_shards,
|
||||
BitrotReaderSetupMode::ReadQuorum,
|
||||
prefer_data_blocks_first_reader_setup,
|
||||
None,
|
||||
Some(BitrotReaderSetupAttribution {
|
||||
path: metrics_path,
|
||||
object_class: metrics_object_class,
|
||||
size_bucket: metrics_size_bucket,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let reader_setup_elapsed = reader_setup_stage_start.elapsed();
|
||||
};
|
||||
let mut setup_result = None;
|
||||
// A stale prefetch (part mismatch) is dropped by the failed let
|
||||
// chain, which aborts its task via the guard.
|
||||
if let Some((prefetched_part, handle)) = prefetched.take()
|
||||
&& prefetched_part == current_part
|
||||
{
|
||||
setup_result = handle.join().await;
|
||||
if setup_result.is_none() {
|
||||
warn!(
|
||||
bucket,
|
||||
object,
|
||||
part_index = current_part,
|
||||
"Multipart reader-setup prefetch task did not complete; retrying synchronously"
|
||||
);
|
||||
}
|
||||
}
|
||||
let (reader_setup, reader_setup_elapsed) = match setup_result {
|
||||
Some(result) => result,
|
||||
None => {
|
||||
setup_multipart_part_readers(
|
||||
&files,
|
||||
&disks,
|
||||
bucket,
|
||||
object,
|
||||
sync_spec,
|
||||
erasure.shard_size(),
|
||||
erasure.data_shards,
|
||||
erasure.parity_shards,
|
||||
skip_verify_bitrot,
|
||||
use_mmap_read,
|
||||
prefer_data_blocks_first_reader_setup,
|
||||
metrics_path,
|
||||
metrics_object_class,
|
||||
metrics_size_bucket,
|
||||
)
|
||||
.await
|
||||
}
|
||||
};
|
||||
|
||||
// Kick off the next part's reader setup before decoding this one
|
||||
// so the disk opens overlap with decode + client writeback
|
||||
// (backlog#870).
|
||||
let remaining_after_current = length - total_read - part_length;
|
||||
if prefetch_enabled && remaining_after_current > 0 && current_part < last_part_index {
|
||||
let next_part = current_part + 1;
|
||||
let next_number = fi.parts[next_part].number;
|
||||
let next_size = fi.parts[next_part].size;
|
||||
let next_length = next_size.min(remaining_after_current);
|
||||
let spec = PartReaderSetupSpec {
|
||||
part_number: next_number,
|
||||
read_offset: 0,
|
||||
read_length: erasure.shard_file_offset(0, next_length, next_size),
|
||||
checksum_algo: multipart_part_checksum_algo(&fi, next_number),
|
||||
};
|
||||
let files = Arc::clone(&files);
|
||||
let disks = Arc::clone(&disks);
|
||||
let bucket = bucket.to_owned();
|
||||
let object = object.to_owned();
|
||||
let shard_size = erasure.shard_size();
|
||||
let data_shards = erasure.data_shards;
|
||||
let parity_shards = erasure.parity_shards;
|
||||
let handle = tokio::task::spawn(async move {
|
||||
setup_multipart_part_readers(
|
||||
&files,
|
||||
&disks,
|
||||
&bucket,
|
||||
&object,
|
||||
spec,
|
||||
shard_size,
|
||||
data_shards,
|
||||
parity_shards,
|
||||
skip_verify_bitrot,
|
||||
use_mmap_read,
|
||||
prefer_data_blocks_first_reader_setup,
|
||||
metrics_path,
|
||||
metrics_object_class,
|
||||
metrics_size_bucket,
|
||||
)
|
||||
.await
|
||||
});
|
||||
prefetched = Some((next_part, PrefetchedReaderSetup::new(handle)));
|
||||
}
|
||||
rustfs_io_metrics::record_get_object_shard_reader_setup_duration(reader_setup_elapsed.as_secs_f64());
|
||||
rustfs_io_metrics::record_get_object_stage_duration_by_size(
|
||||
metrics_path,
|
||||
@@ -1022,36 +1113,78 @@ impl SetDisks {
|
||||
return Err(Error::other("codec streaming multipart part sizes do not match object size"));
|
||||
}
|
||||
|
||||
let mut readers = Vec::with_capacity(fi.parts.len());
|
||||
for part in &fi.parts {
|
||||
match Self::build_codec_streaming_part_reader(
|
||||
bucket,
|
||||
object,
|
||||
fi,
|
||||
&files,
|
||||
&disks,
|
||||
&erasure,
|
||||
part.number,
|
||||
0,
|
||||
part.size,
|
||||
part.size,
|
||||
skip_verify_bitrot,
|
||||
metrics_object_class,
|
||||
metrics_size_bucket,
|
||||
false,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
GetCodecStreamingReaderBuildOutcome::Reader(reader) => readers.push(reader),
|
||||
GetCodecStreamingReaderBuildOutcome::Fallback(reason) => {
|
||||
return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason));
|
||||
}
|
||||
// Lazy multipart construction (backlog#871): only the first part's
|
||||
// shard readers are opened before streaming starts, so TTFB no longer
|
||||
// pays for `parts x disks` file opens and an early client disconnect
|
||||
// never touches the remaining parts. The first part stays eager so the
|
||||
// dominant fallback conditions (missing shards, read quorum) are still
|
||||
// detected before any byte is streamed and the whole request can fall
|
||||
// back to the legacy duplex path.
|
||||
let first_part = &fi.parts[0];
|
||||
let first_reader = match Self::build_codec_streaming_part_reader(
|
||||
bucket,
|
||||
object,
|
||||
fi,
|
||||
&files,
|
||||
&disks,
|
||||
&erasure,
|
||||
first_part.number,
|
||||
0,
|
||||
first_part.size,
|
||||
first_part.size,
|
||||
skip_verify_bitrot,
|
||||
metrics_object_class,
|
||||
metrics_size_bucket,
|
||||
false,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
GetCodecStreamingReaderBuildOutcome::Reader(reader) => reader,
|
||||
GetCodecStreamingReaderBuildOutcome::Fallback(reason) => {
|
||||
return Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason));
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(MultipartCodecStreamingReader::new(
|
||||
readers,
|
||||
))))
|
||||
let remaining_parts: Vec<(usize, usize)> = fi.parts[1..].iter().map(|part| (part.number, part.size)).collect();
|
||||
let total_parts = fi.parts.len();
|
||||
let ctx = Arc::new(LazyCodecPartContext {
|
||||
bucket: bucket.to_owned(),
|
||||
object: object.to_owned(),
|
||||
fi: fi.clone(),
|
||||
files,
|
||||
disks,
|
||||
erasure,
|
||||
skip_verify_bitrot,
|
||||
metrics_object_class,
|
||||
metrics_size_bucket,
|
||||
});
|
||||
let builder: LazyPartBuilder = Box::new(move |remaining_index| {
|
||||
let ctx = Arc::clone(&ctx);
|
||||
let (part_number, part_size) = remaining_parts[remaining_index];
|
||||
tokio::task::spawn(async move {
|
||||
SetDisks::build_codec_streaming_part_reader(
|
||||
&ctx.bucket,
|
||||
&ctx.object,
|
||||
&ctx.fi,
|
||||
&ctx.files,
|
||||
&ctx.disks,
|
||||
&ctx.erasure,
|
||||
part_number,
|
||||
0,
|
||||
part_size,
|
||||
part_size,
|
||||
ctx.skip_verify_bitrot,
|
||||
ctx.metrics_object_class,
|
||||
ctx.metrics_size_bucket,
|
||||
false,
|
||||
)
|
||||
.await
|
||||
})
|
||||
});
|
||||
|
||||
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(
|
||||
LazyMultipartCodecStreamingReader::new(first_reader, total_parts, builder, get_codec_streaming_metrics_path()),
|
||||
)))
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
@@ -1165,6 +1298,243 @@ impl SetDisks {
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-part parameters for a multipart bitrot reader setup.
|
||||
struct PartReaderSetupSpec {
|
||||
part_number: usize,
|
||||
read_offset: usize,
|
||||
read_length: usize,
|
||||
checksum_algo: HashAlgorithm,
|
||||
}
|
||||
|
||||
/// Resolve the bitrot checksum algorithm for one part, honoring the legacy
|
||||
/// HighwayHash flag.
|
||||
fn multipart_part_checksum_algo(fi: &FileInfo, part_number: usize) -> HashAlgorithm {
|
||||
let checksum_info = fi.erasure.get_checksum_info(part_number);
|
||||
if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
|
||||
HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
}
|
||||
}
|
||||
|
||||
/// Run one part's bitrot reader setup and measure its wall-clock duration.
|
||||
///
|
||||
/// Shared by the synchronous path and the prefetch task in
|
||||
/// `get_object_with_fileinfo` (backlog#870) so both report the same
|
||||
/// stage-duration semantics.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn setup_multipart_part_readers(
|
||||
files: &[FileInfo],
|
||||
disks: &[Option<DiskStore>],
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
spec: PartReaderSetupSpec,
|
||||
shard_size: usize,
|
||||
data_shards: usize,
|
||||
parity_shards: usize,
|
||||
skip_verify_bitrot: bool,
|
||||
use_mmap_read: bool,
|
||||
prefer_data_blocks_first: bool,
|
||||
metrics_path: &'static str,
|
||||
metrics_object_class: &'static str,
|
||||
metrics_size_bucket: &'static str,
|
||||
) -> (BitrotReaderSetup, Duration) {
|
||||
let started = Instant::now();
|
||||
let setup = create_bitrot_readers_until_quorum_with_preference(
|
||||
files,
|
||||
disks,
|
||||
bucket,
|
||||
object,
|
||||
spec.part_number,
|
||||
spec.read_offset,
|
||||
spec.read_length,
|
||||
shard_size,
|
||||
spec.checksum_algo,
|
||||
skip_verify_bitrot,
|
||||
use_mmap_read,
|
||||
data_shards,
|
||||
parity_shards,
|
||||
BitrotReaderSetupMode::ReadQuorum,
|
||||
prefer_data_blocks_first,
|
||||
None,
|
||||
Some(BitrotReaderSetupAttribution {
|
||||
path: metrics_path,
|
||||
object_class: metrics_object_class,
|
||||
size_bucket: metrics_size_bucket,
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
(setup, started.elapsed())
|
||||
}
|
||||
|
||||
/// Guard around an in-flight prefetch of the next part's reader setup
|
||||
/// (backlog#870). Dropping the guard without consuming it aborts the task so
|
||||
/// error returns and early breaks stop the background disk IO.
|
||||
struct PrefetchedReaderSetup(Option<tokio::task::JoinHandle<(BitrotReaderSetup, Duration)>>);
|
||||
|
||||
impl PrefetchedReaderSetup {
|
||||
fn new(handle: tokio::task::JoinHandle<(BitrotReaderSetup, Duration)>) -> Self {
|
||||
Self(Some(handle))
|
||||
}
|
||||
|
||||
/// Wait for the prefetch to finish; `None` means the task was cancelled
|
||||
/// or panicked and the caller must set up synchronously.
|
||||
async fn join(mut self) -> Option<(BitrotReaderSetup, Duration)> {
|
||||
let handle = self.0.take()?;
|
||||
handle.await.ok()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for PrefetchedReaderSetup {
|
||||
fn drop(&mut self) {
|
||||
if let Some(handle) = self.0.take() {
|
||||
handle.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Owned context for lazily constructing codec streaming part readers after
|
||||
/// the first part has started streaming (backlog#871).
|
||||
struct LazyCodecPartContext {
|
||||
bucket: String,
|
||||
object: String,
|
||||
fi: FileInfo,
|
||||
files: Vec<FileInfo>,
|
||||
disks: Vec<Option<DiskStore>>,
|
||||
erasure: coding::Erasure,
|
||||
skip_verify_bitrot: bool,
|
||||
metrics_object_class: &'static str,
|
||||
metrics_size_bucket: &'static str,
|
||||
}
|
||||
|
||||
type LazyPartBuildHandle = tokio::task::JoinHandle<Result<GetCodecStreamingReaderBuildOutcome>>;
|
||||
type LazyPartBuilder = Box<dyn FnMut(usize) -> LazyPartBuildHandle + Send + Sync>;
|
||||
|
||||
/// Multipart codec streaming reader that constructs part readers on demand.
|
||||
///
|
||||
/// The first part reader is built eagerly by the caller so the dominant
|
||||
/// fallback conditions are detected before any byte is streamed; every
|
||||
/// subsequent part is only built once the previous part reaches EOF. If a
|
||||
/// later part hits a fallback condition mid-stream, the reader surfaces a read
|
||||
/// error instead: data has already been streamed, so switching the whole
|
||||
/// request to the legacy path is no longer possible. The next request detects
|
||||
/// the condition on its eager first-part setup and falls back cleanly.
|
||||
struct LazyMultipartCodecStreamingReader {
|
||||
current: Option<Box<dyn AsyncRead + Unpin + Send + Sync>>,
|
||||
pending: Option<LazyPartBuildHandle>,
|
||||
dispatched_remaining: usize,
|
||||
total_parts: usize,
|
||||
builder: LazyPartBuilder,
|
||||
metrics_path: &'static str,
|
||||
}
|
||||
|
||||
impl LazyMultipartCodecStreamingReader {
|
||||
fn new(
|
||||
first_reader: Box<dyn AsyncRead + Unpin + Send + Sync>,
|
||||
total_parts: usize,
|
||||
builder: LazyPartBuilder,
|
||||
metrics_path: &'static str,
|
||||
) -> Self {
|
||||
Self {
|
||||
current: Some(first_reader),
|
||||
pending: None,
|
||||
dispatched_remaining: 0,
|
||||
total_parts,
|
||||
builder,
|
||||
metrics_path,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncRead for LazyMultipartCodecStreamingReader {
|
||||
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
|
||||
if buf.remaining() == 0 {
|
||||
return Poll::Ready(Ok(()));
|
||||
}
|
||||
|
||||
let this = self.get_mut();
|
||||
loop {
|
||||
if let Some(reader) = this.current.as_mut() {
|
||||
let filled_before = buf.filled().len();
|
||||
match Pin::new(reader).poll_read(cx, buf) {
|
||||
Poll::Ready(Ok(())) if buf.filled().len() == filled_before => {
|
||||
// Part EOF: drop its shard readers before building the
|
||||
// next part.
|
||||
this.current = None;
|
||||
}
|
||||
result => return result,
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if let Some(handle) = this.pending.as_mut() {
|
||||
match Pin::new(handle).poll(cx) {
|
||||
Poll::Pending => return Poll::Pending,
|
||||
Poll::Ready(join_result) => {
|
||||
this.pending = None;
|
||||
match join_result {
|
||||
Ok(Ok(GetCodecStreamingReaderBuildOutcome::Reader(reader))) => {
|
||||
this.current = Some(reader);
|
||||
}
|
||||
Ok(Ok(GetCodecStreamingReaderBuildOutcome::Fallback(reason))) => {
|
||||
// KNOWN LIMITATION (backlog#871 follow-up): once
|
||||
// earlier parts have streamed we can no longer
|
||||
// hand the whole request back to the legacy
|
||||
// duplex path, so a degraded later part surfaces
|
||||
// as a read error even though legacy per-part
|
||||
// decode could still reconstruct it. This only
|
||||
// affects the OPT-IN multipart codec streaming
|
||||
// path (RUSTFS_GET_CODEC_STREAMING_MULTIPART_ENABLE,
|
||||
// default off): the first part stays eager so the
|
||||
// common case where part 1 is already degraded
|
||||
// still falls back cleanly before any byte ships.
|
||||
// A proper fix (in-place per-part legacy
|
||||
// degradation) is tracked as a follow-up.
|
||||
record_get_object_pipeline_failure_for_path(
|
||||
this.metrics_path,
|
||||
GET_STAGE_READER_SETUP,
|
||||
GetObjectFailureReason::ReadQuorum,
|
||||
);
|
||||
warn!(
|
||||
metrics_path = this.metrics_path,
|
||||
fallback_reason = ?reason,
|
||||
state = "codec_streaming_mid_stream_fallback",
|
||||
"Lazy multipart part construction hit a fallback condition mid-stream; surfacing read error"
|
||||
);
|
||||
return Poll::Ready(Err(std::io::Error::other(format!(
|
||||
"codec streaming multipart reader cannot fall back mid-stream: {reason:?}"
|
||||
))));
|
||||
}
|
||||
Ok(Err(err)) => return Poll::Ready(Err(std::io::Error::other(err))),
|
||||
Err(join_err) => return Poll::Ready(Err(std::io::Error::other(join_err))),
|
||||
}
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if this.dispatched_remaining + 1 < this.total_parts {
|
||||
let handle = (this.builder)(this.dispatched_remaining);
|
||||
this.dispatched_remaining += 1;
|
||||
this.pending = Some(handle);
|
||||
continue;
|
||||
}
|
||||
|
||||
return Poll::Ready(Ok(()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for LazyMultipartCodecStreamingReader {
|
||||
fn drop(&mut self) {
|
||||
// Abort an in-flight part construction so an early client disconnect
|
||||
// does not keep opening shard readers in the background.
|
||||
if let Some(handle) = self.pending.take() {
|
||||
handle.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOptions, read_data: bool) -> Option<&'static str> {
|
||||
if !read_data {
|
||||
return Some(GET_METADATA_CACHE_REASON_NOT_READ_DATA);
|
||||
@@ -2145,12 +2515,47 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_early_stop_gate_defaults_to_disabled() {
|
||||
fn metadata_early_stop_gate_defaults_to_enabled() {
|
||||
temp_env::with_var(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, None::<&str>, || {
|
||||
assert!(!is_get_metadata_early_stop_enabled());
|
||||
assert!(is_get_metadata_early_stop_enabled());
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_early_stop_gate_honors_explicit_opt_out() {
|
||||
temp_env::with_var(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false"), || {
|
||||
assert!(!is_get_metadata_early_stop_enabled());
|
||||
// With the gate off, even safe metadata-only requests must fall
|
||||
// back to the full-wait fanout.
|
||||
assert!(!should_allow_metadata_early_stop(false, "", false, false));
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_early_stop_permitted_respects_caller_opt_out() {
|
||||
temp_env::with_vars(
|
||||
[
|
||||
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
|
||||
(ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE, Some("true")),
|
||||
],
|
||||
|| {
|
||||
// Read-before-write callers (object tagging) pass
|
||||
// caller_allows_early_stop=false and must never early-stop, even
|
||||
// for an otherwise-eligible safe metadata-only read. The
|
||||
// early-stop subset would fail write quorum (backlog#872).
|
||||
assert!(!metadata_early_stop_permitted(false, true, false, "", false, false));
|
||||
assert!(!metadata_early_stop_permitted(false, true, false, "version-id", false, false));
|
||||
// With the caller allowing it and every other condition safe,
|
||||
// the fast path is permitted.
|
||||
assert!(metadata_early_stop_permitted(true, true, false, "", false, false));
|
||||
// observe=false (non-observed fanout) also disables early-stop.
|
||||
assert!(!metadata_early_stop_permitted(true, false, false, "", false, false));
|
||||
// Data reads are never eligible regardless of caller opt-in.
|
||||
assert!(!metadata_early_stop_permitted(true, true, true, "", false, false));
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_early_stop_rejects_data_reads() {
|
||||
temp_env::with_vars(
|
||||
@@ -2458,6 +2863,86 @@ mod tests {
|
||||
assert_eq!(drops.load(Ordering::SeqCst), 2);
|
||||
}
|
||||
|
||||
fn lazy_test_builder(
|
||||
parts: Vec<&'static [u8]>,
|
||||
builds: Arc<AtomicUsize>,
|
||||
) -> Box<dyn FnMut(usize) -> LazyPartBuildHandle + Send + Sync> {
|
||||
Box::new(move |remaining_index| {
|
||||
builds.fetch_add(1, Ordering::SeqCst);
|
||||
let data = parts[remaining_index];
|
||||
tokio::task::spawn(
|
||||
async move { Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(Cursor::new(data.to_vec())))) },
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn lazy_multipart_codec_streaming_reader_reads_parts_in_order() {
|
||||
let builds = Arc::new(AtomicUsize::new(0));
|
||||
let builder = lazy_test_builder(vec![b"multi", b"part"], Arc::clone(&builds));
|
||||
let mut reader =
|
||||
LazyMultipartCodecStreamingReader::new(Box::new(Cursor::new(b"hello ".to_vec())), 3, builder, "codec_streaming");
|
||||
|
||||
let mut output = Vec::new();
|
||||
reader
|
||||
.read_to_end(&mut output)
|
||||
.await
|
||||
.expect("lazy multipart reader should stream all parts");
|
||||
|
||||
assert_eq!(output, b"hello multipart");
|
||||
assert_eq!(builds.load(Ordering::SeqCst), 2, "both remaining parts should be built exactly once");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn lazy_multipart_codec_streaming_reader_defers_construction_until_needed() {
|
||||
let builds = Arc::new(AtomicUsize::new(0));
|
||||
let builder = lazy_test_builder(vec![b"never"], Arc::clone(&builds));
|
||||
{
|
||||
let mut reader =
|
||||
LazyMultipartCodecStreamingReader::new(Box::new(Cursor::new(b"abcdef".to_vec())), 2, builder, "codec_streaming");
|
||||
|
||||
let mut first = [0u8; 6];
|
||||
reader
|
||||
.read_exact(&mut first)
|
||||
.await
|
||||
.expect("first part should satisfy the read");
|
||||
assert_eq!(&first, b"abcdef");
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
builds.load(Ordering::SeqCst),
|
||||
0,
|
||||
"dropping the reader before crossing the part boundary must not build later parts"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn lazy_multipart_codec_streaming_reader_surfaces_mid_stream_fallback_as_error() {
|
||||
let builds = Arc::new(AtomicUsize::new(0));
|
||||
let builds_clone = Arc::clone(&builds);
|
||||
let builder: Box<dyn FnMut(usize) -> LazyPartBuildHandle + Send + Sync> = Box::new(move |_| {
|
||||
builds_clone.fetch_add(1, Ordering::SeqCst);
|
||||
tokio::task::spawn(async move {
|
||||
Ok(GetCodecStreamingReaderBuildOutcome::Fallback(GetCodecStreamingFallbackReason::Multipart))
|
||||
})
|
||||
});
|
||||
let mut reader =
|
||||
LazyMultipartCodecStreamingReader::new(Box::new(Cursor::new(b"first".to_vec())), 2, builder, "codec_streaming");
|
||||
|
||||
let mut output = Vec::new();
|
||||
let err = reader
|
||||
.read_to_end(&mut output)
|
||||
.await
|
||||
.expect_err("mid-stream fallback must surface as a read error");
|
||||
|
||||
assert_eq!(output, b"first", "bytes streamed before the fallback stay intact");
|
||||
assert!(
|
||||
err.to_string().contains("cannot fall back mid-stream"),
|
||||
"error should explain the mid-stream fallback: {err}"
|
||||
);
|
||||
assert_eq!(builds.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
fn inline_reader_setup_fileinfo(data: Option<&'static [u8]>) -> FileInfo {
|
||||
let mut fi = FileInfo::new("object", 2, 2);
|
||||
fi.volume = "bucket".to_string();
|
||||
|
||||
@@ -39,14 +39,15 @@ use futures::future::join_all;
|
||||
use rand::seq::SliceRandom;
|
||||
use rustfs_filemeta::{
|
||||
FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntriesSortedResult, MetaCacheEntry,
|
||||
MetacacheReader, MetadataResolutionParams, is_io_eof, merge_file_meta_versions,
|
||||
MetacacheReader, MetadataResolutionParams, is_io_eof,
|
||||
};
|
||||
use rustfs_io_metrics::{
|
||||
LIST_OBJECTS_GATHER_OUTCOME_INPUT_CLOSED, LIST_OBJECTS_GATHER_OUTCOME_LIMIT_REACHED, LIST_OBJECTS_SOURCE_WALKER,
|
||||
ListObjectsGatherObservation, ListObjectsIndexPageObservation,
|
||||
};
|
||||
use rustfs_utils::path::{self, SLASH_SEPARATOR, base_dir_from_prefix};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::cmp::Reverse;
|
||||
use std::collections::{BinaryHeap, HashMap, HashSet};
|
||||
use std::future::Future;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{
|
||||
@@ -4376,7 +4377,7 @@ impl ECStore {
|
||||
continue;
|
||||
}
|
||||
|
||||
let fvs = match if opts.include_free_versions {
|
||||
let mut fvs = match if opts.include_free_versions {
|
||||
entry.file_info_versions_with_free_versions(&bucket_clone)
|
||||
} else {
|
||||
entry.file_info_versions(&bucket_clone)
|
||||
@@ -4395,8 +4396,13 @@ impl ECStore {
|
||||
}
|
||||
};
|
||||
|
||||
// `FileMeta` keeps versions newest-first (`sort_by_mod_time`
|
||||
// sorts descending), so ascending emission is the exact
|
||||
// reverse. Callers such as replication resync walk with the
|
||||
// default (ascending) order so that re-applied versions
|
||||
// preserve the original version-stack order.
|
||||
if opts.versions_sort == WalkVersionsSortOrder::Ascending {
|
||||
//TODO: SORT
|
||||
fvs.versions.reverse();
|
||||
}
|
||||
|
||||
for fi in fvs.versions.iter() {
|
||||
@@ -4639,28 +4645,64 @@ async fn gather_results(
|
||||
Ok(GatherResultsState::InputClosed)
|
||||
}
|
||||
|
||||
async fn select_from(
|
||||
rx: &CancellationToken,
|
||||
in_channels: &mut [Receiver<MetaCacheEntry>],
|
||||
/// Head entry of one input channel inside the k-way merge.
|
||||
///
|
||||
/// `cleaned` caches `path::clean(name)` only when it differs from the raw
|
||||
/// name, so heap comparisons never allocate (walker-produced names are
|
||||
/// already clean in the common case).
|
||||
struct MergeHead {
|
||||
entry: MetaCacheEntry,
|
||||
cleaned: Option<String>,
|
||||
idx: usize,
|
||||
top: &mut [Option<MetaCacheEntry>],
|
||||
n_done: &mut usize,
|
||||
) -> Result<bool> {
|
||||
let entry = tokio::select! {
|
||||
entry = in_channels[idx].recv() => entry,
|
||||
_ = rx.cancelled() => return Ok(false),
|
||||
};
|
||||
}
|
||||
|
||||
match entry {
|
||||
Some(entry) => {
|
||||
top[idx] = Some(entry);
|
||||
}
|
||||
None => {
|
||||
top[idx] = None;
|
||||
*n_done += 1;
|
||||
}
|
||||
impl MergeHead {
|
||||
fn new(entry: MetaCacheEntry, idx: usize) -> Self {
|
||||
let cleaned = path::clean(&entry.name);
|
||||
let cleaned = if cleaned == entry.name { None } else { Some(cleaned) };
|
||||
Self { entry, cleaned, idx }
|
||||
}
|
||||
|
||||
fn sort_key(&self) -> &str {
|
||||
self.cleaned.as_deref().unwrap_or(&self.entry.name)
|
||||
}
|
||||
}
|
||||
|
||||
impl PartialEq for MergeHead {
|
||||
fn eq(&self, other: &Self) -> bool {
|
||||
self.cmp(other) == std::cmp::Ordering::Equal
|
||||
}
|
||||
}
|
||||
|
||||
impl Eq for MergeHead {}
|
||||
|
||||
impl PartialOrd for MergeHead {
|
||||
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
|
||||
Some(self.cmp(other))
|
||||
}
|
||||
}
|
||||
|
||||
impl Ord for MergeHead {
|
||||
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
|
||||
// Ascending channel index preserves the legacy scan order for equal keys.
|
||||
self.sort_key().cmp(other.sort_key()).then_with(|| self.idx.cmp(&other.idx))
|
||||
}
|
||||
}
|
||||
|
||||
enum MergePoll {
|
||||
Entry(Box<MergeHead>),
|
||||
Closed,
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
async fn poll_merge_head(rx: &CancellationToken, in_channels: &mut [Receiver<MetaCacheEntry>], idx: usize) -> MergePoll {
|
||||
tokio::select! {
|
||||
entry = in_channels[idx].recv() => match entry {
|
||||
Some(entry) => MergePoll::Entry(Box::new(MergeHead::new(entry, idx))),
|
||||
None => MergePoll::Closed,
|
||||
},
|
||||
_ = rx.cancelled() => MergePoll::Cancelled,
|
||||
}
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
async fn send_or_cancel(rx: &CancellationToken, out_channel: &Sender<MetaCacheEntry>, entry: MetaCacheEntry) -> Result<bool> {
|
||||
@@ -4707,147 +4749,100 @@ async fn merge_entry_channels(
|
||||
}
|
||||
}
|
||||
|
||||
let mut top: Vec<Option<MetaCacheEntry>> = vec![None; in_channels.len()];
|
||||
let mut n_done = 0;
|
||||
|
||||
let in_channels_len = in_channels.len();
|
||||
|
||||
for idx in 0..in_channels_len {
|
||||
if !select_from(&rx, &mut in_channels, idx, &mut top, &mut n_done).await? {
|
||||
return Ok(());
|
||||
// K-way merge across per-channel sorted streams via a min-heap keyed by
|
||||
// the path-cleaned entry name: advancing the merge costs O(log channels)
|
||||
// per entry and comparisons are allocation-free (see `MergeHead`).
|
||||
//
|
||||
// Note on same-name resolution: prefix-dir groups collapse into a single
|
||||
// emission and objects shadow same-named prefix dirs. The legacy
|
||||
// `merge_file_meta_versions` block that used to live here was dead code:
|
||||
// it only ran for prefix-dir groups, whose entries have empty metadata,
|
||||
// so `xl_meta()` always failed. Cross-drive version merging happens in
|
||||
// the metadata resolve path (`MetaCacheEntries::resolve`), not here.
|
||||
let mut heap: BinaryHeap<Reverse<Box<MergeHead>>> = BinaryHeap::with_capacity(in_channels.len());
|
||||
for idx in 0..in_channels.len() {
|
||||
match poll_merge_head(&rx, &mut in_channels, idx).await {
|
||||
MergePoll::Entry(head) => heap.push(Reverse(head)),
|
||||
MergePoll::Closed => {}
|
||||
MergePoll::Cancelled => {
|
||||
info!("merge_entry_channels rx.recv() cancel");
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut last = String::new();
|
||||
let mut to_merge: Vec<usize> = Vec::new();
|
||||
loop {
|
||||
if n_done == in_channels.len() {
|
||||
let mut group: Vec<Box<MergeHead>> = Vec::new();
|
||||
let mut refill: Vec<usize> = Vec::with_capacity(in_channels.len());
|
||||
|
||||
while let Some(Reverse(first)) = heap.pop() {
|
||||
// Collect every channel head that resolves to the same cleaned key so
|
||||
// the duplicate rules below see the whole same-name group at once.
|
||||
group.clear();
|
||||
refill.clear();
|
||||
refill.push(first.idx);
|
||||
while heap.peek().is_some_and(|Reverse(next)| next.sort_key() == first.sort_key()) {
|
||||
let Some(Reverse(next)) = heap.pop() else { break };
|
||||
refill.push(next.idx);
|
||||
group.push(next);
|
||||
}
|
||||
|
||||
// Legacy same-name resolution rules (heads arrive in ascending
|
||||
// channel order):
|
||||
// - prefix dir vs prefix dir with the same suffix shape: the first
|
||||
// head wins and the rest collapse;
|
||||
// - prefix dir vs object: the object wins;
|
||||
// - object vs object: the later channel wins;
|
||||
// - both dirs with different suffix shape: the smaller raw name wins.
|
||||
let mut winner = first;
|
||||
for other in group.drain(..) {
|
||||
let dir_matches = winner.entry.is_dir() && other.entry.is_dir();
|
||||
let suffix_matches = winner.entry.name.ends_with(SLASH_SEPARATOR) == other.entry.name.ends_with(SLASH_SEPARATOR);
|
||||
|
||||
if dir_matches && suffix_matches {
|
||||
continue;
|
||||
}
|
||||
|
||||
if !dir_matches {
|
||||
if other.entry.is_dir() {
|
||||
continue;
|
||||
}
|
||||
winner = other;
|
||||
continue;
|
||||
}
|
||||
|
||||
if winner.entry.name > other.entry.name {
|
||||
winner = other;
|
||||
}
|
||||
}
|
||||
|
||||
// Gate emission on the same cleaned key the heap orders by, not the raw
|
||||
// name. The heap pops in non-decreasing cleaned order, so comparing the
|
||||
// raw name here could drop a legitimate entry whose cleaned order and
|
||||
// raw order disagree (e.g. redundant slashes or `./` segments).
|
||||
let emit = winner.sort_key() > last.as_str();
|
||||
if emit {
|
||||
last.clear();
|
||||
last.push_str(winner.sort_key());
|
||||
}
|
||||
let MergeHead { entry, .. } = *winner;
|
||||
if emit && !send_or_cancel(&rx, &out_channel, entry).await? {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let mut best = top[0].clone();
|
||||
let mut best_idx = 0;
|
||||
to_merge.clear();
|
||||
|
||||
// Note: `select_from` mutates `top[idx]` during the inner loop, but this is safe
|
||||
// because each borrow from `top[other_idx]` is only used before any later
|
||||
// `select_from` call that can mutate that slot.
|
||||
|
||||
for other_idx in 1..top.len() {
|
||||
if let Some(other_entry) = &top[other_idx] {
|
||||
if let Some(best_entry) = &best {
|
||||
if path::clean(&best_entry.name) == path::clean(&other_entry.name) {
|
||||
let dir_matches = best_entry.is_dir() && other_entry.is_dir();
|
||||
let suffix_matches =
|
||||
best_entry.name.ends_with(SLASH_SEPARATOR) == other_entry.name.ends_with(SLASH_SEPARATOR);
|
||||
|
||||
if dir_matches && suffix_matches {
|
||||
to_merge.push(other_idx);
|
||||
continue;
|
||||
}
|
||||
|
||||
if !dir_matches {
|
||||
// dir and object has the save name
|
||||
if other_entry.is_dir() {
|
||||
if !select_from(&rx, &mut in_channels, other_idx, &mut top, &mut n_done).await? {
|
||||
return Ok(());
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
to_merge.clear();
|
||||
|
||||
best = Some(other_entry.clone());
|
||||
best_idx = other_idx;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
if best_entry.name > other_entry.name {
|
||||
to_merge.clear();
|
||||
best = Some(other_entry.clone());
|
||||
best_idx = other_idx;
|
||||
}
|
||||
} else {
|
||||
best = Some(other_entry.clone());
|
||||
best_idx = other_idx;
|
||||
for &idx in &refill {
|
||||
match poll_merge_head(&rx, &mut in_channels, idx).await {
|
||||
MergePoll::Entry(head) => heap.push(Reverse(head)),
|
||||
MergePoll::Closed => {}
|
||||
MergePoll::Cancelled => {
|
||||
info!("merge_entry_channels rx.recv() cancel");
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !to_merge.is_empty() {
|
||||
if let Some(entry) = &best {
|
||||
let mut versions = Vec::with_capacity(to_merge.len() + 1);
|
||||
|
||||
let mut has_xl = { entry.clone().xl_meta().ok() };
|
||||
|
||||
if let Some(x) = &has_xl {
|
||||
versions.push(x.versions.clone());
|
||||
}
|
||||
|
||||
for &idx in to_merge.iter() {
|
||||
let has_entry = top[idx].clone();
|
||||
|
||||
if let Some(entry) = has_entry {
|
||||
let xl2 = match entry.clone().xl_meta() {
|
||||
Ok(res) => res,
|
||||
Err(_) => {
|
||||
if !select_from(&rx, &mut in_channels, idx, &mut top, &mut n_done).await? {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
versions.push(xl2.versions.clone());
|
||||
|
||||
if has_xl.is_none() {
|
||||
if !select_from(&rx, &mut in_channels, best_idx, &mut top, &mut n_done).await? {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
best_idx = idx;
|
||||
best = Some(entry.clone());
|
||||
has_xl = Some(xl2);
|
||||
} else {
|
||||
if !select_from(&rx, &mut in_channels, best_idx, &mut top, &mut n_done).await? {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(xl) = has_xl.as_mut()
|
||||
&& !versions.is_empty()
|
||||
{
|
||||
xl.versions = merge_file_meta_versions(read_quorum, true, 0, &versions);
|
||||
|
||||
if let Ok(meta) = xl.marshal_msg()
|
||||
&& let Some(b) = best.as_mut()
|
||||
{
|
||||
b.metadata = meta;
|
||||
b.cached = Some(xl.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
to_merge.clear();
|
||||
}
|
||||
|
||||
if let Some(best_entry) = &best
|
||||
&& best_entry.name > last
|
||||
{
|
||||
if !send_or_cancel(&rx, &out_channel, best_entry.clone()).await? {
|
||||
return Ok(());
|
||||
}
|
||||
last = best_entry.name.clone();
|
||||
}
|
||||
|
||||
if !select_from(&rx, &mut in_channels, best_idx, &mut top, &mut n_done).await? {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl Sets {
|
||||
@@ -9230,6 +9225,178 @@ mod test {
|
||||
assert_eq!(results, vec!["obj-a", "obj-b"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn walk_ascending_versions_contract_reverses_newest_first_metadata() {
|
||||
// Documents the invariant the walk `versions_sort` handling relies on:
|
||||
// `FileMeta` keeps versions newest-first, so ascending emission is the
|
||||
// exact reverse of `file_info_versions` output.
|
||||
let older = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let newer = time::OffsetDateTime::from_unix_timestamp(1_705_312_500).expect("valid timestamp");
|
||||
let entry =
|
||||
test_object_meta_entry_with_erasure_versions("obj-a", &[(older, "etag-old", 4, 2), (newer, "etag-new", 4, 2)]);
|
||||
|
||||
let fvs = entry.file_info_versions("bucket").expect("versions should parse");
|
||||
assert_eq!(fvs.versions.len(), 2, "both versions should be visible");
|
||||
assert!(
|
||||
fvs.versions[0].mod_time >= fvs.versions[1].mod_time,
|
||||
"file_info_versions must yield newest-first"
|
||||
);
|
||||
assert!(fvs.versions[0].is_latest, "the first (newest) version carries is_latest");
|
||||
|
||||
let mut ascending = fvs.versions;
|
||||
ascending.reverse();
|
||||
assert!(
|
||||
ascending[0].mod_time <= ascending[1].mod_time,
|
||||
"reversing newest-first output must produce ascending mod_time order"
|
||||
);
|
||||
assert!(
|
||||
ascending.last().expect("versions present").is_latest,
|
||||
"ascending emission ends with the latest version"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_merges_interleaved_channels_in_global_order() {
|
||||
let mut txs = Vec::new();
|
||||
let mut rxs = Vec::new();
|
||||
for _ in 0..4 {
|
||||
let (tx, rx) = mpsc::channel(16);
|
||||
txs.push(tx);
|
||||
rxs.push(rx);
|
||||
}
|
||||
|
||||
// Round-robin keys so every channel stays sorted while the global
|
||||
// stream interleaves across all four channels.
|
||||
let names: Vec<String> = (0..24).map(|i| format!("obj-{i:02}")).collect();
|
||||
for (i, name) in names.iter().enumerate() {
|
||||
txs[i % 4].send(test_meta_entry(name)).await.unwrap();
|
||||
}
|
||||
// The same trailing key on three channels must be emitted once.
|
||||
for tx in txs.iter().take(3) {
|
||||
tx.send(test_meta_entry("obj-99")).await.unwrap();
|
||||
}
|
||||
drop(txs);
|
||||
|
||||
let (out_tx, mut out_rx) = mpsc::channel(64);
|
||||
let cancel = CancellationToken::new();
|
||||
let handle = tokio::spawn(merge_entry_channels(cancel, rxs, out_tx, 1));
|
||||
|
||||
let mut results = Vec::new();
|
||||
while let Some(entry) = out_rx.recv().await {
|
||||
results.push(entry.name.clone());
|
||||
}
|
||||
handle.await.unwrap().unwrap();
|
||||
|
||||
let mut expected = names;
|
||||
expected.push("obj-99".to_owned());
|
||||
assert_eq!(results, expected);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_emits_when_cleaned_order_differs_from_raw_order() {
|
||||
// Regression: `a//c` cleans to `a/c` and `a/b` is already clean. In
|
||||
// cleaned order `a/b` < `a/c`, but raw byte order is the opposite
|
||||
// (`a//c` < `a/b` because '/' < 'b'). The heap pops in cleaned order,
|
||||
// so gating emission on the raw name would drop `a//c` after `a/b` is
|
||||
// emitted. Both distinct keys must survive.
|
||||
let (tx_a, rx_a) = mpsc::channel(4);
|
||||
let (tx_b, rx_b) = mpsc::channel(4);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(8);
|
||||
|
||||
tx_a.send(test_meta_entry("a/b")).await.unwrap();
|
||||
drop(tx_a);
|
||||
tx_b.send(test_meta_entry("a//c")).await.unwrap();
|
||||
drop(tx_b);
|
||||
|
||||
let cancel = CancellationToken::new();
|
||||
let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1));
|
||||
|
||||
let mut results = Vec::new();
|
||||
while let Some(entry) = out_rx.recv().await {
|
||||
results.push(entry.name.clone());
|
||||
}
|
||||
handle.await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(
|
||||
results,
|
||||
vec!["a/b", "a//c"],
|
||||
"distinct cleaned keys must both be emitted in cleaned order"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_object_wins_over_same_named_prefix_dir() {
|
||||
let (tx_a, rx_a) = mpsc::channel(4);
|
||||
let (tx_b, rx_b) = mpsc::channel(4);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(8);
|
||||
|
||||
// `path::clean("a/") == "a"`, so the prefix dir and the object group
|
||||
// under the same key and the object must win.
|
||||
tx_a.send(test_dir_meta_entry("a/")).await.unwrap();
|
||||
drop(tx_a);
|
||||
tx_b.send(test_object_meta_entry("a")).await.unwrap();
|
||||
drop(tx_b);
|
||||
|
||||
let cancel = CancellationToken::new();
|
||||
let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1));
|
||||
|
||||
let mut results = Vec::new();
|
||||
while let Some(entry) = out_rx.recv().await {
|
||||
assert!(entry.is_object(), "the object entry must shadow the same-named prefix dir");
|
||||
results.push(entry.name.clone());
|
||||
}
|
||||
handle.await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(results, vec!["a"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_collapses_equivalent_prefix_dirs() {
|
||||
let (tx_a, rx_a) = mpsc::channel(4);
|
||||
let (tx_b, rx_b) = mpsc::channel(4);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(8);
|
||||
|
||||
tx_a.send(test_dir_meta_entry("x/")).await.unwrap();
|
||||
drop(tx_a);
|
||||
tx_b.send(test_dir_meta_entry("x/")).await.unwrap();
|
||||
drop(tx_b);
|
||||
|
||||
let cancel = CancellationToken::new();
|
||||
let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1));
|
||||
|
||||
let mut results = Vec::new();
|
||||
while let Some(entry) = out_rx.recv().await {
|
||||
results.push(entry.name.clone());
|
||||
}
|
||||
handle.await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(results, vec!["x/"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_normalizes_uncleaned_names_before_grouping() {
|
||||
let (tx_a, rx_a) = mpsc::channel(4);
|
||||
let (tx_b, rx_b) = mpsc::channel(4);
|
||||
let (out_tx, mut out_rx) = mpsc::channel(8);
|
||||
|
||||
// Both names clean to "a/b"; they must merge into a single emission.
|
||||
tx_a.send(test_meta_entry("a//b")).await.unwrap();
|
||||
drop(tx_a);
|
||||
tx_b.send(test_meta_entry("a/b")).await.unwrap();
|
||||
drop(tx_b);
|
||||
|
||||
let cancel = CancellationToken::new();
|
||||
let handle = tokio::spawn(merge_entry_channels(cancel, vec![rx_a, rx_b], out_tx, 1));
|
||||
|
||||
let mut results = Vec::new();
|
||||
while let Some(entry) = out_rx.recv().await {
|
||||
results.push(entry.name.clone());
|
||||
}
|
||||
handle.await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(results, vec!["a/b"]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_respects_cancellation() {
|
||||
let (tx, rx) = mpsc::channel::<MetaCacheEntry>(4);
|
||||
|
||||
Reference in New Issue
Block a user