diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 979e9623a..165005ca3 100644 --- a/weed/topology/volume_layout.go +++ b/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 - lastGrowCount atomic.Uint32 - rp *super_block.ReplicaPlacement - ttl *needle.TTL - diskType types.DiskType - vid2location map[needle.VolumeId]*VolumeLocationList - writables []needle.VolumeId // transient array of writable volume id + growRequest atomic.Bool + lastGrowCount atomic.Uint32 + rp *super_block.ReplicaPlacement + ttl *needle.TTL + diskType types.DiskType + vid2location map[needle.VolumeId]*VolumeLocationList + 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 - break + glog.V(0).Infoln("Volume", vid, "becomes unwritable") + 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 v == vid { - return false - } + if _, ok := vl.writableMembers[vid]; ok { + return false } glog.V(1).Infoln("Volume", vid, "becomes writable") vl.writables = append(vl.writables, vid) + vl.writableMembers[vid] = struct{}{} return true } diff --git a/weed/topology/volume_layout_pick_test.go b/weed/topology/volume_layout_pick_test.go index 78c741fb1..4fff615a6 100644 --- a/weed/topology/volume_layout_pick_test.go +++ b/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") } } -