This is an automated email from the ASF dual-hosted git repository.
lxfeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 1a031ae3 fix: synchronize consistent hash refresh (#1081)
1a031ae3 is described below
commit 1a031ae35ad9d68e7226f99302860716db853c69
Author: Zhifan C <[email protected]>
AuthorDate: Thu Apr 2 11:22:19 2026 +0800
fix: synchronize consistent hash refresh (#1081)
* fix: synchronize consistent hash refresh
* test: add concurrent race coverage for consistent hash refresh
---------
Co-authored-by: Konnyaku <[email protected]>
---
.../loadbalance/consistent_hash_loadbalance.go | 95 +++++++--------
.../consistent_hash_loadbalance_test.go | 132 +++++++++++++++++++++
2 files changed, 180 insertions(+), 47 deletions(-)
diff --git a/pkg/remoting/loadbalance/consistent_hash_loadbalance.go
b/pkg/remoting/loadbalance/consistent_hash_loadbalance.go
index 54f696c3..6875e445 100644
--- a/pkg/remoting/loadbalance/consistent_hash_loadbalance.go
+++ b/pkg/remoting/loadbalance/consistent_hash_loadbalance.go
@@ -40,12 +40,6 @@ type Consistent struct {
sortedHashNodes []int64
}
-func (c *Consistent) put(key int64, session getty.Session) {
- c.Lock()
- defer c.Unlock()
- c.hashCircle[key] = session
-}
-
func (c *Consistent) hash(key string) int64 {
hashByte := md5.Sum([]byte(key))
var res int64
@@ -57,28 +51,42 @@ 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
@@ -86,32 +94,42 @@ func (c *Consistent) pick(sessions *sync.Map, key string)
getty.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]
})
+ c.Lock()
c.sortedHashNodes = sortedHashNodes
c.hashCircle = hashCircle
+ c.Unlock()
}
func (c *Consistent) firstKey() getty.Session {
@@ -128,27 +146,10 @@ func (c *Consistent) firstKey() getty.Session {
func newConsistenceInstance(sessions *sync.Map) *Consistent {
once.Do(func() {
consistentInstance = &Consistent{
- hashCircle: make(map[int64]getty.Session),
+ virtualNodeCount: defaultVirtualNodeNumber,
+ hashCircle: make(map[int64]getty.Session),
}
- // construct hash circle
- sessions.Range(func(key, value interface{}) bool {
- session := key.(getty.Session)
- for i := 0; i < defaultVirtualNodeNumber; i++ {
- if !session.IsClosed() {
- position :=
consistentInstance.hash(fmt.Sprintf("%s%d", session.RemoteAddr(), i))
- consistentInstance.put(position,
session)
- consistentInstance.sortedHashNodes =
append(consistentInstance.sortedHashNodes, position)
- } else {
- sessions.Delete(key)
- }
- }
- return true
- })
-
- // virtual node sort
- sort.Slice(consistentInstance.sortedHashNodes, func(i, j int)
bool {
- return consistentInstance.sortedHashNodes[i] <
consistentInstance.sortedHashNodes[j]
- })
+ consistentInstance.refreshHashCircle(sessions)
})
return consistentInstance
diff --git a/pkg/remoting/loadbalance/consistent_hash_loadbalance_test.go
b/pkg/remoting/loadbalance/consistent_hash_loadbalance_test.go
index 250a2638..24e60edf 100644
--- a/pkg/remoting/loadbalance/consistent_hash_loadbalance_test.go
+++ b/pkg/remoting/loadbalance/consistent_hash_loadbalance_test.go
@@ -20,15 +20,25 @@ package loadbalance
import (
"fmt"
"sync"
+ "sync/atomic"
"testing"
+ getty "github.com/apache/dubbo-getty"
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
"seata.apache.org/seata-go/v2/pkg/remoting/mock"
)
+func resetConsistentHashForTest() {
+ consistentInstance = nil
+ once = sync.Once{}
+}
+
func TestConsistentHashLoadBalance(t *testing.T) {
+ resetConsistentHashForTest()
+ defer resetConsistentHashForTest()
+
ctrl := gomock.NewController(t)
sessions := &sync.Map{}
@@ -50,3 +60,125 @@ func TestConsistentHashLoadBalance(t *testing.T) {
return true
})
}
+
+func TestConsistentPick_RefreshesClosedSession(t *testing.T) {
+ resetConsistentHashForTest()
+ defer resetConsistentHashForTest()
+
+ ctrl := gomock.NewController(t)
+ sessions := &sync.Map{}
+
+ closedSession := mock.NewMockTestSession(ctrl)
+ closedSession.EXPECT().IsClosed().AnyTimes().Return(true)
+
+ openSession := mock.NewMockTestSession(ctrl)
+ openSession.EXPECT().IsClosed().AnyTimes().Return(false)
+ openSession.EXPECT().RemoteAddr().AnyTimes().Return("127.0.0.1:8001")
+
+ sessions.Store(closedSession, "closed")
+ sessions.Store(openSession, "open")
+
+ c := &Consistent{
+ virtualNodeCount: defaultVirtualNodeNumber,
+ hashCircle: map[int64]getty.Session{
+ 1: closedSession,
+ },
+ sortedHashNodes: []int64{1},
+ }
+
+ result := c.pick(sessions, "test_xid")
+ assert.NotNil(t, result)
+ assert.Equal(t, openSession, result)
+
+ _, stillExists := sessions.Load(closedSession)
+ assert.False(t, stillExists)
+}
+
+func TestConsistentPick_ConcurrentPickAndRefresh(t *testing.T) {
+ resetConsistentHashForTest()
+ defer resetConsistentHashForTest()
+
+ ctrl := gomock.NewController(t)
+ sessions := &sync.Map{}
+
+ stableSession := mock.NewMockTestSession(ctrl)
+ stableSession.EXPECT().IsClosed().AnyTimes().Return(false)
+ stableSession.EXPECT().RemoteAddr().AnyTimes().Return("127.0.0.1:9000")
+ sessions.Store(stableSession, "stable")
+
+ type flappingSession struct {
+ session getty.Session
+ addr string
+ closed atomic.Bool
+ }
+
+ flapping := make([]*flappingSession, 0, 2)
+ for i := 0; i < 2; i++ {
+ addr := fmt.Sprintf("127.0.0.1:900%d", i+1)
+ state := &flappingSession{addr: addr}
+
+ session := mock.NewMockTestSession(ctrl)
+ session.EXPECT().IsClosed().AnyTimes().DoAndReturn(func() bool {
+ return state.closed.Load()
+ })
+ session.EXPECT().RemoteAddr().AnyTimes().Return(addr)
+
+ state.session = session
+ flapping = append(flapping, state)
+ sessions.Store(session, addr)
+ }
+
+ c := &Consistent{
+ virtualNodeCount: defaultVirtualNodeNumber,
+ hashCircle: make(map[int64]getty.Session),
+ }
+ c.refreshHashCircle(sessions)
+
+ const (
+ pickers = 8
+ iterations = 200
+ )
+
+ start := make(chan struct{})
+ var wg sync.WaitGroup
+
+ for pickerID := 0; pickerID < pickers; pickerID++ {
+ pickerID := pickerID
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ for i := 0; i < iterations; i++ {
+ session := c.pick(sessions,
fmt.Sprintf("xid-%d-%d", pickerID, i))
+ if session == nil {
+ continue
+ }
+ _ = session.IsClosed()
+ }
+ }()
+ }
+
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ for i := 0; i < iterations; i++ {
+ state := flapping[i%len(flapping)]
+ if i%2 == 0 {
+ state.closed.Store(true)
+ } else {
+ state.closed.Store(false)
+ sessions.Store(state.session, state.addr)
+ }
+
+ c.refreshHashCircle(sessions)
+ }
+ }()
+
+ close(start)
+ wg.Wait()
+
+ result := c.pick(sessions, "final-xid")
+ assert.NotNil(t, result)
+ assert.False(t, result.IsClosed())
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
