diff --git a/seaweed-volume/Cargo.toml b/seaweed-volume/Cargo.toml index 1e4a58f2a2..29c37d4ff3 100644 --- a/seaweed-volume/Cargo.toml +++ b/seaweed-volume/Cargo.toml @@ -157,6 +157,8 @@ windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } [dev-dependencies] tempfile = "3" +# Paused-time tests drive the write-queue batch window deterministically. +tokio = { version = "1", features = ["full", "test-util"] } # Already a transitive dependency of aws-sdk-s3 at a single locked version; # needed directly only for the canned HttpClient in remote_storage::s3 tests. aws-smithy-runtime-api = "1" diff --git a/seaweed-volume/src/config.rs b/seaweed-volume/src/config.rs index a50e52e093..cf73731cc1 100644 --- a/seaweed-volume/src/config.rs +++ b/seaweed-volume/src/config.rs @@ -233,6 +233,25 @@ pub struct Cli { #[arg(long = "volume.allowUntrustedRemoteEndpoints", default_value_t = false)] pub allow_untrusted_remote_endpoints: bool, + /// Minimum bytes an fsync batch gathers before a drained queue flushes + /// without waiting. + #[arg(long = "fsyncMinBatchBytes", default_value_t = 1024 * 1024)] + pub fsync_min_batch_bytes: u64, + + /// Minimum requests an fsync batch gathers before a drained queue flushes + /// without waiting. + #[arg(long = "fsyncMinBatchRequests", default_value_t = 128)] + pub fsync_min_batch_requests: usize, + + /// How long a below-minimum fsync batch waits for more requests once the + /// queue drains; 0 flushes immediately. + #[arg( + long = "fsyncBatchWait", + default_value = "500us", + value_parser = parse_go_duration + )] + pub fsync_batch_wait: std::time::Duration, + /// A file of command line options, each line in optionName=optionValue format. #[arg(long = "options", default_value = "")] pub options: String, @@ -308,6 +327,14 @@ pub struct VolumeServerConfig { /// If true, FetchAndWriteNeedle skips remote S3 endpoint validation /// (allows loopback / link-local / metadata hosts). pub allow_untrusted_remote_endpoints: bool, + /// Minimum bytes an fsync batch gathers before a drained queue flushes + /// without waiting. + pub fsync_min_batch_bytes: u64, + /// Minimum requests an fsync batch gathers before a drained queue flushes + /// without waiting. + pub fsync_min_batch_requests: usize, + /// How long a below-minimum fsync batch waits for stragglers. + pub fsync_batch_wait: std::time::Duration, } pub use crate::storage::needle_map::NeedleMapKind; @@ -938,6 +965,9 @@ fn resolve_config_with_env(cli: Cli, env: EnvLookup<'_>) -> VolumeServerConfig { concurrent_download_limit: cli.concurrent_download_limit_mb as i64 * 1024 * 1024, inflight_upload_data_timeout: cli.inflight_upload_data_timeout, inflight_download_data_timeout: cli.inflight_download_data_timeout, + fsync_min_batch_bytes: cli.fsync_min_batch_bytes, + fsync_min_batch_requests: cli.fsync_min_batch_requests, + fsync_batch_wait: cli.fsync_batch_wait, has_slow_read: cli.has_slow_read, read_buffer_size_mb: cli.read_buffer_size_mb, ldb_timeout: cli.ldb_timeout, diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs index c57a92e3b8..46fe45289d 100644 --- a/seaweed-volume/src/main.rs +++ b/seaweed-volume/src/main.rs @@ -23,7 +23,7 @@ use seaweed_volume::server::request_id::GrpcRequestIdLayer; use seaweed_volume::server::volume_server::{ RuntimeMetricsConfig, VolumeServerState, build_metrics_router, }; -use seaweed_volume::server::write_queue::WriteQueue; +use seaweed_volume::server::write_queue::{self, WriteQueue}; use seaweed_volume::storage::store::Store; use seaweed_volume::storage::types::DiskType; @@ -436,7 +436,17 @@ async fn run( // 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 min_bytes = usize::try_from(config.fsync_min_batch_bytes) + .map_err(|_| "fsyncMinBatchBytes exceeds this platform's maximum usize value")?; + let wq = WriteQueue::new( + state.clone(), + 128, + write_queue::SyncBatchOption { + min_bytes, + min_requests: config.fsync_min_batch_requests, + wait: config.fsync_batch_wait, + }, + ); let _ = state.write_queue.set(wq); // Set initial metric gauges for concurrent limits and max volumes diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index 45a0170744..9567e43658 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/seaweed-volume/src/server/grpc_client.rs @@ -393,6 +393,9 @@ mod tests { fix_jpg_orientation: false, read_mode: ReadMode::Local, allow_untrusted_remote_endpoints: false, + fsync_min_batch_bytes: 1024 * 1024, + fsync_min_batch_requests: 128, + fsync_batch_wait: std::time::Duration::from_micros(500), cpu_profile: String::new(), mem_profile: String::new(), compaction_byte_per_second: 0, diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 6accfed9fe..fdba467c7c 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -7308,6 +7308,7 @@ mod tests { .set(crate::server::write_queue::WriteQueue::new( state.clone(), 128, + crate::server::write_queue::SyncBatchOption::default(), )) .is_ok() ); diff --git a/seaweed-volume/src/server/profiling.rs b/seaweed-volume/src/server/profiling.rs index 9d5398b22c..5622e9d428 100644 --- a/seaweed-volume/src/server/profiling.rs +++ b/seaweed-volume/src/server/profiling.rs @@ -132,6 +132,9 @@ mod tests { fix_jpg_orientation: false, read_mode: ReadMode::Local, allow_untrusted_remote_endpoints: false, + fsync_min_batch_bytes: 1024 * 1024, + fsync_min_batch_requests: 128, + fsync_batch_wait: std::time::Duration::from_micros(500), cpu_profile: String::new(), mem_profile: String::new(), compaction_byte_per_second: 0, diff --git a/seaweed-volume/src/server/write_queue.rs b/seaweed-volume/src/server/write_queue.rs index 01c4f69e72..dc649e0d72 100644 --- a/seaweed-volume/src/server/write_queue.rs +++ b/seaweed-volume/src/server/write_queue.rs @@ -56,6 +56,27 @@ const MAX_CONCURRENT_VOLUME_WRITES: usize = 8; /// retirement is the volume leaving the store. const IDLE_WORKER_POLL: std::time::Duration = std::time::Duration::from_secs(60); +/// Tunes the per-volume durable-write worker. Once the queue drains with the +/// batch still below min_bytes and min_requests, the batch stays open up to +/// wait for stragglers instead of paying an fsync per request. Mirrors Go's +/// storage.SyncBatchOption. +#[derive(Clone, Copy)] +pub struct SyncBatchOption { + pub min_bytes: usize, + pub min_requests: usize, + pub wait: std::time::Duration, +} + +impl Default for SyncBatchOption { + fn default() -> Self { + SyncBatchOption { + min_bytes: 1024 * 1024, + min_requests: MAX_BATCH_SIZE, + wait: std::time::Duration::from_micros(500), + } + } +} + /// Handle for submitting write requests to the per-volume workers. #[derive(Clone)] pub struct WriteQueue { @@ -67,23 +88,35 @@ pub struct WriteQueue { /// alone would multiply the old single queue's capacity by volume count. queue_permits: Arc, workers: Arc>>>, + sync_batch: SyncBatchOption, /// Submissions accepted, for tests asserting which writes route here. #[cfg(test)] submits: Arc, + /// Times a worker armed the straggler window, for tests that must land a + /// write inside an already-open batch. + #[cfg(test)] + windows_armed: Arc, } impl WriteQueue { /// Create a new write queue. Per-volume workers are spawned lazily on the /// first submission for that volume. - pub fn new(state: Arc, capacity: usize) -> Self { + pub fn new( + state: Arc, + capacity: usize, + sync_batch: SyncBatchOption, + ) -> Self { WriteQueue { state, capacity, io_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_VOLUME_WRITES)), queue_permits: Arc::new(Semaphore::new(capacity)), workers: Arc::new(Mutex::new(HashMap::new())), + sync_batch, #[cfg(test)] submits: Arc::new(std::sync::atomic::AtomicU64::new(0)), + #[cfg(test)] + windows_armed: Arc::new(std::sync::atomic::AtomicU64::new(0)), } } @@ -93,6 +126,12 @@ impl WriteQueue { self.submits.load(std::sync::atomic::Ordering::Relaxed) } + /// How many batch windows workers have armed so far. + #[cfg(test)] + fn windows_armed_for_test(&self) -> u64 { + self.windows_armed.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(); @@ -107,6 +146,9 @@ impl WriteQueue { state: self.state.clone(), io_permits: self.io_permits.clone(), workers: self.workers.clone(), + sync_batch: self.sync_batch, + #[cfg(test)] + windows_armed: self.windows_armed.clone(), } .run(), ); @@ -191,6 +233,9 @@ struct VolumeWriteWorker { state: Arc, io_permits: Arc, workers: Arc>>>, + sync_batch: SyncBatchOption, + #[cfg(test)] + windows_armed: Arc, } impl VolumeWriteWorker { @@ -238,14 +283,40 @@ impl VolumeWriteWorker { let mut batch = Vec::with_capacity(MAX_BATCH_SIZE); let mut batch_bytes: usize = first.needle.data.len(); batch.push(first); + // Armed once the queue first drains below the minimum batch; holds the + // batch open for stragglers so the wait bound is per batch, not per + // lull (Go's batchWait). + let mut window_deadline = None; while batch.len() < MAX_BATCH_SIZE && batch_bytes < MAX_BATCH_BYTES { - match self.rx.try_recv() { - Ok(req) => { - batch_bytes += req.needle.data.len(); - batch.push(req); - } - Err(_) => break, + if window_deadline.is_some_and(|d| tokio::time::Instant::now() >= d) { + break; } + let req = match self.rx.try_recv() { + Ok(req) => req, + Err(mpsc::error::TryRecvError::Empty) => { + let sync = self.sync_batch; + if sync.wait.is_zero() + || batch_bytes >= sync.min_bytes + || batch.len() >= sync.min_requests + { + // a drained queue flushes right away once it holds enough + break; + } + let deadline = *window_deadline.get_or_insert_with(|| { + #[cfg(test)] + self.windows_armed + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + tokio::time::Instant::now() + sync.wait + }); + match tokio::time::timeout_at(deadline, self.rx.recv()).await { + Ok(Some(req)) => req, + Ok(None) | Err(_) => break, + } + } + Err(mpsc::error::TryRecvError::Disconnected) => break, + }; + batch_bytes += req.needle.data.len(); + batch.push(req); } debug!("processing write batch of {} requests", batch.len()); @@ -380,7 +451,7 @@ mod tests { async fn test_write_queue_submit_no_volume() { // Submit a write to a non-existent volume -- should return VolumeError::NotFound. let state = make_test_state(); - let queue = WriteQueue::new(state, MAX_BATCH_SIZE); + let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default()); let needle = Needle { id: 1.into(), @@ -402,7 +473,7 @@ mod tests { async fn test_write_queue_concurrent_submissions() { // Submit multiple concurrent writes -- all should complete (with errors since no volume). let state = make_test_state(); - let queue = WriteQueue::new(state, MAX_BATCH_SIZE); + let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default()); let mut handles = Vec::new(); for i in 0..10u64 { @@ -430,7 +501,7 @@ mod tests { async fn test_write_queue_batching() { // Verify that many concurrent writes get processed (testing the batching path). let state = make_test_state(); - let queue = WriteQueue::new(state, MAX_BATCH_SIZE); + let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default()); // Submit MAX_BATCH_SIZE requests concurrently let mut handles = Vec::new(); @@ -522,7 +593,7 @@ mod tests { async fn test_write_queue_dropped_sender() { // When the queue is dropped, subsequent submits should fail gracefully. let state = make_test_state(); - let queue = WriteQueue::new(state, 1); + let queue = WriteQueue::new(state, 1, SyncBatchOption::default()); // Clone then drop the original -- the worker keeps running via its rx handle. let queue2 = queue.clone(); @@ -665,7 +736,7 @@ mod tests { .unwrap(); } } - let queue = WriteQueue::new(state, MAX_BATCH_SIZE); + let queue = WriteQueue::new(state, MAX_BATCH_SIZE, SyncBatchOption::default()); let in_append = Arc::new(Barrier::new(2)); let release = Arc::new(Barrier::new(2)); @@ -734,4 +805,85 @@ mod tests { assert!(a1.await.unwrap().is_ok()); assert!(a2.await.unwrap().is_ok()); } + + /// A durable write below the minimum batch rides out its window so a + /// straggler joins the same sync instead of each paying an fsync alone — + /// Go's batchWait. Paused time makes the hold observable: a lone write can + /// only complete once the runtime auto-advances to the window's deadline. + #[tokio::test(start_paused = true)] + async fn test_batch_window_defers_flush_for_stragglers() { + use crate::config::MinFreeSpace; + use crate::storage::types::{DiskType, NeedleId}; + use crate::storage::volume::VolumeSpec; + use std::time::Duration; + + let tmp = tempfile::TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let state = make_test_state(); + { + let mut store = state.store.write().unwrap(); + store + .add_location( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + store + .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default()) + .unwrap(); + } + + let queue = WriteQueue::new( + state, + MAX_BATCH_SIZE, + SyncBatchOption { + min_bytes: usize::MAX, + min_requests: MAX_BATCH_SIZE, + wait: Duration::from_secs(30), + }, + ); + + let needle = |id: u64| Needle { + id: NeedleId(id), + cookie: 0x1111.into(), + data: vec![id as u8; 8], + data_size: 8, + ..Needle::default() + }; + + let start = tokio::time::Instant::now(); + queue.submit(VolumeId(1), needle(1), true).await.unwrap(); + assert!(start.elapsed() >= Duration::from_secs(30)); + + // Wait for the worker to take the second write and open its batch + // window, so the straggler provably lands inside an open batch + // instead of arriving during the initial drain. + let armed = queue.windows_armed_for_test(); + let second = tokio::spawn({ + let queue = queue.clone(); + async move { queue.submit(VolumeId(1), needle(2), true).await } + }); + for _ in 0..10_000 { + if queue.windows_armed_for_test() > armed { + break; + } + tokio::task::yield_now().await; + } + assert!(queue.windows_armed_for_test() > armed); + let third = tokio::spawn({ + let queue = queue.clone(); + async move { queue.submit(VolumeId(1), needle(3), true).await } + }); + let (second, third) = tokio::join!(second, third); + second.unwrap().unwrap(); + third.unwrap().unwrap(); + // One shared window ends both writes ~30s after their batch armed; + // back-to-back windows would keep the third out to ~90s. + let elapsed = start.elapsed(); + assert!(elapsed >= Duration::from_secs(60) && elapsed < Duration::from_secs(90)); + } } diff --git a/seaweed-volume/tests/http_integration.rs b/seaweed-volume/tests/http_integration.rs index 7622fab739..b90234a5ad 100644 --- a/seaweed-volume/tests/http_integration.rs +++ b/seaweed-volume/tests/http_integration.rs @@ -1340,11 +1340,11 @@ async fn non_ascii_fid_and_ttl_are_rejected_not_panicked() { #[tokio::test] async fn write_queue_upload_returns_same_etag_as_direct_write() { - use seaweed_volume::server::write_queue::WriteQueue; + use seaweed_volume::server::write_queue::{SyncBatchOption, WriteQueue}; let (direct_state, _direct_tmp) = test_state(); let (queued_state, _queued_tmp) = test_state(); - let wq = WriteQueue::new(queued_state.clone(), 128); + let wq = WriteQueue::new(queued_state.clone(), 128, SyncBatchOption::default()); let _ = queued_state.write_queue.set(wq); let compressible = "seaweedfs ".repeat(200).into_bytes(); diff --git a/weed/command/mini.go b/weed/command/mini.go index 0d2660438e..7cc848d3b7 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -517,6 +517,9 @@ func initMiniVolumeFlags() { miniOptions.v.hasSlowRead = cmdMini.Flag.Bool("volume.hasSlowRead", true, "if true, prevents slow reads from blocking other requests") miniOptions.v.readBufferSizeMB = cmdMini.Flag.Int("volume.readBufferSizeMB", 4, "read buffer size in MB") miniOptions.v.allowUntrustedRemoteEndpoints = cmdMini.Flag.Bool("volume.allowUntrustedRemoteEndpoints", false, "if true, FetchAndWriteNeedle accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints.") + miniOptions.v.fsyncMinBatchBytes = cmdMini.Flag.Int64("volume.fsyncMinBatchBytes", 1*1024*1024, "minimum bytes an fsync batch gathers before a drained queue flushes without waiting") + miniOptions.v.fsyncMinBatchRequests = cmdMini.Flag.Int("volume.fsyncMinBatchRequests", 128, "minimum requests an fsync batch gathers before a drained queue flushes without waiting") + miniOptions.v.fsyncBatchWait = cmdMini.Flag.Duration("volume.fsyncBatchWait", 500*time.Microsecond, "how long a below-minimum fsync batch waits for more requests once the queue drains; 0 flushes immediately") miniOptions.v.preStopSeconds = cmdMini.Flag.Int("volume.preStopSeconds", 1, "number of seconds between stop send heartbeats and stop volume server (default: 1 for mini)") miniOptions.v.setDiskIOProbeDefaults() } diff --git a/weed/command/server.go b/weed/command/server.go index 5e41c6e3bb..f64f3b8973 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -165,6 +165,9 @@ func init() { serverOptions.v.hasSlowRead = cmdServer.Flag.Bool("volume.hasSlowRead", true, " if true, this prevents slow reads from blocking other requests, but large file read P99 latency will increase.") serverOptions.v.readBufferSizeMB = cmdServer.Flag.Int("volume.readBufferSizeMB", 4, " larger values can optimize query performance but will increase some memory usage,Use with hasSlowRead normally") serverOptions.v.allowUntrustedRemoteEndpoints = cmdServer.Flag.Bool("volume.allowUntrustedRemoteEndpoints", false, "if true, FetchAndWriteNeedle accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints.") + serverOptions.v.fsyncMinBatchBytes = cmdServer.Flag.Int64("volume.fsyncMinBatchBytes", 1*1024*1024, "minimum bytes an fsync batch gathers before a drained queue flushes without waiting") + serverOptions.v.fsyncMinBatchRequests = cmdServer.Flag.Int("volume.fsyncMinBatchRequests", 128, "minimum requests an fsync batch gathers before a drained queue flushes without waiting") + serverOptions.v.fsyncBatchWait = cmdServer.Flag.Duration("volume.fsyncBatchWait", 500*time.Microsecond, "how long a below-minimum fsync batch waits for more requests once the queue drains; 0 flushes immediately") serverOptions.v.setDiskIOProbeDefaults() s3Options.port = cmdServer.Flag.Int("s3.port", 8333, "s3 server http listen port") diff --git a/weed/command/volume.go b/weed/command/volume.go index e3244bf473..95b76409c1 100644 --- a/weed/command/volume.go +++ b/weed/command/volume.go @@ -76,6 +76,9 @@ type VolumeServerOptions struct { readBufferSizeMB *int ldbTimeout *int64 allowUntrustedRemoteEndpoints *bool + fsyncMinBatchBytes *int64 + fsyncMinBatchRequests *int + fsyncBatchWait *time.Duration debug *bool debugPort *int diskIOProbe *bool @@ -147,6 +150,22 @@ func (v *VolumeServerOptions) setDiskIOProbeDefaults() { } } +// syncBatchOption resolves the fsync batch window, defaulting each field so +// commands or embedders that leave the option pointers unset still start. +func (v *VolumeServerOptions) syncBatchOption() storage.SyncBatchOption { + option := storage.DefaultSyncBatchOption + if v.fsyncMinBatchBytes != nil { + option.MinBytes = *v.fsyncMinBatchBytes + } + if v.fsyncMinBatchRequests != nil { + option.MinRequests = *v.fsyncMinBatchRequests + } + if v.fsyncBatchWait != nil { + option.Wait = *v.fsyncBatchWait + } + return option +} + func (v *VolumeServerOptions) applyDiskIOProbeConfig() { v.setDiskIOProbeDefaults() @@ -230,6 +249,9 @@ func init() { v.hasSlowRead = cmdVolume.Flag.Bool("hasSlowRead", true, " if true, this prevents slow reads from blocking other requests, but large file read P99 latency will increase.") v.readBufferSizeMB = cmdVolume.Flag.Int("readBufferSizeMB", 4, " larger values can optimize query performance but will increase some memory usage,Use with hasSlowRead normally.") v.allowUntrustedRemoteEndpoints = cmdVolume.Flag.Bool("volume.allowUntrustedRemoteEndpoints", false, "if true, FetchAndWriteNeedle accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints.") + v.fsyncMinBatchBytes = cmdVolume.Flag.Int64("fsyncMinBatchBytes", 1*1024*1024, "minimum bytes an fsync batch gathers before a drained queue flushes without waiting") + v.fsyncMinBatchRequests = cmdVolume.Flag.Int("fsyncMinBatchRequests", 128, "minimum requests an fsync batch gathers before a drained queue flushes without waiting") + v.fsyncBatchWait = cmdVolume.Flag.Duration("fsyncBatchWait", 500*time.Microsecond, "how long a below-minimum fsync batch waits for more requests once the queue drains; 0 flushes immediately") v.debug = cmdVolume.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port") v.debugPort = cmdVolume.Flag.Int("debug.port", 6060, "http port for debugging") v.setDiskIOProbeDefaults() @@ -460,6 +482,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v *v.ldbTimeout, *v.allowUntrustedRemoteEndpoints, diskProbeConfig, + v.syncBatchOption(), ) // starting grpc server grpcS := v.startGrpcService(volumeServer) diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index ff1c85d044..93351eca44 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -93,6 +93,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, ldbTimeout int64, allowUntrustedRemoteEndpoints bool, diskProbeConfig stats.DiskIOProbeConfig, + syncBatch storage.SyncBatchOption, ) *VolumeServer { v := util.GetViper() @@ -149,6 +150,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout, diskProbeConfig) vs.store.MasterFn = vs.getCurrentMaster + vs.store.SetSyncBatchOption(syncBatch) vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec) handleStaticResources(adminMux) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index ce76412a76..e904bae7dc 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -58,6 +58,7 @@ type DiskLocation struct { isDiskSpaceLow atomic.Bool isDiskUnavailable atomic.Bool + syncBatch SyncBatchOption closeCh chan struct{} } @@ -119,6 +120,7 @@ func NewDiskLocation(dir string, maxVolumeCount int32, minFreeSpace util.MinFree MaxVolumeCount: maxVolumeCount, OriginalMaxVolumeCount: maxVolumeCount, MinFreeSpace: minFreeSpace, + syncBatch: DefaultSyncBatchOption, } location.volumes = make(map[needle.VolumeId]*Volume) location.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume) diff --git a/weed/storage/store.go b/weed/storage/store.go index 0d2b51b50d..d213ee6f9e 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -1359,6 +1359,14 @@ func (s *Store) GetVolumeSizeLimit() uint64 { return s.volumeSizeLimit.Load() } +// SetSyncBatchOption applies the durable-write batch window to every disk. +// It runs at startup, before any volume takes a write. +func (s *Store) SetSyncBatchOption(syncBatch SyncBatchOption) { + for _, location := range s.Locations { + location.syncBatch = syncBatch + } +} + func (s *Store) SetPreallocate(x bool) { s.preallocate.Store(x) } diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 88072c9aa3..f119425fff 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "os" + "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/backend" @@ -661,6 +662,29 @@ func (v *Volume) processBatchRun(run []*needle.AsyncRequest) { } } +const ( + syncBatchMaxBytes = 4 * 1024 * 1024 + syncBatchMaxRequests = 128 +) + +// SyncBatchOption tunes the volume's durable-write worker. Once the queue +// drains with the batch still below MinBytes and MinRequests, the batch stays +// open up to Wait for stragglers instead of paying an fsync per request. +type SyncBatchOption struct { + MinBytes int64 + MinRequests int + Wait time.Duration +} + +var DefaultSyncBatchOption = SyncBatchOption{ + MinBytes: 1 * 1024 * 1024, + MinRequests: syncBatchMaxRequests, + Wait: 500 * time.Microsecond, +} + +// syncBatchAfter arms the batch window; a var so tests can drive it. +var syncBatchAfter = time.After + // startWorker returns the volume's batch-write channel, creating it and its // goroutine on first use, and nil once stopWorker has run. func (v *Volume) startWorker() chan *needle.AsyncRequest { @@ -672,8 +696,12 @@ func (v *Volume) startWorker() chan *needle.AsyncRequest { if v.asyncRequestsChan != nil { return v.asyncRequestsChan } - requests := make(chan *needle.AsyncRequest, 128) + requests := make(chan *needle.AsyncRequest, syncBatchMaxRequests) v.asyncRequestsChan = requests + syncBatch := DefaultSyncBatchOption + if v.location != nil { + syncBatch = v.location.syncBatch + } go func() { chanClosed := false for { @@ -681,15 +709,25 @@ func (v *Volume) startWorker() chan *needle.AsyncRequest { if chanClosed { break } - currentRequests := make([]*needle.AsyncRequest, 0, 128) + currentRequests := make([]*needle.AsyncRequest, 0, syncBatchMaxRequests) currentBytesToWrite := int64(0) + var batchDeadline time.Time + var batchWait <-chan time.Time for { - request, ok := <-requests + var request *needle.AsyncRequest + queueOpen := true + select { + case request, queueOpen = <-requests: + case <-batchWait: + } // volume may be closed - if !ok { + if !queueOpen { chanClosed = true break } + if request == nil { + break + } if MaxPossibleVolumeSize < v.ContentSize()+uint64(currentBytesToWrite+request.ActualSize) { request.Complete(0, 0, false, fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.ContentSize())) @@ -698,10 +736,27 @@ func (v *Volume) startWorker() chan *needle.AsyncRequest { currentRequests = append(currentRequests, request) currentBytesToWrite += request.ActualSize // submit at most 4M bytes or 128 requests at one time to decrease request delay. - // it also need to break if there is no data in channel to avoid io hang. - if currentBytesToWrite >= 4*1024*1024 || len(currentRequests) >= 128 || len(requests) == 0 { + if currentBytesToWrite >= syncBatchMaxBytes || len(currentRequests) >= syncBatchMaxRequests { break } + // a refilled queue must not outwait the armed deadline + if !batchDeadline.IsZero() && !time.Now().Before(batchDeadline) { + break + } + if len(requests) > 0 { + continue + } + if syncBatch.Wait <= 0 || currentBytesToWrite >= syncBatch.MinBytes || len(currentRequests) >= syncBatch.MinRequests { + // a drained queue flushes right away once it holds enough + break + } + // flushing now would sync a singleton batch; hold the window + // open for stragglers, armed once per batch so the added + // latency stays bounded by Wait + if batchWait == nil { + batchDeadline = time.Now().Add(syncBatch.Wait) + batchWait = syncBatchAfter(syncBatch.Wait) + } } if len(currentRequests) == 0 { continue diff --git a/weed/storage/volume_write_test.go b/weed/storage/volume_write_test.go index d35cdda2b2..2fa479067b 100644 --- a/weed/storage/volume_write_test.go +++ b/weed/storage/volume_write_test.go @@ -6,12 +6,14 @@ import ( "fmt" "math" "os" + "sync/atomic" "testing" "time" "github.com/stretchr/testify/assert" "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/types" @@ -497,3 +499,85 @@ func TestDoWriteRequestRefusesReadOnlyBeforeAppend(t *testing.T) { t.Fatalf("idx size changed; needle must not index") } } + +type syncCountingBackend struct { + backend.BackendStorageFile + syncs atomic.Int64 +} + +func (b *syncCountingBackend) Sync() error { + b.syncs.Add(1) + return b.BackendStorageFile.Sync() +} + +// A drained queue used to flush whatever it held, turning durable writes into +// one fsync per request at low concurrency. With a batch window configured, a +// straggler that lands inside the window joins the same sync. +func TestSyncBatchDefersFlushForStragglers(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + counting := &syncCountingBackend{BackendStorageFile: v.DataBackend} + v.DataBackend = counting + writeBytes := int64(needle.GetActualSize(16, v.Version())) + v.location = &DiskLocation{syncBatch: SyncBatchOption{ + MinBytes: writeBytes * 2, + MinRequests: syncBatchMaxRequests, + Wait: time.Minute, + }} + + armed := make(chan chan time.Time, 4) + prevAfter := syncBatchAfter + syncBatchAfter = func(time.Duration) <-chan time.Time { + window := make(chan time.Time, 1) + armed <- window + return window + } + defer func() { syncBatchAfter = prevAfter }() + + done := make(chan error, 3) + writeAsync := func(id uint64) { + n := new(needle.Needle) + n.Data = make([]byte, 16) + n.Checksum = needle.NewCRC(n.Data) + n.Id = types.Uint64ToNeedleId(id) + _, _, _, writeErr := v.writeNeedle2(n, true, true, false) + done <- writeErr + } + windowFor := func(id uint64) chan time.Time { + go writeAsync(id) + window := <-armed + select { + case <-done: + t.Fatalf("write %d flushed while its batch window was still open", id) + default: + } + return window + } + + windowFor(1) + // the straggler fills the batch to MinBytes, closing the window early + go writeAsync(2) + for i := 0; i < 2; i++ { + if writeErr := <-done; writeErr != nil { + t.Fatalf("write needle: %v", writeErr) + } + } + if syncs := counting.syncs.Load(); syncs != 1 { + t.Fatalf("expected both writes to share one .dat sync, got %d", syncs) + } + + window := windowFor(3) + window <- time.Now() + if writeErr := <-done; writeErr != nil { + t.Fatalf("write needle: %v", writeErr) + } + if syncs := counting.syncs.Load(); syncs != 2 { + t.Fatalf("expected a lone write to take its own sync, got %d total", syncs) + } +}