Repository: alibaba/sentinel-golang (go)

Issue:
Rule Manager: Make LoadRule() serialization  and introduce recover logic when panic for LoadRules in core module
## Issue Description

Type: *feature request*

### Describe what feature you want
Currently, the logic of `LoadRules` is below:
1. acquire Write Lock
2. rebuild traffic shaping control from new rules
3. update core module's tcMap
4. release Write Lock
5. defer test recover().

There are two question:
1. The Write Lock take too long
2. If occurs panic when LoadRules, the origin tcMap could not be rollback.

I have a proposal to optimize this issue: (for example: circuit breaker module)
1. Define a channel(no buffer) in circuit breaker  manager, LoadRules/ClearRule/(other write operation) push rules into channel
. It guarantee the single thread to write.
2. init function of  circuit breaker module run a goroutine to receive channel and do Write operation.
3. Write operation copy the origin tcMap(need write lock), and do load logic. If any panic occurs, rollback to copied tcMap.

### Additional context

Scope

- [ ] flow

- [ ] circuit breaker

- [ ] isolation

- [x] hotpot

- [ ] system


Add any other context or screenshots about the feature request here.


Pull request description:
Make flow, breaker, isolation and system LoadRule()  serialization
<!--  Thanks for submitting a pull request! Here are some tips for you:
1. Please make sure you have read and understood the contributing guidelines: https://github.com/alibaba/blob/master/CONTRIBUTING.md
2. Please make sure the PR has a corresponding issue.
-->

### Describe what this PR does / why we need it
Make flow, breaker, isolation and system LoadRule()  serialization

### Does this pull request fix one issue?

<!--If that, add "Fixes #xxxx" below in the next line. For example, Fixes #15. Otherwise, add "NONE" -->
Fixes #248
### Describe how you did it

Likes #333

### Describe how to verify it


### Special notes for reviews

