diff --git a/seaweed-volume/README.md b/seaweed-volume/README.md index 4321f70f3b..2798a0b4fe 100644 --- a/seaweed-volume/README.md +++ b/seaweed-volume/README.md @@ -68,7 +68,10 @@ seaweed-volume --port 8080 --master localhost:9333 --dir /data/vol1 --max 7 \ | `--pprof` | `false` | Enable pprof HTTP handlers | Set `RUST_LOG=debug` (or `trace`, `info`, `warn`) for log level control. -Set `SEAWEED_WRITE_QUEUE=1` to enable batched async write processing. + +Durable (`fsync=true`) uploads always go through per-volume batching workers +that share one `.dat`/`.idx` sync per batch, matching the Go volume server; +non-fsync writes take the synchronous path. ## Features diff --git a/seaweed-volume/src/config.rs b/seaweed-volume/src/config.rs index 21117725ca..a50e52e093 100644 --- a/seaweed-volume/src/config.rs +++ b/seaweed-volume/src/config.rs @@ -303,8 +303,6 @@ pub struct VolumeServerConfig { pub grpc_allowed_wildcard_domain: String, pub grpc_volume_allowed_common_names: Vec, pub tls_policy: TlsPolicy, - /// Enable batched write queue for improved throughput under load. - pub enable_write_queue: bool, /// Path to security.toml — stored for SIGHUP reload. pub security_file: String, /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation @@ -968,8 +966,6 @@ fn resolve_config_with_env(cli: Cli, env: EnvLookup<'_>) -> VolumeServerConfig { grpc_allowed_wildcard_domain: sec.grpc_allowed_wildcard_domain, grpc_volume_allowed_common_names: sec.grpc_volume_allowed_common_names, tls_policy: sec.tls_policy, - enable_write_queue: env_string(env, "SEAWEED_WRITE_QUEUE") - .is_some_and(|v| v == "1" || v == "true"), security_file: cli.security_file, allow_untrusted_remote_endpoints: cli.allow_untrusted_remote_endpoints, } diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index 3f785edb5c..c57a92e3b8 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -433,12 +433,11 @@ async fn run( } } - // Initialize the batched write queue if enabled - if config.enable_write_queue { - info!("Batched write queue enabled"); - let wq = WriteQueue::new(state.clone(), 128); - let _ = state.write_queue.set(wq); - } + // The write queue is always on: durable (fsync) uploads go through it so + // a batch shares one .dat/.idx sync, the way Go's per-volume async writer + // group-commits. Non-fsync writes stay on the synchronous path. + let wq = WriteQueue::new(state.clone(), 128); + let _ = state.write_queue.set(wq); // Set initial metric gauges for concurrent limits and max volumes metrics::CONCURRENT_UPLOAD_LIMIT.set(state.concurrent_upload_limit); diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index 71cc592ef1..45a0170744 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/seaweed-volume/src/server/grpc_client.rs @@ -431,7 +431,6 @@ mod tests { grpc_allowed_wildcard_domain: String::new(), grpc_volume_allowed_common_names: vec![], tls_policy: TlsPolicy::default(), - enable_write_queue: false, security_file: String::new(), } } diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 5a1ee705b8..737fdfafbc 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -2433,6 +2433,9 @@ fn is_zero_u32(v: &u32) -> bool { *v == 0 } +/// Store an uploaded needle on the local volume — durable (fsync) uploads +/// through the per-volume write queue, the rest synchronously — and then +/// replicate it if the volume's placement calls for more copies. pub async fn post_handler( State(state): State>, request: Request, @@ -2953,15 +2956,7 @@ pub async fn post_handler( let needle_etag = n.etag(); let needle_has_name = n.has_name(); - let write_result = if let Some(wq) = state.write_queue.get() { - wq.submit(vid, n, fsync).await - } else { - let vol = { - let store = state.store.read().unwrap(); - store.writable_volume(vid) - }; - vol.and_then(|vol| vol.write_needle(&mut n, true, fsync)) - }; + let write_result = write_upload_needle(&state, vid, n, fsync).await; // Replicate to remote volume servers if this volume has replicas. // Matches Go's GetWritableRemoteReplications: skip if copy_count == 1. @@ -3055,6 +3050,28 @@ pub async fn post_handler( resp } +/// Append an uploaded needle the way Go's `writeNeedle2` does: a durable +/// write goes to the volume's batching worker, which group-commits the +/// batch's .dat/.idx syncs, while anything else — a non-durable write, or a +/// durable one landing while the server is stopping and the workers are +/// winding down — takes the synchronous path and fsyncs inline. +async fn write_upload_needle( + state: &Arc, + vid: VolumeId, + mut n: Needle, + fsync: bool, +) -> Result<(u64, Size, bool), crate::storage::volume::VolumeError> { + let queued = fsync && !*state.is_stopping.read().unwrap(); + if queued && let Some(wq) = state.write_queue.get() { + return wq.submit(vid, n, fsync).await; + } + let vol = { + let store = state.store.read().unwrap(); + store.writable_volume(vid) + }; + vol.and_then(|vol| vol.write_needle(&mut n, true, fsync)) +} + // ============================================================================ // Delete Handler // ============================================================================ @@ -7188,4 +7205,52 @@ mod tests { "the vacuum moves the needle once the stream drains" ); } + + /// Issue #801: an fsync upload takes the per-volume worker so its sync is + /// shared across the batch (Go's `writeNeedle2` split); a non-fsync write, + /// or an fsync write landing while the server is stopping, takes the + /// synchronous path instead. + #[tokio::test] + async fn test_write_upload_needle_routes_fsync_to_queue() { + let tmp = tempfile::TempDir::new().unwrap(); + let state = volume_test_state(&tmp); + assert!( + state + .write_queue + .set(crate::server::write_queue::WriteQueue::new( + state.clone(), + 128, + )) + .is_ok() + ); + let queue = state.write_queue.get().unwrap(); + + let needle = |id: u64| { + let mut n = Needle { + id: NeedleId(id), + cookie: Cookie(TEST_COOKIE), + data: vec![id as u8; 8], + data_size: 8, + ..Needle::default() + }; + n.checksum = CRC::new(&n.data); + n + }; + + write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b01), true) + .await + .unwrap(); + assert_eq!(queue.submits_for_test(), 1); + + write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b02), false) + .await + .unwrap(); + assert_eq!(queue.submits_for_test(), 1); + + *state.is_stopping.write().unwrap() = true; + write_upload_needle(&state, VolumeId(1), needle(0x6e7a_0b03), true) + .await + .unwrap(); + assert_eq!(queue.submits_for_test(), 1); + } } diff --git a/seaweed-volume/src/server/profiling.rs b/seaweed-volume/src/server/profiling.rs index 772ceecd1f..9d5398b22c 100644 --- a/seaweed-volume/src/server/profiling.rs +++ b/seaweed-volume/src/server/profiling.rs @@ -170,7 +170,6 @@ mod tests { grpc_allowed_wildcard_domain: String::new(), grpc_volume_allowed_common_names: vec![], tls_policy: TlsPolicy::default(), - enable_write_queue: false, security_file: String::new(), } } diff --git a/seaweed-volume/src/server/volume_server.rs b/seaweed-volume/src/server/volume_server.rs index 853e4d31dc..a440c2a9e1 100644 --- a/seaweed-volume/src/server/volume_server.rs +++ b/seaweed-volume/src/server/volume_server.rs @@ -77,7 +77,9 @@ pub struct VolumeServerState { pub pre_stop_seconds: u32, /// Notify heartbeat to send an immediate update when volume state changes. pub volume_state_notify: tokio::sync::Notify, - /// Optional batched write queue for improved throughput under load. + /// Per-volume write workers that group-commit durable (fsync) uploads. + /// Set once at startup in production; a test state may leave it unset, + /// which sends every write down the synchronous path. pub write_queue: std::sync::OnceLock, /// Read mode: local, proxy, or redirect for non-local volumes. pub read_mode: ReadMode, diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs index 3ec7bd1ffa..01c4f69e72 100644 --- a/seaweed-volume/src/server/write_queue.rs +++ b/seaweed-volume/src/server/write_queue.rs @@ -1,7 +1,8 @@ //! Async batched write processing for the volume server. //! -//! Instead of each upload handler directly calling `write_needle`, writes are -//! submitted to a queue. Each volume gets its own FIFO worker (Go's +//! Durable (fsync) uploads are submitted here instead of each handler calling +//! `write_needle` directly — Go's `writeNeedle2` makes the same split, so the +//! queue is always constructed. Each volume gets its own FIFO worker (Go's //! volume_write.go keeps one batching worker per volume), which drains its //! channel in batches of up to 128 entries or 4 MiB and shares the .dat/.idx //! flushes across durable writes (see `Volume::write_needles_grouped`). @@ -66,6 +67,9 @@ pub struct WriteQueue { /// alone would multiply the old single queue's capacity by volume count. queue_permits: Arc, workers: Arc>>>, + /// Submissions accepted, for tests asserting which writes route here. + #[cfg(test)] + submits: Arc, } impl WriteQueue { @@ -78,9 +82,17 @@ impl WriteQueue { io_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_VOLUME_WRITES)), queue_permits: Arc::new(Semaphore::new(capacity)), workers: Arc::new(Mutex::new(HashMap::new())), + #[cfg(test)] + submits: Arc::new(std::sync::atomic::AtomicU64::new(0)), } } + /// How many writes were submitted, for tests asserting upload routing. + #[cfg(test)] + pub fn submits_for_test(&self) -> u64 { + self.submits.load(std::sync::atomic::Ordering::Relaxed) + } + /// Sender for the volume's worker, spawning one on first use. fn worker_for(&self, volume_id: VolumeId) -> mpsc::Sender { let mut workers = self.workers.lock().unwrap(); @@ -110,6 +122,9 @@ impl WriteQueue { pub async fn submit(&self, volume_id: VolumeId, needle: Needle, fsync: bool) -> WriteResult { // Workers spawn only for mounted volumes — a bogus id never creates // a channel, so volume-id sprays can't grow the worker map. + #[cfg(test)] + self.submits + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); { let store = self.state.store.read().unwrap(); if store.find_volume(volume_id).is_none() {