enhance: lock protect needed while access metaPartitions info of volu…

…me in master Signed-off-by: leonrayang <[email protected]>
aFlyBird0 · Mar 16, 2023 · 34cd246 · 34cd246
1 parent 2f979d6
commit 34cd246
Show file tree

Hide file tree

Showing 6 changed files with 35 additions and 1 deletion.
diff --git a/master/api_service.go b/master/api_service.go
@@ -1814,6 +1814,9 @@ func newSimpleView(vol *Vol) *proto.SimpleVolView {
 		volInodeCount  uint64
 		volDentryCount uint64
 	)
+	vol.mpsLock.RLock()
+	defer vol.mpsLock.RUnlock()
+
 	for _, mp := range vol.MetaPartitions {
 		volDentryCount = volDentryCount + mp.DentryCount
 		volInodeCount = volInodeCount + mp.InodeCount
@@ -3754,6 +3757,9 @@ func volStat(vol *Vol, countByMeta bool) (stat *proto.VolStatInfo) {
 	}
 
 	stat.UsedRatio = strconv.FormatFloat(float64(stat.UsedSize)/float64(stat.TotalSize), 'f', 2, 32)
+
+	vol.mpsLock.RLock()
+	defer vol.mpsLock.RUnlock()
 	for _, mp := range vol.MetaPartitions {
 		stat.InodeCount += mp.InodeCount
 	}

diff --git a/master/cluster.go b/master/cluster.go
@@ -588,13 +588,16 @@ func (c *Cluster) getNotConsistentIDMetaNodes() (metaNodes []*InvalidNodeView) {
 func (c *Cluster) hasNotConsistentIDMetaPartitions(metanode *MetaNode) (notConsistent bool, oldID uint64) {
 	safeVols := c.allVols()
 	for _, vol := range safeVols {
+		vol.mpsLock.RLock()
 		for _, mp := range vol.MetaPartitions {
 			for _, peer := range mp.Peers {
 				if peer.Addr == metanode.Addr && peer.ID != metanode.ID {
+					vol.mpsLock.RUnlock()
 					return true, peer.ID
 				}
 			}
 		}
+		vol.mpsLock.RUnlock()
 	}
 	return
 }
@@ -1376,6 +1379,7 @@ func (c *Cluster) getAllMetaPartitionByMetaNode(addr string) (partitions []*Meta
 	partitions = make([]*MetaPartition, 0)
 	safeVols := c.allVols()
 	for _, vol := range safeVols {
+		vol.mpsLock.RLock()
 		for _, mp := range vol.MetaPartitions {
 			for _, host := range mp.Hosts {
 				if host == addr {
@@ -1384,6 +1388,7 @@ func (c *Cluster) getAllMetaPartitionByMetaNode(addr string) (partitions []*Meta
 				}
 			}
 		}
+		vol.mpsLock.RUnlock()
 	}
 
 	return
@@ -1411,12 +1416,14 @@ func (c *Cluster) getAllMetaPartitionIDByMetaNode(addr string) (partitionIDs []u
 	safeVols := c.allVols()
 	for _, vol := range safeVols {
 		for _, mp := range vol.MetaPartitions {
+			vol.mpsLock.RLock()
 			for _, host := range mp.Hosts {
 				if host == addr {
 					partitionIDs = append(partitionIDs, mp.PartitionID)
 					break
 				}
 			}
+			vol.mpsLock.RUnlock()
 		}
 	}
 
@@ -1428,12 +1435,14 @@ func (c *Cluster) getAllMetaPartitionsByMetaNode(addr string) (partitions []*Met
 	safeVols := c.allVols()
 	for _, vol := range safeVols {
 		for _, mp := range vol.MetaPartitions {
+			vol.mpsLock.RLock()
 			for _, host := range mp.Hosts {
 				if host == addr {
 					partitions = append(partitions, mp)
 					break
 				}
 			}
+			vol.mpsLock.RUnlock()
 		}
 	}
 	return
@@ -2786,7 +2795,9 @@ func (c *Cluster) getDataPartitionCount() (count int) {
 func (c *Cluster) getMetaPartitionCount() (count int) {
 	vols := c.copyVols()
 	for _, vol := range vols {
+		vol.mpsLock.RLock()
 		count = count + len(vol.MetaPartitions)
+		vol.mpsLock.RUnlock()
 	}
 	return count
 }

diff --git a/master/cluster_stat.go b/master/cluster_stat.go
@@ -179,9 +179,11 @@ func (c *Cluster) updateVolStatInfo() {
 		}
 
 		var inodeCount uint64
+		vol.mpsLock.RLock()
 		for _, mp := range vol.MetaPartitions {
 			inodeCount += mp.InodeCount
 		}
+		vol.mpsLock.RUnlock()
 
 		c.volStatInfo.Store(vol.Name, newVolStatInfo(vol.Name, total, used, cacheTotal, cacheUsed, inodeCount))
 	}

diff --git a/master/cluster_task.go b/master/cluster_task.go
@@ -300,11 +300,13 @@ func (c *Cluster) checkLackReplicaMetaPartitions() (lackReplicaMetaPartitions []
 	lackReplicaMetaPartitions = make([]*MetaPartition, 0)
 	vols := c.copyVols()
 	for _, vol := range vols {
+		vol.mpsLock.RLock()
 		for _, mp := range vol.MetaPartitions {
 			if mp.ReplicaNum > uint8(len(mp.Hosts)) {
 				lackReplicaMetaPartitions = append(lackReplicaMetaPartitions, mp)
 			}
 		}
+		vol.mpsLock.RUnlock()
 	}
 	log.LogInfof("clusterID[%v] lackReplicaMetaPartitions count:[%v]", c.Name, len(lackReplicaMetaPartitions))
 	return

diff --git a/master/monitor_metrics.go b/master/monitor_metrics.go
@@ -246,12 +246,13 @@ func (mm *monitorMetrics) setMpAndDpMetrics() {
 				dpMissingLeaderCount++
 			}
 		}
-
+		vol.mpsLock.RLock()
 		for _, mp := range vol.MetaPartitions {
 			if !mp.isLeaderExist() {
 				mpMissingLeaderCount++
 			}
 		}
+		vol.mpsLock.RUnlock()
 	}
 
 	mm.dataPartitionCount.Set(float64(dpCount))
@@ -356,8 +357,10 @@ func (mm *monitorMetrics) setMpInconsistentErrorMetric() {
 	defer mm.cluster.volMutex.RUnlock()
 
 	for _, vol := range mm.cluster.vols {
+		vol.mpsLock.RLock()
 		for _, mp := range vol.MetaPartitions {
 			if mp.IsRecover || mp.EqualCheckPass {
+				vol.mpsLock.RUnlock()
 				continue
 			}
 			idStr := strconv.FormatUint(mp.PartitionID, 10)
@@ -366,6 +369,7 @@ func (mm *monitorMetrics) setMpInconsistentErrorMetric() {
 			log.LogWarnf("setMpInconsistentErrorMetric.mp %v SetWithLabelValues id %v vol %v", mp.PartitionID, idStr, vol.Name)
 			delete(deleteMps, idStr)
 		}
+		vol.mpsLock.RUnlock()
 	}
 
 	for k, v := range deleteMps {

diff --git a/master/vol.go b/master/vol.go
@@ -287,6 +287,8 @@ func (vol *Vol) initMetaPartitions(c *Cluster, count int) (err error) {
 			break
 		}
 	}
+	vol.mpsLock.RLock()
+	defer vol.mpsLock.RUnlock()
 	if len(vol.MetaPartitions) != count {
 		err = fmt.Errorf("action[initMetaPartitions] vol[%v] init meta partition failed,mpCount[%v],expectCount[%v],err[%v]",
 			vol.Name, len(vol.MetaPartitions), count, err)
@@ -451,6 +453,10 @@ func (vol *Vol) checkMetaPartitions(c *Cluster) {
 
 func (vol *Vol) checkSplitMetaPartition(c *Cluster) {
 	maxPartitionID := vol.maxPartitionID()
+
+	vol.mpsLock.RLock()
+	defer vol.mpsLock.RUnlock()
+
 	partition, ok := vol.MetaPartitions[maxPartitionID]
 	if !ok {
 		return
@@ -687,6 +693,9 @@ func (vol *Vol) sendViewCacheToFollower(c *Cluster) {
 func (vol *Vol) ebsUsedSpace() uint64 {
 
 	size := uint64(0)
+	vol.mpsLock.RLock()
+	defer vol.mpsLock.RUnlock()
+
 	for _, pt := range vol.MetaPartitions {
 		size += pt.dataSize()
 	}