Abstract concurrency-related operations as ConcurrencyStat interface and polish StatNode (#283)

Author: louyuting

core/base/stat.go

@@ -38,18 +38,19 @@ type WriteStat interface {
 	AddCount(event MetricEvent, count int64)
 }
 
+type ConcurrencyStat interface {
+	CurrentConcurrency() int32
+	IncreaseConcurrency()
+	DecreaseConcurrency()
+}
+
 // StatNode holds real-time statistics for resources.
 type StatNode interface {
 	MetricItemRetriever
 
 	ReadStat
 	WriteStat
-
-	CurrentGoroutineNum() int32
-	IncreaseGoroutineNum()
-	DecreaseGoroutineNum()
-
-	Reset()
+	ConcurrencyStat
 
 	// GenerateReadStat generates the readonly metric statistic based on resource level global statistic
 	// If parameters, sampleCount and intervalInMs, are not suitable for resource level global statistic, return (nil, error)

core/base/stat_test.go

@@ -50,22 +50,17 @@ func (m *StatNodeMock) MinRT() float64 {
 	return float64(args.Int(0))
 }
 
-func (m *StatNodeMock) CurrentGoroutineNum() int32 {
+func (m *StatNodeMock) CurrentConcurrency() int32 {
 	args := m.Called()
 	return int32(args.Int(0))
 }
 
-func (m *StatNodeMock) IncreaseGoroutineNum() {
+func (m *StatNodeMock) IncreaseConcurrency() {
 	m.Called()
 	return
 }
 
-func (m *StatNodeMock) DecreaseGoroutineNum() {
-	m.Called()
-	return
-}
-
-func (m *StatNodeMock) Reset() {
+func (m *StatNodeMock) DecreaseConcurrency() {
 	m.Called()
 	return
 }

core/isolation/slot.go

@@ -44,7 +44,7 @@ func checkPass(ctx *base.EntryContext) (bool, *Rule, uint32) {
 	for _, rule := range getRulesOfResource(ctx.Resource.Name()) {
 		threshold := rule.Threshold
 		if rule.MetricType == Concurrency {
-			if cur := statNode.CurrentGoroutineNum(); cur >= 0 {
+			if cur := statNode.CurrentConcurrency(); cur >= 0 {
 				curCount = uint32(cur)
 			} else {
 				curCount = 0

core/stat/base_node.go

@@ -12,7 +12,7 @@ type BaseStatNode struct {
 	sampleCount uint32
 	intervalMs  uint32
 
-	goroutineNum int32
+	concurrency int32
 
 	arr    *sbase.BucketLeapArray
 	metric *sbase.SlidingWindowMetric
@@ -22,11 +22,11 @@ func NewBaseStatNode(sampleCount uint32, intervalInMs uint32) *BaseStatNode {
 	la := sbase.NewBucketLeapArray(config.GlobalStatisticSampleCountTotal(), config.GlobalStatisticIntervalMsTotal())
 	metric, _ := sbase.NewSlidingWindowMetric(sampleCount, intervalInMs, la)
 	return &BaseStatNode{
-		goroutineNum: 0,
-		sampleCount:  sampleCount,
-		intervalMs:   intervalInMs,
-		arr:          la,
-		metric:       metric,
+		concurrency: 0,
+		sampleCount: sampleCount,
+		intervalMs:  intervalInMs,
+		arr:         la,
+		metric:      metric,
 	}
 }
 
@@ -66,21 +66,16 @@ func (n *BaseStatNode) MinRT() float64 {
 	return float64(n.metric.MinRT())
 }
 
-func (n *BaseStatNode) CurrentGoroutineNum() int32 {
-	return atomic.LoadInt32(&(n.goroutineNum))
+func (n *BaseStatNode) CurrentConcurrency() int32 {
+	return atomic.LoadInt32(&(n.concurrency))
 }
 
-func (n *BaseStatNode) IncreaseGoroutineNum() {
-	atomic.AddInt32(&(n.goroutineNum), 1)
+func (n *BaseStatNode) IncreaseConcurrency() {
+	atomic.AddInt32(&(n.concurrency), 1)
 }
 
-func (n *BaseStatNode) DecreaseGoroutineNum() {
-	atomic.AddInt32(&(n.goroutineNum), -1)
-}
-
-func (n *BaseStatNode) Reset() {
-	// TODO: this should be thread-safe, or error may occur
-	panic("to be implemented")
+func (n *BaseStatNode) DecreaseConcurrency() {
+	atomic.AddInt32(&(n.concurrency), -1)
 }
 
 func (n *BaseStatNode) GenerateReadStat(sampleCount uint32, intervalInMs uint32) (base.ReadStat, error) {

core/stat/stat_slot.go

@@ -47,7 +47,7 @@ func (s *Slot) recordPassFor(sn base.StatNode, count uint32) {
 	if sn == nil {
 		return
 	}
-	sn.IncreaseGoroutineNum()
+	sn.IncreaseConcurrency()
 	sn.AddCount(base.MetricEventPass, int64(count))
 }
 
@@ -67,5 +67,5 @@ func (s *Slot) recordCompleteFor(sn base.StatNode, count uint32, rt uint64, err
 	}
 	sn.AddCount(base.MetricEventRt, int64(rt))
 	sn.AddCount(base.MetricEventComplete, int64(count))
-	sn.DecreaseGoroutineNum()
+	sn.DecreaseConcurrency()
 }

core/system/slot.go

@@ -49,7 +49,7 @@ func (s *AdaptiveSlot) doCheckRule(rule *Rule) (bool, float64) {
 		res := qps < threshold
 		return res, qps
 	case Concurrency:
-		n := float64(stat.InboundNode().CurrentGoroutineNum())
+		n := float64(stat.InboundNode().CurrentConcurrency())
 		res := n < threshold
 		return res, n
 	case AvgRT:
@@ -78,7 +78,7 @@ func (s *AdaptiveSlot) doCheckRule(rule *Rule) (bool, float64) {
 }
 
 func checkBbrSimple() bool {
-	concurrency := stat.InboundNode().CurrentGoroutineNum()
+	concurrency := stat.InboundNode().CurrentConcurrency()
 	minRt := stat.InboundNode().MinRT()
 	maxComplete := stat.InboundNode().GetMaxAvg(base.MetricEventComplete)
 	if concurrency > 1 && float64(concurrency) > maxComplete*minRt/1000.0 {

core/system/slot_test.go

@@ -51,11 +51,11 @@ func TestDoCheckRuleConcurrency(t *testing.T) {
 	})
 
 	t.Run("FalseConcurrency", func(t *testing.T) {
-		stat.InboundNode().IncreaseGoroutineNum()
+		stat.InboundNode().IncreaseConcurrency()
 		isOK, v := sas.doCheckRule(rule)
 		assert.True(t, util.Float64Equals(float64(1.0), v))
 		assert.Equal(t, false, isOK)
-		stat.InboundNode().DecreaseGoroutineNum()
+		stat.InboundNode().DecreaseConcurrency()
 	})
 }
 

example/hotspot_param_flow/concurrency/hotspot_params_concurrency_example.go

@@ -41,7 +41,7 @@ func main() {
 	go func() {
 		node := stat.GetOrCreateResourceNode("abc", base.ResTypeCommon)
 		for {
-			logging.Info(fmt.Sprintf("current concurrency:%d", node.CurrentGoroutineNum()))
+			logging.Info(fmt.Sprintf("current concurrency:%d", node.CurrentConcurrency()))
 			time.Sleep(time.Duration(100) * time.Millisecond)
 		}
 	}()