From 65ce9f8b7c4267f3cc2e586a06ec0fcdb1cb3fc9 Mon Sep 17 00:00:00 2001
From: chihe <chihe@oppo.com>
Date: Sun, 15 Sep 2024 21:24:41 +0800
Subject: [PATCH] fix(master): deadlock for datanode and nodeset

Signed-off-by: chihe <chihe@oppo.com>
---
 master/cluster.go      | 14 +++++++++++---
 master/cluster_test.go |  2 ++
 master/topology.go     |  4 +++-
 3 files changed, 16 insertions(+), 4 deletions(-)

diff --git a/master/cluster.go b/master/cluster.go
index 0770cbb60..063feb9c6 100644
--- a/master/cluster.go
+++ b/master/cluster.go
@@ -2916,10 +2916,14 @@ func (c *Cluster) putBadDataPartitionIDs(replica *DataReplica, addr string, part
 func (c *Cluster) clearBadDataPartitionIDS() {
 	c.badPartitionMutex.Lock()
 	defer c.badPartitionMutex.Unlock()
+	keysToDelete := make([]string, 0)
 	c.BadDataPartitionIds.Range(func(key, value interface{}) bool {
-		c.BadDataPartitionIds.Delete(key)
+		keysToDelete = append(keysToDelete, key.(string))
 		return true
 	})
+	for _, key := range keysToDelete {
+		c.BadDataPartitionIds.Delete(key)
+	}
 }
 
 func (c *Cluster) putBadDataPartitionIDsByDiskPath(disk, addr string, partitionID uint64) {
@@ -4391,10 +4395,14 @@ func (c *Cluster) canAutoDecommissionDisk(addr string, diskPath string) (yes boo
 }
 
 func (c *Cluster) handleDataNodeBadDisk(dataNode *DataNode) {
+	badDisks := make([]proto.BadDiskStat, 0)
 	dataNode.RLock()
-	defer dataNode.RUnlock()
-
 	for _, disk := range dataNode.BadDiskStats {
+		badDisks = append(badDisks, disk)
+	}
+	dataNode.RUnlock()
+
+	for _, disk := range badDisks {
 		// TODO:no dp left on bad disk, notify sre to remove this disk
 		// decommission failed, but lack replica for disk err dp is already removed
 		retry := c.RetryDecommissionDisk(dataNode.Addr, disk.DiskPath)
diff --git a/master/cluster_test.go b/master/cluster_test.go
index ce9122b03..bee7b8121 100644
--- a/master/cluster_test.go
+++ b/master/cluster_test.go
@@ -2,6 +2,7 @@ package master
 
 import (
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -364,6 +365,7 @@ func TestMasterClientLeaderChange(t *testing.T) {
 		masterClient: masterSDK.NewMasterClient(nil, false),
 	}
 	cluster.t = newTopology()
+	cluster.BadDataPartitionIds = new(sync.Map)
 	server := &Server{
 		cluster: cluster,
 		leaderInfo: &LeaderInfo{
diff --git a/master/topology.go b/master/topology.go
index 8252845db..110184cff 100644
--- a/master/topology.go
+++ b/master/topology.go
@@ -1174,7 +1174,7 @@ func (ns *nodeSet) traverseDecommissionDisk(c *Cluster) {
 	for {
 		select {
 		case <-ns.doneDecommissionDiskListTraverse:
-			log.LogWarnf("ns %v(%p)  traverse stopped", ns.ID)
+			log.LogWarnf("ns %v(%p)  traverse stopped", ns.ID, ns)
 			ns.ClearDecommissionDisks()
 			return
 		case <-t.C:
@@ -1247,9 +1247,11 @@ func (ns *nodeSet) traverseDecommissionDisk(c *Cluster) {
 		}
 	}
 }
+
 func (ns *nodeSet) stopDecommissionDiskSchedule() {
 	ns.doneDecommissionDiskListTraverse <- struct{}{}
 }
+
 func (t *topology) isSingleZone() bool {
 	t.zoneLock.RLock()
 	defer t.zoneLock.RUnlock()