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

@@ -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,

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)
 }
 

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)
+		}
+	}
+}

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,
 			})

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,
 		})