From 56d2f05ccd01ef2af39829f0f2e40a81af4ce9f7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 25 Sep 2026 02:14:59 +0800 Subject: [PATCH] topology: wake the vacuum dispatcher when a worker frees quota (#11436) * topology: wake the vacuum dispatcher when a worker frees quota The dispatch loop slept a fixed 10s whenever every pending volume was waiting for a per-server quota slot, so a sweep took volumes x 10s regardless of how fast the compactions were. Workers now signal on a buffered channel after crediting quota; the dispatcher waits on it with the 10s sleep kept only as a timeout. * master: add -vacuumIntervalSeconds to tune the automatic sweep interval The 14-minute base interval was a literal inside the refresh loop while every neighbouring vacuum knob was already a flag. Defaults to 840s, unchanged. * topology: keep the 14 minute floor on the vacuum interval A zero-valued MasterOption or a negative -vacuumIntervalSeconds left the sweep sleeping only its jitter, so treat non-positive intervals as the previous default. --- weed/command/master.go | 3 +++ weed/command/master_follower.go | 1 + weed/command/mini.go | 1 + weed/command/server.go | 1 + weed/server/master_server.go | 2 ++ weed/topology/topology_event_handling.go | 7 +++++-- weed/topology/topology_vacuum.go | 10 +++++++++- 7 files changed, 22 insertions(+), 3 deletions(-) diff --git a/weed/command/master.go b/weed/command/master.go index 61bbea730..5b29b914f 100644 --- a/weed/command/master.go +++ b/weed/command/master.go @@ -57,6 +57,7 @@ type MasterOptions struct { fileSizeLimitMB *int volumePreallocate *bool maxParallelVacuumPerServer *int + vacuumIntervalSeconds *int // pulseSeconds *int defaultReplication *string garbageThreshold *float64 @@ -93,6 +94,7 @@ func init() { m.fileSizeLimitMB = cmdMaster.Flag.Int("fileSizeLimitMB", 256, "limit the file size accepted by /submit, should match the volume servers' -fileSizeLimitMB (-volume.fileSizeLimitMB under weed server or weed mini, which set this for you)") m.volumePreallocate = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.") m.maxParallelVacuumPerServer = cmdMaster.Flag.Int("maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel per volume server") + m.vacuumIntervalSeconds = cmdMaster.Flag.Int("vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps") // m.pulseSeconds = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats") m.defaultReplication = cmdMaster.Flag.String("defaultReplication", "", "Default replication type if not specified.") m.garbageThreshold = cmdMaster.Flag.Float64("garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces") @@ -473,6 +475,7 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp FileSizeLimitMB: *m.fileSizeLimitMB, VolumePreallocate: *m.volumePreallocate, MaxParallelVacuumPerServer: *m.maxParallelVacuumPerServer, + VacuumIntervalSeconds: *m.vacuumIntervalSeconds, // PulseSeconds: *m.pulseSeconds, DefaultReplicaPlacement: *m.defaultReplication, GarbageThreshold: *m.garbageThreshold, diff --git a/weed/command/master_follower.go b/weed/command/master_follower.go index 249f51dcf..1a8ffefba 100644 --- a/weed/command/master_follower.go +++ b/weed/command/master_follower.go @@ -47,6 +47,7 @@ func init() { mf.metricsIntervalSec = aws.Int(0) mf.raftResumeState = aws.Bool(false) mf.maxParallelVacuumPerServer = aws.Int(1) + mf.vacuumIntervalSeconds = aws.Int(840) mf.telemetryUrl = aws.String("https://telemetry.seaweedfs.com/api/collect") mf.telemetryEnabled = aws.Bool(false) } diff --git a/weed/command/mini.go b/weed/command/mini.go index 5d8a54915..2cf14a491 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -427,6 +427,7 @@ func initMiniMasterFlags() { miniMasterOptions.volumeSizeLimitMB = cmdMini.Flag.Uint("master.volumeSizeLimitMB", defaultMiniVolumeSizeMB, "Master stops directing writes to oversized volumes (default: 128MB for mini)") miniMasterOptions.volumePreallocate = cmdMini.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.") miniMasterOptions.maxParallelVacuumPerServer = cmdMini.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server") + miniMasterOptions.vacuumIntervalSeconds = cmdMini.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps") miniMasterOptions.defaultReplication = cmdMini.Flag.String("master.defaultReplication", "", "Default replication type if not specified.") miniMasterOptions.garbageThreshold = cmdMini.Flag.Float64("master.garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces") miniMasterOptions.metricsAddress = cmdMini.Flag.String("master.metrics.address", "", "Prometheus gateway address") diff --git a/weed/command/server.go b/weed/command/server.go index 7ff8b9449..b987fe9a3 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -98,6 +98,7 @@ func init() { masterOptions.volumeSizeLimitMB = cmdServer.Flag.Uint("master.volumeSizeLimitMB", util.DefaultVolumeSizeLimitMB, "Master stops directing writes to oversized volumes.") masterOptions.volumePreallocate = cmdServer.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.") masterOptions.maxParallelVacuumPerServer = cmdServer.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server") + masterOptions.vacuumIntervalSeconds = cmdServer.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps") masterOptions.defaultReplication = cmdServer.Flag.String("master.defaultReplication", "", "Default replication type if not specified.") masterOptions.garbageThreshold = cmdServer.Flag.Float64("master.garbageThreshold", 0.3, "threshold to vacuum and reclaim spaces") masterOptions.metricsAddress = cmdServer.Flag.String("master.metrics.address", "", "Prometheus gateway address") diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 3a1364c11..73106c99b 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -51,6 +51,7 @@ type MasterOption struct { FileSizeLimitMB int VolumePreallocate bool MaxParallelVacuumPerServer int + VacuumIntervalSeconds int // PulseSeconds int DefaultReplicaPlacement string GarbageThreshold float64 @@ -202,6 +203,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se ms.option.MaxParallelVacuumPerServer, topology.VolumeGrowStrategy.Threshold, ms.preallocateSize, + time.Duration(ms.option.VacuumIntervalSeconds)*time.Second, ) ms.ProcessGrowRequest() diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index 665dfe16c..c5c031a67 100644 --- a/weed/topology/topology_event_handling.go +++ b/weed/topology/topology_event_handling.go @@ -13,7 +13,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage" ) -func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, garbageThreshold float64, concurrentVacuumLimitPerVolumeServer int, growThreshold float64, preallocate int64) { +func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, garbageThreshold float64, concurrentVacuumLimitPerVolumeServer int, growThreshold float64, preallocate int64, vacuumInterval time.Duration) { + if vacuumInterval <= 0 { + vacuumInterval = 14 * time.Minute + } go func() { for { if t.IsLeader() { @@ -41,7 +44,7 @@ func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, g } else { stats.MasterReplicaPlacementMismatch.Reset() } - time.Sleep(14*time.Minute + time.Duration(120*rand.Float32())*time.Second) + time.Sleep(vacuumInterval + time.Duration(120*rand.Float32())*time.Second) } }(garbageThreshold) go func() { diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index 0205459d5..171f848ce 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -285,6 +285,7 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL executor := util.NewLimitedConcurrentExecutor(100) var wg sync.WaitGroup + quotaFreed := make(chan struct{}, 1) for len(todoVolumeMap) > 0 { pendingVolumeMap := make(map[needle.VolumeId]*VolumeLocationList) @@ -321,6 +322,10 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL limiter[dn.Id()]++ limiterLock.Unlock() } + select { + case quotaFreed <- struct{}{}: + default: + } }) if automatic && t.IsVacuumDisabled() { break @@ -330,7 +335,10 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL break } if len(todoVolumeMap) == len(pendingVolumeMap) { - time.Sleep(10 * time.Second) + select { + case <-quotaFreed: + case <-time.After(10 * time.Second): + } } todoVolumeMap = pendingVolumeMap }