topology: mirror the writable volume list in a set (#11076)

Membership was a linear scan over a slice, and ensureCorrectWritables runs it
for every volume on every heartbeat, so the master's steady-state cost per
volume server is quadratic in that server's volume count.

BenchmarkSyncDataNodeRegistration, median of 3:

  1000 volumes     565.7us -> 535.0us    -5.4%
  100000 volumes    1.665s -> 55.3ms     -96.7%

Allocations are unchanged at both sizes, so the difference is the scan.

Claude-Session: https://claude.ai/code/session_01P3pE6J2UPFp6G3ksfMV4s1

weed/topology/volume_layout.go

@@ -53,13 +53,16 @@ const capacityRecoveryDelay = 30 * time.Second
 
 // mapping from volume to its locations, inverted from server to volume
 type VolumeLayout struct {
-	growRequest      atomic.Bool
+	growRequest   atomic.Bool
-	lastGrowCount    atomic.Uint32
+	lastGrowCount atomic.Uint32
-	rp               *super_block.ReplicaPlacement
+	rp            *super_block.ReplicaPlacement
-	ttl              *needle.TTL
+	ttl           *needle.TTL
-	diskType         types.DiskType
+	diskType      types.DiskType
-	vid2location     map[needle.VolumeId]*VolumeLocationList
+	vid2location  map[needle.VolumeId]*VolumeLocationList
-	writables        []needle.VolumeId // transient array of writable volume id
+	writables     []needle.VolumeId // transient array of writable volume id
+	// writableMembers mirrors writables for membership tests, which run for
+	// every volume on every heartbeat through ensureCorrectWritables.
+	writableMembers  map[needle.VolumeId]struct{}
 	crowded          map[needle.VolumeId]struct{}
 	vacuumedVolumes  map[needle.VolumeId]time.Time
 	volumeSizeLimit  uint64
@@ -86,6 +89,7 @@ func NewVolumeLayout(rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType
 		diskType:         diskType,
 		vid2location:     make(map[needle.VolumeId]*VolumeLocationList),
 		writables:        *new([]needle.VolumeId),
+		writableMembers:  make(map[needle.VolumeId]struct{}),
 		crowded:          make(map[needle.VolumeId]struct{}),
 		vacuumedVolumes:  make(map[needle.VolumeId]time.Time),
 		volumeSizeLimit:  volumeSizeLimit,
@@ -896,28 +900,26 @@ func (vl *VolumeLayout) CountUnderReplicatedVolumes() int {
 }
 
 func (vl *VolumeLayout) removeFromWritable(vid needle.VolumeId) bool {
-	toDeleteIndex := -1
+	if _, ok := vl.writableMembers[vid]; !ok {
+		return false
+	}
 	for k, id := range vl.writables {
 		if id == vid {
-			toDeleteIndex = k
+			glog.V(0).Infoln("Volume", vid, "becomes unwritable")
-			break
+			vl.writables = append(vl.writables[0:k], vl.writables[k+1:]...)
+			delete(vl.writableMembers, vid)
+			return true
 		}
 	}
-	if toDeleteIndex >= 0 {
-		glog.V(0).Infoln("Volume", vid, "becomes unwritable")
-		vl.writables = append(vl.writables[0:toDeleteIndex], vl.writables[toDeleteIndex+1:]...)
-		return true
-	}
 	return false
 }
 func (vl *VolumeLayout) setVolumeWritable(vid needle.VolumeId) bool {
-	for _, v := range vl.writables {
+	if _, ok := vl.writableMembers[vid]; ok {
-		if v == vid {
+		return false
-			return false
-		}
 	}
 	glog.V(1).Infoln("Volume", vid, "becomes writable")
 	vl.writables = append(vl.writables, vid)
+	vl.writableMembers[vid] = struct{}{}
 	return true
 }
 

weed/topology/volume_layout_pick_test.go

@@ -103,7 +103,7 @@ func TestPickForWriteWeightedDistribution(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	counts := make(map[needle.VolumeId]int)
 	option := &VolumeGrowOption{}
@@ -155,7 +155,7 @@ func TestPickForWriteWithPendingSize(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	// Add large pending to vid 1, making it effectively 9000/10000
 	vl.RecordAssign(1, 8000)
@@ -268,7 +268,7 @@ func TestPickForWriteSingleWritable(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	option := &VolumeGrowOption{}
 	for i := 0; i < 100; i++ {
@@ -298,7 +298,7 @@ func TestPickForWriteAllNearFull(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	option := &VolumeGrowOption{}
 	for i := 0; i < 100; i++ {
@@ -338,7 +338,7 @@ func TestPickForWriteConstrainedWeighted(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	counts := make(map[needle.VolumeId]int)
 	option := &VolumeGrowOption{DataCenter: "dc1"}
@@ -373,7 +373,7 @@ func TestRecordAssignMarksCrowded(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	// Volume at 85% — not crowded yet (threshold is 90%)
 	_, crowded := vl.GetWritableVolumeCount()
@@ -591,7 +591,7 @@ func TestHeartbeatDecaysPendingSize(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	// vid2size starts at 1000 (reported). Add 8000 pending → 9000.
 	vl.RecordAssign(1, 8000)
@@ -770,7 +770,7 @@ func TestShouldGrowVolumesByDcAndRack_WithPendingSize(t *testing.T) {
   }
 }
 `
-	_, vl := setupPickTest(t, layout,10000)
+	_, vl := setupPickTest(t, layout, 10000)
 
 	writables := vl.CloneWritableVolumes()
 	if vl.ShouldGrowVolumesByDcAndRack(&writables, "dc1", "rack1") {
@@ -784,4 +784,3 @@ func TestShouldGrowVolumesByDcAndRack_WithPendingSize(t *testing.T) {
 		t.Error("should grow after pending pushes volume past crowded threshold")
 	}
 }
-