master: cap writable volumes per collection via -maxWritableVolumes (#11923)

* master: cap writable volumes per collection via -maxWritableVolumes

Some deployments serialize disk writes (e.g. an Unraid array where
concurrent writes to multiple disks collapse throughput). With more
than one writable volume in a collection, the master's weighted pick
spreads assignments over all of them, so several disks get written at
once.

-maxWritableVolumes=N narrows the pick candidates to the N fullest
writable volumes, so a collection's writes concentrate on one (or few)
volumes at a time. 0 keeps the current spread-everywhere behavior.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* master: give master.follower a maxWritableVolumes default

* master: cap writable volumes inside the requested location

* master: pick the capped volumes without sorting the writable list

* master: assert the capped candidate set instead of random pick outcomes

* master: describe -maxWritableVolumes as a per-layout cap

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/command/master.go

@@ -57,6 +57,7 @@ type MasterOptions struct {
 	fileSizeLimitMB               *int
 	volumePreallocate             *bool
 	maxParallelVacuumPerServer    *int
+	maxWritableVolumes            *int
 	vacuumIntervalSeconds         *int
 	vacuumDeleteEmptyAfterSeconds *int
 	// pulseSeconds       *int
@@ -95,6 +96,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.maxWritableVolumes = cmdMaster.Flag.Int("maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit")
 	m.vacuumIntervalSeconds = cmdMaster.Flag.Int("vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps")
 	m.vacuumDeleteEmptyAfterSeconds = cmdMaster.Flag.Int("vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables")
 	// m.pulseSeconds = cmdMaster.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats")
@@ -477,6 +479,7 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp
 		FileSizeLimitMB:               *m.fileSizeLimitMB,
 		VolumePreallocate:             *m.volumePreallocate,
 		MaxParallelVacuumPerServer:    *m.maxParallelVacuumPerServer,
+		MaxWritableVolumes:            *m.maxWritableVolumes,
 		VacuumIntervalSeconds:         *m.vacuumIntervalSeconds,
 		VacuumDeleteEmptyAfterSeconds: *m.vacuumDeleteEmptyAfterSeconds,
 		// PulseSeconds:            *m.pulseSeconds,

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.maxWritableVolumes = aws.Int(0)
 	mf.vacuumIntervalSeconds = aws.Int(840)
 	mf.vacuumDeleteEmptyAfterSeconds = aws.Int(0)
 	mf.telemetryUrl = aws.String("https://telemetry.seaweedfs.com/api/collect")

weed/command/mini.go

@@ -446,6 +446,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.maxWritableVolumes = cmdMini.Flag.Int("master.maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit")
 	miniMasterOptions.vacuumIntervalSeconds = cmdMini.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps")
 	miniMasterOptions.vacuumDeleteEmptyAfterSeconds = cmdMini.Flag.Int("master.vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables")
 	miniMasterOptions.defaultReplication = cmdMini.Flag.String("master.defaultReplication", "", "Default replication type if not specified.")

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.maxWritableVolumes = cmdServer.Flag.Int("master.maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit")
 	masterOptions.vacuumIntervalSeconds = cmdServer.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps")
 	masterOptions.vacuumDeleteEmptyAfterSeconds = cmdServer.Flag.Int("master.vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables")
 	masterOptions.defaultReplication = cmdServer.Flag.String("master.defaultReplication", "", "Default replication type if not specified.")

weed/server/master_server.go

@@ -54,6 +54,7 @@ type MasterOption struct {
 	FileSizeLimitMB               int
 	VolumePreallocate             bool
 	MaxParallelVacuumPerServer    int
+	MaxWritableVolumes            int
 	VacuumIntervalSeconds         int
 	VacuumDeleteEmptyAfterSeconds int
 	// PulseSeconds            int
@@ -162,6 +163,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 		glog.Fatalf("create sequencer failed.")
 	}
 	ms.Topo = topology.NewTopology("topo", seq, uint64(ms.option.VolumeSizeLimitMB)*1024*1024, 5, replicationAsMin)
+	ms.Topo.SetMaxWritableVolumes(ms.option.MaxWritableVolumes)
 	ms.Topo.SetGrowthFreezeConfig(
 		v.GetBool("master.volume_growth.freeze_on_topology_mismatch"),
 		v.GetDuration("master.volume_growth.state_loss_recovery_wait"))

weed/topology/collection.go

@@ -14,6 +14,7 @@ type Collection struct {
 	Name                     string
 	volumeSizeLimit          uint64
 	replicationAsMin         bool
+	maxWritableVolumes       int
 	storageType2VolumeLayout *util.ConcurrentReadMap
 }
 
@@ -40,7 +41,9 @@ func (c *Collection) GetOrCreateVolumeLayout(rp *super_block.ReplicaPlacement, t
 		keyString += string(diskType)
 	}
 	vl := c.storageType2VolumeLayout.Get(keyString, func() interface{} {
-		return NewVolumeLayout(rp, ttl, diskType, c.volumeSizeLimit, c.replicationAsMin)
+		layout := NewVolumeLayout(rp, ttl, diskType, c.volumeSizeLimit, c.replicationAsMin)
+		layout.maxWritableVolumes = c.maxWritableVolumes
+		return layout
 	})
 	return vl.(*VolumeLayout)
 }

weed/topology/topology.go

@@ -50,6 +50,7 @@ type Topology struct {
 
 	volumeSizeLimit          uint64
 	replicationAsMin         bool
+	maxWritableVolumes       int
 	vacuumDisabledByOperator atomic.Bool // true when operator manually disables vacuum
 	vacuumDisabledByPlugin   atomic.Bool // true when disabled by the vacuum plugin monitor
 	adminServerConnectedFunc func() bool // optional callback to check admin server presence
@@ -532,10 +533,20 @@ func (t *Topology) PickForWrite(requestedCount uint64, option *VolumeGrowOption,
 
 func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) *VolumeLayout {
 	return t.collectionMap.Get(collectionName, func() interface{} {
-		return NewCollection(collectionName, t.volumeSizeLimit, t.replicationAsMin)
+		c := NewCollection(collectionName, t.volumeSizeLimit, t.replicationAsMin)
+		c.maxWritableVolumes = t.maxWritableVolumes
+		return c
 	}).(*Collection).GetOrCreateVolumeLayout(rp, ttl, diskType)
 }
 
+// SetMaxWritableVolumes caps how many writable volumes in a volume layout
+// (collection + replica placement + ttl + disk type) assignments are spread
+// over; 0 means every writable volume is a candidate. It only affects
+// collections created afterwards, so call it before serving.
+func (t *Topology) SetMaxWritableVolumes(n int) {
+	t.maxWritableVolumes = n
+}
+
 // FindVolumeLayout is the read-only counterpart of GetVolumeLayout: it returns
 // nil when the collection or its layout does not exist, instead of creating
 // one. Lets callers check a layout's state without polluting the collection
@@ -797,7 +808,6 @@ func ttlEqual(a, b *needle.TTL) bool {
 	return a.ToUint32() == b.ToUint32()
 }
 
-
 // findDuplicateVolumeLayouts reports every layout that already holds v.Id on
 // data nodes other than dn. Volume ids are global, so the same id held by two
 // nodes under two layouts means the allocator reused an id.

weed/topology/volume_layout.go

@@ -68,8 +68,11 @@ type VolumeLayout struct {
 	deletingVolumes  map[needle.VolumeId]struct{}
 	volumeSizeLimit  uint64
 	replicationAsMin bool
-	accessLock       sync.RWMutex
-	sizeTracking     map[needle.VolumeId]*volumeSizeTracking
+	// maxWritableVolumes caps how many writables PickForWrite spreads over;
+	// 0 means every writable volume is a candidate.
+	maxWritableVolumes int
+	accessLock         sync.RWMutex
+	sizeTracking       map[needle.VolumeId]*volumeSizeTracking
 	// dropped: the layout went away with its collection; late registrations must re-resolve.
 	dropped bool
 }
@@ -646,13 +649,40 @@ func (vl *VolumeLayout) PickForWrite(count uint64, option *VolumeGrowOption) (vi
 		return 0, 0, nil, true, fmt.Errorf("%s", NoWritableVolumes)
 	}
 	if option.DataCenter == "" && option.Rack == "" && option.DataNode == "" {
-		vid, locationList = vl.pickWeightedByRemaining(vl.writables)
+		writables := vl.writables
+		if vl.maxWritableVolumes > 0 && len(writables) > vl.maxWritableVolumes {
+			writables = vl.fullestWritable(writables)
+		}
+		vid, locationList = vl.pickWeightedByRemaining(writables)
 		if locationList == nil || len(locationList.list) == 0 {
 			return 0, 0, nil, false, fmt.Errorf("Strangely vid %s is on no machine!", vid.String())
 		}
 		return vid, count, locationList.Copy(), false, nil
 	}
 
+	if vl.maxWritableVolumes > 0 {
+		// The cap must hold inside the requested location, so collect every
+		// matching writable before narrowing to the fullest.
+		var matches []needle.VolumeId
+		for _, writableVolumeId := range vl.writables {
+			volumeLocationList := vl.vid2location[writableVolumeId]
+			for _, dn := range volumeLocationList.list {
+				if dn.matchesLocation(option) {
+					matches = append(matches, writableVolumeId)
+					break
+				}
+			}
+		}
+		if len(matches) == 0 {
+			return vid, count, locationList, true, fmt.Errorf("%s in DataCenter:%v Rack:%v DataNode:%v", NoWritableVolumes, option.DataCenter, option.Rack, option.DataNode)
+		}
+		if len(matches) > vl.maxWritableVolumes {
+			matches = vl.fullestWritable(matches)
+		}
+		vid, locationList = vl.weightedPick(matches)
+		return vid, count, locationList.Copy(), false, nil
+	}
+
 	// Scan from a random offset to collect up to pickSampleSize matching
 	// candidates, avoiding a full scan + allocation in the common case.
 	var sample [pickSampleSize]needle.VolumeId
@@ -700,6 +730,35 @@ func (vl *VolumeLayout) pickWeightedByRemaining(candidates []needle.VolumeId) (n
 	return vl.weightedPick(sample[:])
 }
 
+// fullestWritable narrows candidates to the maxWritableVolumes fullest ones,
+// so assignments concentrate on the fullest writable volume instead of
+// spreading across many disks at once. The cap is expected to be small, so the
+// top-n are kept by insertion instead of sorting the whole candidate list.
+func (vl *VolumeLayout) fullestWritable(candidates []needle.VolumeId) []needle.VolumeId {
+	n := vl.maxWritableVolumes
+	top := make([]needle.VolumeId, 0, n)
+	remaining := make([]uint64, 0, n)
+	for _, vid := range candidates {
+		r := vl.remainingSize(vid)
+		i := len(top)
+		for i > 0 && (remaining[i-1] > r || (remaining[i-1] == r && top[i-1] > vid)) {
+			i--
+		}
+		if i == n {
+			continue
+		}
+		top = append(top, 0)
+		remaining = append(remaining, 0)
+		copy(top[i+1:], top[i:])
+		copy(remaining[i+1:], remaining[i:])
+		top[i], remaining[i] = vid, r
+		if len(top) > n {
+			top, remaining = top[:n], remaining[:n]
+		}
+	}
+	return top
+}
+
 func (vl *VolumeLayout) weightedPick(candidates []needle.VolumeId) (needle.VolumeId, *VolumeLocationList) {
 	if len(candidates) == 1 {
 		vid := candidates[0]

weed/topology/volume_layout_max_writable_test.go

@@ -0,0 +1,127 @@
+package topology
+
+import (
+	"slices"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+func maxWritableLayout(t *testing.T, n int) (*Topology, *DataNode, *VolumeLayout) {
+	t.Helper()
+	topo := NewTopology("mw", nil, 32*1024, 5, false)
+	topo.SetMaxWritableVolumes(n)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100})
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	return topo, dn, topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType)
+}
+
+// With a cap, assignments concentrate on the fullest writable volumes so at
+// most cap volumes are written in parallel.
+func TestPickForWriteCapsWritableVolumes(t *testing.T) {
+	topo, dn, vl := maxWritableLayout(t, 1)
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		sizeTrackingVolume(1, 30000, false),
+		sizeTrackingVolume(2, 1000, false),
+		sizeTrackingVolume(3, 500, false),
+	}, dn)
+
+	for i := 0; i < 20; i++ {
+		vid, _, _, _, err := vl.PickForWrite(1, &VolumeGrowOption{})
+		if err != nil {
+			t.Fatalf("PickForWrite: %v", err)
+		}
+		if vid != needle.VolumeId(1) {
+			t.Fatalf("picked volume %d, want the fullest writable 1", vid)
+		}
+	}
+}
+
+// Once the fullest volume leaves the writable list, the next fullest takes over.
+func TestPickForWriteCapFallsToNextWritable(t *testing.T) {
+	topo, dn, vl := maxWritableLayout(t, 1)
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		sizeTrackingVolume(1, 30000, false),
+		sizeTrackingVolume(2, 1000, false),
+	}, dn)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		sizeTrackingVolume(1, 30000, true),
+		sizeTrackingVolume(2, 1000, false),
+	}, dn)
+
+	vid, _, _, _, err := vl.PickForWrite(1, &VolumeGrowOption{})
+	if err != nil {
+		t.Fatalf("PickForWrite: %v", err)
+	}
+	if vid != needle.VolumeId(2) {
+		t.Fatalf("picked volume %d, want 2 after 1 went read-only", vid)
+	}
+}
+
+// A cap of two still concentrates on the two fullest writables; zero disables
+// the cap so every writable volume is a candidate.
+func TestPickForWriteCapOfTwoAndDisabled(t *testing.T) {
+	topo, dn, vl := maxWritableLayout(t, 2)
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		sizeTrackingVolume(1, 30000, false),
+		sizeTrackingVolume(2, 20000, false),
+		sizeTrackingVolume(3, 500, false),
+	}, dn)
+
+	// the capped candidate set is deterministic even though the pick among
+	// them is weighted-random
+	if got := vl.fullestWritable(vl.writables); !slices.Equal(got, []needle.VolumeId{1, 2}) {
+		t.Fatalf("capped candidates %v, want [1 2]", got)
+	}
+	for i := 0; i < 40; i++ {
+		vid, _, _, _, err := vl.PickForWrite(1, &VolumeGrowOption{})
+		if err != nil {
+			t.Fatalf("PickForWrite: %v", err)
+		}
+		if vid != needle.VolumeId(1) && vid != needle.VolumeId(2) {
+			t.Fatalf("capped pick reached volume %d, want one of 1,2", vid)
+		}
+	}
+
+	vl.maxWritableVolumes = 0
+	for i := 0; i < 40; i++ {
+		if _, _, _, _, err := vl.PickForWrite(1, &VolumeGrowOption{}); err != nil {
+			t.Fatalf("PickForWrite: %v", err)
+		}
+	}
+}
+
+// The cap applies inside the requested location: with cap 1, a write naming
+// dc2 must still find dc2's writable volume even though the fullest is in dc1.
+func TestPickForWriteCapWithinRequestedLocation(t *testing.T) {
+	topo := NewTopology("mw", nil, 32*1024, 5, false)
+	topo.SetMaxWritableVolumes(1)
+	dn1 := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100})
+	dn2 := topo.GetOrCreateDataCenter("dc2").GetOrCreateRack("rack2").
+		GetOrCreateDataNode("127.0.0.2", 8080, 18080, "", "", map[string]uint32{"": 100})
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType)
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		sizeTrackingVolume(1, 30000, false), // fullest, in dc1
+	}, dn1)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		sizeTrackingVolume(2, 1000, false), // in dc2
+	}, dn2)
+
+	vid, _, _, _, err := vl.PickForWrite(1, &VolumeGrowOption{DataCenter: "dc2"})
+	if err != nil {
+		t.Fatalf("PickForWrite dc2: %v", err)
+	}
+	if vid != needle.VolumeId(2) {
+		t.Fatalf("picked volume %d, want dc2's volume 2", vid)
+	}
+}