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
The following commit(s) were added to refs/heads/develop by this push:
new 5e87b0673 fix(registry): add lock protection for serviceListeners map
access (#3442)
5e87b0673 is described below
commit 5e87b0673f70273e1986e3b9508f7ea007a88fb2
Author: aias00 <[email protected]>
AuthorDate: Sun Jul 26 18:41:24 2026 -0700
fix(registry): add lock protection for serviceListeners map access (#3442)
* fix(registry): add lock protection for serviceListeners map access
serviceListeners map was read/written without holding the lock in
SubscribeURL and UnSubscribe, while the same struct's
serviceMappingListeners
already had proper lock protection. This could cause data races when
concurrent subscribe/unsubscribe operations access the map.
- SubscribeURL: use s.lock.Lock() to protect the check-then-act read+write
of serviceListeners (write lock needed to prevent TOCTOU race)
- UnSubscribe: use s.lock.RLock() for the read access of serviceListeners
Co-Authored-By: Claude <[email protected]>
* fix(registry): use defer+lambda for serviceListeners lock to prevent
deadlock on panic
Wrap lock/unlock in anonymous functions with defer to ensure the lock
is always released even if a panic occurs between Lock() and Unlock(),
as suggested in PR review.
Co-Authored-By: Claude <[email protected]>
* fix(registry): narrow service-discovery lock to map check/install only
Address [P1] review on PR #3442: the write lock previously wrapped
GetInstances and listener.OnEvent, which may perform external RPC /
metadata-report calls and block every other subscribe/unsubscribe on
the same registry.
- Move GetInstances/OnEvent out of the lock; build the listener first.
- Install under a short write lock with a double-check so a concurrent
subscriber for the same key does not install a duplicate listener.
- UnSubscribe uses the new getServiceListener helper (short RLock) in
place of the inline lambda.
- Add TestServiceDiscoveryRegistrySubscribeURLDoesNotHoldLockOnExternalCall
which blocks GetInstances for one service and asserts that subscribing a
different service is not stalled. The test fails on the previous
lock-during-GetInstances implementation.
Co-Authored-By: Claude <[email protected]>
* chore: trigger CI rerun on Integration Test
Re-runs the Integration Test job, which previously failed due to a
transient 500 from sum.golang.org while indexing the freshly-pushed
fork commit (not a code issue; the CI unit-test job passed).
Co-Authored-By: Claude <[email protected]>
* ci: skip sumdb for fork integration test replace
---------
Co-authored-by: Claude <[email protected]>
---
integrate_test.sh | 1 +
.../servicediscovery/service_discovery_registry.go | 70 +++++++++++-----
.../service_discovery_registry_test.go | 93 ++++++++++++++++++++++
3 files changed, 145 insertions(+), 19 deletions(-)
diff --git a/integrate_test.sh b/integrate_test.sh
index 6cc07122f..9a973a80e 100755
--- a/integrate_test.sh
+++ b/integrate_test.sh
@@ -31,6 +31,7 @@ git clone -b $3
https://github.com/apache/dubbo-go-samples.git samples --depth=1
if [ "$1" == "apache/dubbo-go" ]; then
go mod edit
-replace=dubbo.apache.org/dubbo-go/v3=dubbo.apache.org/dubbo-go/v3@"$2"
else
+ export GONOSUMDB="${GONOSUMDB:+${GONOSUMDB},}github.com/$1/v3"
go mod edit -replace=dubbo.apache.org/dubbo-go/v3=github.com/"$1"/v3@"$2"
fi
diff --git a/registry/servicediscovery/service_discovery_registry.go
b/registry/servicediscovery/service_discovery_registry.go
index 71088d4a1..52c0c6850 100644
--- a/registry/servicediscovery/service_discovery_registry.go
+++ b/registry/servicediscovery/service_discovery_registry.go
@@ -221,8 +221,7 @@ func (s *serviceDiscoveryRegistry) UnSubscribe(url
*common.URL, listener registr
return nil
}
serviceNamesKey := sortServices(services)
- l := s.serviceListeners[serviceNamesKey]
- if l != nil {
+ if l := s.getServiceListener(serviceNamesKey); l != nil {
l.RemoveListener(url.ServiceKey())
}
s.stopListen(url)
@@ -525,36 +524,69 @@ func (s *serviceDiscoveryRegistry) Subscribe(url
*common.URL, notify registry.No
}
func (s *serviceDiscoveryRegistry) SubscribeURL(url *common.URL, notify
registry.NotifyListener, services *gxset.HashSet) {
- var err error
serviceNamesKey := sortServices(services)
protocol := constant.TriProtocol // consume "tri" protocol by default,
other protocols need to be specified on reference/consumer explicitly
if url.Protocol != "" {
protocol = url.Protocol
}
protocolServiceKey := url.ServiceKey() + ":" + protocol
- listener := s.serviceListeners[serviceNamesKey]
- if listener == nil {
- listener =
NewServiceInstancesChangedListener(url.GetParam(constant.ApplicationKey, ""),
s.url.GetParam(constant.RegistryIdKey, constant.DefaultKey), services)
- for _, serviceNameTmp := range services.Values() {
- serviceName := serviceNameTmp.(string)
- instances :=
s.serviceDiscovery.GetInstances(serviceName)
- logger.Infof("[Registry][ServiceDiscovery] synchronized
instance notification on application %s subscription, instance list size %s",
serviceName, len(instances))
- err =
listener.OnEvent(®istry.ServiceInstancesChangedEvent{
- ServiceName: serviceName,
- Instances: instances,
- })
- if err != nil {
- logger.Warnf("[Registry][ServiceDiscovery]
ServiceInstancesChangedListenerImpl handle error, err=%v", err)
- }
+
+ // Fast path: reuse an already installed listener without touching
external calls.
+ if listener := s.getServiceListener(serviceNamesKey); listener != nil {
+ s.subscribeAndNotify(url, serviceNamesKey, protocolServiceKey,
listener, notify)
+ return
+ }
+
+ // Build the listener and load its initial instances outside s.lock.
+ // GetInstances and OnEvent may perform external RPC / metadata-report
+ // calls; holding the registry write lock across them would block every
+ // other subscribe/unsubscribe on this registry. The lock below only
guards
+ // the serviceListeners check/install, never the external work.
+ listener :=
NewServiceInstancesChangedListener(url.GetParam(constant.ApplicationKey, ""),
s.url.GetParam(constant.RegistryIdKey, constant.DefaultKey), services)
+ for _, serviceNameTmp := range services.Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized
instance notification on application %s subscription, instance list size %s",
serviceName, len(instances))
+ if err :=
listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery]
ServiceInstancesChangedListenerImpl handle error, err=%v", err)
}
}
- s.serviceListeners[serviceNamesKey] = listener
+
+ // Install under a short write lock with a double-check so a concurrent
+ // subscriber for the same key does not install a duplicate listener.
+ s.lock.Lock()
+ if existing := s.serviceListeners[serviceNamesKey]; existing != nil {
+ listener = existing
+ } else {
+ s.serviceListeners[serviceNamesKey] = listener
+ }
+ s.lock.Unlock()
+
+ s.subscribeAndNotify(url, serviceNamesKey, protocolServiceKey,
listener, notify)
+}
+
+// getServiceListener returns the listener installed for serviceNamesKey, or
nil
+// if none has been installed yet, acquired under a read lock.
+func (s *serviceDiscoveryRegistry) getServiceListener(serviceNamesKey string)
registry.ServiceInstancesChangedListener {
+ s.lock.RLock()
+ defer s.lock.RUnlock()
+ return s.serviceListeners[serviceNamesKey]
+}
+
+// subscribeAndNotify registers the notify callback and asynchronously wires
the
+// listener into the service discovery so the caller does not block on it.
+func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL,
serviceNamesKey, protocolServiceKey string,
+ listener registry.ServiceInstancesChangedListener, notify
registry.NotifyListener,
+) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
event :=
metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to
registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err = s.serviceDiscovery.AddListener(listener)
+ err := s.serviceDiscovery.AddListener(listener)
event.Succ = err != nil
event.End = time.Now()
event.Attachment[constant.InterfaceKey] = url.Interface()
diff --git a/registry/servicediscovery/service_discovery_registry_test.go
b/registry/servicediscovery/service_discovery_registry_test.go
index 2cce2a696..15e4d8b04 100644
--- a/registry/servicediscovery/service_discovery_registry_test.go
+++ b/registry/servicediscovery/service_discovery_registry_test.go
@@ -1062,3 +1062,96 @@ func TestServiceDiscoveryRegistryUnRegister_Concurrent(t
*testing.T) {
// unregister should be called
assert.True(t, mockSD.unregisterCalled)
}
+
+// TestServiceDiscoveryRegistrySubscribeURLDoesNotHoldLockOnExternalCall
+// verifies that a stalled GetInstances (e.g. an external RPC or
metadata-report
+// fetch) for one service does not block subscription for a different service.
+// The registry write lock must only guard the serviceListeners map, not the
+// external GetInstances / OnEvent calls. See review feedback [P1] on PR #3442.
+func TestServiceDiscoveryRegistrySubscribeURLDoesNotHoldLockOnExternalCall(t
*testing.T) {
+ mockSD, _ := setupEnvironment(t)
+ regID := fmt.Sprintf("mock-reg-%s-%d", t.Name(), time.Now().UnixNano())
+
+ registryURL, err := common.NewURL(testRegistryURL,
+ common.WithParamsValue(constant.RegistryKey, "mock"),
+ common.WithParamsValue(constant.RegistryIdKey, regID))
+ require.NoError(t, err)
+
+ reg, err := newServiceDiscoveryRegistry(registryURL)
+ require.NoError(t, err)
+ sdReg := reg.(*serviceDiscoveryRegistry)
+
+ unblock := make(chan struct{})
+ started := make(chan struct{}, 1)
+ sdReg.serviceDiscovery = &blockingMockServiceDiscovery{
+ mockServiceDiscovery: mockSD,
+ blockName: "appA",
+ unblock: unblock,
+ started: started,
+ }
+
+ // service A targets appA (GetInstances blocks); service B targets appB
(returns at once).
+ urlA, _ := common.NewURL("dubbo://127.0.0.1:20000/",
+ common.WithInterface(testInterface),
+ common.WithParamsValue(constant.GroupKey, "A"),
+ common.WithParamsValue(constant.SideKey, constant.SideConsumer))
+ urlB, _ := common.NewURL("dubbo://127.0.0.1:20001/",
+ common.WithInterface("org.apache.dubbo.test.OtherService"),
+ common.WithParamsValue(constant.GroupKey, "B"),
+ common.WithParamsValue(constant.SideKey, constant.SideConsumer))
+
+ doneA := make(chan struct{})
+ go func() {
+ sdReg.SubscribeURL(urlA, &mockNotifyListener{},
gxset.NewSet("appA"))
+ close(doneA)
+ }()
+
+ // wait until A is blocked inside GetInstances("appA")
+ select {
+ case <-started:
+ case <-time.After(2 * time.Second):
+ t.Fatal("GetInstances for service A was never invoked")
+ }
+
+ // B must complete promptly even though A is still blocked in
GetInstances.
+ // If the registry write lock were held across GetInstances, B would
stall here.
+ completedB := make(chan struct{})
+ go func() {
+ sdReg.SubscribeURL(urlB, &mockNotifyListener{},
gxset.NewSet("appB"))
+ close(completedB)
+ }()
+ select {
+ case <-completedB:
+ case <-time.After(2 * time.Second):
+ t.Fatal("subscribe for B was blocked by A's stalled external
call; lock must not cover GetInstances")
+ }
+
+ close(unblock)
+ <-doneA
+}
+
+// blockingMockServiceDiscovery wraps mockServiceDiscovery and blocks
+// GetInstances for blockName until unblock is closed, to assert that a stalled
+// external call does not hold the registry lock and stall other keys.
AddListener
+// is overridden so the async wiring goroutine does not touch the mock's
WaitGroup.
+type blockingMockServiceDiscovery struct {
+ *mockServiceDiscovery
+ blockName string
+ unblock chan struct{}
+ started chan struct{}
+}
+
+func (m *blockingMockServiceDiscovery) GetInstances(name string)
[]registry.ServiceInstance {
+ if name == m.blockName {
+ select {
+ case m.started <- struct{}{}:
+ default:
+ }
+ <-m.unblock
+ }
+ return m.mockServiceDiscovery.GetInstances(name)
+}
+
+func (m *blockingMockServiceDiscovery)
AddListener(registry.ServiceInstancesChangedListener) error {
+ return nil
+}
