From 168789d5f3c5077a5a8d493df34d18be2ac94f9f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 23 Oct 2026 17:04:36 -0700 Subject: [PATCH] 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 | 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/collection.go | 5 +- weed/topology/topology.go | 14 +- weed/topology/volume_layout.go | 65 ++++++++- .../volume_layout_max_writable_test.go | 127 ++++++++++++++++++ 9 files changed, 213 insertions(+), 6 deletions(-) create mode 100644 weed/topology/volume_layout_max_writable_test.go diff --git a/weed/command/master.go b/weed/command/master.go index 81ffa6db94..2a5a409678 100644 --- a/weed/command/master.go +++ b/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, diff --git a/weed/command/master_follower.go b/weed/command/master_follower.go index 4e0983eb96..95c065775e 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.maxWritableVolumes = aws.Int(0) mf.vacuumIntervalSeconds = aws.Int(840) mf.vacuumDeleteEmptyAfterSeconds = aws.Int(0) mf.telemetryUrl = aws.String("https://telemetry.seaweedfs.com/api/collect") diff --git a/weed/command/mini.go b/weed/command/mini.go index 9fd8d0963f..381e438e9c 100644 --- a/weed/command/mini.go +++ b/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.") diff --git a/weed/command/server.go b/weed/command/server.go index 5c800b3004..0277ccf421 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.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.") diff --git a/weed/server/master_server.go b/weed/server/master_server.go index b36679ebcd..6a5c5e7ac9 100644 --- a/weed/server/master_server.go +++ b/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")) diff --git a/weed/topology/collection.go b/weed/topology/collection.go index 9fe355a083..254c951dc7 100644 --- a/weed/topology/collection.go +++ b/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) } diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 3c68d3586b..cee0018584 100644 --- a/weed/topology/topology.go +++ b/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. diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 4219c8eff3..b0a48e750b 100644 --- a/weed/topology/volume_layout.go +++ b/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] diff --git a/weed/topology/volume_layout_max_writable_test.go b/weed/topology/volume_layout_max_writable_test.go new file mode 100644 index 0000000000..26b718801e --- /dev/null +++ b/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) + } +}