From 9886ac0bcfb2e8de2c1020b3c593d63a3432c30f Mon Sep 17 00:00:00 2001 From: Mansur Akhmadi Date: Wed, 11 Nov 2026 03:17:21 +0700 Subject: [PATCH] topology: take dn.Lock when a heartbeat links a new disk type (#12535) Co-authored-by: Mansur Akhmadi <10739460+mansur-akhmadi@users.noreply.github.com> --- weed/topology/data_node.go | 13 ++- weed/topology/data_node_ec.go | 4 +- weed/topology/data_node_new_disk_type_test.go | 92 +++++++++++++++++++ weed/topology/topology_event_handling.go | 2 +- weed/topology/volume_layout.go | 2 +- 5 files changed, 107 insertions(+), 6 deletions(-) create mode 100644 weed/topology/data_node_new_disk_type_test.go diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index 5e90dff418..74ae33b12e 100644 --- a/weed/topology/data_node.go +++ b/weed/topology/data_node.go @@ -102,6 +102,15 @@ func (dn *DataNode) getOrCreateDisk(diskType string) *Disk { return disk } +// getOrCreateDiskWithLock is getOrCreateDisk for callers that do not hold +// dn.Lock: linking a new disk writes dn.children, which readers range under +// dn.RLock. +func (dn *DataNode) getOrCreateDiskWithLock(diskType string) *Disk { + dn.Lock() + defer dn.Unlock() + return dn.getOrCreateDisk(diskType) +} + func (dn *DataNode) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged, tierTransition bool) { disk := dn.getOrCreateDisk(v.DiskType) return disk.AddOrUpdateVolume(v) @@ -223,7 +232,7 @@ func (dn *DataNode) AdjustMaxVolumeCounts(maxVolumeCounts map[string]uint32) { if currentDiskUsageMaxVolumeCount == int64(maxVolumeCount) { continue } - disk := dn.getOrCreateDisk(dt.String()) + disk := dn.getOrCreateDiskWithLock(dt.String()) disk.UpAdjustDiskUsageDelta(dt, &DiskUsageCounts{ maxVolumeCount: int64(maxVolumeCount) - currentDiskUsageMaxVolumeCount, }) @@ -249,7 +258,7 @@ func (dn *DataNode) AdjustDiskUsageBytes(diskTotalBytes, diskFreeBytes map[strin if currentTotal == newTotal && currentFree == newFree { continue } - disk := dn.getOrCreateDisk(dt.String()) + disk := dn.getOrCreateDiskWithLock(dt.String()) disk.UpAdjustDiskUsageDelta(dt, &DiskUsageCounts{ diskTotalBytes: newTotal - currentTotal, diskFreeBytes: newFree - currentFree, diff --git a/weed/topology/data_node_ec.go b/weed/topology/data_node_ec.go index 621045bf0e..03cca552cf 100644 --- a/weed/topology/data_node_ec.go +++ b/weed/topology/data_node_ec.go @@ -216,12 +216,12 @@ func (dn *DataNode) DeltaUpdateEcShards(newShards, deletedShards []*erasure_codi } func (dn *DataNode) AddOrUpdateEcShard(s *erasure_coding.EcVolumeInfo) { - disk := dn.getOrCreateDisk(s.DiskType) + disk := dn.getOrCreateDiskWithLock(s.DiskType) disk.AddOrUpdateEcShard(s) } func (dn *DataNode) DeleteEcShard(s *erasure_coding.EcVolumeInfo) { - disk := dn.getOrCreateDisk(s.DiskType) + disk := dn.getOrCreateDiskWithLock(s.DiskType) disk.DeleteEcShard(s) } diff --git a/weed/topology/data_node_new_disk_type_test.go b/weed/topology/data_node_new_disk_type_test.go new file mode 100644 index 0000000000..73c569531b --- /dev/null +++ b/weed/topology/data_node_new_disk_type_test.go @@ -0,0 +1,92 @@ +package topology + +import ( + "fmt" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func newDiskTypeTestNode() *DataNode { + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1") + return rack.GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1:8080", "", map[string]uint32{"hdd": 10}) +} + +func ecShardOnDiskType(vid needle.VolumeId, diskType string) *erasure_coding.EcVolumeInfo { + s := &erasure_coding.EcVolumeInfo{VolumeId: vid, DiskType: diskType, ShardsInfo: erasure_coding.NewShardsInfo()} + s.ShardsInfo.Set(erasure_coding.ShardInfo{Id: 0, Size: 100}) + return s +} + +// A heartbeat that reports a disk type the node does not have yet links a new +// Disk into dn.children. AdjustMaxVolumeCounts, AdjustDiskUsageBytes and the +// EC shard deltas did that without dn.Lock while lookups, assigns and the UI +// range dn.children under dn.RLock, which the Go runtime can abort the master +// for ("concurrent map read and map write"). Run with -race to reproduce. +func TestNewDiskTypeFromHeartbeatDoesNotRaceReaders(t *testing.T) { + dn := newDiskTypeTestNode() + const rounds = 100 + + done := make(chan struct{}) + var readers sync.WaitGroup + for range 2 { + readers.Add(1) + go func() { + defer readers.Done() + for { + select { + case <-done: + return + default: + } + _ = dn.GetVolumes() + _ = dn.GetEcShards() + _ = dn.Children() + } + }() + } + + for i := range rounds { + dn.AdjustMaxVolumeCounts(map[string]uint32{fmt.Sprintf("max%d", i): 3}) + dn.AdjustDiskUsageBytes(map[string]uint64{fmt.Sprintf("bytes%d", i): 1 << 30}, map[string]uint64{fmt.Sprintf("bytes%d", i): 1 << 20}) + dn.AddOrUpdateEcShard(ecShardOnDiskType(needle.VolumeId(i+1), fmt.Sprintf("ec%d", i))) + dn.DeleteEcShard(ecShardOnDiskType(needle.VolumeId(i+1), fmt.Sprintf("ec%d", i))) + } + close(done) + readers.Wait() + + // 1 initial hdd disk + 3 new disk types per round + if got, want := len(dn.Children()), 1+3*rounds; got != want { + t.Fatalf("children = %d, want %d", got, want) + } +} + +// Several heartbeats introducing the same disk type at once must end up on +// one linked Disk. A second Disk created in parallel is never linked, so +// whatever is added to it is lost from the topology. +func TestConcurrentFirstUseOfDiskTypeSharesOneDisk(t *testing.T) { + for round := range 50 { + dn := newDiskTypeTestNode() + const writers = 8 + var wg sync.WaitGroup + start := make(chan struct{}) + for w := range writers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + dn.AddOrUpdateEcShard(ecShardOnDiskType(needle.VolumeId(w+1), "ssd")) + }() + } + close(start) + wg.Wait() + + if got := len(dn.GetEcShards()); got != writers { + t.Fatalf("round %d: node holds %d ec volumes, want %d", round, got, writers) + } + } +} diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index 825ab7c36c..a27d8fcdb5 100644 --- a/weed/topology/topology_event_handling.go +++ b/weed/topology/topology_event_handling.go @@ -76,7 +76,7 @@ func (t *Topology) SetVolumeCapacityFull(volumeInfo storage.VolumeInfo) bool { for _, dn := range vidLocations.list { if !volumeInfo.ReadOnly { - disk := dn.getOrCreateDisk(volumeInfo.DiskType) + disk := dn.getOrCreateDiskWithLock(volumeInfo.DiskType) disk.UpAdjustDiskUsageDelta(types.ToDiskType(volumeInfo.DiskType), &DiskUsageCounts{ activeVolumeCount: -1, }) diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index bd961565ae..f900d170fa 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -586,7 +586,7 @@ func (vl *VolumeLayout) adjustActiveVolumeCount(vid needle.VolumeId, delta int64 diskTypeStr := string(vl.diskType) for _, dn := range nodes { - disk := dn.getOrCreateDisk(diskTypeStr) + disk := dn.getOrCreateDiskWithLock(diskTypeStr) disk.UpAdjustDiskUsageDelta(vl.diskType, &DiskUsageCounts{ activeVolumeCount: delta, })