AlexStocks commented on code in PR #1081:
URL:
https://github.com/apache/incubator-seata-go/pull/1081#discussion_r3004257734
##########
pkg/remoting/loadbalance/consistent_hash_loadbalance.go:
##########
@@ -57,61 +51,85 @@ func (c *Consistent) hash(key string) int64 {
return res
}
-// pick get a node
-func (c *Consistent) pick(sessions *sync.Map, key string) getty.Session {
- hashKey := c.hash(key)
+func (c *Consistent) pickByHash(hashKey int64) getty.Session {
Review Comment:
[P0] 修复数据竞争。原代码 pick() 读取 sortedHashNodes 无锁保护，修复后 pickByHash() 使用 RLock
保护读取。
##########
pkg/remoting/loadbalance/consistent_hash_loadbalance.go:
##########
@@ -57,61 +51,85 @@ func (c *Consistent) hash(key string) int64 {
return res
}
-// pick get a node
-func (c *Consistent) pick(sessions *sync.Map, key string) getty.Session {
- hashKey := c.hash(key)
+func (c *Consistent) pickByHash(hashKey int64) getty.Session {
+ c.RLock()
+ defer c.RUnlock()
+
+ if len(c.sortedHashNodes) == 0 {
+ return nil
+ }
+
index := sort.Search(len(c.sortedHashNodes), func(i int) bool {
return c.sortedHashNodes[i] >= hashKey
})
-
if index == len(c.sortedHashNodes) {
- return RandomLoadBalance(sessions, key)
+ index = 0
}
- c.RLock()
- session, ok := c.hashCircle[c.sortedHashNodes[index]]
- if !ok {
- c.RUnlock()
- return RandomLoadBalance(sessions, key)
+ return c.hashCircle[c.sortedHashNodes[index]]
+}
+
+// pick get a node
+func (c *Consistent) pick(sessions *sync.Map, key string) getty.Session {
+ hashKey := c.hash(key)
+ session := c.pickByHash(hashKey)
+ if session == nil {
+ c.refreshHashCircle(sessions)
+ session = c.pickByHash(hashKey)
+ if session == nil {
+ return RandomLoadBalance(sessions, key)
+ }
}
- c.RUnlock()
if session.IsClosed() {
- go c.refreshHashCircle(sessions)
- return c.firstKey()
+ c.refreshHashCircle(sessions)
+ session = c.pickByHash(hashKey)
+ if session == nil || session.IsClosed() {
+ return RandomLoadBalance(sessions, key)
+ }
}
return session
}
// refreshHashCircle refresh hashCircle
func (c *Consistent) refreshHashCircle(sessions *sync.Map) {
- var sortedHashNodes []int64
- hashCircle := make(map[int64]getty.Session)
- var session getty.Session
- c.RLock()
- defer c.RUnlock()
+ var (
+ sortedHashNodes []int64
+ hashCircle = make(map[int64]getty.Session)
+ closedSessions []interface{}
+ )
+
sessions.Range(func(key, value interface{}) bool {
- session = key.(getty.Session)
+ session := key.(getty.Session)
+ if session.IsClosed() {
+ closedSessions = append(closedSessions, key)
+ return true
+ }
+
for i := 0; i < defaultVirtualNodeNumber; i++ {
if !session.IsClosed() {
position := c.hash(fmt.Sprintf("%s%d",
session.RemoteAddr(), i))
hashCircle[position] = session
sortedHashNodes = append(sortedHashNodes,
position)
- } else {
- sessions.Delete(key)
}
}
return true
})
+ for _, session := range closedSessions {
+ sessions.Delete(session)
+ }
+
// virtual node sort
sort.Slice(sortedHashNodes, func(i, j int) bool {
return sortedHashNodes[i] < sortedHashNodes[j]
})
Review Comment:
[P0] 修复写锁问题。原代码 refreshHashCircle() 使用 RLock 写入共享状态，修复后使用 Lock 保护写入。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
