You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by al...@apache.org on 2026/05/21 06:06:25 UTC
(dubbo-go) 04/14: fix(triple): make protocol initialization and invoker tracking race-safe (#3307)
This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
commit 5ead6439c3e8ea16e72b731b5d67e3d66f35c7e4
Author: 承潜 <29...@qq.com>
AuthorDate: Wed Apr 29 21:05:50 2026 +0800
fix(triple): make protocol initialization and invoker tracking race-safe (#3307)
---
protocol/base/base_protocol.go | 28 +++++++++++----
protocol/base/base_protocol_test.go | 66 ++++++++++++++++++++++++++++++++++-
protocol/triple/active_notify_test.go | 32 ++++++++++++-----
protocol/triple/triple.go | 9 +++--
protocol/triple/triple_test.go | 57 +++++++++++++++++++++++++++---
5 files changed, 169 insertions(+), 23 deletions(-)
diff --git a/protocol/base/base_protocol.go b/protocol/base/base_protocol.go
index 964e10620..f07502017 100644
--- a/protocol/base/base_protocol.go
+++ b/protocol/base/base_protocol.go
@@ -44,8 +44,10 @@ type RegistryUnregisterer interface {
// BaseProtocol is default protocol implement.
type BaseProtocol struct {
- exporterMap *sync.Map
- invokers []Invoker
+ exporterMap *sync.Map
+ invokersLock sync.RWMutex
+ // invokers is protected by invokersLock; never expose the backing slice to callers.
+ invokers []Invoker
}
// NewBaseProtocol creates a new BaseProtocol
@@ -67,12 +69,20 @@ func (bp *BaseProtocol) ExporterMap() *sync.Map {
// SetInvokers sets invoker into local memory
func (bp *BaseProtocol) SetInvokers(invoker Invoker) {
+ bp.invokersLock.Lock()
+ defer bp.invokersLock.Unlock()
+
bp.invokers = append(bp.invokers, invoker)
}
-// Invokers gets all invokers
+// Invokers gets a snapshot of all invokers.
func (bp *BaseProtocol) Invokers() []Invoker {
- return bp.invokers
+ bp.invokersLock.RLock()
+ defer bp.invokersLock.RUnlock()
+
+ invokers := make([]Invoker, len(bp.invokers))
+ copy(invokers, bp.invokers)
+ return invokers
}
// Export is default export implement.
@@ -87,13 +97,19 @@ func (bp *BaseProtocol) Refer(url *common.URL) Invoker {
// Destroy will destroy all invoker and exporter, so it only is called once.
func (bp *BaseProtocol) Destroy() {
+ // Do not hold invokersLock while calling Invoker.Destroy(); implementations may run protocol callbacks.
+ bp.invokersLock.Lock()
+ invokers := make([]Invoker, len(bp.invokers))
+ copy(invokers, bp.invokers)
+ bp.invokers = []Invoker{}
+ bp.invokersLock.Unlock()
+
// destroy invokers
- for _, invoker := range bp.invokers {
+ for _, invoker := range invokers {
if invoker != nil {
invoker.Destroy()
}
}
- bp.invokers = []Invoker{}
// un export exporters
bp.exporterMap.Range(func(key, exporter any) bool {
diff --git a/protocol/base/base_protocol_test.go b/protocol/base/base_protocol_test.go
index 2c35820e3..0aed7ea82 100644
--- a/protocol/base/base_protocol_test.go
+++ b/protocol/base/base_protocol_test.go
@@ -18,6 +18,8 @@
package base
import (
+ "strconv"
+ "sync"
"testing"
)
@@ -32,7 +34,6 @@ import (
func TestNewBaseProtocol(t *testing.T) {
protocol := NewBaseProtocol()
- assert.NotNil(t, protocol)
assert.NotNil(t, protocol.exporterMap)
assert.Empty(t, protocol.invokers)
}
@@ -75,6 +76,54 @@ func TestBaseProtocol_SetInvokers(t *testing.T) {
assert.Contains(t, invokers, invoker2)
}
+func TestBaseProtocol_SetInvokersConcurrent(t *testing.T) {
+ protocol := NewBaseProtocol()
+ const goroutines = 128
+
+ start := make(chan struct{})
+ var writers sync.WaitGroup
+ for i := 0; i < goroutines; i++ {
+ i := i
+ writers.Add(1)
+ go func() {
+ defer writers.Done()
+ <-start
+ url := common.NewURLWithOptions(
+ common.WithProtocol("dubbo"),
+ common.WithIp("localhost"),
+ common.WithPort(strconv.Itoa(9000+i)),
+ )
+ protocol.SetInvokers(NewBaseInvoker(url))
+ }()
+ }
+
+ const readers = 8
+ stopReaders := make(chan struct{})
+ var readerWG sync.WaitGroup
+ for i := 0; i < readers; i++ {
+ readerWG.Add(1)
+ go func() {
+ defer readerWG.Done()
+ <-start
+ for {
+ select {
+ case <-stopReaders:
+ return
+ default:
+ _ = protocol.Invokers()
+ }
+ }
+ }()
+ }
+
+ close(start)
+ writers.Wait()
+ close(stopReaders)
+ readerWG.Wait()
+
+ assert.Len(t, protocol.Invokers(), goroutines)
+}
+
func TestBaseProtocol_Invokers(t *testing.T) {
protocol := NewBaseProtocol()
url, _ := common.NewURL("dubbo://localhost:9090")
@@ -87,6 +136,21 @@ func TestBaseProtocol_Invokers(t *testing.T) {
assert.Equal(t, invoker, invokers[0])
}
+func TestBaseProtocol_InvokersReturnsSnapshot(t *testing.T) {
+ protocol := NewBaseProtocol()
+ url1, _ := common.NewURL("dubbo://localhost:9090")
+ invoker1 := NewBaseInvoker(url1)
+ url2, _ := common.NewURL("dubbo://localhost:9091")
+ invoker2 := NewBaseInvoker(url2)
+
+ protocol.SetInvokers(invoker1)
+
+ invokers := protocol.Invokers()
+ invokers[0] = invoker2
+
+ assert.Equal(t, invoker1, protocol.Invokers()[0])
+}
+
func TestBaseProtocol_Export(t *testing.T) {
protocol := NewBaseProtocol()
url, _ := common.NewURL("dubbo://localhost:9090")
diff --git a/protocol/triple/active_notify_test.go b/protocol/triple/active_notify_test.go
index 83730b221..6ffcf7c03 100644
--- a/protocol/triple/active_notify_test.go
+++ b/protocol/triple/active_notify_test.go
@@ -42,14 +42,28 @@ import (
)
type testTripleClosingEventHandler struct {
+ lock sync.Mutex
events []gracefulshutdown.ClosingEvent
}
func (h *testTripleClosingEventHandler) HandleClosingEvent(event gracefulshutdown.ClosingEvent) bool {
+ h.lock.Lock()
+ defer h.lock.Unlock()
+
h.events = append(h.events, event)
return true
}
+// Events returns a snapshot because health-watch callbacks run in a goroutine.
+func (h *testTripleClosingEventHandler) Events() []gracefulshutdown.ClosingEvent {
+ h.lock.Lock()
+ defer h.lock.Unlock()
+
+ events := make([]gracefulshutdown.ClosingEvent, len(h.events))
+ copy(events, h.events)
+ return events
+}
+
func TestTripleInvokerHandleHealthStatusNotServing(t *testing.T) {
url, err := common.NewURL("tri://127.0.0.1:20000/org.apache.dubbo-go.mockService?group=group&version=1.0.0")
if err != nil {
@@ -61,10 +75,11 @@ func TestTripleInvokerHandleHealthStatusNotServing(t *testing.T) {
handled := invoker.handleHealthStatus(grpc_health_v1.HealthCheckResponse_NOT_SERVING, handler)
assert.True(t, handled)
- if assert.Len(t, handler.events, 1) {
- assert.Equal(t, "triple-health-watch", handler.events[0].Source)
- assert.Equal(t, url.GetCacheInvokerMapKey(), handler.events[0].InstanceKey)
- assert.Equal(t, url.ServiceKey(), handler.events[0].ServiceKey)
+ events := handler.Events()
+ if assert.Len(t, events, 1) {
+ assert.Equal(t, "triple-health-watch", events[0].Source)
+ assert.Equal(t, url.GetCacheInvokerMapKey(), events[0].InstanceKey)
+ assert.Equal(t, url.ServiceKey(), events[0].ServiceKey)
}
}
@@ -145,10 +160,11 @@ func TestTripleHealthWatchEmitsClosingEvent(t *testing.T) {
close(notServing)
require.Eventually(t, func() bool {
- return len(handler.events) == 1
+ return len(handler.Events()) == 1
}, 3*time.Second, 20*time.Millisecond)
- assert.Equal(t, "triple-health-watch", handler.events[0].Source)
- assert.Equal(t, url.GetCacheInvokerMapKey(), handler.events[0].InstanceKey)
- assert.Equal(t, serviceKey, handler.events[0].ServiceKey)
+ events := handler.Events()
+ assert.Equal(t, "triple-health-watch", events[0].Source)
+ assert.Equal(t, url.GetCacheInvokerMapKey(), events[0].InstanceKey)
+ assert.Equal(t, serviceKey, events[0].ServiceKey)
}
diff --git a/protocol/triple/triple.go b/protocol/triple/triple.go
index 2b394a9ca..c90ef65d9 100644
--- a/protocol/triple/triple.go
+++ b/protocol/triple/triple.go
@@ -42,7 +42,8 @@ const (
)
var (
- tripleProtocol *TripleProtocol
+ tripleProtocol *TripleProtocol
+ tripleProtocolOnce sync.Once
)
var tripleServerGracefulStop = func(server *Server) {
@@ -204,10 +205,12 @@ func NewTripleProtocol() *TripleProtocol {
}
}
+// GetProtocol returns the shared Triple protocol instance.
+// The protocol is initialized lazily because extension registration stores this factory callback.
func GetProtocol() base.Protocol {
- if tripleProtocol == nil {
+ tripleProtocolOnce.Do(func() {
tripleProtocol = NewTripleProtocol()
- }
+ })
return tripleProtocol
}
diff --git a/protocol/triple/triple_test.go b/protocol/triple/triple_test.go
index ab561f7bf..6af85ff40 100644
--- a/protocol/triple/triple_test.go
+++ b/protocol/triple/triple_test.go
@@ -19,6 +19,7 @@ package triple
import (
"context"
+ "sync"
"sync/atomic"
"testing"
"time"
@@ -30,9 +31,25 @@ import (
import (
"dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
tri "dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol"
)
+func resetTripleProtocolForTest(t *testing.T) {
+ t.Helper()
+
+ tripleProtocol = nil
+ tripleProtocolOnce = sync.Once{}
+}
+
+func setTripleProtocolForTest(t *testing.T, protocol *TripleProtocol) {
+ t.Helper()
+
+ tripleProtocol = protocol
+ tripleProtocolOnce = sync.Once{}
+ tripleProtocolOnce.Do(func() {})
+}
+
func TestNewTripleProtocol(t *testing.T) {
tp := NewTripleProtocol()
@@ -42,8 +59,7 @@ func TestNewTripleProtocol(t *testing.T) {
}
func TestGetProtocol(t *testing.T) {
- // reset singleton for test isolation
- tripleProtocol = nil
+ resetTripleProtocolForTest(t)
p1 := GetProtocol()
assert.NotNil(t, p1)
@@ -53,6 +69,38 @@ func TestGetProtocol(t *testing.T) {
assert.Same(t, p1, p2)
}
+func TestGetProtocolConcurrent(t *testing.T) {
+ resetTripleProtocolForTest(t)
+
+ const goroutines = 128
+ start := make(chan struct{})
+ results := make(chan base.Protocol, goroutines)
+
+ var wg sync.WaitGroup
+ for i := 0; i < goroutines; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ results <- GetProtocol()
+ }()
+ }
+
+ close(start)
+ wg.Wait()
+ close(results)
+
+ var first base.Protocol
+ for protocol := range results {
+ assert.NotNil(t, protocol)
+ if first == nil {
+ first = protocol
+ continue
+ }
+ assert.Same(t, first, protocol)
+ }
+}
+
func TestTripleProtocolRegistration(t *testing.T) {
// verify protocol is registered via init()
p := extension.GetProtocol(TRIPLE)
@@ -68,12 +116,11 @@ func TestTripleGracefulShutdownCallbackRegistration(t *testing.T) {
assert.True(t, ok)
assert.NotNil(t, cb)
- original := tripleProtocol
tp := NewTripleProtocol()
tp.serverMap["graceful-test"] = &Server{triServer: tri.NewServer("", nil)}
- tripleProtocol = tp
+ setTripleProtocolForTest(t, tp)
t.Cleanup(func() {
- tripleProtocol = original
+ resetTripleProtocolForTest(t)
})
assert.NotPanics(t, func() {