This is an automated email from the ASF dual-hosted git repository.
thunguo 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 e74ab61d [GSOC] Wire registry change snapshots into remoting address
lifecycle (#1138)
e74ab61d is described below
commit e74ab61d7efc331efcda4da911a0ab8e4e326437
Author: CAICAII <[email protected]>
AuthorDate: Mon Aug 24 01:28:37 2026 +0800
[GSOC] Wire registry change snapshots into remoting address lifecycle
(#1138)
* feat(discovery): add registry change subscriptions
Signed-off-by: CAICAIIs <[email protected]>
* fix(discovery): address subscription review feedback
Signed-off-by: CAICAIIs <[email protected]>
* feat(discovery): adapt namingserver registry subscriptions
Signed-off-by: CAICAIIs <[email protected]>
* feat(remoting): consume registry address subscriptions
Signed-off-by: CAICAIIs <[email protected]>
* feat(remoting): consume registry changes in grpc
Signed-off-by: CAICAIIs <[email protected]>
* feat(discovery): add raft registry subscriptions
Signed-off-by: CAICAIIs <[email protected]>
* fix(remoting): validate cached grpc channels
Signed-off-by: CAICAIIs <[email protected]>
* fix(registry): harden dynamic address lifecycle
Signed-off-by: CAICAIIs <[email protected]>
* test(remoting): avoid empty critical section in close test
Signed-off-by: CAICAIIs <[email protected]>
---------
Signed-off-by: CAICAIIs <[email protected]>
Co-authored-by: ssshr-66 <[email protected]>
Co-authored-by: Ethan <[email protected]>
Co-authored-by: ThunGuo <[email protected]>
---
pkg/discovery/base.go | 24 +++
pkg/discovery/etcd3.go | 79 ++++++-
pkg/discovery/etcd3_test.go | 44 ++++
pkg/discovery/file_test.go | 11 +
pkg/discovery/metadata/metadata.go | 31 ++-
pkg/discovery/metadata/metadata_test.go | 26 +++
pkg/discovery/naming_server.go | 216 +++++++++++++++++--
pkg/discovery/naming_server_test.go | 133 ++++++++++++
pkg/discovery/raft.go | 147 ++++++++++++-
pkg/discovery/raft_subscription_test.go | 133 ++++++++++++
pkg/discovery/store.go | 11 +-
pkg/discovery/subscription.go | 132 ++++++++++++
pkg/discovery/subscription_test.go | 136 ++++++++++++
pkg/remoting/getty/getty_client.go | 10 +-
pkg/remoting/getty/getty_client_test.go | 14 +-
pkg/remoting/getty/session_manager.go | 284 ++++++++++++++++++++++---
pkg/remoting/getty/session_manager_test.go | 236 +++++++++++++++++++++
pkg/remoting/grpc/channel.go | 22 +-
pkg/remoting/grpc/channel_manager.go | 323 +++++++++++++++++++++++------
pkg/remoting/grpc/channel_manager_test.go | 205 ++++++++++++++++++
pkg/remoting/grpc/listener.go | 12 +-
21 files changed, 2081 insertions(+), 148 deletions(-)
diff --git a/pkg/discovery/base.go b/pkg/discovery/base.go
index 9a30b04e..0c3745b2 100644
--- a/pkg/discovery/base.go
+++ b/pkg/discovery/base.go
@@ -45,3 +45,27 @@ type RegisterableRegistryService interface {
Register(instance *ServiceInstance) error
Unregister(instance *ServiceInstance) error
}
+
+// RegistryChangeEvent is a backend-neutral snapshot of service instances for a
+// transaction service group.
+type RegistryChangeEvent struct {
+ Key string
+ Instances []*ServiceInstance
+}
+
+// RegistryChangeListener receives backend-neutral registry change snapshots.
+type RegistryChangeListener func(event RegistryChangeEvent)
+
+// RegistrySubscription cancels a registry change subscription. Unsubscribe is
+// safe to call multiple times. A callback already in progress may still run.
+type RegistrySubscription interface {
+ Unsubscribe()
+}
+
+// RegistrySubscriber is an optional capability for registries that can report
+// address changes. Listeners receive the current snapshot first, followed by
+// backend-neutral snapshots. Calls for one subscription are serialized. Slow
+// listeners may skip intermediate snapshots.
+type RegistrySubscriber interface {
+ Subscribe(key string, listener RegistryChangeListener)
(RegistrySubscription, error)
+}
diff --git a/pkg/discovery/etcd3.go b/pkg/discovery/etcd3.go
index 3997ab88..0f2547d8 100644
--- a/pkg/discovery/etcd3.go
+++ b/pkg/discovery/etcd3.go
@@ -39,13 +39,19 @@ const (
type EtcdRegistryService struct {
client *etcd3.Client
cfg etcd3.Config
- vgroupMapping map[string]string
+ vgroupMapping map[string]string // copied during construction;
read-only afterwards
store *AddressStore
stopCh chan struct{}
closeOnce sync.Once
+
+ subscriptionsMu sync.Mutex
+ subscriptions map[*registryChangeSubscription]struct{}
+ closed bool
}
+var _ RegistrySubscriber = (*EtcdRegistryService)(nil)
+
func newEtcdRegistryService(config *ServiceConfig, etcd3Config *Etcd3Config)
(RegistryService, error) {
if config == nil {
return nil, fmt.Errorf("service config is nil")
@@ -62,7 +68,10 @@ func newEtcdRegistryService(config *ServiceConfig,
etcd3Config *Etcd3Config) (Re
return nil, fmt.Errorf("failed to create etcd3 client: %w", err)
}
- vgroupMapping := config.VgroupMapping
+ vgroupMapping := make(map[string]string, len(config.VgroupMapping))
+ for key, cluster := range config.VgroupMapping {
+ vgroupMapping[key] = cluster
+ }
etcdRegistryService := &EtcdRegistryService{
client: cli,
@@ -104,8 +113,11 @@ func (s *EtcdRegistryService) watch(key string) {
}
}
- // watch the changes of endpoints
- watchCh := s.client.Watch(ctx, key, etcd3.WithPrefix())
+ watchOptions := []etcd3.OpOption{etcd3.WithPrefix()}
+ if resp != nil && resp.Header != nil {
+ watchOptions = append(watchOptions,
etcd3.WithRev(resp.Header.Revision+1))
+ }
+ watchCh := s.client.Watch(ctx, key, watchOptions...)
for {
select {
@@ -198,14 +210,71 @@ func getClusterAndAddress(key []byte) (string, string,
int, error) {
func (s *EtcdRegistryService) Lookup(key string) ([]*ServiceInstance, error) {
cluster := s.vgroupMapping[key]
if cluster == "" {
- return nil, fmt.Errorf("cluster doesnt exit")
+ return nil, fmt.Errorf("cluster doesn't exist")
}
return s.store.Snapshot(cluster), nil
}
+func (s *EtcdRegistryService) Subscribe(key string, listener
RegistryChangeListener) (RegistrySubscription, error) {
+ if listener == nil {
+ return nil, fmt.Errorf("registry change listener is nil")
+ }
+
+ cluster := s.vgroupMapping[key]
+ if cluster == "" {
+ return nil, fmt.Errorf("cluster doesn't exist")
+ }
+
+ subscription := newRegistryChangeSubscription(listener)
+ // Register before checking closed so taking the initial snapshot
cannot miss
+ // an update. If Close wins the race, the check below removes the
callback.
+ initial, unsubscribe := s.store.subscribeWithSnapshot(cluster,
func(changedCluster string, instances []*ServiceInstance) {
+ if changedCluster == cluster {
+ subscription.publish(RegistryChangeEvent{Key: key,
Instances: instances})
+ }
+ })
+ subscription.initialize(RegistryChangeEvent{Key: key, Instances:
initial}, func() {
+ unsubscribe()
+ s.removeSubscription(subscription)
+ })
+
+ s.subscriptionsMu.Lock()
+ if s.closed {
+ s.subscriptionsMu.Unlock()
+ subscription.Unsubscribe()
+ return nil, fmt.Errorf("registry service is closed")
+ }
+ if s.subscriptions == nil {
+ s.subscriptions = make(map[*registryChangeSubscription]struct{})
+ }
+ s.subscriptions[subscription] = struct{}{}
+ s.subscriptionsMu.Unlock()
+
+ subscription.start()
+ return subscription, nil
+}
+
+func (s *EtcdRegistryService) removeSubscription(subscription
*registryChangeSubscription) {
+ s.subscriptionsMu.Lock()
+ delete(s.subscriptions, subscription)
+ s.subscriptionsMu.Unlock()
+}
+
func (s *EtcdRegistryService) Close() {
s.closeOnce.Do(func() {
+ s.subscriptionsMu.Lock()
+ s.closed = true
+ subscriptions := make([]*registryChangeSubscription, 0,
len(s.subscriptions))
+ for subscription := range s.subscriptions {
+ subscriptions = append(subscriptions, subscription)
+ }
+ s.subscriptions = nil
+ s.subscriptionsMu.Unlock()
+
+ for _, subscription := range subscriptions {
+ subscription.Unsubscribe()
+ }
if s.stopCh != nil {
close(s.stopCh)
}
diff --git a/pkg/discovery/etcd3_test.go b/pkg/discovery/etcd3_test.go
index e8322688..346954d1 100644
--- a/pkg/discovery/etcd3_test.go
+++ b/pkg/discovery/etcd3_test.go
@@ -25,6 +25,7 @@ import (
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
+ "go.etcd.io/etcd/api/v3/etcdserverpb"
"go.etcd.io/etcd/api/v3/mvccpb"
clientv3 "go.etcd.io/etcd/client/v3"
@@ -199,6 +200,39 @@ func TestEtcd3RegistryService_CloseIsRepeatable(t
*testing.T) {
}
}
+func TestEtcd3RegistryService_WatchContinuesAfterSnapshotRevision(t
*testing.T) {
+ ctrl := gomock.NewController(t)
+ mockEtcdClient := mock.NewMockEtcdClient(ctrl)
+ service := &EtcdRegistryService{
+ client: newTestEtcdClient(mockEtcdClient),
+ store: NewAddressStore(),
+ stopCh: make(chan struct{}),
+ }
+
+ mockEtcdClient.EXPECT().Get(gomock.Any(), gomock.Any(),
gomock.Any()).Return(&clientv3.GetResponse{
+ Header: &etcdserverpb.ResponseHeader{Revision: 41},
+ }, nil)
+ watchStarted := make(chan struct{})
+ watchCh := make(chan clientv3.WatchResponse)
+ mockEtcdClient.EXPECT().Watch(gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any()).
+ DoAndReturn(func(_ context.Context, key string, options
...clientv3.OpOption) clientv3.WatchChan {
+ op := clientv3.OpGet(key, options...)
+ assert.Equal(t, int64(42), op.Rev())
+ close(watchStarted)
+ return watchCh
+ })
+ mockEtcdClient.EXPECT().Close().Return(nil)
+
+ watchDone := make(chan struct{})
+ go func() {
+ service.watch(etcdClusterPrefix)
+ close(watchDone)
+ }()
+ waitForSignal(t, watchStarted, "etcd watch to start")
+ service.Close()
+ waitForSignal(t, watchDone, "etcd watch to stop")
+}
+
func newTestEtcdClient(client mock.EtcdClient) *clientv3.Client {
return clientv3.NewCtxClient(
context.Background(),
@@ -208,3 +242,13 @@ func newTestEtcdClient(client mock.EtcdClient)
*clientv3.Client {
},
)
}
+
+func waitForSignal(t *testing.T, signal <-chan struct{}, name string) {
+ t.Helper()
+
+ select {
+ case <-signal:
+ case <-time.After(time.Second):
+ t.Fatalf("timed out waiting for %s", name)
+ }
+}
diff --git a/pkg/discovery/file_test.go b/pkg/discovery/file_test.go
index 9d653c13..55c27bd0 100644
--- a/pkg/discovery/file_test.go
+++ b/pkg/discovery/file_test.go
@@ -245,3 +245,14 @@ func TestFileRegistryService_Lookup(t *testing.T) {
})
}
}
+
+func TestFileRegistryService_DoesNotImplementRegistrySubscriber(t *testing.T) {
+ service := newFileRegistryService(&ServiceConfig{
+ VgroupMapping: map[string]string{"default_tx_group": "default"},
+ Grouplist: map[string]string{"default": "127.0.0.1:8091"},
+ })
+
+ if _, ok := service.(RegistrySubscriber); ok {
+ t.Fatal("file registry should keep lookup-only behavior")
+ }
+}
diff --git a/pkg/discovery/metadata/metadata.go
b/pkg/discovery/metadata/metadata.go
index 3e9b4ed4..2134c94d 100644
--- a/pkg/discovery/metadata/metadata.go
+++ b/pkg/discovery/metadata/metadata.go
@@ -134,14 +134,24 @@ func (m *Metadata) GetClusterTerm(clusterName string)
map[string]int64 {
}
func (m *Metadata) RefreshMetadata(clusterName string, response
MetadataResponse) {
+ m.refreshMetadata(clusterName, "", response)
+}
+
+// RefreshGroupMetadata replaces the metadata for one queried group.
+func (m *Metadata) RefreshGroupMetadata(clusterName, group string, response
MetadataResponse) {
+ m.refreshMetadata(clusterName, group, response)
+}
+
+func (m *Metadata) refreshMetadata(clusterName, queriedGroup string, response
MetadataResponse) {
nodesByGroup := make(map[string][]*Node)
for _, node := range response.Nodes {
- nodesByGroup[node.Group] = append(nodesByGroup[node.Group],
node)
- if node.Role == LEADER {
- groupMapAny, _ := m.leaders.LoadOrStore(clusterName,
&sync.Map{})
- groupMap := groupMapAny.(*sync.Map)
- groupMap.Store(node.Group, node)
+ if node == nil {
+ continue
}
+ nodesByGroup[node.Group] = append(nodesByGroup[node.Group],
node)
+ }
+ if len(nodesByGroup) == 0 && queriedGroup != "" {
+ nodesByGroup[queriedGroup] = nil
}
switch response.StoreMode {
@@ -152,9 +162,20 @@ func (m *Metadata) RefreshMetadata(clusterName string,
response MetadataResponse
}
if len(nodesByGroup) > 0 {
+ leaderMapAny, _ := m.leaders.LoadOrStore(clusterName,
&sync.Map{})
+ leaderMap := leaderMapAny.(*sync.Map)
termMapAny, _ := m.clusterTerm.LoadOrStore(clusterName,
&sync.Map{})
termMap := termMapAny.(*sync.Map)
for group, nodes := range nodesByGroup {
+ leaderMap.Delete(group)
+ clusterNodesAny, _ :=
m.clusterNodes.LoadOrStore(clusterName, &sync.Map{})
+ clusterNodes := clusterNodesAny.(*sync.Map)
+ clusterNodes.Delete(group)
+ for _, node := range nodes {
+ if node.Role == LEADER {
+ leaderMap.Store(group, node)
+ }
+ }
m.SetNodes(clusterName, group, nodes)
termMap.Store(group, response.Term)
}
diff --git a/pkg/discovery/metadata/metadata_test.go
b/pkg/discovery/metadata/metadata_test.go
index 01914634..ed584fcb 100644
--- a/pkg/discovery/metadata/metadata_test.go
+++ b/pkg/discovery/metadata/metadata_test.go
@@ -60,3 +60,29 @@ func TestRefreshMetadataGroupsNodesAndTermsByGroup(t
*testing.T) {
assert.Equal(t, int64(7), m.GetClusterTerm("test-cluster")["group-b"])
assert.Equal(t, RAFT, m.storeMode)
}
+
+func TestRefreshGroupMetadataReplacesRemovedNodes(t *testing.T) {
+ m := NewMetadata()
+ m.RefreshGroupMetadata("test-cluster", "group-a", MetadataResponse{
+ Term: 1,
+ Nodes: []*Node{
+ {Transaction: &Endpoint{Host: "127.0.0.1", Port: 8001},
Group: "group-a", Role: LEADER},
+ {Transaction: &Endpoint{Host: "127.0.0.1", Port: 8002},
Group: "group-a", Role: FOLLOWER},
+ },
+ })
+
+ m.RefreshGroupMetadata("test-cluster", "group-a", MetadataResponse{
+ Term: 2,
+ Nodes: []*Node{
+ {Transaction: &Endpoint{Host: "127.0.0.1", Port: 8003},
Group: "group-a", Role: LEADER},
+ },
+ })
+
+ nodes := m.GetNodes("test-cluster", "group-a")
+ assert.Len(t, nodes, 1)
+ assert.Equal(t, 8003, m.GetLeader("test-cluster").Transaction.Port)
+
+ m.RefreshGroupMetadata("test-cluster", "group-a",
MetadataResponse{Term: 3})
+ assert.Empty(t, m.GetNodes("test-cluster", "group-a"))
+ assert.Nil(t, m.GetLeader("test-cluster"))
+}
diff --git a/pkg/discovery/naming_server.go b/pkg/discovery/naming_server.go
index 711d7224..6d52f4ba 100644
--- a/pkg/discovery/naming_server.go
+++ b/pkg/discovery/naming_server.go
@@ -112,12 +112,14 @@ type NamingServerClient struct {
vgroupAddressMap sync.Map
listenerServiceMap sync.Map
subscribedVGroups sync.Map
+ nextListenerID uint64
tokenMu sync.RWMutex
jwtToken string
healthCheckTicker *time.Ticker
closeChan chan struct{}
+ closeOnce sync.Once
wg sync.WaitGroup
httpClient *http.Client
@@ -128,11 +130,27 @@ type NamingListener interface {
OnEvent(vGroup string) error
}
+type namingListenerEntry struct {
+ id uint64
+ listener NamingListener
+}
+
+type namingWatchState struct {
+ stopCh chan struct{}
+ listeners int
+ permanent bool
+}
+
type NamingServerRegistryService struct {
- client *NamingServerClient
+ client *NamingServerClient
+ closeOnce sync.Once
+ subscriptionsMu sync.Mutex
+ subscriptions map[*registryChangeSubscription]struct{}
+ closed bool
}
var _ NamingServerRegistry = (*NamingServerRegistryService)(nil)
+var _ RegistrySubscriber = (*NamingServerRegistryService)(nil)
func buildNamingServerURL(addr string, elem ...string) (string, error) {
baseURL := strings.TrimSpace(addr)
@@ -149,8 +167,82 @@ func (n *NamingServerRegistryService) Lookup(key string)
([]*ServiceInstance, er
return n.client.Lookup(key)
}
+// Subscribe reports NamingServer registry snapshots for a transaction service
group.
+func (n *NamingServerRegistryService) Subscribe(key string, listener
RegistryChangeListener) (RegistrySubscription, error) {
+ if listener == nil {
+ return nil, fmt.Errorf("registry change listener is nil")
+ }
+ if n == nil || n.client == nil {
+ return nil, fmt.Errorf("naming server client is nil")
+ }
+ if n.isClosed() {
+ return nil, fmt.Errorf("registry service is closed")
+ }
+
+ subscription := newRegistryChangeSubscription(listener)
+ unsubscribe, err := n.client.subscribe(key,
&namingRegistryChangeListener{
+ key: key,
+ client: n.client,
+ subscription: subscription,
+ }, false)
+ if err != nil {
+ return nil, err
+ }
+
+ instances, err := n.client.lookupInstances(key)
+ if err != nil {
+ unsubscribe()
+ return nil, err
+ }
+ subscription.initialize(RegistryChangeEvent{Key: key, Instances:
instances}, func() {
+ unsubscribe()
+ n.removeSubscription(subscription)
+ })
+
+ n.subscriptionsMu.Lock()
+ if n.closed {
+ n.subscriptionsMu.Unlock()
+ subscription.Unsubscribe()
+ return nil, fmt.Errorf("registry service is closed")
+ }
+ if n.subscriptions == nil {
+ n.subscriptions = make(map[*registryChangeSubscription]struct{})
+ }
+ n.subscriptions[subscription] = struct{}{}
+ n.subscriptionsMu.Unlock()
+
+ subscription.start()
+ return subscription, nil
+}
+
func (n *NamingServerRegistryService) Close() {
- n.client.Close()
+ n.closeOnce.Do(func() {
+ n.subscriptionsMu.Lock()
+ n.closed = true
+ subscriptions := make([]*registryChangeSubscription, 0,
len(n.subscriptions))
+ for subscription := range n.subscriptions {
+ subscriptions = append(subscriptions, subscription)
+ }
+ n.subscriptions = nil
+ n.subscriptionsMu.Unlock()
+
+ for _, subscription := range subscriptions {
+ subscription.Unsubscribe()
+ }
+ n.client.Close()
+ })
+}
+
+func (n *NamingServerRegistryService) removeSubscription(subscription
*registryChangeSubscription) {
+ n.subscriptionsMu.Lock()
+ delete(n.subscriptions, subscription)
+ n.subscriptionsMu.Unlock()
+}
+
+func (n *NamingServerRegistryService) isClosed() bool {
+ n.subscriptionsMu.Lock()
+ defer n.subscriptionsMu.Unlock()
+ return n.closed
}
func newNamingServerRegistryService(_ *ServiceConfig, cfg *NamingServerConfig)
RegistryService {
@@ -286,6 +378,10 @@ func (c *NamingServerClient) Lookup(vGroup string)
([]*ServiceInstance, error) {
return nil, fmt.Errorf("subscribe failed: %w", err)
}
+ return c.lookupInstances(vGroup)
+}
+
+func (c *NamingServerClient) lookupInstances(vGroup string)
([]*ServiceInstance, error) {
val, ok := c.vgroupAddressMap.Load(vGroup)
if !ok {
if err := c.RefreshGroup(vGroup); err != nil {
@@ -565,25 +661,88 @@ func (c *NamingServerClient) handleMetadata(metaResp
*MetaResponse, vGroup strin
}
func (c *NamingServerClient) Subscribe(vGroup string, listener NamingListener)
error {
+ _, err := c.subscribe(vGroup, listener, true)
+ return err
+}
+
+func (c *NamingServerClient) subscribe(vGroup string, listener NamingListener,
permanent bool) (func(), error) {
+ var listenerID uint64
+ var state *namingWatchState
+ startWatch := false
+
c.mu.Lock()
if listener != nil {
- var listeners []NamingListener
+ listenerID = atomic.AddUint64(&c.nextListenerID, 1)
+ var listeners []namingListenerEntry
if val, ok := c.listenerServiceMap.Load(vGroup); ok {
- listeners = append(listeners, val.([]NamingListener)...)
+ listeners = append(listeners,
val.([]namingListenerEntry)...)
}
- listeners = append(listeners, listener)
+ listeners = append(listeners, namingListenerEntry{id:
listenerID, listener: listener})
c.listenerServiceMap.Store(vGroup, listeners)
}
+ if val, ok := c.subscribedVGroups.Load(vGroup); ok {
+ state = val.(*namingWatchState)
+ } else {
+ state = &namingWatchState{stopCh: make(chan struct{})}
+ c.subscribedVGroups.Store(vGroup, state)
+ startWatch = true
+ }
+ if listenerID != 0 {
+ state.listeners++
+ }
+ if permanent {
+ state.permanent = true
+ }
c.mu.Unlock()
- if _, loaded := c.subscribedVGroups.LoadOrStore(vGroup, struct{}{});
!loaded {
+ if startWatch {
c.wg.Add(1)
- go c.watchLoop(vGroup)
+ go c.watchLoop(vGroup, state.stopCh)
+ }
+
+ return func() {
+ if listenerID != 0 {
+ c.unsubscribe(vGroup, listenerID)
+ }
+ }, nil
+}
+
+func (c *NamingServerClient) unsubscribe(vGroup string, listenerID uint64) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+
+ val, ok := c.listenerServiceMap.Load(vGroup)
+ if !ok {
+ return
+ }
+ listeners := val.([]namingListenerEntry)
+ next := make([]namingListenerEntry, 0, len(listeners))
+ for _, entry := range listeners {
+ if entry.id != listenerID {
+ next = append(next, entry)
+ }
+ }
+ if len(next) == 0 {
+ c.listenerServiceMap.Delete(vGroup)
+ } else {
+ c.listenerServiceMap.Store(vGroup, next)
+ }
+
+ val, ok = c.subscribedVGroups.Load(vGroup)
+ if !ok {
+ return
+ }
+ state := val.(*namingWatchState)
+ if state.listeners > 0 {
+ state.listeners--
+ }
+ if state.listeners == 0 && !state.permanent {
+ c.subscribedVGroups.Delete(vGroup)
+ close(state.stopCh)
}
- return nil
}
-func (c *NamingServerClient) watchLoop(vGroup string) {
+func (c *NamingServerClient) watchLoop(vGroup string, stopCh <-chan struct{}) {
defer c.wg.Done()
var retryCount int
@@ -591,6 +750,8 @@ func (c *NamingServerClient) watchLoop(vGroup string) {
select {
case <-c.closeChan:
return
+ case <-stopCh:
+ return
default:
}
@@ -604,6 +765,8 @@ func (c *NamingServerClient) watchLoop(vGroup string) {
case <-time.After(time.Duration(retryDelayMs) *
time.Millisecond):
case <-c.closeChan:
return
+ case <-stopCh:
+ return
}
} else {
c.logger.Warn("watch error, will retry
immediately", zap.Error(err))
@@ -622,8 +785,8 @@ func (c *NamingServerClient) watchLoop(vGroup string) {
if !ok {
continue
}
- for _, listener := range val.([]NamingListener) {
- if err := listener.OnEvent(vGroup); err != nil {
+ for _, entry := range val.([]namingListenerEntry) {
+ if err := entry.listener.OnEvent(vGroup); err
!= nil {
c.logger.Warn("listener callback
failed", zap.Error(err))
}
}
@@ -692,10 +855,18 @@ func (c *NamingServerClient) Watch(vGroup string) (bool,
error) {
}
func (c *NamingServerClient) Close() {
- close(c.closeChan)
- c.healthCheckTicker.Stop()
- c.wg.Wait()
- c.logger.Info("naming server client closed")
+ c.closeOnce.Do(func() {
+ if c.closeChan != nil {
+ close(c.closeChan)
+ }
+ if c.healthCheckTicker != nil {
+ c.healthCheckTicker.Stop()
+ }
+ c.wg.Wait()
+ if c.logger != nil {
+ c.logger.Info("naming server client closed")
+ }
+ })
}
func (n *NamingServerRegistryService) Register(instance *ServiceInstance)
error {
@@ -721,3 +892,18 @@ func (n *NamingServerRegistryService) RefreshGroup(vGroup
string) error {
func (n *NamingServerRegistryService) Watch(vGroup string) (bool, error) {
return n.client.Watch(vGroup)
}
+
+type namingRegistryChangeListener struct {
+ key string
+ client *NamingServerClient
+ subscription *registryChangeSubscription
+}
+
+func (l *namingRegistryChangeListener) OnEvent(vGroup string) error {
+ instances, err := l.client.lookupInstances(vGroup)
+ if err != nil {
+ return err
+ }
+ l.subscription.publish(RegistryChangeEvent{Key: l.key, Instances:
instances})
+ return nil
+}
diff --git a/pkg/discovery/naming_server_test.go
b/pkg/discovery/naming_server_test.go
index 8a2517c2..f8c0a96d 100644
--- a/pkg/discovery/naming_server_test.go
+++ b/pkg/discovery/naming_server_test.go
@@ -802,6 +802,97 @@ func TestSubscribeStartsWatchLoopPerVGroup(t *testing.T) {
}, 2*time.Second, 20*time.Millisecond)
}
+func TestNamingServerRegistryService_SubscribePublishesSnapshots(t *testing.T)
{
+ client := newSubscriptionTestNamingClient(t)
+ client.vgroupAddressMap.Store("default_tx_group", []NamingServerNode{{
+ Healthy: true,
+ Transaction: Endpoint{Host: "127.0.0.1", Port: 8091},
+ }})
+ service := &NamingServerRegistryService{client: client}
+
+ events := make(chan RegistryChangeEvent, 2)
+ subscription, err := service.Subscribe("default_tx_group", func(event
RegistryChangeEvent) {
+ events <- event
+ })
+ if err != nil {
+ t.Fatalf("subscribe failed: %v", err)
+ }
+ defer subscription.Unsubscribe()
+
+ assert.Equal(t, RegistryChangeEvent{
+ Key: "default_tx_group",
+ Instances: []*ServiceInstance{{Addr: "127.0.0.1", Port: 8091}},
+ }, nextRegistryChangeEvent(t, events))
+
+ client.vgroupAddressMap.Store("default_tx_group", []NamingServerNode{{
+ Healthy: true,
+ Transaction: Endpoint{Host: "127.0.0.2", Port: 8092},
+ }})
+ fireNamingListeners(t, client, "default_tx_group")
+ assert.Equal(t, RegistryChangeEvent{
+ Key: "default_tx_group",
+ Instances: []*ServiceInstance{{Addr: "127.0.0.2", Port: 8092}},
+ }, nextRegistryChangeEvent(t, events))
+
+ subscription.Unsubscribe()
+ if _, ok := client.listenerServiceMap.Load("default_tx_group"); ok {
+ t.Fatal("naming server listener was not removed after
unsubscribe")
+ }
+ if _, ok := client.subscribedVGroups.Load("default_tx_group"); ok {
+ t.Fatal("naming server watch was not removed after unsubscribe")
+ }
+}
+
+func TestNamingServerRegistryService_CloseUnsubscribesListeners(t *testing.T) {
+ client := newSubscriptionTestNamingClient(t)
+ client.vgroupAddressMap.Store("default_tx_group", []NamingServerNode{{
+ Healthy: true,
+ Transaction: Endpoint{Host: "127.0.0.1", Port: 8091},
+ }})
+ service := &NamingServerRegistryService{client: client}
+
+ events := make(chan RegistryChangeEvent, 1)
+ subscription, err := service.Subscribe("default_tx_group", func(event
RegistryChangeEvent) {
+ events <- event
+ })
+ if err != nil {
+ t.Fatalf("subscribe failed: %v", err)
+ }
+ nextRegistryChangeEvent(t, events)
+
+ service.Close()
+ concreteSubscription := subscription.(*registryChangeSubscription)
+ waitForSignal(t, concreteSubscription.doneCh, "naming server registry
subscription to close")
+ if _, ok := client.listenerServiceMap.Load("default_tx_group"); ok {
+ t.Fatal("naming server listener was not removed after close")
+ }
+ if _, ok := client.subscribedVGroups.Load("default_tx_group"); ok {
+ t.Fatal("naming server watch was not removed after close")
+ }
+ service.Close()
+}
+
+func TestNamingServerRegistryService_SubscribeAfterCloseDoesNotStartWatch(t
*testing.T) {
+ client := newSubscriptionTestNamingClient(t)
+ client.vgroupAddressMap.Store("default_tx_group", []NamingServerNode{{
+ Healthy: true,
+ Transaction: Endpoint{Host: "127.0.0.1", Port: 8091},
+ }})
+ service := &NamingServerRegistryService{client: client}
+
+ service.Close()
+ subscription, err := service.Subscribe("default_tx_group",
func(RegistryChangeEvent) {})
+ if err == nil {
+ t.Fatal("expected subscribe after close to fail")
+ }
+ if subscription != nil {
+ t.Fatal("expected subscribe after close to return nil
subscription")
+ }
+ if _, ok := client.subscribedVGroups.Load("default_tx_group"); ok {
+ t.Fatal("subscribe after close started a naming server watch")
+ }
+}
+
func TestNamingServerRegistryService_RegisterDeregisterNotSupported(t
*testing.T) {
service := &NamingServerRegistryService{}
@@ -812,3 +903,45 @@ func
TestNamingServerRegistryService_RegisterDeregisterNotSupported(t *testing.T
t.Fatal("expected deregister to return an error")
}
}
+
+func newSubscriptionTestNamingClient(t *testing.T) *NamingServerClient {
+ t.Helper()
+
+ mockServer := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/naming/v1/watch":
+ w.WriteHeader(http.StatusNotModified)
+ default:
+ w.WriteHeader(http.StatusNotFound)
+ }
+ }))
+ t.Cleanup(mockServer.Close)
+
+ mockAddr := mockServer.Listener.Addr().String()
+ client := &NamingServerClient{
+ config: &NamingServerConfig{ServerAddr: mockAddr,
Namespace: "public"},
+ logger: zap.NewNop(),
+ closeChan: make(chan struct{}),
+ healthCheckTicker: time.NewTicker(time.Hour),
+ httpClient: &http.Client{Timeout: 100 *
time.Millisecond},
+ longPollClient: &http.Client{Timeout: 100 *
time.Millisecond},
+ }
+ client.availableNamingMap.Store(mockAddr, int32(0))
+ client.namingAddrCache = mockAddr
+ t.Cleanup(client.Close)
+ return client
+}
+
+func fireNamingListeners(t *testing.T, client *NamingServerClient, vGroup
string) {
+ t.Helper()
+
+ val, ok := client.listenerServiceMap.Load(vGroup)
+ if !ok {
+ t.Fatalf("no listener registered for vGroup %s", vGroup)
+ }
+ for _, entry := range val.([]namingListenerEntry) {
+ if err := entry.listener.OnEvent(vGroup); err != nil {
+ t.Fatalf("listener callback failed: %v", err)
+ }
+ }
+}
diff --git a/pkg/discovery/raft.go b/pkg/discovery/raft.go
index f724ccbe..21448e5c 100644
--- a/pkg/discovery/raft.go
+++ b/pkg/discovery/raft.go
@@ -43,10 +43,11 @@ const (
)
type RaftRegistryService struct {
- cfg *RaftConfig
- metadata *metadata.Metadata
- initAddresses sync.Map // clusterName ->
[]*ServiceInstance
- aliveNodes sync.Map // transactionServiceGroup ->
[]*ServiceInstance
+ cfg *RaftConfig
+ metadata *metadata.Metadata
+ initAddresses sync.Map // clusterName -> []*ServiceInstance
+ aliveNodes sync.Map // transactionServiceGroup -> []*ServiceInstance
+ // vgroupMapping is copied during construction and read-only afterward.
vgroupMapping map[string]string
namingserverAddress string
username string
@@ -56,14 +57,29 @@ type RaftRegistryService struct {
currentTransactionServiceGroup string
currentTransactionClusterName string
mu sync.RWMutex
+ stateMu sync.RWMutex
stopCh chan struct{}
refreshOnce sync.Once
+ closeOnce sync.Once
+ subscriptionsMu sync.Mutex
+ subscriptions
map[*registryChangeSubscription]raftSubscription
+ closed bool
httpClient *http.Client
random *rand.Rand
}
+type raftSubscription struct {
+ key string
+ cluster string
+}
+
+var _ RegistrySubscriber = (*RaftRegistryService)(nil)
+
func NewRaftRegistryService(config *ServiceConfig, raftConfig *RegistryConfig)
*RaftRegistryService {
- vgroupMapping := config.VgroupMapping
+ vgroupMapping := make(map[string]string, len(config.VgroupMapping))
+ for key, value := range config.VgroupMapping {
+ vgroupMapping[key] = value
+ }
r := &RaftRegistryService{
cfg: &raftConfig.Raft,
@@ -83,6 +99,9 @@ func NewRaftRegistryService(config *ServiceConfig, raftConfig
*RegistryConfig) *
}
func (r *RaftRegistryService) Lookup(key string) ([]*ServiceInstance, error) {
+ if r.isClosed() {
+ return nil, fmt.Errorf("registry service is closed")
+ }
clusterName := r.vgroupMapping[key]
if clusterName == "" {
return nil, fmt.Errorf("cluster doesn't exist for
serviceGroup=%s", key)
@@ -125,6 +144,13 @@ func (r *RaftRegistryService) Lookup(key string)
([]*ServiceInstance, error) {
r.startQueryMetadata()
}
}
+ r.stateMu.RLock()
+ instances, err := r.snapshotForClusterLocked(clusterName)
+ r.stateMu.RUnlock()
+ return instances, err
+}
+
+func (r *RaftRegistryService) snapshotForClusterLocked(clusterName string)
([]*ServiceInstance, error) {
leader := r.metadata.GetLeader(clusterName)
if leader != nil {
endpoint, err := r.selectEndpoint(transactionEndpoint, leader)
@@ -136,6 +162,63 @@ func (r *RaftRegistryService) Lookup(key string)
([]*ServiceInstance, error) {
return r.getServiceInstances(clusterName, "")
}
+// Subscribe reports Raft registry snapshots for a transaction service group.
+func (r *RaftRegistryService) Subscribe(key string, listener
RegistryChangeListener) (RegistrySubscription, error) {
+ if listener == nil {
+ return nil, fmt.Errorf("registry change listener is nil")
+ }
+ if r.isClosed() {
+ return nil, fmt.Errorf("registry service is closed")
+ }
+ clusterName := r.vgroupMapping[key]
+ if clusterName == "" {
+ return nil, fmt.Errorf("cluster doesn't exist for
serviceGroup=%s", key)
+ }
+
+ if _, err := r.Lookup(key); err != nil {
+ return nil, err
+ }
+
+ subscription := newRegistryChangeSubscription(listener)
+ r.stateMu.Lock()
+ r.subscriptionsMu.Lock()
+ if r.closed {
+ r.subscriptionsMu.Unlock()
+ r.stateMu.Unlock()
+ return nil, fmt.Errorf("registry service is closed")
+ }
+ if r.subscriptions == nil {
+ r.subscriptions =
make(map[*registryChangeSubscription]raftSubscription)
+ }
+ instances, err := r.snapshotForClusterLocked(clusterName)
+ if err != nil {
+ r.subscriptionsMu.Unlock()
+ r.stateMu.Unlock()
+ return nil, err
+ }
+ r.subscriptions[subscription] = raftSubscription{key: key, cluster:
clusterName}
+ subscription.initialize(RegistryChangeEvent{Key: key, Instances:
instances}, func() {
+ r.removeSubscription(subscription)
+ })
+ r.subscriptionsMu.Unlock()
+ r.stateMu.Unlock()
+
+ subscription.start()
+ return subscription, nil
+}
+
+func (r *RaftRegistryService) removeSubscription(subscription
*registryChangeSubscription) {
+ r.subscriptionsMu.Lock()
+ delete(r.subscriptions, subscription)
+ r.subscriptionsMu.Unlock()
+}
+
+func (r *RaftRegistryService) isClosed() bool {
+ r.subscriptionsMu.Lock()
+ defer r.subscriptionsMu.Unlock()
+ return r.closed
+}
+
func (r *RaftRegistryService) getServiceInstances(clusterName, group string)
([]*ServiceInstance, error) {
nodes := r.metadata.GetNodes(clusterName, group)
if len(nodes) > 0 {
@@ -179,11 +262,23 @@ func (r *RaftRegistryService)
RefreshAliveLookup(transactionServiceGroup string,
}
func (r *RaftRegistryService) Close() {
- select {
- case <-r.stopCh:
- default:
- close(r.stopCh)
- }
+ r.closeOnce.Do(func() {
+ r.subscriptionsMu.Lock()
+ r.closed = true
+ subscriptions := make([]*registryChangeSubscription, 0,
len(r.subscriptions))
+ for subscription := range r.subscriptions {
+ subscriptions = append(subscriptions, subscription)
+ }
+ r.subscriptions = nil
+ r.subscriptionsMu.Unlock()
+
+ for _, subscription := range subscriptions {
+ subscription.Unsubscribe()
+ }
+ if r.stopCh != nil {
+ close(r.stopCh)
+ }
+ })
}
func (r *RaftRegistryService) selectEndpoint(t string, n *metadata.Node)
(*ServiceInstance, error) {
@@ -368,7 +463,10 @@ func (r *RaftRegistryService)
acquireClusterMetaData(clusterName, group string)
if err = json.Unmarshal(body, &mr); err != nil {
return fmt.Errorf("unmarshal metadataResponse failed:
%w", err)
}
- r.metadata.RefreshMetadata(clusterName, mr)
+ r.stateMu.Lock()
+ r.metadata.RefreshGroupMetadata(clusterName, group, mr)
+ r.publishClusterSnapshotLocked(clusterName)
+ r.stateMu.Unlock()
return nil
} else if resp.StatusCode == http.StatusUnauthorized {
if err = r.refreshToken(); err != nil {
@@ -379,6 +477,33 @@ func (r *RaftRegistryService)
acquireClusterMetaData(clusterName, group string)
return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
}
+func (r *RaftRegistryService) publishClusterSnapshotLocked(clusterName string)
{
+ instances, err := r.snapshotForClusterLocked(clusterName)
+ if err != nil {
+ log.Warnf("build registry snapshot failed for cluster=%s: %v",
clusterName, err)
+ return
+ }
+
+ r.subscriptionsMu.Lock()
+ subscriptions := make([]struct {
+ subscription *registryChangeSubscription
+ key string
+ }, 0)
+ for subscription, state := range r.subscriptions {
+ if state.cluster == clusterName {
+ subscriptions = append(subscriptions, struct {
+ subscription *registryChangeSubscription
+ key string
+ }{subscription: subscription, key: state.key})
+ }
+ }
+ r.subscriptionsMu.Unlock()
+
+ for _, item := range subscriptions {
+ item.subscription.publish(RegistryChangeEvent{Key: item.key,
Instances: cloneServiceInstances(instances)})
+ }
+}
+
/* -------------------- Token management -------------------- */
func (r *RaftRegistryService) isTokenExpired() bool {
diff --git a/pkg/discovery/raft_subscription_test.go
b/pkg/discovery/raft_subscription_test.go
new file mode 100644
index 00000000..cdcdadba
--- /dev/null
+++ b/pkg/discovery/raft_subscription_test.go
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package discovery
+
+import (
+ "math/rand"
+ "net/http"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "seata.apache.org/seata-go/v2/pkg/discovery/metadata"
+)
+
+func TestRaftRegistryServiceSubscribePublishesMetadataChanges(t *testing.T) {
+ service := newSubscriptionTestRaftRegistry()
+ service.metadata.RefreshGroupMetadata("test-cluster", "",
metadata.MetadataResponse{
+ Term: 1,
+ Nodes: []*metadata.Node{{
+ Transaction: &metadata.Endpoint{Host: "127.0.0.1",
Port: 8001},
+ Group: "",
+ Role: metadata.LEADER,
+ }},
+ })
+
+ events := make(chan RegistryChangeEvent, 2)
+ subscription, err := service.Subscribe("default_tx_group", func(event
RegistryChangeEvent) {
+ events <- event
+ })
+ require.NoError(t, err)
+
+ initial := nextRaftRegistryEvent(t, events)
+ assert.Equal(t, "default_tx_group", initial.Key)
+ assert.Equal(t, []*ServiceInstance{{Addr: "127.0.0.1", Port: 8001}},
initial.Instances)
+
+ service.metadata.RefreshGroupMetadata("test-cluster", "",
metadata.MetadataResponse{
+ Term: 2,
+ Nodes: []*metadata.Node{{
+ Transaction: &metadata.Endpoint{Host: "127.0.0.1",
Port: 8002},
+ Group: "",
+ Role: metadata.LEADER,
+ }},
+ })
+ service.stateMu.Lock()
+ service.publishClusterSnapshotLocked("test-cluster")
+ service.stateMu.Unlock()
+
+ changed := nextRaftRegistryEvent(t, events)
+ assert.Equal(t, []*ServiceInstance{{Addr: "127.0.0.1", Port: 8002}},
changed.Instances)
+
+ subscription.Unsubscribe()
+ subscription.Unsubscribe()
+ service.Close()
+ service.Close()
+}
+
+func TestRaftRegistryServiceCloseStopsSubscription(t *testing.T) {
+ service := newSubscriptionTestRaftRegistry()
+ service.metadata.RefreshGroupMetadata("test-cluster", "",
metadata.MetadataResponse{
+ Nodes: []*metadata.Node{{
+ Transaction: &metadata.Endpoint{Host: "127.0.0.1",
Port: 8001},
+ Group: "",
+ Role: metadata.LEADER,
+ }},
+ })
+
+ events := make(chan RegistryChangeEvent, 2)
+ _, err := service.Subscribe("default_tx_group", func(event
RegistryChangeEvent) {
+ events <- event
+ })
+ require.NoError(t, err)
+ _ = nextRaftRegistryEvent(t, events)
+
+ service.Close()
+ service.stateMu.Lock()
+ service.publishClusterSnapshotLocked("test-cluster")
+ service.stateMu.Unlock()
+
+ assert.Equal(t, 0, len(events))
+}
+
+func TestRaftRegistryServiceSubscribeAfterCloseDoesNotLookup(t *testing.T) {
+ service := newSubscriptionTestRaftRegistry()
+ service.Close()
+
+ subscription, err := service.Subscribe("default_tx_group",
func(RegistryChangeEvent) {})
+ assert.Nil(t, subscription)
+ assert.EqualError(t, err, "registry service is closed")
+
+ instances, err := service.Lookup("default_tx_group")
+ assert.Nil(t, instances)
+ assert.EqualError(t, err, "registry service is closed")
+}
+
+func newSubscriptionTestRaftRegistry() *RaftRegistryService {
+ return &RaftRegistryService{
+ cfg: &RaftConfig{},
+ metadata: metadata.NewMetadata(),
+ vgroupMapping: map[string]string{"default_tx_group":
"test-cluster"},
+ stopCh: make(chan struct{}),
+ httpClient: &http.Client{},
+ random: rand.New(rand.NewSource(1)),
+ }
+}
+
+func nextRaftRegistryEvent(t *testing.T, events <-chan RegistryChangeEvent)
RegistryChangeEvent {
+ t.Helper()
+
+ select {
+ case event := <-events:
+ return event
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for registry change event")
+ }
+ return RegistryChangeEvent{}
+}
diff --git a/pkg/discovery/store.go b/pkg/discovery/store.go
index fd53d194..928a197e 100644
--- a/pkg/discovery/store.go
+++ b/pkg/discovery/store.go
@@ -50,21 +50,28 @@ func (s *AddressStore) Update(cluster string, instances
[]*ServiceInstance) {
}
func (s *AddressStore) Subscribe(subscriber AddressStoreSubscriber) func() {
+ _, unsubscribe := s.subscribeWithSnapshot("", subscriber)
+ return unsubscribe
+}
+
+func (s *AddressStore) subscribeWithSnapshot(cluster string, subscriber
AddressStoreSubscriber) ([]*ServiceInstance, func()) {
if subscriber == nil {
- return func() {}
+ return s.Snapshot(cluster), func() {}
}
s.mu.Lock()
id := s.nextID
s.nextID++
s.subscribers[id] = subscriber
+ snapshot := cloneServiceInstances(s.clusters[cluster])
s.mu.Unlock()
- return func() {
+ unsubscribe := func() {
s.mu.Lock()
delete(s.subscribers, id)
s.mu.Unlock()
}
+ return snapshot, unsubscribe
}
func (s *AddressStore) upsert(cluster string, instance *ServiceInstance) {
diff --git a/pkg/discovery/subscription.go b/pkg/discovery/subscription.go
new file mode 100644
index 00000000..c6bd5085
--- /dev/null
+++ b/pkg/discovery/subscription.go
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package discovery
+
+import "sync"
+
+type registryChangeSubscription struct {
+ once sync.Once
+ listener RegistryChangeListener
+ unsubscribe func()
+
+ mu sync.Mutex
+ initial *RegistryChangeEvent
+ latest *RegistryChangeEvent
+ closed bool
+ wakeCh chan struct{}
+ doneCh chan struct{}
+}
+
+func newRegistryChangeSubscription(listener RegistryChangeListener)
*registryChangeSubscription {
+ return ®istryChangeSubscription{
+ listener: listener,
+ wakeCh: make(chan struct{}, 1),
+ doneCh: make(chan struct{}),
+ }
+}
+
+func (s *registryChangeSubscription) initialize(initial RegistryChangeEvent,
unsubscribe func()) {
+ s.mu.Lock()
+ s.initial = &initial
+ s.unsubscribe = unsubscribe
+ s.mu.Unlock()
+}
+
+func (s *registryChangeSubscription) start() {
+ s.mu.Lock()
+ if s.closed {
+ s.mu.Unlock()
+ return
+ }
+ s.mu.Unlock()
+
+ go s.dispatch()
+ s.signal()
+}
+
+func (s *registryChangeSubscription) dispatch() {
+ for {
+ select {
+ case <-s.wakeCh:
+ for {
+ event, ok := s.nextEvent()
+ if !ok {
+ break
+ }
+ s.listener(event)
+ }
+ case <-s.doneCh:
+ return
+ }
+ }
+}
+
+func (s *registryChangeSubscription) publish(event RegistryChangeEvent) {
+ s.mu.Lock()
+ if s.closed {
+ s.mu.Unlock()
+ return
+ }
+ s.latest = &event
+ s.mu.Unlock()
+
+ s.signal()
+}
+
+func (s *registryChangeSubscription) signal() {
+ select {
+ case s.wakeCh <- struct{}{}:
+ default:
+ }
+}
+
+func (s *registryChangeSubscription) nextEvent() (RegistryChangeEvent, bool) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ if s.closed {
+ return RegistryChangeEvent{}, false
+ }
+ if s.initial != nil {
+ event := *s.initial
+ s.initial = nil
+ return event, true
+ }
+ if s.latest != nil {
+ event := *s.latest
+ s.latest = nil
+ return event, true
+ }
+ return RegistryChangeEvent{}, false
+}
+
+func (s *registryChangeSubscription) Unsubscribe() {
+ s.once.Do(func() {
+ s.mu.Lock()
+ s.closed = true
+ s.initial = nil
+ s.latest = nil
+ unsubscribe := s.unsubscribe
+ s.mu.Unlock()
+
+ if unsubscribe != nil {
+ unsubscribe()
+ }
+ close(s.doneCh)
+ })
+}
diff --git a/pkg/discovery/subscription_test.go
b/pkg/discovery/subscription_test.go
new file mode 100644
index 00000000..17a5e932
--- /dev/null
+++ b/pkg/discovery/subscription_test.go
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package discovery
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func TestEtcd3RegistryService_Subscribe(t *testing.T) {
+ store := NewAddressStore()
+ store.Update("default", []*ServiceInstance{{Addr: "127.0.0.1", Port:
8091}})
+ service := newTestEtcdRegistryService(store)
+
+ events := make(chan RegistryChangeEvent, 4)
+ subscription, err := service.Subscribe("default_tx_group", func(event
RegistryChangeEvent) {
+ events <- event
+ })
+ require.NoError(t, err)
+ defer subscription.Unsubscribe()
+
+ assert.Equal(t, RegistryChangeEvent{
+ Key: "default_tx_group",
+ Instances: []*ServiceInstance{{Addr: "127.0.0.1", Port: 8091}},
+ }, nextRegistryChangeEvent(t, events))
+
+ store.Update("other", []*ServiceInstance{{Addr: "127.0.0.2", Port:
8092}})
+ store.Update("default", []*ServiceInstance{
+ {Addr: "127.0.0.1", Port: 8091},
+ {Addr: "127.0.0.3", Port: 8093},
+ })
+ assert.Equal(t, []*ServiceInstance{
+ {Addr: "127.0.0.1", Port: 8091},
+ {Addr: "127.0.0.3", Port: 8093},
+ }, nextRegistryChangeEvent(t, events).Instances)
+
+ subscription.Unsubscribe()
+ subscription.Unsubscribe()
+ store.mu.RLock()
+ assert.Empty(t, store.subscribers)
+ store.mu.RUnlock()
+}
+
+func TestEtcd3RegistryService_SubscribeValidatesInputs(t *testing.T) {
+ service := newTestEtcdRegistryService(NewAddressStore())
+
+ subscription, err := service.Subscribe("default_tx_group", nil)
+ assert.Nil(t, subscription)
+ assert.EqualError(t, err, "registry change listener is nil")
+
+ subscription, err = service.Subscribe("missing_tx_group",
func(RegistryChangeEvent) {})
+ assert.Nil(t, subscription)
+ assert.EqualError(t, err, "cluster doesn't exist")
+}
+
+func TestEtcd3RegistryService_SubscribeCoalescesSlowListener(t *testing.T) {
+ store := NewAddressStore()
+ service := newTestEtcdRegistryService(store)
+
+ block := make(chan struct{})
+ events := make(chan RegistryChangeEvent, 4)
+ subscription, err := service.Subscribe("default_tx_group", func(event
RegistryChangeEvent) {
+ events <- event
+ <-block
+ })
+ require.NoError(t, err)
+ defer subscription.Unsubscribe()
+
+ nextRegistryChangeEvent(t, events)
+ store.Update("default", []*ServiceInstance{{Addr: "127.0.0.1", Port:
8091}})
+ store.Update("default", []*ServiceInstance{{Addr: "127.0.0.2", Port:
8092}})
+ store.Update("default", []*ServiceInstance{{Addr: "127.0.0.3", Port:
8093}})
+ close(block)
+
+ assert.Equal(t, []*ServiceInstance{{Addr: "127.0.0.3", Port: 8093}},
nextRegistryChangeEvent(t, events).Instances)
+}
+
+func TestEtcd3RegistryService_CloseUnsubscribesListeners(t *testing.T) {
+ store := NewAddressStore()
+ service := newTestEtcdRegistryService(store)
+
+ events := make(chan RegistryChangeEvent, 1)
+ subscription, err := service.Subscribe("default_tx_group", func(event
RegistryChangeEvent) {
+ events <- event
+ })
+ require.NoError(t, err)
+ nextRegistryChangeEvent(t, events)
+
+ service.Close()
+ concreteSubscription := subscription.(*registryChangeSubscription)
+ waitForSignal(t, concreteSubscription.doneCh, "registry subscription to
close")
+ store.mu.RLock()
+ assert.Empty(t, store.subscribers)
+ store.mu.RUnlock()
+
+ subscription, err = service.Subscribe("default_tx_group",
func(RegistryChangeEvent) {})
+ assert.Nil(t, subscription)
+ assert.EqualError(t, err, "registry service is closed")
+}
+
+func newTestEtcdRegistryService(store *AddressStore) *EtcdRegistryService {
+ return &EtcdRegistryService{
+ vgroupMapping: map[string]string{"default_tx_group": "default"},
+ store: store,
+ }
+}
+
+func nextRegistryChangeEvent(t *testing.T, events <-chan RegistryChangeEvent)
RegistryChangeEvent {
+ t.Helper()
+
+ select {
+ case event := <-events:
+ return event
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for registry change event")
+ }
+ return RegistryChangeEvent{}
+}
diff --git a/pkg/remoting/getty/getty_client.go
b/pkg/remoting/getty/getty_client.go
index a945ac0a..4b60a9b5 100644
--- a/pkg/remoting/getty/getty_client.go
+++ b/pkg/remoting/getty/getty_client.go
@@ -52,20 +52,24 @@ func GetGettyRemotingClient() *GettyRemotingClient {
}
func (client *GettyRemotingClient) SendAsyncRequest(msg interface{}) error {
+ rpcMessage := newAsyncRequestMessage(int32(client.idGenerator.Inc()),
msg)
+ return client.gettyRemoting.SendAsync(rpcMessage, nil,
client.asyncCallback)
+}
+
+func newAsyncRequestMessage(id int32, msg interface{}) message.RpcMessage {
var msgType message.RequestType
if _, ok := msg.(message.HeartBeatMessage); ok {
msgType = message.RequestTypeHeartbeatRequest
} else {
msgType = message.RequestTypeRequestOneway
}
- rpcMessage := message.RpcMessage{
- ID: int32(client.idGenerator.Inc()),
+ return message.RpcMessage{
+ ID: id,
Type: msgType,
Codec: byte(codec.CodecTypeSeata),
Compressor: 0,
Body: msg,
}
- return client.gettyRemoting.SendAsync(rpcMessage, nil,
client.asyncCallback)
}
func (client *GettyRemotingClient) SendAsyncResponse(msgID int32, msg
interface{}) error {
diff --git a/pkg/remoting/getty/getty_client_test.go
b/pkg/remoting/getty/getty_client_test.go
index 432f31d5..a6ec8585 100644
--- a/pkg/remoting/getty/getty_client_test.go
+++ b/pkg/remoting/getty/getty_client_test.go
@@ -82,16 +82,10 @@ func TestGettyRemotingClient_SendAsyncRequest(t *testing.T)
{
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
- var capturedType message.RequestType
- patches :=
gomonkey.ApplyMethod(reflect.TypeOf(GetGettyRemotingClient().gettyRemoting),
"SendAsync",
- func(_ *GettyRemoting, msg message.RpcMessage,
s getty.Session, callback callbackMethod) error {
- capturedType = msg.Type
- return nil
- })
- defer patches.Reset()
- err :=
GetGettyRemotingClient().SendAsyncRequest(test.message)
- assert.Empty(t, err)
- assert.Equal(t, test.expectedType, capturedType)
+ msg := newAsyncRequestMessage(1, test.message)
+ assert.Equal(t, test.expectedType, msg.Type)
+ assert.Equal(t, byte(codec.CodecTypeSeata), msg.Codec)
+ assert.Equal(t, test.message, msg.Body)
})
}
}
diff --git a/pkg/remoting/getty/session_manager.go
b/pkg/remoting/getty/session_manager.go
index a467cd6a..8bb4bb16 100644
--- a/pkg/remoting/getty/session_manager.go
+++ b/pkg/remoting/getty/session_manager.go
@@ -51,11 +51,55 @@ var (
type SessionManager struct {
// serverAddress -> rpc_client.Session -> bool
- serverSessions sync.Map
- allSessions sync.Map
- sessionSize int32
- gettyConf *config.Config
- seataConfig *config.SeataConfig
+ serverSessions sync.Map
+ allSessions sync.Map
+ sessionSize int32
+ gettyConf *config.Config
+ seataConfig *config.SeataConfig
+ registrySubscription discovery.RegistrySubscription
+ serverClients sync.Map
+ serverAddressMu sync.RWMutex
+ serverAddressSnapshot map[string]struct{}
+ serverAddressReady bool
+ startClient func(*discovery.ServiceInstance) closeableClient
+}
+
+type closeableClient interface {
+ Close()
+}
+
+type serverClientEntry struct {
+ mu sync.Mutex
+ client closeableClient
+ closed bool
+}
+
+func (e *serverClientEntry) setClient(client closeableClient) bool {
+ e.mu.Lock()
+ if e.closed {
+ e.mu.Unlock()
+ client.Close()
+ return false
+ }
+ e.client = client
+ e.mu.Unlock()
+ return true
+}
+
+func (e *serverClientEntry) Close() {
+ e.mu.Lock()
+ if e.closed {
+ e.mu.Unlock()
+ return
+ }
+ e.closed = true
+ client := e.client
+ e.client = nil
+ e.mu.Unlock()
+
+ if client != nil {
+ client.Close()
+ }
}
func initSessionManager(gettyConfig *config.Config, seataConfig
*config.SeataConfig) {
@@ -73,31 +117,158 @@ func initSessionManager(gettyConfig *config.Config,
seataConfig *config.SeataCon
}
func (g *SessionManager) init() {
- addressList := g.getAvailServerList()
+ g.initWithRegistry(discovery.GetRegistry())
+}
+
+func (g *SessionManager) initWithRegistry(registryService
discovery.RegistryService) {
+ if registryService == nil {
+ log.Warn("registry service not initialized")
+ return
+ }
+ if g.subscribeRegistry(registryService) {
+ return
+ }
+
+ addressList := g.getAvailServerList(registryService)
if len(addressList) == 0 {
log.Warn("no have valid seata server list")
}
- for _, address := range addressList {
- gettyClient := getty.NewTCPClient(
- getty.WithServerAddress(net.JoinHostPort(address.Addr,
strconv.Itoa(address.Port))),
- // todo if read c.gettyConf.ConnectionNum, will cause
the connect to fail
- getty.WithConnectionNumber(1),
-
getty.WithReconnectInterval(g.gettyConf.ReconnectInterval),
- getty.WithClientTaskPool(gxsync.NewTaskPoolSimple(0)),
- )
- go gettyClient.RunEventLoop(g.newSession)
+ g.refreshServerList(addressList)
+}
+
+func (g *SessionManager) subscribeRegistry(registryService
discovery.RegistryService) bool {
+ subscriber, ok := registryService.(discovery.RegistrySubscriber)
+ if !ok {
+ return false
}
+ subscription, err := subscriber.Subscribe(g.seataConfig.TxServiceGroup,
func(event discovery.RegistryChangeEvent) {
+ if event.Key != g.seataConfig.TxServiceGroup {
+ return
+ }
+ g.refreshServerList(event.Instances)
+ })
+ if err != nil {
+ log.Warnf("subscribe registry changes failed: %v", err)
+ return false
+ }
+ g.registrySubscription = subscription
+ return true
}
-func (g *SessionManager) getAvailServerList() []*discovery.ServiceInstance {
- registryService := discovery.GetRegistry()
+func (g *SessionManager) getAvailServerList(registryService
discovery.RegistryService) []*discovery.ServiceInstance {
instances, err := registryService.Lookup(g.seataConfig.TxServiceGroup)
if err != nil {
+ log.Warnf("lookup seata server list failed: %v", err)
return nil
}
return instances
}
+func (g *SessionManager) refreshServerList(instances
[]*discovery.ServiceInstance) {
+ servers := make(map[string]*discovery.ServiceInstance, len(instances))
+ for _, instance := range instances {
+ if instance == nil || instance.Addr == "" || instance.Port <= 0
{
+ continue
+ }
+ clone := &discovery.ServiceInstance{Addr: instance.Addr, Port:
instance.Port}
+ servers[serverAddress(clone)] = clone
+ }
+
+ removedAddresses := g.replaceServerAddressSnapshot(servers)
+ for _, address := range removedAddresses {
+ g.releaseServerAddress(address)
+ }
+
+ for _, instance := range servers {
+ g.ensureServerClient(instance)
+ }
+}
+
+func (g *SessionManager) replaceServerAddressSnapshot(servers
map[string]*discovery.ServiceInstance) []string {
+ g.serverAddressMu.Lock()
+ var removedAddresses []string
+ for address := range g.serverAddressSnapshot {
+ if _, ok := servers[address]; !ok {
+ removedAddresses = append(removedAddresses, address)
+ }
+ }
+ g.serverAddressSnapshot = make(map[string]struct{}, len(servers))
+ for address := range servers {
+ g.serverAddressSnapshot[address] = struct{}{}
+ }
+ g.serverAddressReady = true
+ g.serverAddressMu.Unlock()
+ return removedAddresses
+}
+
+func (g *SessionManager) ensureServerClient(instance
*discovery.ServiceInstance) {
+ address := serverAddress(instance)
+ entry := &serverClientEntry{}
+ if _, loaded := g.serverClients.LoadOrStore(address, entry); loaded {
+ return
+ }
+ var client closeableClient
+ if g.startClient != nil {
+ client = g.startClient(instance)
+ } else {
+ client = g.startGettyClient(instance)
+ }
+ if client == nil {
+ g.serverClients.CompareAndDelete(address, entry)
+ return
+ }
+ if !entry.setClient(client) {
+ return
+ }
+ if !g.isServerAddressAvailable(address) ||
!g.isServerClientEntryCurrent(address, entry) {
+ g.releaseServerClientEntry(address, entry)
+ return
+ }
+}
+
+func (g *SessionManager) isServerClientEntryCurrent(address string, entry
*serverClientEntry) bool {
+ current, ok := g.serverClients.Load(address)
+ return ok && current == entry
+}
+
+func (g *SessionManager) releaseServerClientEntry(address string, entry
*serverClientEntry) {
+ g.serverClients.CompareAndDelete(address, entry)
+ entry.Close()
+}
+
+func (g *SessionManager) startGettyClient(instance *discovery.ServiceInstance)
getty.Client {
+ gettyClient := getty.NewTCPClient(
+ getty.WithServerAddress(serverAddress(instance)),
+ // todo if read c.gettyConf.ConnectionNum, will cause the
connect to fail
+ getty.WithConnectionNumber(1),
+ getty.WithReconnectInterval(g.gettyConf.ReconnectInterval),
+ getty.WithClientTaskPool(gxsync.NewTaskPoolSimple(0)),
+ )
+ go gettyClient.RunEventLoop(g.newSession)
+ return gettyClient
+}
+
+func (g *SessionManager) releaseServerAddress(address string) {
+ if clientAny, loaded := g.serverClients.LoadAndDelete(address); loaded {
+ if client, ok := clientAny.(closeableClient); ok && client !=
nil {
+ client.Close()
+ }
+ }
+ if sessionsAny, ok := g.serverSessions.LoadAndDelete(address); ok {
+ sessions := sessionsAny.(*sync.Map)
+ sessions.Range(func(key, _ interface{}) bool {
+ if session, ok := key.(getty.Session); ok {
+ g.releaseSession(session)
+ }
+ return true
+ })
+ }
+}
+
+func serverAddress(instance *discovery.ServiceInstance) string {
+ return net.JoinHostPort(instance.Addr, strconv.Itoa(instance.Port))
+}
+
func (g *SessionManager) setSessionConfig(session getty.Session) {
session.SetName(g.gettyConf.SessionConfig.SessionName)
session.SetMaxMsgLen(g.gettyConf.SessionConfig.MaxMsgLen)
@@ -160,27 +331,22 @@ func (g *SessionManager) newSession(session
getty.Session) error {
}
func (g *SessionManager) selectSession(msg interface{}) getty.Session {
- selected := loadbalance.Select(loadbalance.GetLoadBalanceConfig().Type,
&g.allSessions, g.getXid(msg))
+ sessions := g.selectableSessions()
+ selected := loadbalance.Select(loadbalance.GetLoadBalanceConfig().Type,
sessions, g.getXid(msg))
session, ok := selected.(getty.Session)
if ok && session != nil {
return session
}
- if g.sessionSize == 0 {
+ if selectableConnectionCount(sessions) == 0 {
ticker := time.NewTicker(time.Duration(checkAliveInternal) *
time.Millisecond)
defer ticker.Stop()
for i := 0; i < maxCheckAliveRetry; i++ {
<-ticker.C
- g.allSessions.Range(func(key, value interface{}) bool {
- session = key.(getty.Session)
- if session.IsClosed() {
- g.releaseSession(session)
- } else {
- return false
- }
- return true
- })
- if session != nil {
+ sessions = g.selectableSessions()
+ selected =
loadbalance.Select(loadbalance.GetLoadBalanceConfig().Type, sessions,
g.getXid(msg))
+ session, ok = selected.(getty.Session)
+ if ok && session != nil {
return session
}
}
@@ -188,6 +354,36 @@ func (g *SessionManager) selectSession(msg interface{})
getty.Session {
return nil
}
+func (g *SessionManager) selectableSessions() *sync.Map {
+ sessions := &sync.Map{}
+ g.allSessions.Range(func(key, value interface{}) bool {
+ session, ok := key.(getty.Session)
+ if !ok {
+ return true
+ }
+ if session.IsClosed() {
+ g.releaseSession(session)
+ return true
+ }
+ if g.isServerAddressAvailable(session.RemoteAddr()) {
+ sessions.Store(session, value)
+ }
+ return true
+ })
+ return sessions
+}
+
+func (g *SessionManager) isServerAddressAvailable(address string) bool {
+ g.serverAddressMu.RLock()
+ defer g.serverAddressMu.RUnlock()
+
+ if !g.serverAddressReady {
+ return true
+ }
+ _, ok := g.serverAddressSnapshot[address]
+ return ok
+}
+
func (g *SessionManager) getXid(msg interface{}) string {
var xid string
if tmpMsg, ok := msg.(message.AbstractGlobalEndRequest); ok {
@@ -210,20 +406,42 @@ func (g *SessionManager) getXid(msg interface{}) string {
}
func (g *SessionManager) releaseSession(session getty.Session) {
- g.allSessions.Delete(session)
- if !session.IsClosed() {
- m, _ := g.serverSessions.LoadOrStore(session.RemoteAddr(),
&sync.Map{})
+ if session == nil {
+ return
+ }
+ if _, loaded := g.allSessions.LoadAndDelete(session); !loaded {
+ return
+ }
+ if m, ok := g.serverSessions.Load(session.RemoteAddr()); ok {
sMap := m.(*sync.Map)
sMap.Delete(session)
+ }
+ if !session.IsClosed() {
session.Close()
}
atomic.AddInt32(&g.sessionSize, -1)
}
func (g *SessionManager) registerSession(session getty.Session) {
- g.allSessions.Store(session, true)
+ if !g.isServerAddressAvailable(session.RemoteAddr()) {
+ log.Warnf("skip session for removed server address: %s",
session.RemoteAddr())
+ session.Close()
+ return
+ }
+ if _, loaded := g.allSessions.LoadOrStore(session, true); loaded {
+ return
+ }
m, _ := g.serverSessions.LoadOrStore(session.RemoteAddr(), &sync.Map{})
sMap := m.(*sync.Map)
sMap.Store(session, true)
atomic.AddInt32(&g.sessionSize, 1)
}
+
+func selectableConnectionCount(connections *sync.Map) int {
+ count := 0
+ connections.Range(func(_, _ interface{}) bool {
+ count++
+ return true
+ })
+ return count
+}
diff --git a/pkg/remoting/getty/session_manager_test.go
b/pkg/remoting/getty/session_manager_test.go
new file mode 100644
index 00000000..963e9829
--- /dev/null
+++ b/pkg/remoting/getty/session_manager_test.go
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package getty
+
+import (
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ getty "github.com/apache/dubbo-getty"
+ "github.com/golang/mock/gomock"
+ "github.com/stretchr/testify/assert"
+
+ "seata.apache.org/seata-go/v2/pkg/discovery"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/remoting/config"
+ "seata.apache.org/seata-go/v2/pkg/remoting/loadbalance"
+ "seata.apache.org/seata-go/v2/pkg/remoting/mock"
+)
+
+func TestSessionManagerRefreshesServerListFromRegistrySubscription(t
*testing.T) {
+ registry := &fakeSubscriberRegistry{
+ fakeLookupRegistry: fakeLookupRegistry{
+ instances: []*discovery.ServiceInstance{{Addr:
"127.0.0.1", Port: 8091}},
+ },
+ }
+ started := make(chan string, 2)
+ closed := make(chan string, 1)
+ manager := newTestSessionManager(func(instance
*discovery.ServiceInstance) closeableClient {
+ address := serverAddress(instance)
+ started <- address
+ return &fakeServerClient{address: address, closed: closed}
+ })
+
+ manager.initWithRegistry(registry)
+ assert.Equal(t, "default_tx_group", registry.subscribeKey)
+ assert.Equal(t, "127.0.0.1:8091", nextStartedServer(t, started))
+
+ registry.publish([]*discovery.ServiceInstance{{Addr: "127.0.0.2", Port:
8092}})
+ assert.Equal(t, "127.0.0.1:8091", nextClosedServer(t, closed))
+ assert.Equal(t, "127.0.0.2:8092", nextStartedServer(t, started))
+ assert.False(t, manager.isServerAddressAvailable("127.0.0.1:8091"))
+ assert.True(t, manager.isServerAddressAvailable("127.0.0.2:8092"))
+}
+
+func TestSessionManagerFallsBackToLookupWhenRegistryDoesNotSubscribe(t
*testing.T) {
+ registry := &fakeLookupRegistry{
+ instances: []*discovery.ServiceInstance{{Addr: "127.0.0.1",
Port: 8091}},
+ }
+ started := make(chan string, 1)
+ manager := newTestSessionManager(func(instance
*discovery.ServiceInstance) closeableClient {
+ address := serverAddress(instance)
+ started <- address
+ return &fakeServerClient{address: address}
+ })
+
+ manager.initWithRegistry(registry)
+ assert.Equal(t, "default_tx_group", registry.lookupKey)
+ assert.Equal(t, "127.0.0.1:8091", nextStartedServer(t, started))
+ assert.True(t, manager.isServerAddressAvailable("127.0.0.1:8091"))
+}
+
+func TestSessionManagerSelectSessionSkipsRemovedServerAddress(t *testing.T) {
+ previousLoadBalance := loadbalance.GetLoadBalanceConfig()
+ loadbalance.InitLoadBalanceConfig(loadbalance.Config{Type:
"RoundRobinLoadBalance"})
+ t.Cleanup(func() {
+ loadbalance.InitLoadBalanceConfig(previousLoadBalance)
+ })
+
+ ctrl := gomock.NewController(t)
+ manager := newTestSessionManager(nil)
+ manager.refreshServerList([]*discovery.ServiceInstance{
+ {Addr: "127.0.0.1", Port: 8091},
+ {Addr: "127.0.0.2", Port: 8092},
+ })
+
+ removed := mock.NewMockTestSession(ctrl)
+ removed.EXPECT().IsClosed().Return(false).AnyTimes()
+ removed.EXPECT().RemoteAddr().Return("127.0.0.1:8091").AnyTimes()
+ removed.EXPECT().Close().Times(1)
+ kept := mock.NewMockTestSession(ctrl)
+ kept.EXPECT().IsClosed().Return(false).AnyTimes()
+ kept.EXPECT().RemoteAddr().Return("127.0.0.2:8092").AnyTimes()
+
+ manager.registerSession(removed)
+ manager.registerSession(kept)
+ manager.refreshServerList([]*discovery.ServiceInstance{{Addr:
"127.0.0.2", Port: 8092}})
+
+ selected :=
manager.selectSession(message.GlobalBeginRequest{TransactionName: "tx"})
+ assert.Equal(t, getty.Session(kept), selected)
+ assert.Equal(t, int32(1), atomic.LoadInt32(&manager.sessionSize))
+ if _, ok := manager.serverClients.Load("127.0.0.1:8091"); ok {
+ t.Fatal("removed server client is still tracked")
+ }
+}
+
+func TestSessionManagerClosesStaleClientWhenAddressRemovedDuringStart(t
*testing.T) {
+ started := make(chan string, 1)
+ resumeStart := make(chan struct{})
+ closed := make(chan string, 1)
+ manager := newTestSessionManager(func(instance
*discovery.ServiceInstance) closeableClient {
+ address := serverAddress(instance)
+ started <- address
+ <-resumeStart
+ return &fakeServerClient{address: address, closed: closed}
+ })
+
+ refreshDone := make(chan struct{})
+ go func() {
+ manager.refreshServerList([]*discovery.ServiceInstance{{Addr:
"127.0.0.1", Port: 8091}})
+ close(refreshDone)
+ }()
+ assert.Equal(t, "127.0.0.1:8091", nextStartedServer(t, started))
+
+ manager.refreshServerList(nil)
+ close(resumeStart)
+
+ select {
+ case <-refreshDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for stale server client start to
finish")
+ }
+ assert.Equal(t, "127.0.0.1:8091", nextClosedServer(t, closed))
+ if _, ok := manager.serverClients.Load("127.0.0.1:8091"); ok {
+ t.Fatal("stale server client is still tracked")
+ }
+}
+
+func newTestSessionManager(startClient func(*discovery.ServiceInstance)
closeableClient) *SessionManager {
+ if startClient == nil {
+ startClient = func(*discovery.ServiceInstance) closeableClient {
+ return &fakeServerClient{}
+ }
+ }
+ return &SessionManager{
+ gettyConf: &config.Config{},
+ seataConfig: &config.SeataConfig{TxServiceGroup:
"default_tx_group"},
+ startClient: startClient,
+ }
+}
+
+func nextStartedServer(t *testing.T, started <-chan string) string {
+ t.Helper()
+
+ select {
+ case address := <-started:
+ return address
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for server client start")
+ }
+ return ""
+}
+
+func nextClosedServer(t *testing.T, closed <-chan string) string {
+ t.Helper()
+
+ select {
+ case address := <-closed:
+ return address
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for server client close")
+ }
+ return ""
+}
+
+type fakeLookupRegistry struct {
+ lookupKey string
+ instances []*discovery.ServiceInstance
+}
+
+func (r *fakeLookupRegistry) Lookup(key string) ([]*discovery.ServiceInstance,
error) {
+ r.lookupKey = key
+ return cloneTestServiceInstances(r.instances), nil
+}
+
+func (r *fakeLookupRegistry) Close() {}
+
+type fakeSubscriberRegistry struct {
+ fakeLookupRegistry
+ subscribeKey string
+ listener discovery.RegistryChangeListener
+}
+
+func (r *fakeSubscriberRegistry) Subscribe(key string, listener
discovery.RegistryChangeListener) (discovery.RegistrySubscription, error) {
+ r.subscribeKey = key
+ r.listener = listener
+ listener(discovery.RegistryChangeEvent{Key: key, Instances:
cloneTestServiceInstances(r.instances)})
+ return fakeRegistrySubscription{}, nil
+}
+
+func (r *fakeSubscriberRegistry) publish(instances
[]*discovery.ServiceInstance) {
+ r.listener(discovery.RegistryChangeEvent{Key: r.subscribeKey,
Instances: cloneTestServiceInstances(instances)})
+}
+
+type fakeRegistrySubscription struct{}
+
+func (fakeRegistrySubscription) Unsubscribe() {}
+
+func cloneTestServiceInstances(instances []*discovery.ServiceInstance)
[]*discovery.ServiceInstance {
+ clones := make([]*discovery.ServiceInstance, 0, len(instances))
+ for _, instance := range instances {
+ clone := *instance
+ clones = append(clones, &clone)
+ }
+ return clones
+}
+
+type fakeServerClient struct {
+ address string
+ closed chan<- string
+ once sync.Once
+}
+
+func (c *fakeServerClient) Close() {
+ c.once.Do(func() {
+ if c.closed != nil {
+ c.closed <- c.address
+ }
+ })
+}
diff --git a/pkg/remoting/grpc/channel.go b/pkg/remoting/grpc/channel.go
index b480a066..c6093829 100644
--- a/pkg/remoting/grpc/channel.go
+++ b/pkg/remoting/grpc/channel.go
@@ -62,14 +62,26 @@ func (c *Channel) IsClosed() bool {
func (c *Channel) close() {
c.mu.Lock()
- defer c.mu.Unlock()
- if !c.IsClosed() {
- close(c.closeCh)
- if err := c.stream.CloseSend(); err != nil {
+ if c.IsClosed() {
+ c.mu.Unlock()
+ return
+ }
+ close(c.closeCh)
+ stream := c.stream
+ conn := c.conn
+ c.mu.Unlock()
+
+ if stream != nil {
+ if err := stream.CloseSend(); err != nil {
log.Debugf("CloseSend error: %v", err)
}
- c.wg.Wait()
}
+ if conn != nil {
+ if err := conn.Close(); err != nil {
+ log.Debugf("ClientConn close error: %v", err)
+ }
+ }
+ c.wg.Wait()
}
func (c *Channel) softReconnect() error {
diff --git a/pkg/remoting/grpc/channel_manager.go
b/pkg/remoting/grpc/channel_manager.go
index a1859a67..085352bb 100644
--- a/pkg/remoting/grpc/channel_manager.go
+++ b/pkg/remoting/grpc/channel_manager.go
@@ -58,16 +58,26 @@ type ChannelManager struct {
//addr:map[Channel]bool
serverChannels sync.Map
//stream:bool
- allChannels sync.Map
- clientSize int32
- config *config.Config
+ allChannels sync.Map
+ clientSize int32
+ config *config.Config
+ seataConfig *config.SeataConfig
+ registrySubscription discovery.RegistrySubscription
+ startedAddresses sync.Map
+ serverAddressMu sync.RWMutex
+ serverAddressSnapshot map[string]struct{}
+ serverAddressReady bool
+ startChannel func(*discovery.ServiceInstance)
}
-func initChannelManager(config *config.Config) {
+type channelStartEntry struct{}
+
+func initChannelManager(grpcConfig *config.Config) {
if channelManager == nil {
onceChannelManager.Do(func() {
channelManager = &ChannelManager{
- config: config,
+ config: grpcConfig,
+ seataConfig: config.GetSeataConfig(),
allChannels: sync.Map{},
serverChannels: sync.Map{},
}
@@ -77,50 +87,162 @@ func initChannelManager(config *config.Config) {
}
func (g *ChannelManager) init() {
- addressList := g.getAvailServerList()
+ g.initWithRegistry(discovery.GetRegistry())
+}
+
+func (g *ChannelManager) initWithRegistry(registryService
discovery.RegistryService) {
+ if registryService == nil {
+ log.Warn("registry service not initialized")
+ return
+ }
+ if g.seataConfig == nil || g.seataConfig.TxServiceGroup == "" {
+ log.Warn("transaction service group not initialized")
+ return
+ }
+ if g.subscribeRegistry(registryService) {
+ return
+ }
+
+ addressList := g.getAvailServerList(registryService)
if len(addressList) == 0 {
log.Warn("no have valid seata server list")
}
- for _, address := range addressList {
- addr := net.JoinHostPort(address.Addr,
strconv.Itoa(address.Port))
- if conn, err := g.newConn(addr); err != nil {
- log.Errorf("failed to dial gRPC addr %s: %v", addr, err)
+ g.refreshServerList(addressList)
+}
+
+func (g *ChannelManager) subscribeRegistry(registryService
discovery.RegistryService) bool {
+ subscriber, ok := registryService.(discovery.RegistrySubscriber)
+ if !ok {
+ return false
+ }
+ subscription, err := subscriber.Subscribe(g.seataConfig.TxServiceGroup,
func(event discovery.RegistryChangeEvent) {
+ if event.Key != g.seataConfig.TxServiceGroup {
+ return
+ }
+ g.refreshServerList(event.Instances)
+ })
+ if err != nil {
+ log.Warnf("subscribe registry changes failed: %v", err)
+ return false
+ }
+ g.registrySubscription = subscription
+ return true
+}
+
+func (g *ChannelManager) refreshServerList(instances
[]*discovery.ServiceInstance) {
+ servers := make(map[string]*discovery.ServiceInstance, len(instances))
+ for _, instance := range instances {
+ if instance == nil || instance.Addr == "" || instance.Port <= 0
{
continue
- } else {
- regLock := sync.Mutex{}
- registered := atomic.Bool{}
- // todo if read g.config.ConnectionNum, will cause the
connect to fail
- for i := 1; i <= 1; i++ {
- channel := &Channel{
- addr: addr,
- conn: conn,
- sendCh: make(chan
*pb.GrpcMessageProto, defaultSendChBuffer),
- closeCh: make(chan struct{}),
- wg: sync.WaitGroup{},
- mu: sync.Mutex{},
- regLock: ®Lock,
- registered: ®istered,
- }
-
- channel, err = g.initChannel(channel)
- if err != nil {
- log.Errorf("failed to create gRPC
stream error: %v", err)
- continue
- }
- g.registerChannel(channel)
- }
- if err = g.registerTm(addr); err != nil {
- log.Errorf("%v", err)
- g.releaseChannelByAddr(addr)
- }
}
+ clone := &discovery.ServiceInstance{Addr: instance.Addr, Port:
instance.Port}
+ servers[serverAddress(clone)] = clone
+ }
+
+ removedAddresses := g.replaceServerAddressSnapshot(servers)
+ for _, address := range removedAddresses {
+ g.releaseChannelByAddr(address)
+ }
+
+ for _, instance := range servers {
+ g.ensureChannel(instance)
}
}
-func (g *ChannelManager) getAvailServerList() []*discovery.ServiceInstance {
- registryService := discovery.GetRegistry()
- instances, err :=
registryService.Lookup(config.GetSeataConfig().TxServiceGroup)
+func (g *ChannelManager) replaceServerAddressSnapshot(servers
map[string]*discovery.ServiceInstance) []string {
+ g.serverAddressMu.Lock()
+ var removedAddresses []string
+ for address := range g.serverAddressSnapshot {
+ if _, ok := servers[address]; !ok {
+ removedAddresses = append(removedAddresses, address)
+ }
+ }
+ g.serverAddressSnapshot = make(map[string]struct{}, len(servers))
+ for address := range servers {
+ g.serverAddressSnapshot[address] = struct{}{}
+ }
+ g.serverAddressReady = true
+ g.serverAddressMu.Unlock()
+ return removedAddresses
+}
+
+func (g *ChannelManager) ensureChannel(instance *discovery.ServiceInstance) {
+ address := serverAddress(instance)
+ entry := &channelStartEntry{}
+ if _, loaded := g.startedAddresses.LoadOrStore(address, entry); loaded {
+ return
+ }
+ if g.startChannel != nil {
+ g.startChannel(instance)
+ return
+ }
+ go g.startGrpcChannel(instance, entry)
+}
+
+func (g *ChannelManager) startGrpcChannel(instance *discovery.ServiceInstance,
entry *channelStartEntry) {
+ addr := serverAddress(instance)
+ if !g.isServerAddressAvailable(addr) || !g.isChannelStartCurrent(addr,
entry) {
+ g.deleteChannelStart(addr, entry)
+ return
+ }
+
+ conn, err := g.newConn(addr)
if err != nil {
+ log.Errorf("failed to dial gRPC addr %s: %v", addr, err)
+ g.deleteChannelStart(addr, entry)
+ return
+ }
+
+ regLock := sync.Mutex{}
+ registered := atomic.Bool{}
+ // todo if read g.config.ConnectionNum, will cause the connect to fail
+ channel := &Channel{
+ addr: addr,
+ conn: conn,
+ sendCh: make(chan *pb.GrpcMessageProto,
defaultSendChBuffer),
+ closeCh: make(chan struct{}),
+ wg: sync.WaitGroup{},
+ mu: sync.Mutex{},
+ regLock: ®Lock,
+ registered: ®istered,
+ }
+
+ channel, err = g.initChannel(channel)
+ if err != nil {
+ log.Errorf("failed to create gRPC stream error: %v", err)
+ _ = conn.Close()
+ g.deleteChannelStart(addr, entry)
+ return
+ }
+ if !g.isServerAddressAvailable(addr) || !g.isChannelStartCurrent(addr,
entry) {
+ channel.close()
+ g.deleteChannelStart(addr, entry)
+ return
+ }
+ if !g.registerChannel(channel) {
+ _ = conn.Close()
+ g.deleteChannelStart(addr, entry)
+ return
+ }
+ if err = g.registerTm(addr); err != nil {
+ log.Errorf("%v", err)
+ g.releaseChannelByAddr(addr)
+ }
+}
+
+func (g *ChannelManager) isChannelStartCurrent(address string, entry
*channelStartEntry) bool {
+ current, ok := g.startedAddresses.Load(address)
+ return ok && current == entry
+}
+
+func (g *ChannelManager) deleteChannelStart(address string, entry
*channelStartEntry) {
+ g.startedAddresses.CompareAndDelete(address, entry)
+}
+
+func (g *ChannelManager) getAvailServerList(registryService
discovery.RegistryService) []*discovery.ServiceInstance {
+ instances, err := registryService.Lookup(g.seataConfig.TxServiceGroup)
+ if err != nil {
+ log.Warnf("lookup seata server list failed: %v", err)
return nil
}
return instances
@@ -190,26 +312,19 @@ func (g *ChannelManager) registerTm(addr string) error {
}
func (g *ChannelManager) selectChannel(msg interface{}) *Channel {
- selected := loadbalance.Select(loadbalance.GetLoadBalanceConfig().Type,
&g.allChannels, g.getXid(msg))
- channel, ok := selected.(*Channel)
- if ok && channel != nil {
+ channels := g.selectableChannels()
+ channel := g.selectAvailableChannel(channels, msg)
+ if channel != nil {
return channel
}
- if g.clientSize == 0 {
+ if selectableConnectionCount(channels) == 0 {
ticker := time.NewTicker(time.Duration(checkAliveInternal) *
time.Millisecond)
defer ticker.Stop()
for i := 0; i < maxCheckAliveRetry; i++ {
<-ticker.C
- g.allChannels.Range(func(key, value interface{}) bool {
- channel = key.(*Channel)
- if channel.IsClosed() {
- g.releaseChannel(channel)
- } else {
- return false
- }
- return true
- })
+ channels = g.selectableChannels()
+ channel = g.selectAvailableChannel(channels, msg)
if channel != nil {
return channel
}
@@ -218,6 +333,31 @@ func (g *ChannelManager) selectChannel(msg interface{})
*Channel {
return nil
}
+func (g *ChannelManager) selectAvailableChannel(channels *sync.Map, msg
interface{}) *Channel {
+ selected := loadbalance.Select(loadbalance.GetLoadBalanceConfig().Type,
channels, g.getXid(msg))
+ channel, ok := selected.(*Channel)
+ if ok && g.isChannelSelectable(channels, channel) {
+ return channel
+ }
+
+ // Some load balancers cache their result. Re-select from the current
+ // snapshot if a cached result points to a removed server address.
+ selected = loadbalance.Select("RandomLoadBalance", channels,
g.getXid(msg))
+ channel, ok = selected.(*Channel)
+ if ok && g.isChannelSelectable(channels, channel) {
+ return channel
+ }
+ return nil
+}
+
+func (g *ChannelManager) isChannelSelectable(channels *sync.Map, channel
*Channel) bool {
+ if channel == nil || channel.IsClosed() ||
!g.isServerAddressAvailable(channel.addr) {
+ return false
+ }
+ _, ok := channels.Load(channel)
+ return ok
+}
+
func (g *ChannelManager) getXid(msg interface{}) string {
switch tmpMsg := msg.(type) {
case *pb.AbstractGlobalEndRequestProto:
@@ -253,22 +393,43 @@ func (g *ChannelManager) getXid(msg interface{}) string {
}
func (g *ChannelManager) releaseChannel(channel *Channel) {
- g.allChannels.Delete(channel)
- if !channel.IsClosed() {
- m, _ := g.serverChannels.LoadOrStore(channel.addr, &sync.Map{})
+ if _, loaded := g.allChannels.LoadAndDelete(channel); !loaded {
+ return
+ }
+ if m, ok := g.serverChannels.Load(channel.addr); ok {
sMap := m.(*sync.Map)
sMap.Delete(channel)
+ }
+ if !channel.IsClosed() {
channel.close()
}
atomic.AddInt32(&g.clientSize, -1)
+ if g.getAllChannelIsClosedByAddr(channel.addr) {
+ g.startedAddresses.Delete(channel.addr)
+ }
}
-func (g *ChannelManager) registerChannel(channel *Channel) {
- g.allChannels.LoadOrStore(channel, true)
+func (g *ChannelManager) registerChannel(channel *Channel) bool {
+ if !g.isServerAddressAvailable(channel.addr) {
+ log.Warnf("skip channel for removed server address: %s",
channel.addr)
+ if _, loaded := g.allChannels.Load(channel); loaded {
+ g.releaseChannel(channel)
+ } else if !channel.IsClosed() {
+ channel.close()
+ }
+ return false
+ }
+ if _, loaded := g.allChannels.LoadOrStore(channel, true); loaded {
+ return true
+ }
m, _ := g.serverChannels.LoadOrStore(channel.addr, &sync.Map{})
sMap := m.(*sync.Map)
sMap.Store(channel, true)
+ if _, loaded := g.startedAddresses.Load(channel.addr); !loaded {
+ g.startedAddresses.Store(channel.addr, &channelStartEntry{})
+ }
atomic.AddInt32(&g.clientSize, 1)
+ return true
}
func (g *ChannelManager) releaseChannelByAddr(addr string) {
@@ -279,7 +440,10 @@ func (g *ChannelManager) releaseChannelByAddr(addr string)
{
}
return true
})
+ g.serverChannels.Delete(addr)
+ g.startedAddresses.Delete(addr)
}
+
func (g *ChannelManager) getAllChannelIsClosedByAddr(addr string) bool {
flag := true
g.allChannels.Range(func(key, value any) bool {
@@ -294,3 +458,46 @@ func (g *ChannelManager) getAllChannelIsClosedByAddr(addr
string) bool {
})
return flag
}
+
+func (g *ChannelManager) selectableChannels() *sync.Map {
+ channels := &sync.Map{}
+ g.allChannels.Range(func(key, value interface{}) bool {
+ channel, ok := key.(*Channel)
+ if !ok {
+ return true
+ }
+ if channel.IsClosed() {
+ g.releaseChannel(channel)
+ return true
+ }
+ if g.isServerAddressAvailable(channel.addr) {
+ channels.Store(channel, value)
+ }
+ return true
+ })
+ return channels
+}
+
+func selectableConnectionCount(connections *sync.Map) int {
+ count := 0
+ connections.Range(func(_, _ interface{}) bool {
+ count++
+ return true
+ })
+ return count
+}
+
+func (g *ChannelManager) isServerAddressAvailable(address string) bool {
+ g.serverAddressMu.RLock()
+ defer g.serverAddressMu.RUnlock()
+
+ if !g.serverAddressReady {
+ return true
+ }
+ _, ok := g.serverAddressSnapshot[address]
+ return ok
+}
+
+func serverAddress(instance *discovery.ServiceInstance) string {
+ return net.JoinHostPort(instance.Addr, strconv.Itoa(instance.Port))
+}
diff --git a/pkg/remoting/grpc/channel_manager_test.go
b/pkg/remoting/grpc/channel_manager_test.go
new file mode 100644
index 00000000..fb23f193
--- /dev/null
+++ b/pkg/remoting/grpc/channel_manager_test.go
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package grpc
+
+import (
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+
+ "seata.apache.org/seata-go/v2/pkg/discovery"
+ "seata.apache.org/seata-go/v2/pkg/remoting/config"
+ "seata.apache.org/seata-go/v2/pkg/remoting/loadbalance"
+)
+
+func TestChannelManagerRefreshesServerListFromRegistrySubscription(t
*testing.T) {
+ registry := &fakeSubscriberRegistry{
+ fakeLookupRegistry: fakeLookupRegistry{
+ instances: []*discovery.ServiceInstance{{Addr:
"127.0.0.1", Port: 8091}},
+ },
+ }
+ started := make(chan string, 2)
+ startCount := make(map[string]int)
+ var startMu sync.Mutex
+ manager := newTestChannelManager(func(instance
*discovery.ServiceInstance) {
+ address := serverAddress(instance)
+ startMu.Lock()
+ startCount[address]++
+ startMu.Unlock()
+ started <- address
+ })
+
+ manager.initWithRegistry(registry)
+ assert.Equal(t, "default_tx_group", registry.subscribeKey)
+ assert.Equal(t, "127.0.0.1:8091", nextStartedChannel(t, started))
+
+ registry.publish([]*discovery.ServiceInstance{{Addr: "127.0.0.2", Port:
8092}})
+ assert.Equal(t, "127.0.0.2:8092", nextStartedChannel(t, started))
+ assert.False(t, manager.isServerAddressAvailable("127.0.0.1:8091"))
+ assert.True(t, manager.isServerAddressAvailable("127.0.0.2:8092"))
+
+ registry.publish([]*discovery.ServiceInstance{{Addr: "127.0.0.2", Port:
8092}})
+ startMu.Lock()
+ assert.Equal(t, 1, startCount["127.0.0.2:8092"])
+ startMu.Unlock()
+
+ registry.publish([]*discovery.ServiceInstance{{Addr: "127.0.0.1", Port:
8091}})
+ assert.Equal(t, "127.0.0.1:8091", nextStartedChannel(t, started))
+ startMu.Lock()
+ assert.Equal(t, 2, startCount["127.0.0.1:8091"])
+ startMu.Unlock()
+}
+
+func TestChannelManagerFallsBackToLookupWhenRegistryDoesNotSubscribe(t
*testing.T) {
+ registry := &fakeLookupRegistry{
+ instances: []*discovery.ServiceInstance{{Addr: "127.0.0.1",
Port: 8091}},
+ }
+ started := make(chan string, 1)
+ manager := newTestChannelManager(func(instance
*discovery.ServiceInstance) {
+ started <- serverAddress(instance)
+ })
+
+ manager.initWithRegistry(registry)
+ assert.Equal(t, "default_tx_group", registry.lookupKey)
+ assert.Equal(t, "127.0.0.1:8091", nextStartedChannel(t, started))
+ assert.True(t, manager.isServerAddressAvailable("127.0.0.1:8091"))
+}
+
+func TestChannelManagerSelectChannelSkipsRemovedServerAddress(t *testing.T) {
+ previousLoadBalance := loadbalance.GetLoadBalanceConfig()
+ loadbalance.InitLoadBalanceConfig(loadbalance.Config{Type:
"RoundRobinLoadBalance"})
+ t.Cleanup(func() {
+ loadbalance.InitLoadBalanceConfig(previousLoadBalance)
+ })
+
+ manager := newTestChannelManager(nil)
+ manager.refreshServerList([]*discovery.ServiceInstance{
+ {Addr: "127.0.0.1", Port: 8091},
+ {Addr: "127.0.0.2", Port: 8092},
+ })
+
+ removed := newSelectableTestChannel("127.0.0.1:8091")
+ kept := newSelectableTestChannel("127.0.0.2:8092")
+ assert.True(t, manager.registerChannel(removed))
+ assert.True(t, manager.registerChannel(kept))
+ manager.refreshServerList([]*discovery.ServiceInstance{{Addr:
"127.0.0.2", Port: 8092}})
+
+ assert.True(t, removed.IsClosed())
+ assert.Equal(t, kept, manager.selectChannel(&struct{ TransactionName
string }{TransactionName: "tx"}))
+ if _, ok := manager.startedAddresses.Load("127.0.0.1:8091"); ok {
+ t.Fatal("removed channel address is still tracked")
+ }
+}
+
+func TestChannelCloseDoesNotWaitWithMutexHeld(t *testing.T) {
+ channel := &Channel{closeCh: make(chan struct{})}
+ channel.wg.Add(1)
+ observedClosed := make(chan bool, 1)
+
+ go func() {
+ defer channel.wg.Done()
+ <-channel.closeCh
+ channel.mu.Lock()
+ observedClosed <- channel.IsClosed()
+ channel.mu.Unlock()
+ }()
+
+ closed := make(chan struct{})
+ go func() {
+ channel.close()
+ close(closed)
+ }()
+
+ select {
+ case <-closed:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for channel close")
+ }
+ assert.True(t, <-observedClosed)
+}
+
+func newTestChannelManager(startChannel func(*discovery.ServiceInstance))
*ChannelManager {
+ if startChannel == nil {
+ startChannel = func(*discovery.ServiceInstance) {}
+ }
+ return &ChannelManager{
+ config: &config.Config{},
+ seataConfig: &config.SeataConfig{TxServiceGroup:
"default_tx_group"},
+ startChannel: startChannel,
+ }
+}
+
+func newSelectableTestChannel(addr string) *Channel {
+ return &Channel{addr: addr, closeCh: make(chan struct{})}
+}
+
+func nextStartedChannel(t *testing.T, started <-chan string) string {
+ t.Helper()
+
+ select {
+ case address := <-started:
+ return address
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for gRPC channel start")
+ }
+ return ""
+}
+
+type fakeLookupRegistry struct {
+ lookupKey string
+ instances []*discovery.ServiceInstance
+}
+
+func (r *fakeLookupRegistry) Lookup(key string) ([]*discovery.ServiceInstance,
error) {
+ r.lookupKey = key
+ return cloneTestServiceInstances(r.instances), nil
+}
+
+func (r *fakeLookupRegistry) Close() {}
+
+type fakeSubscriberRegistry struct {
+ fakeLookupRegistry
+ subscribeKey string
+ listener discovery.RegistryChangeListener
+}
+
+func (r *fakeSubscriberRegistry) Subscribe(key string, listener
discovery.RegistryChangeListener) (discovery.RegistrySubscription, error) {
+ r.subscribeKey = key
+ r.listener = listener
+ listener(discovery.RegistryChangeEvent{Key: key, Instances:
cloneTestServiceInstances(r.instances)})
+ return fakeRegistrySubscription{}, nil
+}
+
+func (r *fakeSubscriberRegistry) publish(instances
[]*discovery.ServiceInstance) {
+ r.listener(discovery.RegistryChangeEvent{Key: r.subscribeKey,
Instances: cloneTestServiceInstances(instances)})
+}
+
+type fakeRegistrySubscription struct{}
+
+func (fakeRegistrySubscription) Unsubscribe() {}
+
+func cloneTestServiceInstances(instances []*discovery.ServiceInstance)
[]*discovery.ServiceInstance {
+ clones := make([]*discovery.ServiceInstance, 0, len(instances))
+ for _, instance := range instances {
+ clone := *instance
+ clones = append(clones, &clone)
+ }
+ return clones
+}
diff --git a/pkg/remoting/grpc/listener.go b/pkg/remoting/grpc/listener.go
index f4d2915f..500ad1dd 100644
--- a/pkg/remoting/grpc/listener.go
+++ b/pkg/remoting/grpc/listener.go
@@ -66,6 +66,10 @@ func (g *grpcClientHandler) monitorStreamHealth(ctx
context.Context, channel *Ch
for {
select {
case <-ticker.C:
+ if
!channelManager.isServerAddressAvailable(channel.addr) {
+ channelManager.releaseChannel(channel)
+ return
+ }
err := g.transferHeartBeat(channel,
&pb.HeartbeatMessageProto{Ping: true})
if err != nil {
heartBeatRetryTimes++
@@ -80,6 +84,10 @@ func (g *grpcClientHandler) monitorStreamHealth(ctx
context.Context, channel *Ch
flag := false
var reconnectErr error
for !flag {
+ if
!channelManager.isServerAddressAvailable(channel.addr) {
+
channelManager.releaseChannel(channel)
+ return
+ }
state := channel.conn.GetState()
switch state {
case connectivity.Shutdown:
@@ -98,7 +106,9 @@ func (g *grpcClientHandler) monitorStreamHealth(ctx
context.Context, channel *Ch
channel.conn.WaitForStateChange(ctx, state)
continue
}
-
channelManager.registerChannel(channel)
+ if
!channelManager.registerChannel(channel) {
+ return
+ }
flag = true
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
