diff --git a/VOLUME_SERVER_RUST_PLAN.md b/VOLUME_SERVER_RUST_PLAN.md index 1c402336f..4097b4d9f 100644 --- a/VOLUME_SERVER_RUST_PLAN.md +++ b/VOLUME_SERVER_RUST_PLAN.md @@ -169,6 +169,7 @@ seaweed-volume/ - `--memprofile` - `--compactionMBps` (default 0) - `--maintenanceMBps` (default 0) + - `--compactionMaxParallelCommits` (default 0) - `--fileSizeLimitMB` (default 256) - `--concurrentUploadLimitMB` (default 0) - `--concurrentDownloadLimitMB` (default 0) diff --git a/seaweed-volume/README.md b/seaweed-volume/README.md index 039cb436b..4321f70f3 100644 --- a/seaweed-volume/README.md +++ b/seaweed-volume/README.md @@ -64,6 +64,7 @@ seaweed-volume --port 8080 --master localhost:9333 --dir /data/vol1 --max 7 \ | `--whiteList` | | Comma-separated IPs with write permission | | `--preStopSeconds` | `10` | Graceful drain period before shutdown | | `--compactionMBps` | `0` (unlimited) | Compaction I/O rate limit | +| `--compactionMaxParallelCommits` | `0` (unlimited) | Max concurrent vacuum commits | | `--pprof` | `false` | Enable pprof HTTP handlers | Set `RUST_LOG=debug` (or `trace`, `info`, `warn`) for log level control. diff --git a/seaweed-volume/src/config.rs b/seaweed-volume/src/config.rs index b4e4da552..21117725c 100644 --- a/seaweed-volume/src/config.rs +++ b/seaweed-volume/src/config.rs @@ -125,6 +125,11 @@ pub struct Cli { #[arg(long = "maintenanceMBps", default_value_t = 0)] pub maintenance_mb_per_second: u32, + /// Limit how many vacuum commits run at once on this server; queued + /// commits hold no volume lock. 0 means unlimited. + #[arg(long = "compactionMaxParallelCommits", default_value_t = 0)] + pub max_parallel_vacuum_commits: usize, + /// Limit file size to avoid out of memory. #[arg(long = "fileSizeLimitMB", default_value_t = 256)] pub file_size_limit_mb: u32, @@ -263,6 +268,8 @@ pub struct VolumeServerConfig { pub mem_profile: String, pub compaction_byte_per_second: i64, pub maintenance_byte_per_second: i64, + /// Max concurrent vacuum commits; 0 means unlimited. + pub max_parallel_vacuum_commits: usize, pub file_size_limit_bytes: i64, pub concurrent_upload_limit: i64, pub concurrent_download_limit: i64, @@ -927,6 +934,7 @@ fn resolve_config_with_env(cli: Cli, env: EnvLookup<'_>) -> VolumeServerConfig { mem_profile: cli.mem_profile, compaction_byte_per_second: cli.compaction_mb_per_second as i64 * 1024 * 1024, maintenance_byte_per_second: cli.maintenance_mb_per_second as i64 * 1024 * 1024, + max_parallel_vacuum_commits: cli.max_parallel_vacuum_commits, file_size_limit_bytes: cli.file_size_limit_mb as i64 * 1024 * 1024, concurrent_upload_limit: cli.concurrent_upload_limit_mb as i64 * 1024 * 1024, concurrent_download_limit: cli.concurrent_download_limit_mb as i64 * 1024 * 1024, diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index 96d77a0b0..3f785edb5 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -340,6 +340,14 @@ async fn run( rack: config.rack.clone(), file_size_limit_bytes: config.file_size_limit_bytes, maintenance_byte_per_second: config.maintenance_byte_per_second, + vacuum_commit_slots: if config.max_parallel_vacuum_commits > 0 { + Some(tokio::sync::Semaphore::new( + config.max_parallel_vacuum_commits, + )) + } else { + None + }, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), // Go sets isHeartbeating: true unconditionally at startup is_heartbeating: std::sync::atomic::AtomicBool::new(true), has_master: !config.masters.is_empty(), diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index e8c35f4d9..71cc592ef 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/seaweed-volume/src/server/grpc_client.rs @@ -397,6 +397,7 @@ mod tests { mem_profile: String::new(), compaction_byte_per_second: 0, maintenance_byte_per_second: 0, + max_parallel_vacuum_commits: 0, file_size_limit_bytes: 0, concurrent_upload_limit: 0, concurrent_download_limit: 0, diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index ee3db150b..6ca502385 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -53,6 +53,29 @@ fn needle_not_found(needle_id: NeedleId) -> Status { Status::not_found(format!("needle not found {}", needle_id.0)) } +/// Deadline from the caller's `grpc-timeout` header (TimeoutValue + +/// H|M|S|m|u|n unit, per the gRPC-over-HTTP2 spec). Tonic's GrpcTimeout layer +/// drops an over-deadline handler future anyway; parsing it here lets a queued +/// vacuum commit refuse politely instead of being dropped mid-wait. +fn grpc_timeout_deadline(request: &Request) -> Option { + let value = request.metadata().get("grpc-timeout")?.to_str().ok()?; + let (digits, unit) = value.split_at(value.len().checked_sub(1)?); + if digits.len() > 8 { + return None; + } + let n: u64 = digits.parse().ok()?; + let dur = match unit { + "H" => std::time::Duration::from_secs(n * 60 * 60), + "M" => std::time::Duration::from_secs(n * 60), + "S" => std::time::Duration::from_secs(n), + "m" => std::time::Duration::from_millis(n), + "u" => std::time::Duration::from_micros(n), + "n" => std::time::Duration::from_nanos(n), + _ => return None, + }; + Some(tokio::time::Instant::now() + dur) +} + fn unix_now_seconds() -> f64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -1262,8 +1285,12 @@ impl VolumeServer for VolumeGrpcService { ) -> Result, Status> { self.check_grpc_admin_auth(&request)?; self.state.check_maintenance()?; + let deadline = grpc_timeout_deadline(&request); let vid = VolumeId(request.into_inner().volume_id); + let _permit = self.state.acquire_vacuum_commit_slot(deadline).await?; + self.state.check_maintenance()?; + // Match Go's store_vacuum.go CommitCompactVolume: skip commit if stopping if *self.state.is_stopping.read().unwrap() { return Err(Status::internal(format!( @@ -7347,6 +7374,8 @@ mod tests { rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: std::sync::atomic::AtomicBool::new(true), has_master: false, pre_stop_seconds: 0, @@ -7461,6 +7490,8 @@ mod tests { rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: std::sync::atomic::AtomicBool::new(true), has_master: false, pre_stop_seconds: 0, @@ -7572,6 +7603,271 @@ mod tests { ); } + fn make_commit_limited_service( + collection: &str, + max_parallel_commits: usize, + ) -> (VolumeGrpcService, TempDir) { + let (mut service, tmp) = make_local_service_with_volume(collection, None); + if max_parallel_commits > 0 { + Arc::get_mut(&mut service.state) + .unwrap() + .vacuum_commit_slots = Some(tokio::sync::Semaphore::new(max_parallel_commits)); + } + (service, tmp) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] + async fn test_vacuum_commit_slots_bound_concurrent_commits() { + async fn peak_holders(limit: usize) -> i32 { + let (service, _tmp) = make_commit_limited_service("bound_commits", limit); + let current = Arc::new(std::sync::atomic::AtomicI32::new(0)); + let peak = Arc::new(std::sync::atomic::AtomicI32::new(0)); + let mut workers = Vec::new(); + for _ in 0..8 { + let state = service.state.clone(); + let current = current.clone(); + let peak = peak.clone(); + workers.push(tokio::spawn(async move { + let Ok(Some(_permit)) = state.acquire_vacuum_commit_slot(None).await else { + return; + }; + let n = current.fetch_add(1, Ordering::Relaxed) + 1; + peak.fetch_max(n, Ordering::Relaxed); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + current.fetch_sub(1, Ordering::Relaxed); + })); + } + for w in workers { + w.await.unwrap(); + } + peak.load(Ordering::Relaxed) + } + assert_eq!(peak_holders(1).await, 1); + assert!(peak_holders(3).await <= 3); + } + + #[tokio::test] + async fn test_vacuum_commit_slots_unlimited_by_default() { + let (service, _tmp) = make_commit_limited_service("unlimited_commits", 0); + for i in 0..8 { + service + .state + .acquire_vacuum_commit_slot(Some( + tokio::time::Instant::now() + std::time::Duration::from_secs(1), + )) + .await + .unwrap_or_else(|e| { + panic!("acquisition {i} must not wait when no limit is configured: {e}") + }); + } + } + + #[tokio::test] + async fn test_vacuum_commit_slot_not_granted_to_expired_deadline() { + let (service, _tmp) = make_commit_limited_service("expired_commit", 1); + let slots = service.state.vacuum_commit_slots.as_ref().unwrap(); + let past = tokio::time::Instant::now() - std::time::Duration::from_secs(1); + for _ in 0..100 { + let err = service + .state + .acquire_vacuum_commit_slot(Some(past)) + .await + .expect_err("an expired deadline must not get a slot"); + assert_eq!(err.code(), tonic::Code::DeadlineExceeded); + assert_eq!( + slots.available_permits(), + 1, + "a refused acquisition must not keep the slot" + ); + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_vacuum_commit_deadline_while_queued() { + let (service, _tmp) = make_commit_limited_service("queued_deadline", 1); + let state = service.state.clone(); + let held = state + .vacuum_commit_slots + .as_ref() + .unwrap() + .acquire() + .await + .unwrap(); + + let mut request = + Request::new(volume_server_pb::VacuumVolumeCommitRequest { volume_id: 1 }); + request + .metadata_mut() + .insert("grpc-timeout", "20m".parse().unwrap()); + let err = service + .vacuum_volume_commit(request) + .await + .expect_err("the queued commit must expire at its deadline"); + assert_eq!(err.code(), tonic::Code::DeadlineExceeded); + assert_eq!( + state + .vacuum_commit_slots + .as_ref() + .unwrap() + .available_permits(), + 0, + "a refused commit must not consume the held slot" + ); + drop(held); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_queued_vacuum_commit_holds_no_store_lock() { + let (service, _tmp) = make_commit_limited_service("no_lock_commit", 1); + service + .state + .store + .write() + .unwrap() + .compact_volume(VolumeId(1), 0, 0, |_| true) + .unwrap(); + let state = service.state.clone(); + let held = state + .vacuum_commit_slots + .as_ref() + .unwrap() + .acquire() + .await + .unwrap(); + + let committed = tokio::spawn(async move { + service + .vacuum_volume_commit(Request::new(volume_server_pb::VacuumVolumeCommitRequest { + volume_id: 1, + })) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while state.vacuum_commits_waiting.load(Ordering::Relaxed) == 0 { + tokio::time::sleep(std::time::Duration::from_millis(1)).await; + } + }) + .await + .expect("the commit must pass the entry checks and wait for a slot"); + + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if let Ok(_guard) = state.store.try_read() { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(1)).await; + } + }) + .await + .expect("store read blocked behind a queued commit"); + + drop(held); + tokio::time::timeout(std::time::Duration::from_secs(5), committed) + .await + .expect("queued commit did not run after the slot freed") + .unwrap() + .unwrap(); + assert_eq!( + state + .vacuum_commit_slots + .as_ref() + .unwrap() + .available_permits(), + 1, + "the commit must return its slot" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_vacuum_commit_releases_slot_on_error() { + let (service, _tmp) = make_commit_limited_service("commit_err", 1); + service + .vacuum_volume_commit(Request::new(volume_server_pb::VacuumVolumeCommitRequest { + volume_id: 42, + })) + .await + .expect_err("committing an unknown volume must fail"); + assert_eq!( + service + .state + .vacuum_commit_slots + .as_ref() + .unwrap() + .available_permits(), + 1, + "a failed commit must return its slot" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_queued_vacuum_commit_does_not_start_in_maintenance() { + let (service, _tmp) = make_commit_limited_service("maint_commit", 1); + service + .state + .store + .write() + .unwrap() + .compact_volume(VolumeId(1), 0, 0, |_| true) + .unwrap(); + let revision_before = { + let store = service.state.store.read().unwrap(); + let (_, v) = store.find_volume(VolumeId(1)).unwrap(); + v.super_block.compaction_revision + }; + let state = service.state.clone(); + let held = state + .vacuum_commit_slots + .as_ref() + .unwrap() + .acquire() + .await + .unwrap(); + + let committed = tokio::spawn(async move { + service + .vacuum_volume_commit(Request::new(volume_server_pb::VacuumVolumeCommitRequest { + volume_id: 1, + })) + .await + }); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while state.vacuum_commits_waiting.load(Ordering::Relaxed) == 0 { + assert!( + std::time::Instant::now() < deadline, + "the commit must pass the entry maintenance check and wait for a slot" + ); + tokio::time::sleep(std::time::Duration::from_millis(1)).await; + } + + state.maintenance.store(true, Ordering::Relaxed); + drop(held); + + let err = tokio::time::timeout(std::time::Duration::from_secs(5), committed) + .await + .expect("queued commit did not return after the slot freed") + .unwrap() + .expect_err("a queued commit must not run in maintenance mode"); + assert!(err.message().contains("maintenance mode")); + let revision_after = { + let store = state.store.read().unwrap(); + let (_, v) = store.find_volume(VolumeId(1)).unwrap(); + v.super_block.compaction_revision + }; + assert_eq!( + revision_before, revision_after, + "the volume must not be committed" + ); + assert_eq!( + state + .vacuum_commit_slots + .as_ref() + .unwrap() + .available_permits(), + 1, + "the refused commit must return its slot" + ); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_volume_incremental_copy_streams_remote_only_volume_data() { let (service, _tmp, shutdown_tx, dat_bytes, super_block_size, _delete_count) = @@ -9628,6 +9924,8 @@ mod tests { rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: std::sync::atomic::AtomicBool::new(true), has_master: false, pre_stop_seconds: 0, @@ -11462,6 +11760,8 @@ mod tests { rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: std::sync::atomic::AtomicBool::new(true), has_master: false, pre_stop_seconds: 0, diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 2e56f3ed1..257f2a7cb 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -5285,6 +5285,8 @@ mod tests { rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: AtomicBool::new(false), has_master: false, pre_stop_seconds: 0, diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index cfa23c38f..2cff3bd9b 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -1485,6 +1485,8 @@ mod tests { rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: std::sync::atomic::AtomicBool::new(false), has_master: true, pre_stop_seconds: 0, diff --git a/seaweed-volume/src/server/profiling.rs b/seaweed-volume/src/server/profiling.rs index 1ede830c9..772ceecd1 100644 --- a/seaweed-volume/src/server/profiling.rs +++ b/seaweed-volume/src/server/profiling.rs @@ -136,6 +136,7 @@ mod tests { mem_profile: String::new(), compaction_byte_per_second: 0, maintenance_byte_per_second: 0, + max_parallel_vacuum_commits: 0, file_size_limit_bytes: 0, concurrent_upload_limit: 0, concurrent_download_limit: 0, diff --git a/seaweed-volume/src/server/volume_server.rs b/seaweed-volume/src/server/volume_server.rs index cf848df16..853e4d31d 100644 --- a/seaweed-volume/src/server/volume_server.rs +++ b/seaweed-volume/src/server/volume_server.rs @@ -63,6 +63,12 @@ pub struct VolumeServerState { pub file_size_limit_bytes: i64, /// Default IO rate limit for maintenance copy/replication work. pub maintenance_byte_per_second: i64, + /// Bounds concurrent vacuum commits; None means unlimited. Queued commits + /// wait on the semaphore, not on the store lock, so reads keep flowing. + /// Mirrors Go's `vacuumCommitSlots`. + pub vacuum_commit_slots: Option, + /// Vacuum commits currently waiting for a slot. + pub vacuum_commits_waiting: AtomicU32, /// Whether the server is connected to master (heartbeat active). pub is_heartbeating: AtomicBool, /// Whether master addresses are configured. @@ -144,6 +150,40 @@ impl VolumeServerState { Ok(()) } + /// Wait for a vacuum commit slot without touching the store lock; the wait + /// counts against the caller's `grpc-timeout`. An already-expired deadline + /// is refused even when a slot is free, matching Go's post-acquire + /// `ctx.Err()` check in `acquireVacuumCommitSlot`. + pub async fn acquire_vacuum_commit_slot( + &self, + deadline: Option, + ) -> Result>, tonic::Status> { + let Some(slots) = &self.vacuum_commit_slots else { + return Ok(None); + }; + self.vacuum_commits_waiting.fetch_add(1, Ordering::Relaxed); + let acquired = match deadline { + Some(d) => match tokio::time::timeout_at(d, slots.acquire()).await { + Ok(permit) => permit.map_err(|e| tonic::Status::internal(e.to_string())), + Err(_) => Err(tonic::Status::deadline_exceeded( + "timed out waiting for a vacuum commit slot", + )), + }, + None => slots + .acquire() + .await + .map_err(|e| tonic::Status::internal(e.to_string())), + }; + self.vacuum_commits_waiting.fetch_sub(1, Ordering::Relaxed); + let permit = acquired?; + if deadline.is_some_and(|d| tokio::time::Instant::now() >= d) { + return Err(tonic::Status::deadline_exceeded( + "timed out waiting for a vacuum commit slot", + )); + } + Ok(Some(permit)) + } + /// Build the seed master set from a list of raw `host:port[.grpcPort]` /// addresses, normalised the same way Go's `pb.ServerAddress.ToHttpAddress` /// does (drop the `.grpcPort` suffix, preserve everything else). diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs index 80389e802..18932cea0 100644 --- a/seaweed-volume/src/server/write_queue.rs +++ b/seaweed-volume/src/server/write_queue.rs @@ -206,6 +206,8 @@ mod tests { rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: AtomicBool::new(false), has_master: false, pre_stop_seconds: 0, diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs index 723461d4e..574b44be4 100644 --- a/seaweed-volume/tests/http_integration.rs +++ b/seaweed-volume/tests/http_integration.rs @@ -107,6 +107,8 @@ fn build_test_state( rack: String::new(), file_size_limit_bytes: 0, maintenance_byte_per_second: 0, + vacuum_commit_slots: None, + vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0), is_heartbeating: std::sync::atomic::AtomicBool::new(true), ec_decodes_in_flight: std::sync::Mutex::new(std::collections::HashSet::new()), ec_decode_tail: std::sync::Mutex::new(std::collections::HashSet::new()), diff --git a/weed/command/mini.go b/weed/command/mini.go index 183e1961f..99a525537 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -501,6 +501,7 @@ func initMiniVolumeFlags() { miniOptions.v.readMode = cmdMini.Flag.String("volume.readMode", "proxy", "[local|proxy|redirect] how to deal with non-local volume: 'not found|read in remote node|redirect volume location'.") miniOptions.v.compactionMBPerSecond = cmdMini.Flag.Int("volume.compactionMBps", 0, "limit compaction speed in mega bytes per second") miniOptions.v.maintenanceMBPerSecond = cmdMini.Flag.Int("volume.maintenanceMBps", 0, "limit maintenance IO rate in MB/s") + miniOptions.v.maxParallelVacuumCommits = cmdMini.Flag.Int("volume.compactionMaxParallelCommits", 0, "limit how many vacuum commits run at once on this server; queued commits hold no volume lock. 0 means unlimited") miniOptions.v.fileSizeLimitMB = cmdMini.Flag.Int("volume.fileSizeLimitMB", 256, "limit file size to avoid out of memory") miniOptions.v.ldbTimeout = cmdMini.Flag.Int64("volume.index.leveldbTimeout", 0, "alive time for leveldb") miniOptions.v.concurrentUploadLimitMB = cmdMini.Flag.Int("volume.concurrentUploadLimitMB", 0, "limit total concurrent upload size") diff --git a/weed/command/server.go b/weed/command/server.go index 22a3ae4f4..501534854 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -149,6 +149,7 @@ func init() { serverOptions.v.readMode = cmdServer.Flag.String("volume.readMode", "proxy", "[local|proxy|redirect] how to deal with non-local volume: 'not found|read in remote node|redirect volume location'.") serverOptions.v.compactionMBPerSecond = cmdServer.Flag.Int("volume.compactionMBps", 0, "limit compaction speed in mega bytes per second") serverOptions.v.maintenanceMBPerSecond = cmdServer.Flag.Int("volume.maintenanceMBps", 0, "limit maintenance (replication / balance) IO rate in MB/s. Unset is 0, no limitation.") + serverOptions.v.maxParallelVacuumCommits = cmdServer.Flag.Int("volume.compactionMaxParallelCommits", 0, "limit how many vacuum commits run at once on this server; queued commits hold no volume lock. 0 means unlimited") serverOptions.v.fileSizeLimitMB = cmdServer.Flag.Int("volume.fileSizeLimitMB", 256, "limit file size to avoid out of memory") serverOptions.v.ldbTimeout = cmdServer.Flag.Int64("volume.index.leveldbTimeout", 0, "alive time for leveldb (default to 0). If leveldb of volume is not accessed in ldbTimeout hours, it will be off loaded to reduce opened files and memory consumption.") serverOptions.v.concurrentUploadLimitMB = cmdServer.Flag.Int("volume.concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited") diff --git a/weed/command/volume.go b/weed/command/volume.go index 012222798..4646d9130 100644 --- a/weed/command/volume.go +++ b/weed/command/volume.go @@ -61,6 +61,7 @@ type VolumeServerOptions struct { memProfile *string compactionMBPerSecond *int maintenanceMBPerSecond *int + maxParallelVacuumCommits *int fileSizeLimitMB *int concurrentUploadLimitMB *int concurrentDownloadLimitMB *int @@ -215,6 +216,7 @@ func init() { v.memProfile = cmdVolume.Flag.String("memprofile", "", "memory profile output file") v.compactionMBPerSecond = cmdVolume.Flag.Int("compactionMBps", 0, "limit background compaction or copying speed in mega bytes per second") v.maintenanceMBPerSecond = cmdVolume.Flag.Int("maintenanceMBps", 0, "limit maintenance (replication / balance) IO rate in MB/s. Unset is 0, no limitation.") + v.maxParallelVacuumCommits = cmdVolume.Flag.Int("compactionMaxParallelCommits", 0, "limit how many vacuum commits run at once on this server; queued commits hold no volume lock. 0 means unlimited") v.fileSizeLimitMB = cmdVolume.Flag.Int("fileSizeLimitMB", 256, "limit file size to avoid out of memory") v.ldbTimeout = cmdVolume.Flag.Int64("index.leveldbTimeout", 0, "alive time for leveldb (default to 0). If leveldb of volume is not accessed in ldbTimeout hours, it will be off loaded to reduce opened files and memory consumption.") v.concurrentUploadLimitMB = cmdVolume.Flag.Int("concurrentUploadLimitMB", 0, "limit total concurrent upload size, 0 means unlimited") @@ -447,6 +449,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v *v.fixJpgOrientation, *v.readMode, *v.compactionMBPerSecond, *v.maintenanceMBPerSecond, + *v.maxParallelVacuumCommits, *v.fileSizeLimitMB, int64(*v.concurrentUploadLimitMB)*1024*1024, int64(*v.concurrentDownloadLimitMB)*1024*1024, diff --git a/weed/server/volume_grpc_vacuum.go b/weed/server/volume_grpc_vacuum.go index 968966164..5401a1108 100644 --- a/weed/server/volume_grpc_vacuum.go +++ b/weed/server/volume_grpc_vacuum.go @@ -14,6 +14,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "google.golang.org/grpc/status" ) var numCPU = runtime.NumCPU() @@ -96,6 +97,16 @@ func (vs *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_serv return nil, err } + release, err := vs.acquireVacuumCommitSlot(ctx) + if err != nil { + glog.Warningf("commit volume %d not started: %v", req.VolumeId, err) + return nil, status.FromContextError(err).Err() + } + defer release() + if err := vs.CheckMaintenanceMode(); err != nil { + return nil, err + } + start := time.Now() defer func(start time.Time) { stats.VolumeServerVacuumingHistogram.WithLabelValues("commit").Observe(time.Since(start).Seconds()) @@ -117,6 +128,31 @@ func (vs *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_serv } +// acquireVacuumCommitSlot waits for a commit slot holding no volume lock; the +// wait counts against the caller's deadline. +func (vs *VolumeServer) acquireVacuumCommitSlot(ctx context.Context) (release func(), err error) { + if vs.vacuumCommitSlots == nil { + return func() {}, nil + } + vs.vacuumCommitsWaiting.Add(1) + select { + case vs.vacuumCommitSlots <- struct{}{}: + case <-ctx.Done(): + err = ctx.Err() + } + vs.vacuumCommitsWaiting.Add(-1) + if err != nil { + return nil, err + } + release = func() { <-vs.vacuumCommitSlots } + // select picks randomly when the slot and the deadline are ready together. + if err := ctx.Err(); err != nil { + release() + return nil, err + } + return release, nil +} + func (vs *VolumeServer) VacuumVolumeCleanup(ctx context.Context, req *volume_server_pb.VacuumVolumeCleanupRequest) (*volume_server_pb.VacuumVolumeCleanupResponse, error) { if err := vs.checkGrpcAdminAuth(ctx); err != nil { return nil, err diff --git a/weed/server/volume_grpc_vacuum_test.go b/weed/server/volume_grpc_vacuum_test.go new file mode 100644 index 000000000..60c00e355 --- /dev/null +++ b/weed/server/volume_grpc_vacuum_test.go @@ -0,0 +1,225 @@ +package weed_server + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeVacuumCompactStream struct { + grpc.ServerStream + ctx context.Context +} + +func (s *fakeVacuumCompactStream) Context() context.Context { return s.ctx } + +func (s *fakeVacuumCompactStream) Send(*volume_server_pb.VacuumVolumeCompactResponse) error { + return nil +} + +func newVacuumCommitTestServer(t *testing.T, maxParallelCommits int, vids ...needle.VolumeId) *VolumeServer { + t.Helper() + store := newTraversalTestStore(t.TempDir()) + t.Cleanup(store.Close) + for _, vid := range vids { + require.NoError(t, store.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0)) + for i := 1; i <= 3; i++ { + n := new(needle.Needle) + n.Id = types.Uint64ToNeedleId(uint64(i)) + n.Cookie = 0x1234 + n.Data = []byte(fmt.Sprintf("volume-%d-needle-%d", vid, i)) + n.Checksum = needle.NewCRC(n.Data) + _, err := store.WriteVolumeNeedle(vid, n, true, false) + require.NoError(t, err) + } + } + vs := &VolumeServer{store: store} + if maxParallelCommits > 0 { + vs.vacuumCommitSlots = make(chan struct{}, maxParallelCommits) + } + return vs +} + +func compactForTest(vs *VolumeServer, vid needle.VolumeId) error { + stream := &fakeVacuumCompactStream{ctx: context.Background()} + return vs.VacuumVolumeCompact(&volume_server_pb.VacuumVolumeCompactRequest{VolumeId: uint32(vid)}, stream) +} + +func maxConcurrentHolders(vs *VolumeServer, workers int) int32 { + var current, peak int32 + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + release, err := vs.acquireVacuumCommitSlot(context.Background()) + if err != nil { + return + } + n := atomic.AddInt32(¤t, 1) + for { + p := atomic.LoadInt32(&peak) + if n <= p || atomic.CompareAndSwapInt32(&peak, p, n) { + break + } + } + time.Sleep(5 * time.Millisecond) + atomic.AddInt32(¤t, -1) + release() + }() + } + wg.Wait() + return atomic.LoadInt32(&peak) +} + +func TestVacuumCommitSlotsBoundConcurrentCommits(t *testing.T) { + assert.Equal(t, int32(1), maxConcurrentHolders(&VolumeServer{vacuumCommitSlots: make(chan struct{}, 1)}, 8)) + assert.LessOrEqual(t, maxConcurrentHolders(&VolumeServer{vacuumCommitSlots: make(chan struct{}, 3)}, 8), int32(3)) +} + +func TestVacuumCommitSlotsUnlimitedByDefault(t *testing.T) { + vs := &VolumeServer{} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + for i := 0; i < 8; i++ { + _, err := vs.acquireVacuumCommitSlot(ctx) + require.NoError(t, err, "acquisition %d must not wait when no limit is configured", i) + } +} + +func TestVacuumCommitSlotNotGrantedToExpiredContext(t *testing.T) { + vs := &VolumeServer{vacuumCommitSlots: make(chan struct{}, 1)} + ctx, cancel := context.WithCancel(context.Background()) + cancel() + for i := 0; i < 100; i++ { + release, err := vs.acquireVacuumCommitSlot(ctx) + require.ErrorIs(t, err, context.Canceled) + require.Nil(t, release) + require.Empty(t, vs.vacuumCommitSlots, "a refused acquisition must not keep the slot") + } +} + +func TestVacuumCommitLimitLeavesCompactionAndOtherServersAlone(t *testing.T) { + vids := []needle.VolumeId{1, 2, 3, 4} + vs := newVacuumCommitTestServer(t, 1, vids...) + other := newVacuumCommitTestServer(t, 1, 9) + vs.vacuumCommitSlots <- struct{}{} + + errs := make(chan error, len(vids)) + for _, vid := range vids { + go func(vid needle.VolumeId) { + errs <- compactForTest(vs, vid) + }(vid) + } + for range vids { + require.NoError(t, <-errs) + } + + require.NoError(t, compactForTest(other, 9)) + _, err := other.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{VolumeId: 9}) + require.NoError(t, err, "a busy commit slot on one server must not block another server") +} + +func TestQueuedVacuumCommitHoldsNoVolumeLockAndHonoursDeadline(t *testing.T) { + vid := needle.VolumeId(1) + vs := newVacuumCommitTestServer(t, 1, vid) + require.NoError(t, compactForTest(vs, vid)) + vs.vacuumCommitSlots <- struct{}{} + + committed := make(chan error, 1) + go func() { + _, err := vs.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{VolumeId: uint32(vid)}) + committed <- err + }() + require.Eventually(t, func() bool { return vs.vacuumCommitsWaiting.Load() == 1 }, 5*time.Second, time.Millisecond, + "the commit must pass the entry checks and wait for a slot") + + readDone := make(chan error, 1) + go func() { + n := &needle.Needle{Id: types.Uint64ToNeedleId(1), Cookie: 0x1234} + _, err := vs.store.ReadVolumeNeedle(vid, n, nil, nil) + readDone <- err + }() + select { + case err := <-readDone: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("read blocked behind a queued commit") + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := vs.VacuumVolumeCommit(ctx, &volume_server_pb.VacuumVolumeCommitRequest{VolumeId: uint32(vid)}) + assert.Equal(t, codes.DeadlineExceeded, status.Code(err)) + + select { + case err := <-committed: + t.Fatalf("commit ran while the only slot was taken: %v", err) + default: + } + + <-vs.vacuumCommitSlots + select { + case err := <-committed: + require.NoError(t, err) + case <-time.After(5 * time.Second): + t.Fatal("queued commit did not run after the slot freed") + } + assert.Empty(t, vs.vacuumCommitSlots, "the commit must return its slot") +} + +func TestVacuumCommitReleasesSlotOnError(t *testing.T) { + vs := newVacuumCommitTestServer(t, 1) + _, err := vs.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{VolumeId: 42}) + require.Error(t, err) + assert.Empty(t, vs.vacuumCommitSlots, "a failed commit must return its slot") + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + release, err := vs.acquireVacuumCommitSlot(ctx) + require.NoError(t, err) + release() +} + +func TestQueuedVacuumCommitDoesNotStartInMaintenanceMode(t *testing.T) { + vid := needle.VolumeId(1) + vs := newVacuumCommitTestServer(t, 1, vid) + require.NoError(t, compactForTest(vs, vid)) + revisionBefore := vs.store.GetVolume(vid).SuperBlock.CompactionRevision + vs.vacuumCommitSlots <- struct{}{} + + committed := make(chan error, 1) + go func() { + _, err := vs.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{VolumeId: uint32(vid)}) + committed <- err + }() + require.Eventually(t, func() bool { return vs.vacuumCommitsWaiting.Load() == 1 }, 5*time.Second, time.Millisecond, + "the commit must pass the entry maintenance check and wait for a slot") + + require.NoError(t, vs.store.State.Update(&volume_server_pb.VolumeServerState{Maintenance: true})) + <-vs.vacuumCommitSlots + + select { + case err := <-committed: + require.Error(t, err) + assert.Contains(t, err.Error(), "maintenance mode") + case <-time.After(5 * time.Second): + t.Fatal("queued commit did not return after the slot freed") + } + assert.Equal(t, revisionBefore, vs.store.GetVolume(vid).SuperBlock.CompactionRevision, "the volume must not be committed") + assert.Empty(t, vs.vacuumCommitSlots, "the refused commit must return its slot") +} diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index 737d48094..a544d4451 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -4,6 +4,7 @@ import ( "fmt" "net/http" "sync" + "sync/atomic" "time" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -55,6 +56,8 @@ type VolumeServer struct { AllowUntrustedRemoteEndpoints bool compactionBytePerSecond int64 maintenanceBytePerSecond int64 + vacuumCommitSlots chan struct{} + vacuumCommitsWaiting atomic.Int32 metricsAddress string metricsIntervalSec int fileSizeLimitBytes int64 @@ -79,6 +82,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, readMode string, compactionMBPerSecond int, maintenanceMBPerSecond int, + maxParallelVacuumCommits int, fileSizeLimitMB int, concurrentUploadLimit int64, concurrentDownloadLimit int64, @@ -126,6 +130,9 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, whiteList: whiteList, AllowUntrustedRemoteEndpoints: allowUntrustedRemoteEndpoints, } + if maxParallelVacuumCommits > 0 { + vs.vacuumCommitSlots = make(chan struct{}, maxParallelVacuumCommits) + } whiteList = append(whiteList, util.StringSplit(v.GetString("guard.white_list"), ",")...) // Copy the caller's slice so subsequent external mutation cannot desync