mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-13 16:46:55 +00:00
Every GET fans out a `read_version` across all disks to resolve xl.meta. Each fanout allocated an `Arc<ReadOptions>` (3 bools) plus four `Arc<String>` (`Arc::new(x.to_string())` = two allocations each) and cloned them into every spawned task. This trims the per-fanout allocation footprint. - `ReadOptions` is three bools, so it is now `Copy`. The fanout drops the `Arc<ReadOptions>` and hands each spawned task a copy; the two pre-existing `ReadOptions::clone()` sites (set_disk/read.rs, set_disk/ops/heal.rs) stop cloning a `Copy` type. - The four request strings use `Arc::<str>::from(&str)` (one allocation each) instead of `Arc::new(..to_string())` (string buffer + Arc = two each) — four fewer allocations per fanout, transparent to the `read_version(&str)` call. Behavior is unchanged: the fanout still spawns one task per disk (the spawn is deliberate — `read_version_call_counter_observes_spawned_fanout` verifies the process-global counter observes every per-disk increment across workers), quorum / early-stop / full-wait semantics are untouched, and no result ordering or error handling changed. Two larger items from the audit are intentionally NOT in this PR: - `tokio::spawn` -> `FuturesUnordered`: the spawn is a tested, deliberate design (cross-worker counter observation for #1309/#1314), and converting would also change panic isolation. Left as-is. - `vec![FileInfo::default(); N]`: `FileInfo`'s empty containers (String / HashMap / Vec) do not allocate, so this is one `Vec` allocation, not the per-element allocation the audit implied — not a real hot spot. `cargo fmt`, `cargo clippy -p rustfs-ecstore --lib` (0 warnings), `cargo check --lib --tests`, and the 26 fanout / call-counter unit tests pass on macOS (the change is fully cross-platform). Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
@@ -1251,7 +1251,7 @@ pub struct VolumeInfo {
|
||||
pub created: Option<OffsetDateTime>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize, Debug, Default, Clone)]
|
||||
#[derive(Deserialize, Serialize, Debug, Default, Clone, Copy)]
|
||||
pub struct ReadOptions {
|
||||
pub incl_free_versions: bool,
|
||||
pub read_data: bool,
|
||||
|
||||
@@ -2222,18 +2222,18 @@ impl SetDisks {
|
||||
let mut ress = Vec::with_capacity(disks.len());
|
||||
let mut errors = Vec::with_capacity(disks.len());
|
||||
let mut observations = observe.then(|| Vec::with_capacity(disks.len()));
|
||||
let opts = Arc::new(ReadOptions {
|
||||
let opts = ReadOptions {
|
||||
incl_free_versions,
|
||||
read_data,
|
||||
healing,
|
||||
});
|
||||
let org_bucket = Arc::new(org_bucket.to_string());
|
||||
let bucket = Arc::new(bucket.to_string());
|
||||
let object = Arc::new(object.to_string());
|
||||
let version_id = Arc::new(version_id.to_string());
|
||||
};
|
||||
let org_bucket: Arc<str> = Arc::from(org_bucket);
|
||||
let bucket: Arc<str> = Arc::from(bucket);
|
||||
let object: Arc<str> = Arc::from(object);
|
||||
let version_id: Arc<str> = Arc::from(version_id);
|
||||
let futures = disks.iter().enumerate().map(|(disk_index, disk)| {
|
||||
let disk = disk.clone();
|
||||
let opts = opts.clone();
|
||||
let task_opts = opts;
|
||||
let org_bucket = org_bucket.clone();
|
||||
let bucket = bucket.clone();
|
||||
let object = object.clone();
|
||||
@@ -2242,7 +2242,8 @@ impl SetDisks {
|
||||
let response_start = observe.then(Instant::now);
|
||||
let result = if let Some(disk) = disk {
|
||||
Self::record_read_version_call(&object, disk_index);
|
||||
disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await
|
||||
disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
|
||||
.await
|
||||
} else {
|
||||
Err(DiskError::DiskNotFound)
|
||||
};
|
||||
@@ -2307,21 +2308,21 @@ impl SetDisks {
|
||||
let mut observations = Vec::with_capacity(disks.len());
|
||||
let mut accumulator =
|
||||
MetadataQuorumAccumulator::new(disks.len(), default_parity_count, true).with_requested_version_id(version_id);
|
||||
let opts = Arc::new(ReadOptions {
|
||||
let opts = ReadOptions {
|
||||
incl_free_versions,
|
||||
read_data,
|
||||
healing,
|
||||
});
|
||||
let org_bucket = Arc::new(org_bucket.to_string());
|
||||
let bucket = Arc::new(bucket.to_string());
|
||||
let object = Arc::new(object.to_string());
|
||||
let version_id = Arc::new(version_id.to_string());
|
||||
};
|
||||
let org_bucket: Arc<str> = Arc::from(org_bucket);
|
||||
let bucket: Arc<str> = Arc::from(bucket);
|
||||
let object: Arc<str> = Arc::from(object);
|
||||
let version_id: Arc<str> = Arc::from(version_id);
|
||||
let mut join_set = JoinSet::new();
|
||||
let bounded_fanout = is_get_metadata_early_stop_bounded_fanout_enabled();
|
||||
let mut next_disk_index = 0usize;
|
||||
let spawn_read_version =
|
||||
|join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
|
||||
let opts = opts.clone();
|
||||
let task_opts = opts;
|
||||
let org_bucket = org_bucket.clone();
|
||||
let bucket = bucket.clone();
|
||||
let object = object.clone();
|
||||
@@ -2332,7 +2333,8 @@ impl SetDisks {
|
||||
Self::record_read_version_call(&object, index);
|
||||
#[cfg(test)]
|
||||
Self::read_version_fanout_barrier(&object, index).await;
|
||||
disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await
|
||||
disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
|
||||
.await
|
||||
} else {
|
||||
Err(DiskError::DiskNotFound)
|
||||
};
|
||||
|
||||
@@ -362,9 +362,9 @@ impl SetDisks {
|
||||
healing: true,
|
||||
};
|
||||
let checks = target_disks.into_iter().map(|disk| {
|
||||
let read_options = read_options.clone();
|
||||
let task_read_options = read_options;
|
||||
async move {
|
||||
let file_info = match disk.read_version("", bucket, object, version_id, &read_options).await {
|
||||
let file_info = match disk.read_version("", bucket, object, version_id, &task_read_options).await {
|
||||
Ok(file_info) => file_info,
|
||||
Err(
|
||||
DiskError::DiskNotFound
|
||||
|
||||
@@ -224,7 +224,7 @@ impl SetDisks {
|
||||
let bucket = bucket.to_string();
|
||||
let object = object.to_string();
|
||||
let version_id = version_id.to_string();
|
||||
let opts = opts.clone();
|
||||
let opts = *opts;
|
||||
|
||||
let processor = runtime_sources::batch_processors().read_processor();
|
||||
let tasks: Vec<_> = disks
|
||||
@@ -235,9 +235,9 @@ impl SetDisks {
|
||||
let bucket = bucket.clone();
|
||||
let object = object.clone();
|
||||
let version_id = version_id.clone();
|
||||
let opts = opts.clone();
|
||||
let task_opts = opts;
|
||||
|
||||
async move { disk.read_version(&bucket, &bucket, &object, &version_id, &opts).await }
|
||||
async move { disk.read_version(&bucket, &bucket, &object, &version_id, &task_opts).await }
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
Reference in New Issue
Block a user