diff --git a/seaweed-volume/src/config.rs b/seaweed-volume/src/config.rs index cf73731cc1..0fcc08f73f 100644 --- a/seaweed-volume/src/config.rs +++ b/seaweed-volume/src/config.rs @@ -352,6 +352,25 @@ pub enum MinFreeSpace { Bytes(u64), } +impl MinFreeSpace { + /// Mirrors Go's MinFreeSpace.IsLow. + pub fn is_low(&self, free_bytes: u64, total_bytes: u64) -> bool { + match self { + MinFreeSpace::Percent(pct) => free_bytes as f64 / total_bytes as f64 * 100.0 < *pct, + MinFreeSpace::Bytes(min) => free_bytes < *min, + } + } + + /// Free bytes above the floor; mirrors Go's MinFreeSpace.AvailableSpace. + pub fn available_space(&self, free_bytes: u64, total_bytes: u64) -> u64 { + let floor = match self { + MinFreeSpace::Percent(pct) => (total_bytes as f64 * pct / 100.0) as u64, + MinFreeSpace::Bytes(min) => *min, + }; + free_bytes.saturating_sub(floor) + } +} + /// Convert single-dash long options to double-dash for clap compatibility. /// Go's `flag` package uses `-port`, clap expects `--port`. /// This allows both `-port 8080` and `--port 8080` to work. diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 92ae99182e..88dddcbbf6 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -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::>(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 { 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, dest: CopyDestination, + preallocate_size: i64, tx: tokio::sync::mpsc::Sender>, } @@ -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 { 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 diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 5483274f9c..9c548c6355 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -41,6 +41,10 @@ pub struct DiskLocation { ec_volumes: HashMap, pub is_disk_space_low: Arc, pub available_space: AtomicU64, + /// Headroom promised to in-flight copies that statfs cannot see yet. + /// Probes keep it out of available_space so they cannot re-spend it; a + /// copy that lands ends its charge, one that fails removes only its own. + pub reserved_space: AtomicU64, // Physical filesystem capacity from the latest check_disk_space probe, reported // to the master so balancing can see real disk fullness, not just slot counts. pub disk_total_bytes: AtomicU64, @@ -81,6 +85,7 @@ impl DiskLocation { ec_volumes: HashMap::new(), is_disk_space_low: Arc::new(AtomicBool::new(false)), available_space: AtomicU64::new(0), + reserved_space: AtomicU64::new(0), disk_total_bytes: AtomicU64::new(0), disk_free_bytes: AtomicU64::new(0), min_free_space, @@ -761,15 +766,16 @@ impl DiskLocation { return; } let used = total.saturating_sub(free); - let is_low = match &self.min_free_space { - MinFreeSpace::Percent(pct) => { - let free_pct = (free as f64 / total as f64) * 100.0; - free_pct < *pct - } - MinFreeSpace::Bytes(min_bytes) => free < *min_bytes, - }; + let is_low = self.min_free_space.is_low(free, total); self.is_disk_space_low.store(is_low, Ordering::Relaxed); - self.available_space.store(free, Ordering::Relaxed); + // Go's DiskLocation.AvailableSpace counts only free space above the + // -volume.minFreeSpace floor, so space checks stay above it too. + self.available_space.store( + self.min_free_space + .available_space(free, total) + .saturating_sub(self.reserved_space.load(Ordering::Relaxed)), + Ordering::Relaxed, + ); self.disk_total_bytes.store(total, Ordering::Relaxed); self.disk_free_bytes.store(free, Ordering::Relaxed); @@ -1445,6 +1451,73 @@ mod tests { assert!(free > 0, "expected free>0, got {free}"); } + // Mirrors Go's DiskLocation.AvailableSpace: the cached figure a copy is + // checked against counts only free space above the min-free-space floor. + #[test] + fn test_check_disk_space_counts_only_space_above_floor() { + const FLOOR: u64 = 1 << 20; + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let loc = DiskLocation::new( + dir, + dir, + 0, + DiskType::HardDrive, + MinFreeSpace::Bytes(FLOOR), + Vec::new(), + ) + .unwrap(); + loc.check_disk_space(); + assert_eq!( + loc.available_space.load(Ordering::Relaxed), + loc.disk_free_bytes + .load(Ordering::Relaxed) + .saturating_sub(FLOOR) + ); + + let loc = DiskLocation::new( + dir, + dir, + 0, + DiskType::HardDrive, + MinFreeSpace::Bytes(u64::MAX), + Vec::new(), + ) + .unwrap(); + loc.check_disk_space(); + assert_eq!(loc.available_space.load(Ordering::Relaxed), 0); + } + + // Mirrors Go's CheckDiskSpace pending subtraction: an in-flight copy's + // charge stays out of the cache across a probe, so another copy cannot + // re-spend the headroom the live reading still shows as free. + #[test] + fn test_check_disk_space_keeps_pending_reservation_out_of_cache() { + const FLOOR: u64 = 1 << 20; + const RESERVED: u64 = 512 * 1024; + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let loc = DiskLocation::new( + dir, + dir, + 0, + DiskType::HardDrive, + MinFreeSpace::Bytes(FLOOR), + Vec::new(), + ) + .unwrap(); + loc.reserved_space.store(RESERVED, Ordering::Relaxed); + loc.check_disk_space(); + let expected = loc + .min_free_space + .available_space( + loc.disk_free_bytes.load(Ordering::Relaxed), + loc.disk_total_bytes.load(Ordering::Relaxed), + ) + .saturating_sub(RESERVED); + assert_eq!(loc.available_space.load(Ordering::Relaxed), expected); + } + /// When `-dir.idx` is configured the EC `.vif` may live in the idx /// directory; the sweep must look there too, not only the data dir. #[test] diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 4cef145590..8fb92cd005 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -20,7 +20,9 @@ use crate::storage::needle::needle::{Needle, get_actual_size}; use crate::storage::needle_map::NeedleMapKind; use crate::storage::super_block::{ReplicaPlacement, SUPER_BLOCK_SIZE, SuperBlock}; use crate::storage::types::*; -use crate::storage::volume::{CompactionJob, VifVolumeInfo, Volume, VolumeError, VolumeSpec}; +use crate::storage::volume::{ + CompactionJob, PREALLOCATE_SUPPORTED, VifVolumeInfo, Volume, VolumeError, VolumeSpec, +}; /// Mirrors Go's ensureCompactVolumeSpace, per filesystem: the new .dat lands /// next to the old one and the new .idx next to the old index, so when the @@ -52,6 +54,40 @@ fn ensure_compact_volume_space(v: &Volume, preallocate: u64) -> Result<(), Volum } } +// Create-time metadata a location must still fit after a preallocation: .vif +// always lands beside the data file; the .idx a writable volume keeps +// appending counts only when dir_idx shares the filesystem. Holding these +// back keeps an exact-fit reservation from dipping under the floor once the +// metadata is written. +const VOLUME_CREATE_VIF_BYTES: u64 = 1 << 20; +const VOLUME_CREATE_IDX_BYTES: u64 = 4 << 20; + +/// Mirrors Go's hasRoomForNewVolume: a preallocated volume reserves its full +/// size up front, so it must fit in the free space above the min-free-space +/// floor. Reads the disk now rather than the once-a-probe is_disk_space_low +/// flag. A volume without preallocation reserves nothing, as does one on a +/// platform where preallocate_file is a no-op. +fn has_room_for_new_volume( + location: &DiskLocation, + preallocate: u64, + preallocate_supported: bool, + disk_stats_of: fn(&str) -> (u64, u64), +) -> bool { + if preallocate == 0 || !preallocate_supported { + return true; + } + let (all, free) = disk_stats_of(&location.directory); + if location.min_free_space.is_low(free, all) { + return false; + } + let mut need = preallocate + VOLUME_CREATE_VIF_BYTES; + if same_filesystem(&location.directory, &location.idx_directory) { + need += VOLUME_CREATE_IDX_BYTES; + } + location.min_free_space.available_space(free, all) + >= need + location.reserved_space.load(Ordering::Relaxed) +} + /// Top-level storage manager containing all disk locations and their volumes. pub struct Store { pub locations: Vec, @@ -76,6 +112,13 @@ pub struct Store { pub last_seen_topology_id: Mutex, pub vid_stamp_lock: Mutex<()>, pub vid_stamp_dirty: std::sync::atomic::AtomicBool, + /// Reads (total, free) for a directory; a field so a test can stand in + /// for a disk that is full or near its floor. Go: diskStatusOf. + disk_stats_of: fn(&str) -> (u64, u64), + /// Whether a create actually reserves `preallocate` on this platform; a + /// field so a test can exercise the guard off Linux. Go: + /// createVolumeFilePreallocates. + preallocate_supported: bool, } impl Store { @@ -99,6 +142,8 @@ impl Store { last_seen_topology_id: Mutex::new(String::new()), vid_stamp_lock: Mutex::new(()), vid_stamp_dirty: std::sync::atomic::AtomicBool::new(false), + disk_stats_of: super::disk_location::get_disk_stats, + preallocate_supported: PREALLOCATE_SUPPORTED, } } @@ -228,7 +273,7 @@ impl Store { /// Find the location with fewest volumes (load-balance) of the given disk type. /// Matches Go's FindFreeLocation: accounts for EC shards when computing free slots. - fn find_free_location(&self, disk_type: &DiskType) -> Option { + fn find_free_location(&self, disk_type: &DiskType, preallocate: u64) -> Option { use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT; let mut best: Option<(usize, i64)> = None; // (index, effective_free) @@ -253,6 +298,14 @@ impl Store { if loc.is_disk_space_low.load(Ordering::Relaxed) { continue; } + if !has_room_for_new_volume( + loc, + preallocate, + self.preallocate_supported, + self.disk_stats_of, + ) { + continue; + } if best.is_none() || effective_free > best.unwrap().1 { best = Some((i, effective_free)); } @@ -304,6 +357,64 @@ impl Store { best.map(|(i, _)| i) } + /// Mirrors Go's Store.ReserveFreeLocation: a copy's pick and its .dat + /// reservation are separate steps, so verify live space under the store + /// lock and charge need against the location's pending reservations, or + /// a second copy can spend the same headroom before the first lands. + pub fn reserve_free_location( + &self, + need: u64, + pred: F, + replace_vid: Option, + ) -> Option + where + F: Fn(&DiskLocation) -> bool, + { + let loc_idx = self.find_free_location_predicate( + |loc| { + if loc.available_space.load(Ordering::Relaxed) <= need || !pred(loc) { + return false; + } + let (all, free) = (self.disk_stats_of)(&loc.directory); + !loc.min_free_space.is_low(free, all) + && loc.min_free_space.available_space(free, all) + > loc.reserved_space.load(Ordering::Relaxed) + need + }, + replace_vid, + )?; + let loc = &self.locations[loc_idx]; + let (all, free) = (self.disk_stats_of)(&loc.directory); + loc.reserved_space.fetch_add(need, Ordering::Relaxed); + loc.available_space.store( + loc.min_free_space + .available_space(free, all) + .saturating_sub(loc.reserved_space.load(Ordering::Relaxed)), + Ordering::Relaxed, + ); + Some(loc_idx) + } + + /// Mirrors Go's Store.ReleaseFreeLocation: returns space + /// reserve_free_location charged when the copy it paid for does not + /// land. Removing the pending charge first keeps a probe that already + /// rebased the cache from seeing the refund twice. + pub fn release_free_location(&self, loc_idx: usize, need: u64) { + if let Some(loc) = self.locations.get(loc_idx) { + loc.reserved_space.fetch_sub(need, Ordering::Relaxed); + loc.available_space.fetch_add(need, Ordering::Relaxed); + } + } + + /// Mirrors Go's Store.CommitFreeLocation: settles a reservation whose + /// copy landed. The bytes are real on disk, so only the pending charge + /// ends — the cache keeps the charge, which now matches the live usage + /// the next probe will read. + pub fn commit_free_location(&self, loc_idx: usize, need: u64) { + if let Some(loc) = self.locations.get(loc_idx) { + loc.reserved_space.fetch_sub(need, Ordering::Relaxed); + } + } + /// Returns the index of the disk that should receive a new EC /// shard / index file for `(collection, vid)`. Selection order: /// @@ -460,12 +571,14 @@ impl Store { if self.find_volume(vid).is_some() { return Err(VolumeError::AlreadyExists); } - let loc_idx = self.find_free_location(&disk_type).ok_or_else(|| { - VolumeError::Io(io::Error::other(format!( - "no free location for disk type {:?}", - disk_type - ))) - })?; + let loc_idx = self + .find_free_location(&disk_type, spec.preallocate) + .ok_or_else(|| { + VolumeError::Io(io::Error::other(format!( + "no free location for disk type {:?}", + disk_type + ))) + })?; self.locations[loc_idx].create_volume(vid, self.needle_map_kind, spec) } @@ -875,11 +988,7 @@ impl Store { } /// Delete a needle from a volume. - pub fn delete_volume_needle( - &self, - vid: VolumeId, - n: &mut Needle, - ) -> Result { + pub fn delete_volume_needle(&self, vid: VolumeId, n: &mut Needle) -> Result { // Match Go's DeleteVolumeNeedle: check noWriteOrDelete before proceeding. let (_, vol) = self.find_volume(vid).ok_or(VolumeError::NotFound)?; if vol.is_no_write_or_delete() { @@ -941,10 +1050,12 @@ impl Store { for loc in &self.locations { if loc.original_max_volume_count == 0 { let current = loc.max_volume_count.load(Ordering::Relaxed); - let (_, free) = super::disk_location::get_disk_stats(&loc.directory); + let (all, free) = (self.disk_stats_of)(&loc.directory); let unused_space = loc.unused_space(volume_size_limit); - let unclaimed = (free as i64) - (unused_space as i64); + // Free space under -volume.minFreeSpace is not available for new volumes. + let unclaimed = + loc.min_free_space.available_space(free, all) as i64 - unused_space as i64; let vol_count = loc.volumes_len() as i32; let loc_ec_shards = loc.ec_shard_count(); @@ -1761,7 +1872,7 @@ fn update_dat_replica_placement( return Err(VolumeError::Io(io::Error::new( e.kind(), format!("volume {} failed to open {}: {}", vid, dat_path, e), - ))) + ))); } }; let mut header = [0u8; SUPER_BLOCK_SIZE]; @@ -1775,12 +1886,13 @@ fn update_dat_replica_placement( let mut block = header.to_vec(); block.resize(SUPER_BLOCK_SIZE + extra_size, 0); if extra_size > 0 { - file.read_exact(&mut block[SUPER_BLOCK_SIZE..]).map_err(|e| { - VolumeError::Io(io::Error::new( - e.kind(), - format!("volume {} failed to parse superblock: {}", vid, e), - )) - })?; + file.read_exact(&mut block[SUPER_BLOCK_SIZE..]) + .map_err(|e| { + VolumeError::Io(io::Error::new( + e.kind(), + format!("volume {} failed to parse superblock: {}", vid, e), + )) + })?; } let mut super_block = SuperBlock::from_bytes(&block).map_err(|e| { VolumeError::Io(io::Error::new( @@ -2032,6 +2144,351 @@ mod tests { assert_eq!(store.total_volume_count(), 1); } + // fakeDisk stands in for the disk under a location; a preallocating + // create takes its size out of FAKE_FREE, as fallocate does on Linux. + const MIB: u64 = 1 << 20; + static FAKE_ALL: AtomicU64 = AtomicU64::new(0); + static FAKE_FREE: AtomicU64 = AtomicU64::new(0); + static FAKE_DISK_LOCK: Mutex<()> = Mutex::new(()); + + fn fake_disk_stats(_: &str) -> (u64, u64) { + ( + FAKE_ALL.load(Ordering::Relaxed), + FAKE_FREE.load(Ordering::Relaxed), + ) + } + + // Store with one location whose fake disk has the given free space and a + // Bytes floor; the low-disk flag stays false, as between two probes. + fn headroom_store(dir: &str, free: u64, floor: u64) -> Store { + FAKE_ALL.store(10 * 1024 * MIB, Ordering::Relaxed); + FAKE_FREE.store(free, Ordering::Relaxed); + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Bytes(floor), + Vec::new(), + ) + .unwrap(); + store.disk_stats_of = fake_disk_stats; + // The fake disk behaves like fallocate on Linux: pretend creates reserve. + store.preallocate_supported = true; + store + } + + // Mirrors Go's TestAddVolumeRefusesPreallocationThatCrossesFloor: the + // reservation must fit above the floor even though the once-a-probe + // is_disk_space_low flag is still false. + #[test] + fn test_add_volume_refuses_preallocation_crossing_floor() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + // 4 MiB above the floor, so an 8 MiB reservation would cross it. + let mut store = headroom_store(dir, floor + 4 * MIB, floor); + + let spec = VolumeSpec { + preallocate: 8 * MIB, + ..VolumeSpec::default() + }; + assert!( + store + .add_volume(VolumeId(1), DiskType::HardDrive, &spec) + .is_err() + ); + assert!(!store.has_volume(VolumeId(1))); + } + + // Mirrors Go's TestAddVolumeAllowsPreallocationWhereUnsupported: where + // preallocate_file is a no-op the reservation never happens, so it must + // not hold the create back. + #[test] + fn test_add_volume_allows_preallocation_where_unsupported() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let mut store = headroom_store(dir, floor + MIB, floor); + store.preallocate_supported = false; + + let spec = VolumeSpec { + preallocate: 8 * MIB, + ..VolumeSpec::default() + }; + store + .add_volume(VolumeId(1), DiskType::HardDrive, &spec) + .unwrap(); + } + + // Mirrors Go's TestAddVolumeAllowsPlainVolumeAboveFloor and + // TestAddVolumeAllowsMergeVolumeBelowFloor: a volume reserving nothing is + // created either way, so growth is only held back when it would cross + // the floor. + #[test] + fn test_add_volume_allows_plain_volume_anywhere() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let mut store = headroom_store(dir, floor + MIB, floor); + store + .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default()) + .unwrap(); + + FAKE_FREE.store(floor - MIB, Ordering::Relaxed); + store + .add_volume(VolumeId(2), DiskType::HardDrive, &VolumeSpec::default()) + .unwrap(); + } + + // Mirrors Go's TestGrowthLoopStopsAtFloorWhenVolumesPreallocate: the + // master keeps asking for volumes while the disk sits just above its + // floor; each created volume reserves its full size, so the disk must + // not be driven below the floor. + #[test] + fn test_growth_loop_stops_at_floor_when_volumes_preallocate() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let preallocate = MIB; + // Room for 4 preallocated volumes above the floor once each create's + // metadata allowance is held back. + let create_cost = preallocate + VOLUME_CREATE_VIF_BYTES + VOLUME_CREATE_IDX_BYTES; + let mut store = headroom_store(dir, floor + 4 * create_cost, floor); + + let spec = VolumeSpec { + preallocate, + ..VolumeSpec::default() + }; + let mut created = 0; + for i in 1..=20u32 { + if store + .add_volume(VolumeId(i), DiskType::HardDrive, &spec) + .is_ok() + { + created += 1; + FAKE_FREE.fetch_sub(create_cost, Ordering::Relaxed); + } + } + + assert!( + FAKE_FREE.load(Ordering::Relaxed) >= floor, + "free space fell below the floor after {created} volumes" + ); + // Exactly the room above the floor must be used. Checking only an + // upper bound would also pass when every create fails. + assert_eq!(created, 4); + } + + // Mirrors Go's TestReserveFreeLocationChargesPickedReservation: a copy's + // pick and its .dat reservation are separate steps, so the picked + // location's cached free space must carry the charge at once or a second + // copy can spend the same headroom. + #[test] + fn test_reserve_free_location_charges_picked_reservation() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let need = 30 * MIB; + // 40 MiB above the floor: room for one 30 MiB reservation, not two. + let store = headroom_store(dir, floor + 40 * MIB, floor); + store.locations[0] + .available_space + .store(40 * MIB, Ordering::Relaxed); + + let pred = |loc: &DiskLocation| loc.disk_type == DiskType::HardDrive; + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + assert_eq!( + store.locations[0].available_space.load(Ordering::Relaxed), + 10 * MIB + ); + assert_eq!(store.reserve_free_location(need, pred, None), None); + } + + // Mirrors Go's TestReserveFreeLocationRechecksLiveDisk: the cached free + // space is refreshed only once a minute, so a stale-high value must not + // admit a reservation the live disk cannot fit. + #[test] + fn test_reserve_free_location_rechecks_live_disk() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let store = headroom_store(dir, floor + 5 * MIB, floor); + store.locations[0] + .available_space + .store(100 * MIB, Ordering::Relaxed); + + let pred = |loc: &DiskLocation| loc.disk_type == DiskType::HardDrive; + assert_eq!(store.reserve_free_location(30 * MIB, pred, None), None); + } + + // Mirrors Go's TestReleaseFreeLocationRestoresReservation: a failed copy + // must return its reservation, or the headroom stays booked until the + // next disk probe and blocks retries. + #[test] + fn test_release_free_location_restores_reservation() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let need = 30 * MIB; + // 40 MiB above the floor: room for one 30 MiB reservation, not two. + let store = headroom_store(dir, floor + 40 * MIB, floor); + store.locations[0] + .available_space + .store(40 * MIB, Ordering::Relaxed); + + let pred = |loc: &DiskLocation| loc.disk_type == DiskType::HardDrive; + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + store.release_free_location(0, need); + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + } + + // Mirrors Go's TestReleaseFreeLocationAfterProbeRefundsOnlyOwnCharge: a + // probe between reserve and release must not resurrect the charge — the + // live reading already forgot the failed copy's pending bytes, so + // refunding them would let two copies spend the same headroom. + #[test] + fn test_release_free_location_after_probe_refunds_only_own_charge() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let need = 30 * MIB; + let store = headroom_store(dir, floor + 40 * MIB, floor); + store.locations[0] + .available_space + .store(40 * MIB, Ordering::Relaxed); + + let pred = |loc: &DiskLocation| loc.disk_type == DiskType::HardDrive; + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + + // Mid-copy probe stores live-above-floor minus pending (as + // check_disk_space does): the charge survives it. + let loc = &store.locations[0]; + let live_above_floor = loc + .min_free_space + .available_space(FAKE_FREE.load(Ordering::Relaxed), FAKE_ALL.load(Ordering::Relaxed)); + loc.available_space.store( + live_above_floor.saturating_sub(loc.reserved_space.load(Ordering::Relaxed)), + Ordering::Relaxed, + ); + assert_eq!( + loc.available_space.load(Ordering::Relaxed), + 10 * MIB + ); + assert_eq!(store.reserve_free_location(need, pred, None), None); + + // The failed copy removes only its own charge. + store.release_free_location(0, need); + assert_eq!( + loc.available_space.load(Ordering::Relaxed), + 40 * MIB + ); + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + assert_eq!(store.reserve_free_location(need, pred, None), None); + } + + // Mirrors Go's TestCommitFreeLocationSettlesLandedReservation: a copy + // that lands ends its pending charge without refunding the cache — the + // charge and the landed bytes cancel out. + #[test] + fn test_commit_free_location_settles_landed_reservation() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let need = 30 * MIB; + let store = headroom_store(dir, floor + 40 * MIB, floor); + store.locations[0] + .available_space + .store(40 * MIB, Ordering::Relaxed); + + let pred = |loc: &DiskLocation| loc.disk_type == DiskType::HardDrive; + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + + FAKE_FREE.fetch_sub(need, Ordering::Relaxed); // the copied bytes are on disk + store.commit_free_location(0, need); + let loc = &store.locations[0]; + assert_eq!(loc.reserved_space.load(Ordering::Relaxed), 0); + assert_eq!( + loc.available_space.load(Ordering::Relaxed), + 10 * MIB + ); + + // The next probe reads the landed usage and the ended charge agrees. + let live_above_floor = loc + .min_free_space + .available_space(FAKE_FREE.load(Ordering::Relaxed), FAKE_ALL.load(Ordering::Relaxed)); + loc.available_space.store( + live_above_floor.saturating_sub(loc.reserved_space.load(Ordering::Relaxed)), + Ordering::Relaxed, + ); + assert_eq!( + loc.available_space.load(Ordering::Relaxed), + 10 * MIB + ); + } + + // Mirrors Go's TestCommitFreeLocationSettlesAsBytesLand: a running copy + // settles its charge as its bytes land on disk, so the live-space check + // sees the written part once — in the disk reading — not again inside + // the pending charge. + #[test] + fn test_commit_free_location_settles_as_bytes_land() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + let need = 30 * MIB; + let store = headroom_store(dir, floor + 70 * MIB, floor); + store.locations[0] + .available_space + .store(70 * MIB, Ordering::Relaxed); + + let pred = |loc: &DiskLocation| loc.disk_type == DiskType::HardDrive; + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + + // Half the copy's bytes are on disk now; settling that much charge + // keeps the second 30 MiB reservation inside the real headroom. + FAKE_FREE.fetch_sub(15 * MIB, Ordering::Relaxed); + store.commit_free_location(0, 15 * MIB); + + assert_eq!(store.reserve_free_location(need, pred, None), Some(0)); + let loc = &store.locations[0]; + assert_eq!(loc.reserved_space.load(Ordering::Relaxed), 45 * MIB); + assert_eq!(loc.available_space.load(Ordering::Relaxed), 10 * MIB); + } + + // Mirrors Go's TestMaybeAdjustVolumeMaxIgnoresFreeSpaceInsideFloor: space + // inside the floor adds no slots. + #[test] + fn test_maybe_adjust_volume_max_ignores_free_space_inside_floor() { + let _g = FAKE_DISK_LOCK.lock().unwrap(); + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let floor = 1000 * MIB; + // 90 MiB above the floor is 3 x 30 MiB volumes; the rest is headroom. + let mut store = headroom_store(dir, floor + 90 * MIB, floor); + store.volume_size_limit.store(30 * MIB, Ordering::Relaxed); + store.locations[0].original_max_volume_count = 0; + + store.maybe_adjust_volume_max(); + + assert_eq!( + store.locations[0].max_volume_count.load(Ordering::Relaxed), + 3 + ); + } + // Mirrors Go's TestConfigureVolumeUpdatesSuperblock: configure must leave // the .dat header and the .vif both on the new placement. #[test] diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 7fc7be57de..e7efffe908 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -6079,10 +6079,14 @@ pub fn scan_volume_file( Ok(()) } +/// Whether `preallocate_file` reserves blocks on this platform; off Linux it +/// is a no-op, so a create must not be charged for space it will not reserve. +pub(crate) const PREALLOCATE_SUPPORTED: bool = cfg!(target_os = "linux"); + /// Reserve disk blocks for a file without changing its visible size. /// On Linux, uses `fallocate(FALLOC_FL_KEEP_SIZE)` to actually reserve blocks. /// On other platforms, this is a no-op. -fn preallocate_file(file: &File, size: u64) { +pub(crate) fn preallocate_file(file: &File, size: u64) { #[cfg(target_os = "linux")] { use std::os::unix::io::AsRawFd; @@ -6111,6 +6115,21 @@ fn preallocate_file(file: &File, size: u64) { } } +/// Bytes the filesystem holds for the file, which exceeds its size after a +/// FALLOC_FL_KEEP_SIZE preallocate — Go's AllocatedSize. +pub(crate) fn allocated_file_size(file: &File) -> u64 { + #[cfg(unix)] + { + use std::os::unix::fs::MetadataExt; + return file.metadata().map(|m| m.blocks() * 512).unwrap_or(0); + } + #[cfg(not(unix))] + { + let _ = file; + 0 + } +} + // ============================================================================ // Tests // ============================================================================ diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 110759f19e..55bd0d5bbf 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -59,8 +59,9 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre var sourceVolumeStatusAfterCopy *volume_server_pb.VolumeStatusResponse var dataBaseFileName, indexBaseFileName, idxFileName, datFileName string var hasRemoteDatFile bool - err := operation.WithVolumeServerClientOptions(true, pb.ServerAddress(req.SourceDataNode), func(client volume_server_pb.VolumeServerClient) error { - var err error + var copyLocation *storage.DiskLocation + var copyReservedSpace uint64 + err := operation.WithVolumeServerClientOptions(true, pb.ServerAddress(req.SourceDataNode), func(client volume_server_pb.VolumeServerClient) (err error) { sourceVolumeStatus, err = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{ VolumeId: req.VolumeId, }) @@ -82,18 +83,77 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre diskType = req.DiskType } hasRemoteDatFile = volFileInfoResp.VolumeInfo != nil && len(volFileInfoResp.VolumeInfo.Files) > 0 + + var preallocateSize int64 + if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(context.Background()), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { + resp, err := client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) + if err != nil { + return fmt.Errorf("get master %s configuration: %v", vs.GetMaster(context.Background()), err) + } + if resp.VolumePreallocate { + preallocateSize = int64(resp.VolumeSizeLimitMB) * (1 << 20) + } + return nil + }); grpcErr != nil { + glog.V(0).Infof("connect to %s: %v", vs.GetMaster(context.Background()), grpcErr) + } + // a remote-backed volume only lands its .idx/.vif locally; the .dat stays in the tier neededSpace := volFileInfoResp.DatFileSize if hasRemoteDatFile { neededSpace = volFileInfoResp.IdxFileSize + } else if backend.PreallocateSupported && uint64(preallocateSize) > neededSpace { + // the .dat reserves the master's volume size limit up front, so the + // location must fit the reservation, not just the copied bytes + neededSpace = uint64(preallocateSize) } - location := vs.store.FindFreeLocation(func(location *storage.DiskLocation) bool { + location := vs.store.ReserveFreeLocation(neededSpace, func(location *storage.DiskLocation) bool { return location.DiskType == types.ToDiskType(diskType) && location.AvailableSpace.Load() > neededSpace }, needle.VolumeId(req.VolumeId)) if location == nil { return fmt.Errorf("%s %s", util.ErrVolumeNoSpaceLeft, types.ToDiskType(diskType).ReadableString()) } + copyLocation = location + copyReservedSpace = neededSpace + + // The headroom charge covers bytes promised but not yet visible to + // statfs; as the copy's bytes land they stop needing the charge, so a + // slow copy does not count its own .dat against the free space a + // second reservation needs. + commitCopy := func(n uint64) { + if n > copyReservedSpace { + n = copyReservedSpace + } + if n > 0 { + vs.store.CommitFreeLocation(location, n) + copyReservedSpace -= n + } + } + newCopyProgress := func(skip uint64) func(int64) { + counted := int64(skip) + return func(processed int64) { + if processed > counted { + commitCopy(uint64(processed - counted)) + counted = processed + } + } + } + + defer func() { + if err != nil { + if dataBaseFileName != "" { + os.Remove(dataBaseFileName + ".dat") + os.Remove(indexBaseFileName + ".idx") + os.Remove(dataBaseFileName + ".vif") + os.Remove(dataBaseFileName + ".note") + } + // Zero before the outer defer sees the same error, or the + // charge would be released twice. + vs.store.ReleaseFreeLocation(location, copyReservedSpace) + copyReservedSpace = 0 + } + }() // Source is reachable and a destination is reserved: only now is it // safe to drop an existing local replica before overwriting its files. @@ -117,35 +177,20 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre return fmt.Errorf("write .note for volume %d: %w", req.VolumeId, noteErr) } - defer func() { - if err != nil { - os.Remove(dataBaseFileName + ".dat") - os.Remove(indexBaseFileName + ".idx") - os.Remove(dataBaseFileName + ".vif") - os.Remove(dataBaseFileName + ".note") - } - }() - - var preallocateSize int64 - if grpcErr := pb.WithMasterClient(context.Background(), false, vs.GetMaster(context.Background()), vs.grpcDialOption, false, func(client master_pb.SeaweedClient) error { - resp, err := client.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{}) - if err != nil { - return fmt.Errorf("get master %s configuration: %v", vs.GetMaster(context.Background()), err) - } - if resp.VolumePreallocate { - preallocateSize = int64(resp.VolumeSizeLimitMB) * (1 << 20) - } - return nil - }); grpcErr != nil { - glog.V(0).Infof("connect to %s: %v", vs.GetMaster(context.Background()), grpcErr) - } - + var copyDatAllocated uint64 if preallocateSize > 0 && !hasRemoteDatFile { volumeFile := dataBaseFileName + ".dat" _, err := backend.CreateVolumeFile(volumeFile, preallocateSize, 0) if err != nil { return fmt.Errorf("create volume file %s: %v", volumeFile, err) } + // fallocate already holds those bytes on disk; when the + // filesystem did not reserve them the write progress below + // settles the charge as the bytes land instead. + if fi, statErr := os.Stat(volumeFile); statErr == nil { + copyDatAllocated = storage.AllocatedSize(fi) + commitCopy(copyDatAllocated) + } } // println("source:", volFileInfoResp.String()) @@ -163,7 +208,12 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre throttler := util.NewWriteThrottler(ioBytePerSecond) if !hasRemoteDatFile { - if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".dat", false, true, func(processed int64) bool { + datProgress := newCopyProgress(copyDatAllocated) + // Write into the .dat CreateVolumeFile just made: the staged + // .copying file would peak at preallocate+dat on disk and then + // discard the fallocated blocks at rename. + if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.DatFileSize, dataBaseFileName, ".dat", preallocateSize > 0, true, func(processed int64) bool { + datProgress(processed) if processed > nextReportTarget { copyResponse.ProcessedBytes = processed if sendErr = stream.Send(copyResponse); sendErr != nil { @@ -183,14 +233,22 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } } - if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, indexBaseFileName, ".idx", false, false, nil, throttler); err != nil { + idxProgress := newCopyProgress(0) + if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, volFileInfoResp.IdxFileSize, indexBaseFileName, ".idx", false, false, func(processed int64) bool { + idxProgress(processed) + return true + }, throttler); err != nil { return err } if modifiedTsNs > 0 { os.Chtimes(indexBaseFileName+".idx", time.Unix(0, modifiedTsNs), time.Unix(0, modifiedTsNs)) } - if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, 1024*1024, dataBaseFileName, ".vif", false, true, nil, throttler); err != nil { + vifProgress := newCopyProgress(0) + if modifiedTsNs, err = vs.doCopyFileWithThrottler(client, false, req.Collection, req.VolumeId, volFileInfoResp.CompactionRevision, 1024*1024, dataBaseFileName, ".vif", false, true, func(processed int64) bool { + vifProgress(processed) + return true + }, throttler); err != nil { return err } if modifiedTsNs > 0 { @@ -227,10 +285,16 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre datFileName = dataBaseFileName + ".dat" defer func() { - if err != nil && dataBaseFileName != "" { + if copyLocation == nil { + return + } + if err != nil { os.Remove(idxFileName) os.Remove(datFileName) os.Remove(dataBaseFileName + ".vif") + vs.store.ReleaseFreeLocation(copyLocation, copyReservedSpace) + } else { + vs.store.CommitFreeLocation(copyLocation, copyReservedSpace) } }() diff --git a/weed/storage/backend/volume_create.go b/weed/storage/backend/volume_create.go index 6a9e09ef89..5b80ba682b 100644 --- a/weed/storage/backend/volume_create.go +++ b/weed/storage/backend/volume_create.go @@ -8,6 +8,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" ) +// PreallocateSupported reports whether CreateVolumeFile reserves the +// requested size on this platform. +const PreallocateSupported = false + func CreateVolumeFile(fileName string, preallocate int64, memoryMapSizeMB uint32) (BackendStorageFile, error) { file, e := OpenVolumeFile(fileName, os.O_RDWR|os.O_CREATE|os.O_TRUNC) if e != nil { diff --git a/weed/storage/backend/volume_create_linux.go b/weed/storage/backend/volume_create_linux.go index 28aeb6ed1d..c7be3be43f 100644 --- a/weed/storage/backend/volume_create_linux.go +++ b/weed/storage/backend/volume_create_linux.go @@ -10,6 +10,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" ) +// PreallocateSupported reports whether CreateVolumeFile reserves the +// requested size on this platform. +const PreallocateSupported = true + func CreateVolumeFile(fileName string, preallocate int64, memoryMapSizeMB uint32) (BackendStorageFile, error) { file, e := OpenVolumeFile(fileName, os.O_RDWR|os.O_CREATE|os.O_TRUNC) if e != nil { diff --git a/weed/storage/backend/volume_create_windows.go b/weed/storage/backend/volume_create_windows.go index d6eddc81aa..8a735cc3d3 100644 --- a/weed/storage/backend/volume_create_windows.go +++ b/weed/storage/backend/volume_create_windows.go @@ -10,6 +10,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/backend/memory_map/os_overloads" ) +// PreallocateSupported reports whether CreateVolumeFile reserves the +// requested size on this platform. +const PreallocateSupported = false + func CreateVolumeFile(fileName string, preallocate int64, memoryMapSizeMB uint32) (BackendStorageFile, error) { if preallocate > 0 { glog.V(0).Infof("Preallocated disk space for %s is not supported", fileName) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index e904bae7dc..56d6283a9c 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -38,6 +38,10 @@ type DiskLocation struct { OriginalMaxVolumeCount int32 MinFreeSpace util.MinFreeSpace AvailableSpace atomic.Uint64 + // Headroom promised to in-flight copies that statfs cannot see yet. + // Probes keep it out of AvailableSpace so they cannot re-spend it; a + // copy that lands ends its charge, one that fails removes only its own. + ReservedSpace atomic.Uint64 // Physical filesystem capacity from the latest CheckDiskSpace probe, reported // to the master so balancing can see real disk fullness, not just slot counts. diskTotalBytes atomic.Uint64 @@ -738,7 +742,11 @@ func (l *DiskLocation) CheckDiskSpace(config stats.DiskIOProbeConfig) { stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "used").Set(float64(s.Used)) stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "free").Set(float64(s.Free)) stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "avail").Set(float64(available)) - l.AvailableSpace.Store(available) + if pending := l.ReservedSpace.Load(); pending < available { + l.AvailableSpace.Store(available - pending) + } else { + l.AvailableSpace.Store(0) + } l.diskTotalBytes.Store(s.All) l.diskFreeBytes.Store(s.Free) isLow, desc := l.MinFreeSpace.IsLow(s.Free, s.PercentFree) diff --git a/weed/storage/disk_location_linux_test.go b/weed/storage/disk_location_linux_test.go index 7815142c40..9f391743cb 100644 --- a/weed/storage/disk_location_linux_test.go +++ b/weed/storage/disk_location_linux_test.go @@ -22,7 +22,7 @@ func TestUnUsedSpaceCountsPreallocatedVolumesAsFull(t *testing.T) { if err != nil { t.Fatal(err) } - if allocatedSize(fi) < volumeSizeLimit { + if AllocatedSize(fi) < volumeSizeLimit { t.Skipf("filesystem at %s does not support fallocate", dir) } @@ -40,7 +40,7 @@ func TestUnUsedSpaceCountsPreallocatedVolumesAsFull(t *testing.T) { t.Fatal(err) } // The filesystem may round a small file up to a larger allocation unit. - plainSize := max(uint64(fi.Size()), allocatedSize(fi)) + plainSize := max(uint64(fi.Size()), AllocatedSize(fi)) diskLocation := DiskLocation{Directory: dir, IdxDirectory: dir} diskLocation.volumes = make(map[needle.VolumeId]*Volume) diff --git a/weed/storage/store.go b/weed/storage/store.go index d213ee6f9e..5916ee5fb6 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -114,6 +114,9 @@ type Store struct { vidStampReady atomic.Bool vidStampDirty atomic.Bool vidStampLock sync.Mutex + // volumeCreateLock serializes addVolume, so each room check sees the + // preallocation of every volume created before it. + volumeCreateLock sync.Mutex } func (s *Store) String() (str string) { @@ -339,7 +342,58 @@ func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool, rep } return ret } + +// ReserveFreeLocation picks like FindFreeLocation, but under volumeCreateLock +// and against live disk stats rather than the once-a-minute probe, then +// charges need against the location's pending reservations. A VolumeCopy's +// pick and its .dat reservation are separate steps; without this a second +// copy can spend the same headroom and drive the disk under -volume.minFreeSpace. +func (s *Store) ReserveFreeLocation(need uint64, filterFn func(location *DiskLocation) bool, replaceVid ...needle.VolumeId) *DiskLocation { + s.volumeCreateLock.Lock() + defer s.volumeCreateLock.Unlock() + location := s.FindFreeLocation(func(location *DiskLocation) bool { + if location.AvailableSpace.Load() <= need { + return false + } + if filterFn != nil && !filterFn(location) { + return false + } + status := diskStatusOf(location.Directory) + if low, _ := location.MinFreeSpace.IsLow(status.Free, status.PercentFree); low { + return false + } + return location.MinFreeSpace.AvailableSpace(status.Free, status.All) > location.ReservedSpace.Load()+need + }, replaceVid...) + if location == nil { + return nil + } + status := diskStatusOf(location.Directory) + location.ReservedSpace.Add(need) + if available := location.MinFreeSpace.AvailableSpace(status.Free, status.All); available > location.ReservedSpace.Load() { + location.AvailableSpace.Store(available - location.ReservedSpace.Load()) + } else { + location.AvailableSpace.Store(0) + } + return location +} + +// ReleaseFreeLocation returns space ReserveFreeLocation charged when the copy +// it paid for does not land. Removing the pending charge first keeps a probe +// that already rebased the cache from seeing the refund twice. +func (s *Store) ReleaseFreeLocation(location *DiskLocation, need uint64) { + location.ReservedSpace.Add(-need) + location.AvailableSpace.Add(need) +} + +// CommitFreeLocation settles a reservation whose copy landed: the bytes are +// real on disk, so only the pending charge ends — the cache keeps the charge, +// which now matches the live usage the next probe will read. +func (s *Store) CommitFreeLocation(location *DiskLocation, need uint64) { + location.ReservedSpace.Add(-need) +} func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, diskType DiskType, ldbTimeout int64) error { + s.volumeCreateLock.Lock() + defer s.volumeCreateLock.Unlock() s.UpdateMaxVolumeIdSeen(vid) if s.findVolume(vid) != nil { return fmt.Errorf("Volume Id %d already exists!", vid) @@ -350,7 +404,7 @@ func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind var diskId uint32 var minVolCount int for i, loc := range s.Locations { - if loc.DiskType == diskType && s.hasFreeDiskLocation(loc) { + if loc.DiskType == diskType && s.hasFreeDiskLocation(loc) && hasRoomForNewVolume(loc, preallocate) { volCount := loc.LocalVolumesLen() if location == nil || volCount < minVolCount { location = loc @@ -387,6 +441,38 @@ func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind return fmt.Errorf("No more free space left") } +// Create-time metadata a location must still fit after a preallocation: .vif +// always lands beside the data file; the .idx a writable volume keeps +// appending counts only when dir_idx shares the filesystem. Holding these +// back keeps an exact-fit reservation from dipping under the floor once the +// metadata is written. +const ( + volumeCreateVifBytes = 1 << 20 + volumeCreateIdxBytes = 4 << 20 +) + +// hasRoomForNewVolume reports whether a preallocated volume may be created on +// location. A preallocated volume reserves its full size up front, so it must +// fit in the free space above -volume.minFreeSpace; this reads the disk now +// rather than the once-a-minute low-disk flag, so a burst of creates cannot run +// the disk through the floor between two probes. A volume without preallocation +// reserves nothing — nor does one on a platform where CreateVolumeFile cannot +// preallocate — and copying existing data into one (a merge) is always allowed. +func hasRoomForNewVolume(location *DiskLocation, preallocate int64) bool { + if preallocate <= 0 || !createVolumeFilePreallocates { + return true + } + status := diskStatusOf(location.Directory) + if low, _ := location.MinFreeSpace.IsLow(status.Free, status.PercentFree); low { + return false + } + need := uint64(preallocate) + volumeCreateVifBytes + if sameFilesystem(location.Directory, location.IdxDirectory) { + need += volumeCreateIdxBytes + } + return location.MinFreeSpace.AvailableSpace(status.Free, status.All) >= need+location.ReservedSpace.Load() +} + // hasFreeDiskLocation checks if a disk location has free space func (s *Store) hasFreeDiskLocation(location *DiskLocation) bool { // Check if disk space is low first @@ -1375,6 +1461,15 @@ func (s *Store) GetPreallocate() bool { return s.preallocate.Load() } +// diskStatusOf reads the usage of the disk holding a directory. It is a +// variable so a test can stand in for a disk that is full or near its floor. +var diskStatusOf = stats.NewDiskStatus + +// createVolumeFilePreallocates reports whether CreateVolumeFile reserves the +// requested size on this platform. It is a variable so a test can exercise +// the guard off Linux. +var createVolumeFilePreallocates = backend.PreallocateSupported + func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) { volumeSizeLimit := s.GetVolumeSizeLimit() if volumeSizeLimit == 0 { @@ -1384,9 +1479,10 @@ func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) { for _, diskLocation := range s.Locations { if diskLocation.OriginalMaxVolumeCount == 0 { currentMaxVolumeCount := atomic.LoadInt32(&diskLocation.MaxVolumeCount) - diskStatus := stats.NewDiskStatus(diskLocation.Directory) + diskStatus := diskStatusOf(diskLocation.Directory) unusedSpace := diskLocation.UnUsedSpace(volumeSizeLimit) - unclaimedSpaces := int64(diskStatus.Free) - int64(unusedSpace) + // Free space under -volume.minFreeSpace is not available for new volumes. + unclaimedSpaces := int64(diskLocation.MinFreeSpace.AvailableSpace(diskStatus.Free, diskStatus.All)) - int64(unusedSpace) volCount := diskLocation.VolumesLen() ecShardCount := diskLocation.EcShardCount() maxVolumeCount := int32(volCount) + int32((ecShardCount+erasure_coding.DataShardsCount-1)/erasure_coding.DataShardsCount) diff --git a/weed/storage/store_headroom_test.go b/weed/storage/store_headroom_test.go new file mode 100644 index 0000000000..4db6f4c6b0 --- /dev/null +++ b/weed/storage/store_headroom_test.go @@ -0,0 +1,363 @@ +package storage + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +const testMiB = uint64(1 << 20) + +// fakeDisk stands in for the disk under a location. Preallocating a volume +// takes its size out of free space, as fallocate does on Linux. +type fakeDisk struct { + all uint64 + free uint64 +} + +// newHeadroomStore builds a store with one location whose disk has the given +// free space and whose -volume.minFreeSpace floor is floor. The low-disk flag +// stays false, as it does between two once-a-minute probes. +func newHeadroomStore(t *testing.T, free, floor uint64) (*Store, *DiskLocation, *fakeDisk) { + t.Helper() + dir := t.TempDir() + disk := &fakeDisk{all: 10 * 1024 * testMiB, free: free} + + orig := diskStatusOf + diskStatusOf = func(string) *volume_server_pb.DiskStatus { + return &volume_server_pb.DiskStatus{ + All: disk.all, + Free: disk.free, + Used: disk.all - disk.free, + PercentFree: float32(disk.free) * 100 / float32(disk.all), + } + } + t.Cleanup(func() { diskStatusOf = orig }) + + // The fake disk behaves like fallocate on Linux: pretend creates reserve. + origPreallocates := createVolumeFilePreallocates + createVolumeFilePreallocates = true + t.Cleanup(func() { createVolumeFilePreallocates = origPreallocates }) + + loc := &DiskLocation{ + Directory: dir, + IdxDirectory: dir, + DiskType: types.HardDriveType, + MinFreeSpace: util.MinFreeSpace{Type: util.AsBytes, Bytes: floor}, + } + loc.volumes = make(map[needle.VolumeId]*Volume) + // Close created volumes before TempDir removes them; Windows cannot + // delete a file that is still open. + t.Cleanup(func() { + for _, v := range loc.volumes { + v.Close() + } + }) + s := &Store{ + Locations: []*DiskLocation{loc}, + NewVolumesChan: make(chan *master_pb.VolumeShortInformationMessage, 100), + } + return s, loc, disk +} + +// The master sizes a disk's volume slots from its free space. Free space +// inside the -volume.minFreeSpace floor is not available for new volumes, so +// it must not add slots. +func TestMaybeAdjustVolumeMaxIgnoresFreeSpaceInsideFloor(t *testing.T) { + const floor = 1000 * testMiB + const volumeSize = 30 * testMiB + // 90 MiB above the floor is 3 volumes. The other 1000 MiB is headroom. + s, loc, _ := newHeadroomStore(t, floor+90*testMiB, floor) + s.SetVolumeSizeLimit(volumeSize) + s.SetPreallocate(true) + + s.MaybeAdjustVolumeMax() + + if loc.MaxVolumeCount != 3 { + t.Errorf("MaxVolumeCount=%d; want 3 (only the 90 MiB above the floor is usable)", loc.MaxVolumeCount) + } +} + +// A preallocated volume must not be created when its reservation would take +// the disk under the floor, even if the low-disk flag has not been probed yet. +func TestAddVolumeRefusesPreallocationThatCrossesFloor(t *testing.T) { + const floor = 1000 * testMiB + const preallocate = 8 * testMiB + // 4 MiB above the floor, so an 8 MiB reservation would cross it. + s, _, disk := newHeadroomStore(t, floor+4*testMiB, floor) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + err := s.addVolume(1, "", NeedleMapInMemory, rp, needle.EMPTY_TTL, int64(preallocate), needle.GetCurrentVersion(), 0, types.HardDriveType, 0) + + if err == nil { + t.Errorf("addVolume succeeded; want refusal (free %d MiB, floor %d MiB, reserve %d MiB)", + disk.free/testMiB, floor/testMiB, preallocate/testMiB) + } + if s.findVolume(1) != nil { + t.Errorf("volume 1 exists after refused addVolume") + } +} + +// A volume with no preallocation is still created while free space is above +// the floor, so the guard only holds back growth that would cross it. +func TestAddVolumeAllowsPlainVolumeAboveFloor(t *testing.T) { + const floor = 1000 * testMiB + s, _, _ := newHeadroomStore(t, floor+1*testMiB, floor) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + if err := s.addVolume(1, "", NeedleMapInMemory, rp, needle.EMPTY_TTL, 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Errorf("addVolume without preallocation failed above the floor: %v", err) + } +} + +// Copying existing data into a new volume (a merge) reserves nothing and must +// still go through while the disk is below its floor. +func TestAddVolumeAllowsMergeVolumeBelowFloor(t *testing.T) { + const floor = 1000 * testMiB + s, _, _ := newHeadroomStore(t, floor-1*testMiB, floor) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + if err := s.addVolume(1, "", NeedleMapInMemory, rp, needle.EMPTY_TTL, 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Errorf("merge volume refused below the floor: %v", err) + } +} + +// Where CreateVolumeFile cannot preallocate, a create reserves nothing and +// must not be held back by a reservation that will never exist. +func TestAddVolumeAllowsPreallocationWhereUnsupported(t *testing.T) { + const floor = 1000 * testMiB + s, _, _ := newHeadroomStore(t, floor+1*testMiB, floor) + createVolumeFilePreallocates = false + rp, _ := super_block.NewReplicaPlacementFromString("000") + + if err := s.addVolume(1, "", NeedleMapInMemory, rp, needle.EMPTY_TTL, 8*int64(testMiB), needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Errorf("addVolume refused where preallocation is unsupported: %v", err) + } +} + +// Replays the incident in miniature: the master keeps asking for volumes +// while the disk sits just above its floor. Each created volume reserves its +// full size, so the disk must not be driven below the floor. +func TestGrowthLoopStopsAtFloorWhenVolumesPreallocate(t *testing.T) { + const floor = 1000 * testMiB + const preallocate = 1 * testMiB + const attempts = 20 + // Room for 4 preallocated volumes above the floor once each create's + // metadata allowance is held back. + const createCost = preallocate + volumeCreateVifBytes + volumeCreateIdxBytes + s, _, disk := newHeadroomStore(t, floor+4*createCost, floor) + rp, _ := super_block.NewReplicaPlacementFromString("000") + + created := 0 + for i := 1; i <= attempts; i++ { + err := s.addVolume(needle.VolumeId(i), "", NeedleMapInMemory, rp, needle.EMPTY_TTL, int64(preallocate), needle.GetCurrentVersion(), 0, types.HardDriveType, 0) + if err != nil { + continue + } + created++ + // fallocate reserves the volume's full size on disk; the metadata + // allowance it was charged goes with it. + disk.free -= createCost + } + + if disk.free < floor { + t.Errorf("free space fell to %d MiB, below the %d MiB floor, after %d volumes", + disk.free/testMiB, floor/testMiB, created) + } + // Exactly the room above the floor must be used. Checking only an upper + // bound would also pass when every create fails. + if created != 4 { + t.Errorf("created %d preallocated volumes; want exactly 4 (the room above the floor)", created) + } +} + +// A copy's pick and its .dat reservation are separate steps. The picked +// location's cached free space must carry the charge at once, so a second +// copy cannot spend the same headroom before the first reservation lands. +func TestReserveFreeLocationChargesPickedReservation(t *testing.T) { + const floor = 1000 * testMiB + const need = 30 * testMiB + // 40 MiB above the floor: room for one 30 MiB reservation, not two. + s, loc, _ := newHeadroomStore(t, floor+40*testMiB, floor) + loc.MaxVolumeCount = 10 + loc.AvailableSpace.Store(40 * testMiB) + + pick := s.ReserveFreeLocation(need, func(location *DiskLocation) bool { + return location.DiskType == types.HardDriveType + }) + if pick != loc { + t.Fatalf("first pick = %v; want the only location", pick) + } + if got := loc.AvailableSpace.Load(); got != 10*testMiB { + t.Fatalf("AvailableSpace after charge = %d MiB; want 10", got/testMiB) + } + + if pick := s.ReserveFreeLocation(need, func(location *DiskLocation) bool { + return location.DiskType == types.HardDriveType + }); pick != nil { + t.Fatalf("second pick = %v; want nil (the first reservation still holds the headroom)", pick) + } +} + +// The cached free space is refreshed only once a minute, so a stale-high +// value must not admit a reservation the live disk cannot fit. +func TestReserveFreeLocationRechecksLiveDisk(t *testing.T) { + const floor = 1000 * testMiB + const need = 30 * testMiB + // Only 5 MiB above the floor, but the probe last saw 100 MiB of headroom. + s, loc, _ := newHeadroomStore(t, floor+5*testMiB, floor) + loc.MaxVolumeCount = 10 + loc.AvailableSpace.Store(100 * testMiB) + + if pick := s.ReserveFreeLocation(need, func(location *DiskLocation) bool { + return location.DiskType == types.HardDriveType + }); pick != nil { + t.Fatalf("pick = %v; want nil (live disk has no room for the reservation)", pick) + } +} + +// A failed copy must return its reservation, or the headroom stays booked +// until the next disk probe and blocks retries. +func TestReleaseFreeLocationRestoresReservation(t *testing.T) { + const floor = 1000 * testMiB + const need = 30 * testMiB + s, loc, _ := newHeadroomStore(t, floor+40*testMiB, floor) + loc.MaxVolumeCount = 10 + loc.AvailableSpace.Store(40 * testMiB) + + pick := s.ReserveFreeLocation(need, func(location *DiskLocation) bool { + return location.DiskType == types.HardDriveType + }) + if pick != loc { + t.Fatalf("first pick = %v; want the only location", pick) + } + + s.ReleaseFreeLocation(pick, need) + if pick := s.ReserveFreeLocation(need, func(location *DiskLocation) bool { + return location.DiskType == types.HardDriveType + }); pick == nil { + t.Fatal("retry pick = nil; want the refunded reservation to admit a copy") + } +} + +// A probe between reserve and release must not resurrect the charge: the +// probe's live reading already forgot the failed copy's pending bytes, so +// refunding the cached figure would let two copies spend the same headroom. +func TestReleaseFreeLocationAfterProbeRefundsOnlyOwnCharge(t *testing.T) { + const floor = 1000 * testMiB + const need = 30 * testMiB + s, loc, disk := newHeadroomStore(t, floor+40*testMiB, floor) + loc.MaxVolumeCount = 10 + loc.AvailableSpace.Store(40 * testMiB) + + orig := newDiskStatus + newDiskStatus = func(string, stats.DiskIOProbeConfig) *volume_server_pb.DiskStatus { + return &volume_server_pb.DiskStatus{All: disk.all, Free: disk.free} + } + t.Cleanup(func() { newDiskStatus = orig }) + + hdd := func(location *DiskLocation) bool { return location.DiskType == types.HardDriveType } + if pick := s.ReserveFreeLocation(need, hdd); pick == nil { + t.Fatal("first pick = nil; want the only location") + } + + // Mid-copy probe: the live disk still shows the not-yet-written bytes + // free, but the pending charge must keep them out of the cache. + loc.CheckDiskSpace(stats.DiskIOProbeConfig{}) + if got := loc.AvailableSpace.Load(); got != 10*testMiB { + t.Fatalf("AvailableSpace after probe = %d MiB; want 10 (charge survives the probe)", got/testMiB) + } + if pick := s.ReserveFreeLocation(need, hdd); pick != nil { + t.Fatalf("second pick = %v; want nil (the in-flight reservation still holds the headroom)", pick) + } + + // The failed copy removes only its own charge; the probe-rebased cache + // must end up holding exactly the live headroom, not the headroom plus + // a resurrected second charge. + s.ReleaseFreeLocation(loc, need) + if got := loc.AvailableSpace.Load(); got != 40*testMiB { + t.Fatalf("AvailableSpace after refund = %d MiB; want 40", got/testMiB) + } + if pick := s.ReserveFreeLocation(need, hdd); pick == nil { + t.Fatal("retry pick = nil; want the live headroom to admit a copy") + } + if pick := s.ReserveFreeLocation(need, hdd); pick != nil { + t.Fatalf("third pick = %v; want nil (only 40 MiB of headroom exists)", pick) + } +} + +// A copy that lands must end its pending charge without refunding the cache: +// the cache's charge and the landed bytes cancel out, and the next probe +// reads the disk exactly as it is. +func TestCommitFreeLocationSettlesLandedReservation(t *testing.T) { + const floor = 1000 * testMiB + const need = 30 * testMiB + s, loc, disk := newHeadroomStore(t, floor+40*testMiB, floor) + loc.MaxVolumeCount = 10 + loc.AvailableSpace.Store(40 * testMiB) + + orig := newDiskStatus + newDiskStatus = func(string, stats.DiskIOProbeConfig) *volume_server_pb.DiskStatus { + return &volume_server_pb.DiskStatus{All: disk.all, Free: disk.free} + } + t.Cleanup(func() { newDiskStatus = orig }) + + pick := s.ReserveFreeLocation(need, func(location *DiskLocation) bool { + return location.DiskType == types.HardDriveType + }) + if pick != loc { + t.Fatalf("pick = %v; want the only location", pick) + } + + disk.free -= need // the copied bytes are on disk now + s.CommitFreeLocation(pick, need) + if got := loc.ReservedSpace.Load(); got != 0 { + t.Fatalf("ReservedSpace after commit = %d MiB; want 0", got/testMiB) + } + if got := loc.AvailableSpace.Load(); got != 10*testMiB { + t.Fatalf("AvailableSpace after commit = %d MiB; want 10 (no refund)", got/testMiB) + } + + loc.CheckDiskSpace(stats.DiskIOProbeConfig{}) + if got := loc.AvailableSpace.Load(); got != 10*testMiB { + t.Fatalf("AvailableSpace after post-landing probe = %d MiB; want 10", got/testMiB) + } +} + +// A running copy settles its charge as its bytes land on disk. The +// live-space check must then see the written part once — in the disk +// reading — not again inside the pending charge, or a slow copy would block +// reservations the disk has room for. +func TestCommitFreeLocationSettlesAsBytesLand(t *testing.T) { + const floor = 1000 * testMiB + const need = 30 * testMiB + s, loc, disk := newHeadroomStore(t, floor+70*testMiB, floor) + loc.MaxVolumeCount = 10 + loc.AvailableSpace.Store(70 * testMiB) + hdd := func(location *DiskLocation) bool { return location.DiskType == types.HardDriveType } + + if pick := s.ReserveFreeLocation(need, hdd); pick != loc { + t.Fatalf("first pick = %v; want the only location", pick) + } + + // Half the copy's bytes are on disk now; settling that much charge + // keeps the second 30 MiB reservation inside the real headroom. + disk.free -= 15 * testMiB + s.CommitFreeLocation(loc, 15*testMiB) + + if pick := s.ReserveFreeLocation(need, hdd); pick == nil { + t.Fatal("second pick = nil; want the settled headroom to admit the copy") + } + if got := loc.ReservedSpace.Load(); got != 45*testMiB { + t.Fatalf("ReservedSpace after second pick = %d MiB; want 45", got/testMiB) + } + if got := loc.AvailableSpace.Load(); got != 10*testMiB { + t.Fatalf("AvailableSpace after second pick = %d MiB; want 10", got/testMiB) + } +} diff --git a/weed/storage/volume.go b/weed/storage/volume.go index d92a930309..fd615bb452 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -246,7 +246,7 @@ func (v *Volume) datAllocatedSize() uint64 { if err != nil { return 0 } - return allocatedSize(fi) + return AllocatedSize(fi) } func (v *Volume) ContentSize() uint64 { diff --git a/weed/storage/volume_allocated_unix.go b/weed/storage/volume_allocated_unix.go index df267efd83..fa1bc196c0 100644 --- a/weed/storage/volume_allocated_unix.go +++ b/weed/storage/volume_allocated_unix.go @@ -7,9 +7,9 @@ import ( "syscall" ) -// allocatedSize reports the bytes the filesystem holds for a file, which +// AllocatedSize reports the bytes the filesystem holds for a file, which // exceeds its size when the file was preallocated with FALLOC_FL_KEEP_SIZE. -func allocatedSize(fi os.FileInfo) uint64 { +func AllocatedSize(fi os.FileInfo) uint64 { st, ok := fi.Sys().(*syscall.Stat_t) if !ok { return 0 diff --git a/weed/storage/volume_allocated_windows.go b/weed/storage/volume_allocated_windows.go index 6867c1ede9..514e4754d4 100644 --- a/weed/storage/volume_allocated_windows.go +++ b/weed/storage/volume_allocated_windows.go @@ -4,7 +4,7 @@ package storage import "os" -// allocatedSize is unknown on Windows, where volume files are never preallocated. -func allocatedSize(fi os.FileInfo) uint64 { +// AllocatedSize is unknown on Windows, where volume files are never preallocated. +func AllocatedSize(fi os.FileInfo) uint64 { return 0 }