mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-05 11:37:44 +01:00
fix(volume): keep preallocated growth above -volume.minFreeSpace (#12067)
* fix(volume): keep preallocated growth above -volume.minFreeSpace When free space fell below -volume.minFreeSpace, every volume went read-only and the master kept growing new volumes. With preallocation on, each new volume reserved its full size, driving the disk further below the floor (15 Oct incident: 66 empty volumes reserved about 1980 GiB). Two causes, both in weed/storage/store.go: - MaybeAdjustVolumeMax sized a disk's volume slots from raw free space, so space inside the floor was counted as available. It now counts only free space above the floor. - addVolume trusted the once-a-minute low-disk flag. A preallocated volume is now refused when the disk is low now, or when its reservation would not fit above the floor. Volumes without preallocation, including merges that copy existing data, are not affected. addVolume runs under a store-level lock, so two concurrent creates cannot both pass the room check against the same free space. * volume: mirror headroom guard in the rust store find_free_location only trusted the once-a-probe is_disk_space_low flag, so a burst of preallocated creates could still run the disk under -volume.minFreeSpace between probes, and maybe_adjust_volume_max sized slots from raw free space. Adds MinFreeSpace::is_low/available_space to match Go and applies both checks the same way. store.write() already serializes add_volume, so no extra lock is needed. * volume: read rust headroom tests from a fake disk The floor arithmetic used a live free-space reading, so unrelated disk activity could flip a test between pass and fail. Store.disk_stats_of is now the same seam Go uses (diskStatusOf): the growth-loop and floor tests run against a controlled disk, mirroring the Go suite one for one. * volume: hold back create metadata from the headroom check An exact-fit preallocation left free space exactly at the floor, so the .vif and .idx writes that follow a create could dip under it and get the location flagged read-only at the next probe. Reserve room for .vif on the data filesystem, and for .idx growth only when dir_idx shares it. * volume: mirror the metadata allowance in the rust store Same rule as the Go side: charge .vif against the checked filesystem and .idx only when dir_idx shares it, and grow the exact-fit test's fake disk by the per-create cost. * volume: skip the reservation check where create cannot preallocate * volume: mirror the platform preallocation gate in the rust store * volume.copy: count the preallocation reservation in the destination space check The copy path picked a location with room for the copied .dat, then reserved the master's full volume size limit on it. A volume smaller than the limit could land on a disk whose free space above -volume.minFreeSpace fit the copy but not the reservation, driving the disk under the floor the same way the growth loop did. Fetch the preallocation setting before choosing the location and require room for the reservation itself, only where CreateVolumeFile actually reserves blocks. * seaweed-volume: store min-free-space-adjusted available space like Go Go's DiskLocation.AvailableSpace holds free space above the -volume.minFreeSpace floor; the Rust probe cached raw free instead, so the VolumeCopy location check could accept a disk already inside the floor. Store the same adjusted figure and cover it with a test. * seaweed-volume: reserve the copied .dat like Go's CreateVolumeFile VolumeCopy called set_len for the reservation, which only grows a sparse file — no blocks are held, so a disk with room for the copied bytes but not the full reservation only fails mid-copy. preallocate_file fallocates KEEP_SIZE on Linux like Go, reserving the blocks up front; off Linux it is a no-op, matching PREALLOCATE_SUPPORTED. * seaweed-volume: charge a copy's destination pick for the .dat reservation Go's VolumeCopy now sizes its free-location check against the master's volume size limit when the platform preallocates, not just the copied bytes. The Rust pick only measured the copied bytes, so a destination could accept a copy whose reservation then dips under the floor. The config fetch moves ahead of the pick (same order as Go) and feeds both the space check and the reservation itself. * volume.copy: charge the picked location's free space under volumeCreateLock The copy path picked a destination off the once-a-minute AvailableSpace probe and released the decision long before the .dat reservation landed. Two copies could pass the same cached headroom and together drive the disk under -volume.minFreeSpace. ReserveFreeLocation serializes the pick under volumeCreateLock, verifies the disk still has live room for the reservation, and charges it against the location's cached free space until the next probe refreshes it. * seaweed-volume: mirror the copy pick's live-check and headroom charge Go's VolumeCopy now picks through Store.ReserveFreeLocation: live disk stats under the store write lock, then the reservation charged against the location's cached free space. The Rust plan_copy_destination gains the same store.reserve_free_location so two copies cannot spend the same headroom before the first .dat reservation lands. * volume.copy: saturate the free-space charge against a live-space drop The two statfs reads bracketing a pick can race a large external write; a plain subtraction would underflow and leave the location looking roomy until the next probe. * volume.copy: release the headroom reservation when a copy does not land ReserveFreeLocation charges the picked location's cached free space, but a failed copy never returned it, so the headroom stayed booked until the next disk-space probe and blocked retries. ReleaseFreeLocation refunds the charge; the copy's cleanup defer now registers with the reservation so every failure path (existing-replica delete, .note write, stream send) releases it, including the post-transfer mount/validate failures covered by the outer defer. * seaweed-volume: mirror the copy headroom release on failed copies release_free_location refunds reserve_free_location's charge. The copy returns it when the plan's existing-replica delete fails, when the .note write fails, and in cleanup_failed_copy, so a failed copy stops holding the headroom until the next disk-space probe. * volume.copy: keep pending copy headroom out of probe readings ReleaseFreeLocation refunded the reservation into AvailableSpace even when a CheckDiskSpace probe had already rebased the cache on live stats — the probe's reading still shows the in-flight copy's bytes as free, so the refund double-counted headroom and two copies could spend the same space, driving the disk under -volume.minFreeSpace. Track the outstanding charge in DiskLocation.ReservedSpace instead: probes subtract it so they cannot re-spend it, a failed copy removes only its own charge, and a landed copy commits it — the charge and the new usage cancel out. hasRoomForNewVolume respects pending headroom too, so a preallocating create cannot overrun an in-flight copy. * seaweed-volume: mirror pending copy headroom across disk probes reserved_space tracks the in-flight copy's charge separately from the cached free space: check_disk_space subtracts it, release removes only the failed copy's charge, a landed copy commits it, and has_room_for_new_volume respects it — matching the Go side. * volume.copy: settle a copy's headroom as its bytes land on disk The pending charge held the copy's full reservation until commit, but written and fallocated bytes already show up in statfs, so the live-space check counted them twice. A slow copy could then block other copies and volume growth despite free space above -volume.minFreeSpace. Settle the charge instead: commit the bytes CreateVolumeFile actually reserved (a failed fallocate leaves nothing committed), then settle each file's write progress as it lands. The failure paths now release only the unsettled remainder; allocatedSize is exported for the .dat's st_blocks. * seaweed-volume: mirror copy headroom settling as bytes land reserved_space keeps only the promised bytes statfs cannot see yet: preallocate_dat settles whatever blocks the filesystem actually reserved (0 when fallocate cannot), each copied file's progress settles the rest as it lands, and a failed copy releases only the unsettled remainder — matching the Go side. * volume.copy: release a failed copy's headroom charge only once Both defers fired on the same in-closure error: the inner released the unsettled charge and the outer released it again, driving ReservedSpace negative (an atomic.Uint64 underflow) and inflating AvailableSpace — the location would then fail every later headroom check, or worse, pass them with fabricated space. Zero the charge once it is released; the Rust mirror already releases via swap(0) for the same reason. * volume.copy: write into the preallocated .dat instead of staging over it The .dat stream staged into .copying and renamed over the file CreateVolumeFile had just fallocated, so the copy's transient footprint was preallocate+dat — more than the reservation the pick just checked — and the reservation itself was discarded at rename. Writing in place reuses the fallocated extent, keeps the peak at the reserved maximum, and leaves the copied volume preallocated like a created one. * seaweed-volume: mirror the in-place write into the preallocated .dat The .dat copy opened the just-fallocated file with truncate, freeing its blocks at open and spending the headroom again as the stream re-grew it. Append mode preserves them, matching the Go side. --------- Co-authored-by: chrislusf <chrislusf@gmail.com> Co-authored-by: user.email <chrislusf@users.noreply.github.com>
This commit is contained in:
16 files changed
+1332
-135
No files matched your search
@@ -20,7 +20,9 @@ use crate::storage::erasure_coding::ec_shard::{ShardId, shard_id_try_from};
|
||||
use crate::storage::erasure_coding::ecj_merge::EcjIdDecoder;
|
||||
use crate::storage::needle::needle::{self, Needle};
|
||||
use crate::storage::types::*;
|
||||
use crate::storage::volume::VolumeSpec;
|
||||
use crate::storage::volume::{
|
||||
PREALLOCATE_SUPPORTED, VolumeSpec, allocated_file_size, preallocate_file,
|
||||
};
|
||||
|
||||
use super::grpc_client::{
|
||||
GrpcDialOptions, VolumeServerGrpcClient, connect_channel, connect_channel_guarded,
|
||||
@@ -2067,8 +2069,13 @@ impl VolumeServer for VolumeGrpcService {
|
||||
let source_status_before =
|
||||
fetch_source_status_before(&mut client, req.volume_id, vid).await;
|
||||
let source = SourceVolumeStatus::fetch(&mut client, req.volume_id).await?;
|
||||
let dest = self.plan_copy_destination(&req, vid, &source, had_existing_volume)?;
|
||||
dest.write_note(vid, &req.source_data_node)?;
|
||||
let preallocate_size = self.copy_preallocate_size(&source).await;
|
||||
let dest =
|
||||
self.plan_copy_destination(&req, vid, &source, had_existing_volume, preallocate_size)?;
|
||||
if let Err(e) = dest.write_note(vid, &req.source_data_node) {
|
||||
dest.release_reservation(&self.state.store.read().unwrap());
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
let (tx, rx) =
|
||||
tokio::sync::mpsc::channel::<Result<volume_server_pb::VolumeCopyResponse, Status>>(16);
|
||||
@@ -2079,6 +2086,7 @@ impl VolumeServer for VolumeGrpcService {
|
||||
source,
|
||||
source_status_before,
|
||||
dest,
|
||||
preallocate_size,
|
||||
tx,
|
||||
};
|
||||
|
||||
@@ -2088,8 +2096,11 @@ impl VolumeServer for VolumeGrpcService {
|
||||
// races a departing caller is the one case where the existing
|
||||
// file-only cleanup leaves the volume loaded in memory.
|
||||
let mut mounted = false;
|
||||
if let Err(e) = job.run(&mut client, &mut mounted).await {
|
||||
job.cleanup_failed_copy(e, mounted).await;
|
||||
match job.run(&mut client, &mut mounted).await {
|
||||
Ok(()) => job
|
||||
.dest
|
||||
.commit_reservation(&job.state.store.read().unwrap()),
|
||||
Err(e) => job.cleanup_failed_copy(e, mounted).await,
|
||||
}
|
||||
});
|
||||
|
||||
@@ -5625,6 +5636,11 @@ struct CopyDestination {
|
||||
disk_type: DiskType,
|
||||
/// A remote-backed volume only lands its .idx/.vif locally; the .dat stays in the tier.
|
||||
has_remote_dat: bool,
|
||||
/// The disk and headroom charge held by reserve_free_location; the copy
|
||||
/// settles it as its bytes land on disk, and a failed copy releases only
|
||||
/// what is still promised.
|
||||
loc_idx: usize,
|
||||
reserved_space: std::sync::atomic::AtomicU64,
|
||||
}
|
||||
|
||||
impl CopyDestination {
|
||||
@@ -5640,6 +5656,27 @@ impl CopyDestination {
|
||||
.map_err(|e| Status::internal(format!("write .note for volume {}: {}", vid, e)))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Bytes the copy put on disk stop needing the headroom charge, so a
|
||||
/// slow copy does not count its own .dat against the free space a
|
||||
/// second reservation needs.
|
||||
fn settle_reservation(&self, store: &crate::storage::store::Store, n: u64) {
|
||||
let n = n.min(self.reserved_space.load(Ordering::Relaxed));
|
||||
if n > 0 {
|
||||
store.commit_free_location(self.loc_idx, n);
|
||||
self.reserved_space.fetch_sub(n, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
fn release_reservation(&self, store: &crate::storage::store::Store) {
|
||||
store.release_free_location(self.loc_idx, self.reserved_space.swap(0, Ordering::Relaxed));
|
||||
}
|
||||
|
||||
/// The copied bytes are on disk now, so the pending charge ends without
|
||||
/// refunding the cache — the charge and the landed usage cancel out.
|
||||
fn commit_reservation(&self, store: &crate::storage::store::Store) {
|
||||
store.commit_free_location(self.loc_idx, self.reserved_space.swap(0, Ordering::Relaxed));
|
||||
}
|
||||
}
|
||||
|
||||
impl VolumeGrpcService {
|
||||
@@ -5674,6 +5711,36 @@ impl VolumeGrpcService {
|
||||
Ok(volume_server_client(channel))
|
||||
}
|
||||
|
||||
/// Query master for the .dat reservation a copy will make, as Go's
|
||||
/// VolumeCopy does before picking a destination. A remote-backed volume
|
||||
/// lands no .dat, so only a local copy needs the reservation.
|
||||
async fn copy_preallocate_size(&self, source: &SourceVolumeStatus) -> i64 {
|
||||
let has_remote_dat = source
|
||||
.volume_info
|
||||
.as_ref()
|
||||
.map(|vi| !vi.files.is_empty())
|
||||
.unwrap_or(false);
|
||||
if !has_remote_dat {
|
||||
let master = self.state.current_master().await;
|
||||
let grpc_addr = super::heartbeat::to_grpc_address(&master);
|
||||
match super::heartbeat::try_get_master_configuration(
|
||||
&grpc_addr,
|
||||
self.state.outgoing_grpc_tls.as_ref(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(resp) if resp.volume_preallocate => {
|
||||
return resp.volume_size_limit_m_b as i64 * 1024 * 1024;
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
tracing::warn!("get master {} configuration: {}", master, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
0
|
||||
}
|
||||
|
||||
/// Picks the destination disk and, under the same write lock, drops a
|
||||
/// pre-existing local replica so its slot counts as free for the pick.
|
||||
fn plan_copy_destination(
|
||||
@@ -5682,6 +5749,7 @@ impl VolumeGrpcService {
|
||||
vid: VolumeId,
|
||||
vol_info: &SourceVolumeStatus,
|
||||
had_existing_volume: bool,
|
||||
preallocate_size: i64,
|
||||
) -> Result<CopyDestination, Status> {
|
||||
let requested_disk_type = if !req.disk_type.is_empty() {
|
||||
DiskType::from_string(&req.disk_type)
|
||||
@@ -5694,17 +5762,24 @@ impl VolumeGrpcService {
|
||||
.as_ref()
|
||||
.map(|vi| !vi.files.is_empty())
|
||||
.unwrap_or(false);
|
||||
let needed_space = if has_remote_dat {
|
||||
let mut needed_space = if has_remote_dat {
|
||||
vol_info.idx_file_size
|
||||
} else {
|
||||
vol_info.dat_file_size
|
||||
};
|
||||
if !has_remote_dat && PREALLOCATE_SUPPORTED && preallocate_size as u64 > needed_space {
|
||||
// the .dat reserves the master's volume size limit up front, so
|
||||
// the location must fit the reservation, not just the copied
|
||||
// bytes (Go's neededSpace)
|
||||
needed_space = preallocate_size as u64;
|
||||
}
|
||||
|
||||
// Find a free disk location using Go's Store.FindFreeLocation
|
||||
// Find a free disk location using Go's Store.ReserveFreeLocation
|
||||
// semantics; the slot of the replica being replaced counts as free.
|
||||
let (data_base, idx_base, selected_disk_type) = {
|
||||
let ((data_base, idx_base, selected_disk_type), loc_idx) = {
|
||||
let mut store = self.state.store.write().unwrap();
|
||||
let Some(loc_idx) = store.find_free_location_predicate(
|
||||
let Some(loc_idx) = store.reserve_free_location(
|
||||
needed_space,
|
||||
|loc| {
|
||||
loc.disk_type == requested_disk_type
|
||||
&& loc.available_space.load(Ordering::Relaxed) > needed_space
|
||||
@@ -5723,9 +5798,12 @@ impl VolumeGrpcService {
|
||||
loc.disk_type.clone(),
|
||||
);
|
||||
if had_existing_volume {
|
||||
delete_existing_replica(&mut store, vid, vol_info)?;
|
||||
if let Err(e) = delete_existing_replica(&mut store, vid, vol_info) {
|
||||
store.release_free_location(loc_idx, needed_space);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
selected
|
||||
(selected, loc_idx)
|
||||
};
|
||||
if had_existing_volume {
|
||||
self.state.volume_state_notify.notify_one();
|
||||
@@ -5741,6 +5819,8 @@ impl VolumeGrpcService {
|
||||
idx_base_name,
|
||||
disk_type: selected_disk_type,
|
||||
has_remote_dat,
|
||||
loc_idx,
|
||||
reserved_space: std::sync::atomic::AtomicU64::new(needed_space),
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -5753,6 +5833,7 @@ struct VolumeCopyJob {
|
||||
source: SourceVolumeStatus,
|
||||
source_status_before: Option<volume_server_pb::VolumeStatusResponse>,
|
||||
dest: CopyDestination,
|
||||
preallocate_size: i64,
|
||||
tx: tokio::sync::mpsc::Sender<Result<volume_server_pb::VolumeCopyResponse, Status>>,
|
||||
}
|
||||
|
||||
@@ -5782,15 +5863,24 @@ impl VolumeCopyJob {
|
||||
};
|
||||
let mut throttler = WriteThrottler::new(io_byte_per_second);
|
||||
|
||||
self.preallocate_dat().await?;
|
||||
let dat_on_disk = self.preallocate_dat().await?;
|
||||
|
||||
let dest = &self.dest;
|
||||
let store = &self.state.store;
|
||||
let mut settle = |landed: i64| {
|
||||
if landed > 0 {
|
||||
dest.settle_reservation(&store.read().unwrap(), landed as u64);
|
||||
}
|
||||
};
|
||||
let mut progress = CopyProgress {
|
||||
tx,
|
||||
next_report_target: &mut next_report_target,
|
||||
report_interval,
|
||||
throttler: &mut throttler,
|
||||
settle: &mut settle,
|
||||
};
|
||||
self.transfer_files(client, &mut progress).await?;
|
||||
self.transfer_files(client, &mut progress, dat_on_disk)
|
||||
.await?;
|
||||
|
||||
let source_status_after =
|
||||
fetch_source_status_after(client, self.req.volume_id, vid, tx).await?;
|
||||
@@ -5799,64 +5889,39 @@ impl VolumeCopyJob {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Query master for preallocation settings (matching Go VolumeCopy behavior).
|
||||
async fn preallocate_dat(&self) -> Result<(), Status> {
|
||||
let (state, vid, tx) = (&self.state, self.vid, &self.tx);
|
||||
/// Reserve the .dat's full size up front, matching Go's VolumeCopy
|
||||
/// CreateVolumeFile. Returns the bytes the filesystem actually holds —
|
||||
/// settled at once, since statfs already counts them — which a failed
|
||||
/// fallocate reports as 0 so the write progress settles them instead.
|
||||
async fn preallocate_dat(&self) -> Result<u64, Status> {
|
||||
let (data_base_name, has_remote_dat) =
|
||||
(&self.dest.data_base_name, self.dest.has_remote_dat);
|
||||
let mut preallocate_size: i64 = 0;
|
||||
if !has_remote_dat {
|
||||
let master = state.current_master().await;
|
||||
let grpc_addr = super::heartbeat::to_grpc_address(&master);
|
||||
// Race the master-configuration RPC against the caller's
|
||||
// response channel: a stalled master (or a slow leader
|
||||
// election) would otherwise hold the task and its .note
|
||||
// past a departing caller, since the per-chunk checks in
|
||||
// copy_file_from_source are never reached.
|
||||
let config = tokio::select! {
|
||||
res = super::heartbeat::try_get_master_configuration(
|
||||
&grpc_addr,
|
||||
state.outgoing_grpc_tls.as_ref(),
|
||||
) => res,
|
||||
_ = tx.closed() => {
|
||||
return Err(Status::cancelled(format!(
|
||||
"volume {} copy cancelled by caller",
|
||||
vid
|
||||
)));
|
||||
}
|
||||
};
|
||||
match config {
|
||||
Ok(resp) => {
|
||||
if resp.volume_preallocate {
|
||||
preallocate_size = resp.volume_size_limit_m_b as i64 * 1024 * 1024;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("get master {} configuration: {}", master, e);
|
||||
}
|
||||
}
|
||||
|
||||
if preallocate_size > 0 {
|
||||
let dat_path = format!("{}.dat", data_base_name);
|
||||
let file = std::fs::File::create(&dat_path).map_err(|e| {
|
||||
Status::internal(format!(
|
||||
"create preallocated volume file {}: {}",
|
||||
dat_path, e
|
||||
))
|
||||
})?;
|
||||
file.set_len(preallocate_size as u64).map_err(|e| {
|
||||
Status::internal(format!("preallocate volume file {}: {}", dat_path, e))
|
||||
})?;
|
||||
}
|
||||
let preallocate_size = self.preallocate_size;
|
||||
if preallocate_size <= 0 || has_remote_dat {
|
||||
return Ok(0);
|
||||
}
|
||||
Ok(())
|
||||
let dat_path = format!("{}.dat", data_base_name);
|
||||
let file = std::fs::File::create(&dat_path).map_err(|e| {
|
||||
Status::internal(format!(
|
||||
"create preallocated volume file {}: {}",
|
||||
dat_path, e
|
||||
))
|
||||
})?;
|
||||
preallocate_file(&file, preallocate_size as u64);
|
||||
let allocated = allocated_file_size(&file);
|
||||
self.dest
|
||||
.settle_reservation(&self.state.store.read().unwrap(), allocated);
|
||||
Ok(allocated)
|
||||
}
|
||||
|
||||
/// Pull the .dat (unless it stays in the remote tier), .idx and .vif.
|
||||
/// dat_on_disk is how much of the .dat statfs already counts, so only
|
||||
/// writes past it settle the charge.
|
||||
async fn transfer_files(
|
||||
&self,
|
||||
client: &mut VolumeServerGrpcClient,
|
||||
progress: &mut CopyProgress<'_>,
|
||||
dat_on_disk: u64,
|
||||
) -> Result<(), Status> {
|
||||
let (req, vol_info) = (&self.req, &self.source);
|
||||
let (data_base_name, idx_base_name, has_remote_dat) = (
|
||||
@@ -5864,7 +5929,9 @@ impl VolumeCopyJob {
|
||||
&self.dest.idx_base_name,
|
||||
self.dest.has_remote_dat,
|
||||
);
|
||||
// Copy .dat file
|
||||
// Copy .dat file. Writing in place into the just-preallocated file
|
||||
// keeps its fallocated blocks — truncating would free them at open
|
||||
// and re-spend the headroom as the bytes stream back in.
|
||||
if !has_remote_dat {
|
||||
let dat_path = format!("{}.dat", data_base_name);
|
||||
let dat_modified_ts_ns = copy_file_from_source(
|
||||
@@ -5877,9 +5944,10 @@ impl VolumeCopyJob {
|
||||
stop_offset: vol_info.dat_file_size,
|
||||
dest_path: &dat_path,
|
||||
ext: ".dat",
|
||||
is_append: false,
|
||||
is_append: self.preallocate_size > 0,
|
||||
ignore_source_not_found: true,
|
||||
report_progress: true,
|
||||
settle_skip: dat_on_disk as i64,
|
||||
},
|
||||
progress,
|
||||
)
|
||||
@@ -5904,6 +5972,7 @@ impl VolumeCopyJob {
|
||||
is_append: false,
|
||||
ignore_source_not_found: false,
|
||||
report_progress: false,
|
||||
settle_skip: 0,
|
||||
},
|
||||
progress,
|
||||
)
|
||||
@@ -5927,6 +5996,7 @@ impl VolumeCopyJob {
|
||||
is_append: false,
|
||||
ignore_source_not_found: true,
|
||||
report_progress: false,
|
||||
settle_skip: 0,
|
||||
},
|
||||
progress,
|
||||
)
|
||||
@@ -6096,6 +6166,8 @@ impl VolumeCopyJob {
|
||||
let _ = std::fs::remove_file(format!("{}.idx", idx_base_name));
|
||||
let _ = std::fs::remove_file(format!("{}.vif", data_base_name));
|
||||
let _ = std::fs::remove_file(¬e_path);
|
||||
self.dest
|
||||
.release_reservation(&state.store.read().unwrap());
|
||||
let _ = tx.send(Err(e)).await;
|
||||
}
|
||||
}
|
||||
@@ -6294,6 +6366,9 @@ struct CopyFileSpec<'a> {
|
||||
ignore_source_not_found: bool,
|
||||
/// Whether this file's bytes are reported back to the caller as progress.
|
||||
report_progress: bool,
|
||||
/// Bytes of this file statfs already counts (a fallocated .dat); only
|
||||
/// writes past it settle the copy's headroom charge.
|
||||
settle_skip: i64,
|
||||
}
|
||||
|
||||
/// Progress reporting and throttling shared by every file of one volume copy.
|
||||
@@ -6302,6 +6377,8 @@ struct CopyProgress<'a> {
|
||||
next_report_target: &'a mut i64,
|
||||
report_interval: i64,
|
||||
throttler: &'a mut WriteThrottler,
|
||||
/// Settles the headroom charge for bytes newly visible to statfs.
|
||||
settle: &'a mut (dyn FnMut(i64) + Send),
|
||||
}
|
||||
|
||||
/// Mirrors Go's checkVolumeFileExtension: the client-supplied Ext that
|
||||
@@ -6353,6 +6430,7 @@ where
|
||||
is_append,
|
||||
ignore_source_not_found,
|
||||
report_progress,
|
||||
settle_skip,
|
||||
} = *spec;
|
||||
let progress_tx = progress.tx;
|
||||
let copy_req = volume_server_pb::CopyFileRequest {
|
||||
@@ -6411,6 +6489,7 @@ where
|
||||
|
||||
let mut progressed_bytes: i64 = 0;
|
||||
let mut modified_ts_ns: i64 = 0;
|
||||
let mut settled_bytes: i64 = 0;
|
||||
|
||||
// The source stream's message() future is the one await in this loop that
|
||||
// can hang indefinitely: a stalled source (slow disk, network partition,
|
||||
@@ -6445,6 +6524,11 @@ where
|
||||
file.write_all(&resp.file_content)
|
||||
.map_err(|e| Status::internal(format!("write file {}: {}", dest_path, e)))?;
|
||||
progressed_bytes += resp.file_content.len() as i64;
|
||||
// Writes past settle_skip are new usage statfs can see; the
|
||||
// copy's headroom charge settles as they land.
|
||||
let landed = (progressed_bytes - settle_skip).max(0);
|
||||
(progress.settle)(landed - settled_bytes);
|
||||
settled_bytes = landed;
|
||||
// A throttled copy sleeps seconds at a time; wake for a departing
|
||||
// caller instead of finishing the nap first.
|
||||
tokio::select! {
|
||||
@@ -8931,12 +9015,14 @@ mod tests {
|
||||
is_append: false,
|
||||
ignore_source_not_found: true,
|
||||
report_progress: true,
|
||||
settle_skip: 0,
|
||||
},
|
||||
&mut CopyProgress {
|
||||
tx: &tx,
|
||||
next_report_target: &mut next_report_target,
|
||||
report_interval: 128 * 1024 * 1024,
|
||||
throttler: &mut throttler,
|
||||
settle: &mut |_| {},
|
||||
},
|
||||
)
|
||||
.await
|
||||
|
||||
Reference in new issue
Block a user