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

@@ -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,

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)
 }

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")

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")

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()

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() {

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
 	}