Patch:
diff --git a/core/circuitbreaker/rule_manager.go b/core/circuitbreaker/rule_manager.go
index 0ca636b..8f3821d 100644
--- a/core/circuitbreaker/rule_manager.go
+++ b/core/circuitbreaker/rule_manager.go
@@ -30,10 +30,11 @@ type CircuitBreakerGenFunc func(r *Rule, reuseStat interface{}) (CircuitBreaker,
 var (
 	cbGenFuncMap = make(map[Strategy]CircuitBreakerGenFunc, 4)
 
-	breakerRules = make(map[string][]*Rule)
-	breakers     = make(map[string][]CircuitBreaker)
-	updateMux    = &sync.RWMutex{}
-	currentRules = make([]*Rule, 0)
+	breakerRules  = make(map[string][]*Rule)
+	breakers      = make(map[string][]CircuitBreaker)
+	updateMux     = &sync.RWMutex{}
+	currentRules  = make([]*Rule, 0)
+	updateRuleMux = new(sync.Mutex)
 
 	stateChangeListeners = make([]StateChangeListener, 0)
 )
@@ -132,9 +133,9 @@ func ClearRules() error {
 // error: was designed to indicate whether occurs the error.
 func LoadRules(rules []*Rule) (bool, error) {
 	// TODO in order to avoid invalid update, should check consistent with last update rules
-	updateMux.RLock()
+	updateRuleMux.Lock()
+	defer updateRuleMux.Unlock()
 	isEqual := reflect.DeepEqual(currentRules, rules)
-	updateMux.RUnlock()
 	if isEqual {
 		logging.Info("[CircuitBreaker] Load rules is the same with current rules, so ignore load operation.")
 		return false, nil
@@ -230,20 +231,20 @@ func onRuleUpdate(rules []*Rule) (err error) {
 	}
 
 	start := util.CurrentTimeNano()
-	updateMux.Lock()
-	defer func() {
-		updateMux.Unlock()
-		if r := recover(); r != nil {
-			return
-		}
-		logging.Debug("[CircuitBreaker onRuleUpdate] Time statistics(ns) for updating circuit breaker rule", "timeCost", util.CurrentTimeNano()-start)
-		logRuleUpdate(newBreakerRules)
-	}()
+
+	updateMux.RLock()
+	breakersClone := make(map[string][]CircuitBreaker, len(newBreakerRules))
+	for res, tcs := range breakers {
+		resTcClone := make([]CircuitBreaker, 0, len(tcs))
+		resTcClone = append(resTcClone, tcs...)
+		breakersClone[res] = resTcClone
+	}
+	updateMux.RUnlock()
 
 	for res, resRules := range newBreakerRules {
 		emptyCircuitBreakerList := make([]CircuitBreaker, 0, 0)
 		for _, r := range resRules {
-			oldResCbs := breakers[res]
+			oldResCbs := breakersClone[res]
 			if oldResCbs == nil {
 				oldResCbs = emptyCircuitBreakerList
 			}
@@ -255,7 +256,7 @@ func onRuleUpdate(rules []*Rule) (err error) {
 				equalOldCb := oldResCbs[equalIdx]
 				insertCbToCbMap(equalOldCb, res, newBreakers)
 				// remove old cb from oldResCbs
-				breakers[res] = append(oldResCbs[:equalIdx], oldResCbs[equalIdx+1:]...)
+				breakersClone[res] = append(oldResCbs[:equalIdx], oldResCbs[equalIdx+1:]...)
 				continue
 			}
 
@@ -278,7 +279,7 @@ func onRuleUpdate(rules []*Rule) (err error) {
 			}
 
 			if reuseStatIdx >= 0 {
-				breakers[res] = append(oldResCbs[:reuseStatIdx], oldResCbs[reuseStatIdx+1:]...)
+				breakersClone[res] = append(oldResCbs[:reuseStatIdx], oldResCbs[reuseStatIdx+1:]...)
 			}
 			insertCbToCbMap(cb, res, newBreakers)
 		}
@@ -292,10 +293,14 @@ func onRuleUpdate(rules []*Rule) (err error) {
 		}
 	}
 
+	updateMux.Lock()
 	breakerRules = newBreakerRules
 	breakers = newBreakers
 	currentRules = rules
+	updateMux.Unlock()
 
+	logging.Debug("[CircuitBreaker onRuleUpdate] Time statistics(ns) for updating circuit breaker rule", "timeCost", util.CurrentTimeNano()-start)
+	logRuleUpdate(newBreakerRules)
 	return nil
 }
 
diff --git a/core/flow/rule_manager.go b/core/flow/rule_manager.go
index f1f75e4..4d62ed1 100644
--- a/core/flow/rule_manager.go
+++ b/core/flow/rule_manager.go
@@ -49,7 +49,8 @@ var (
 		readOnlyMetric:    base.NopReadStat(),
 		writeOnlyMetric:   base.NopWriteStat(),
 	}
-	currentRules = make([]*Rule, 0)
+	currentRules  = make([]*Rule, 0)
+	updateRuleMux = new(sync.Mutex)
 )
 
 func init() {
@@ -126,12 +127,18 @@ func init() {
 	}
 }
 
-func logRuleUpdate(m TrafficControllerMap) {
-	rs := rulesFrom(m)
-	if len(rs) == 0 {
+func logRuleUpdate(m map[string][]*Rule) {
+	rules := make([]*Rule, 0, 8)
+	for _, rs := range m {
+		if len(rs) == 0 {
+			continue
+		}
+		rules = append(rules, rs...)
+	}
+	if len(rules) == 0 {
 		logging.Info("[FlowRuleManager] Flow rules were cleared")
 	} else {
-		logging.Info("[FlowRuleManager] Flow rules were loaded", "rules", rs)
+		logging.Info("[FlowRuleManager] Flow rules were loaded", "rules", rules)
 	}
 }
 
@@ -160,17 +167,18 @@ func onRuleUpdate(rules []*Rule) (err error) {
 	}
 	m := make(TrafficControllerMap, len(resRulesMap))
 	start := util.CurrentTimeNano()
-	tcMux.Lock()
-	defer func() {
-		tcMux.Unlock()
-		if r := recover(); r != nil {
-			return
-		}
-		logging.Debug("[Flow onRuleUpdate] Time statistic(ns) for updating flow rule", "timeCost", util.CurrentTimeNano()-start)
-		logRuleUpdate(m)
-	}()
+
+	tcMux.RLock()
+	tcMapClone := make(TrafficControllerMap, len(resRulesMap))
+	for res, tcs := range tcMap {
+		resTcClone := make([]*TrafficShapingController, 0, len(tcs))
+		resTcClone = append(resTcClone, tcs...)
+		tcMapClone[res] = resTcClone
+	}
+	tcMux.RUnlock()
+
 	for res, rulesOfRes := range resRulesMap {
-		m[res] = buildRulesOfRes(res, rulesOfRes)
+		m[res] = buildRulesOfRes(res, rulesOfRes, tcMapClone)
 	}
 
 	for res, tcs := range m {
@@ -180,8 +188,14 @@ func onRuleUpdate(rules []*Rule) (err error) {
 			misc.RegisterStatSlotForResource(res, DefaultStandaloneStatSlot)
 		}
 	}
+
+	tcMux.Lock()
 	tcMap = m
 	currentRules = rules
+	tcMux.Unlock()
+
+	logging.Debug("[Flow onRuleUpdate] Time statistic(ns) for updating flow rule", "timeCost", util.CurrentTimeNano()-start)
+	logRuleUpdate(resRulesMap)
 	return nil
 }
 
@@ -189,9 +203,9 @@ func onRuleUpdate(rules []*Rule) (err error) {
 // the first returned value indicates whether do real load operation, if the rules is the same with previous rules, return false
 func LoadRules(rules []*Rule) (bool, error) {
 	// TODO: rethink the design
-	tcMux.RLock()
+	updateRuleMux.Lock()
+	defer updateRuleMux.Unlock()
 	isEqual := reflect.DeepEqual(currentRules, rules)
-	tcMux.RUnlock()
 	if isEqual {
 		logging.Info("[Flow] Load rules is the same with current rules, so ignore load operation.")
 		return false, nil
@@ -411,7 +425,7 @@ func calculateReuseIndexFor(r *Rule, oldResTcs []*TrafficShapingController) (equ
 }
 
 // buildRulesOfRes builds TrafficShapingController slice from rules. the resource of rules must be equals to res
-func buildRulesOfRes(res string, rulesOfRes []*Rule) []*TrafficShapingController {
+func buildRulesOfRes(res string, rulesOfRes []*Rule, m TrafficControllerMap) []*TrafficShapingController {
 	newTcsOfRes := make([]*TrafficShapingController, 0, len(rulesOfRes))
 	emptyTcs := make([]*TrafficShapingController, 0, 0)
 	for _, rule := range rulesOfRes {
@@ -419,7 +433,7 @@ func buildRulesOfRes(res string, rulesOfRes []*Rule) []*TrafficShapingController
 			logging.Error(errors.Errorf("unmatched resource name expect: %s, actual: %s", res, rule.Resource), "Unmatched resource name in flow.buildRulesOfRes()", "rule", rule)
 			continue
 		}
-		oldResTcs, exist := tcMap[res]
+		oldResTcs, exist := m[res]
 		if !exist {
 			oldResTcs = emptyTcs
 		}
@@ -432,7 +446,7 @@ func buildRulesOfRes(res string, rulesOfRes []*Rule) []*TrafficShapingController
 			equalOldTc := oldResTcs[equalIdx]
 			newTcsOfRes = append(newTcsOfRes, equalOldTc)
 			// remove old tc from oldResTcs
-			tcMap[res] = append(oldResTcs[:equalIdx], oldResTcs[equalIdx+1:]...)
+			m[res] = append(oldResTcs[:equalIdx], oldResTcs[equalIdx+1:]...)
 			continue
 		}
 
@@ -458,7 +472,7 @@ func buildRulesOfRes(res string, rulesOfRes []*Rule) []*TrafficShapingController
 		}
 		if reuseStatIdx >= 0 {
 			// remove old tc from oldResTcs
-			tcMap[res] = append(oldResTcs[:reuseStatIdx], oldResTcs[reuseStatIdx+1:]...)
+			m[res] = append(oldResTcs[:reuseStatIdx], oldResTcs[reuseStatIdx+1:]...)
 		}
 		newTcsOfRes = append(newTcsOfRes, tc)
 	}
diff --git a/core/isolation/rule_manager.go b/core/isolation/rule_manager.go
index 2b51f9f..f1654f9 100644
--- a/core/isolation/rule_manager.go
+++ b/core/isolation/rule_manager.go
@@ -25,17 +25,18 @@ import (
 )
 
 var (
-	ruleMap      = make(map[string][]*Rule)
-	rwMux        = &sync.RWMutex{}
-	currentRules = make([]*Rule, 0)
+	ruleMap       = make(map[string][]*Rule)
+	rwMux         = &sync.RWMutex{}
+	currentRules  = make([]*Rule, 0)
+	updateRuleMux = new(sync.Mutex)
 )
 
 // LoadRules loads the given isolation rules to the rule manager, while all previous rules will be replaced.
 // the first returned value indicates whether do real load operation, if the rules is the same with previous rules, return false
 func LoadRules(rules []*Rule) (bool, error) {
-	rwMux.RLock()
+	updateRuleMux.Lock()
+	defer updateRuleMux.Unlock()
 	isEqual := reflect.DeepEqual(currentRules, rules)
-	rwMux.RUnlock()
 	if isEqual {
 		logging.Info("[Isolation] Load rules is the same with current rules, so ignore load operation.")
 		return false, nil
@@ -60,12 +61,6 @@ func onRuleUpdate(rules []*Rule) (err error) {
 	}
 
 	start := util.CurrentTimeNano()
-	rwMux.Lock()
-	defer func() {
-		rwMux.Unlock()
-		logging.Debug("[Isolation LoadRules] Time statistic(ns) for updating isolation rule", "timeCost", util.CurrentTimeNano()-start)
-		logRuleUpdate(m)
-	}()
 
 	for res, rs := range m {
 		if len(rs) > 0 {
@@ -73,8 +68,13 @@ func onRuleUpdate(rules []*Rule) (err error) {
 			misc.RegisterRuleCheckSlotForResource(res, DefaultSlot)
 		}
 	}
+	rwMux.Lock()
 	ruleMap = m
 	currentRules = rules
+	rwMux.Unlock()
+
+	logging.Debug("[Isolation LoadRules] Time statistic(ns) for updating isolation rule", "timeCost", util.CurrentTimeNano()-start)
+	logRuleUpdate(m)
 	return
 }
 
diff --git a/core/system/rule_manager.go b/core/system/rule_manager.go
index 5a73a1d..38a79d4 100644
--- a/core/system/rule_manager.go
+++ b/core/system/rule_manager.go
@@ -15,6 +15,7 @@
 package system
 
 import (
+	"reflect"
 	"sync"
 
 	"github.com/alibaba/sentinel-golang/logging"
@@ -26,8 +27,10 @@ type RuleMap map[MetricType][]*Rule
 
 // const
 var (
-	ruleMap    = make(RuleMap)
-	ruleMapMux = new(sync.RWMutex)
+	ruleMap       = make(RuleMap)
+	ruleMapMux    = new(sync.RWMutex)
+	currentRules  = make([]*Rule, 0)
+	updateRuleMux = new(sync.Mutex)
 )
 
 // GetRules returns all the rules based on copy.
@@ -64,13 +67,21 @@ func getRules() []*Rule {
 
 // LoadRules loads given system rules to the rule manager, while all previous rules will be replaced.
 func LoadRules(rules []*Rule) (bool, error) {
+	updateRuleMux.Lock()
+	defer updateRuleMux.Unlock()
+	isEqual := reflect.DeepEqual(currentRules, rules)
+	if isEqual {
+		logging.Info("[System] Load rules is the same with current rules, so ignore load operation.")
+		return false, nil
+	}
+
 	m := buildRuleMap(rules)
 
 	if err := onRuleUpdate(m); err != nil {
 		logging.Error(err, "Fail to load rules in system.LoadRules()", "rules", rules)
 		return false, err
 	}
-
+	currentRules = rules
 	return true, nil
 }
 
@@ -83,16 +94,15 @@ func ClearRules() error {
 func onRuleUpdate(r RuleMap) error {
 	start := util.CurrentTimeNano()
 	ruleMapMux.Lock()
-	defer func() {
-		ruleMapMux.Unlock()
-		logging.Debug("[System onRuleUpdate] Time statistic(ns) for updating system rule", "timeCost", util.CurrentTimeNano()-start)
-		if len(r) > 0 {
-			logging.Info("[SystemRuleManager] System rules loaded", "rules", r)
-		} else {
-			logging.Info("[SystemRuleManager] System rules were cleared")
-		}
-	}()
 	ruleMap = r
+	ruleMapMux.Unlock()
+
+	logging.Debug("[System onRuleUpdate] Time statistic(ns) for updating system rule", "timeCost", util.CurrentTimeNano()-start)
+	if len(r) > 0 {
+		logging.Info("[SystemRuleManager] System rules loaded", "rules", r)
+	} else {
+		logging.Info("[SystemRuleManager] System rules were cleared")
+	}
 	return nil
 }
 
diff --git a/ext/datasource/helper.go b/ext/datasource/helper.go
index 5466a72..2f07581 100644
--- a/ext/datasource/helper.go
+++ b/ext/datasource/helper.go
@@ -112,8 +112,8 @@ func SystemRulesUpdater(data interface{}) error {
 			fmt.Sprintf("Fail to type assert data to []system.Rule or []*system.Rule, in fact, data: %+v", data),
 		)
 	}
-	succ, err := system.LoadRules(rules)
-	if succ && err == nil {
+	_, err := system.LoadRules(rules)
+	if err == nil {
 		return nil
 	}
 	return NewError(


Test patch:
diff --git a/core/flow/rule_manager_test.go b/core/flow/rule_manager_test.go
index 36144a4..b9e3fac 100644
--- a/core/flow/rule_manager_test.go
+++ b/core/flow/rule_manager_test.go
@@ -263,7 +263,7 @@ func Test_buildRulesOfRes(t *testing.T) {
 			MaxQueueingTimeMs:      10,
 		}
 		assert.True(t, len(tcMap["abc1"]) == 0)
-		tcs := buildRulesOfRes("abc1", []*Rule{r1, r2})
+		tcs := buildRulesOfRes("abc1", []*Rule{r1, r2}, tcMap)
 		assert.True(t, len(tcs) == 2)
 		assert.True(t, tcs[0].BoundRule() == r1)
 		assert.True(t, tcs[1].BoundRule() == r2)
@@ -413,7 +413,7 @@ func Test_buildRulesOfRes(t *testing.T) {
 			StatIntervalInMs:       50000,
 		}
 
-		tcs := buildRulesOfRes("abc1", []*Rule{r12, r22, r32, r42})
+		tcs := buildRulesOfRes("abc1", []*Rule{r12, r22, r32, r42}, tcMap)
 		assert.True(t, len(tcs) == 4)
 
 		assert.True(t, tcs[0].BoundRule() == r12)
diff --git a/core/system/rule_manager_test.go b/core/system/rule_manager_test.go
index e3c12fd..9a25b72 100644
--- a/core/system/rule_manager_test.go
+++ b/core/system/rule_manager_test.go
@@ -73,12 +73,15 @@ func TestClearRules(t *testing.T) {
 	})
 
 	t.Run("NoEmptyOriginRuleMap", func(t *testing.T) {
-		r := map[MetricType][]*Rule{
-			InboundQPS:  {&Rule{MetricType: InboundQPS, TriggerCount: 1}},
-			Concurrency: {&Rule{MetricType: Concurrency, TriggerCount: 2}},
+		r := []*Rule{
+			{MetricType: InboundQPS, TriggerCount: 1},
+			{MetricType: Concurrency, TriggerCount: 2},
 		}
-		ruleMap = r
-		err := ClearRules()
+		isOK, err := LoadRules(r)
+		assert.Equal(t, true, isOK)
+		assert.Nil(t, err)
+		assert.Equal(t, 2, len(ruleMap))
+		err = ClearRules()
 		assert.Nil(t, err)
 		assert.Equal(t, 0, len(ruleMap))
 	})