Skip to content

Commit c745999

Browse files
jieguo-coderjieguo-coder
authored
feat(metadata): add concurrency safety for application-level metadata state (#3353) (#3367)
* feat(metadata): add internal RWMutex for MetadataInfo and fix json serialization (#3353) Add sync.RWMutex to MetadataInfo struct with json:"-" / hessian:"-" tags to skip serialization. All mutating methods (AddService, RemoveService, AddSubscribeURL, RemoveSubscribeURL) acquire the write lock, and read methods (GetExportedServiceURLs, GetSubscribedURLs, GetServices) acquire the read lock. The new GetServices method returns a snapshot copy. Signed-off-by: jieguo-coder <1193249232@qq.com> * feat(metadata): add lock protection for global registry and instances (#3353) Add sync.RWMutex to protect registryMetadataInfo in metadata.go and instances in report_instance.go. Extract getMetadataReportUnsafe helper to avoid reentrant RLock deadlock in GetMetadataReportByRegistry fallback. Fix nacos report_test to use pointer to MetadataInfo for json.Marshal. Signed-off-by: jieguo-coder <1193249232@qq.com> * feat(metadata): fix concurrency safety in listener callbacks and external calls (#3353) Add mutex locking to AddListenerAndNotify and RemoveListener to protect shared fields listeners and serviceUrls. Replace direct access to MetadataInfo.Services with safe GetServices method in OnEvent and convertV2 to prevent unprotected map reads. Signed-off-by: jieguo-coder <1193249232@qq.com> * style: format import blocks using dubbo-go imports-formatter Signed-off-by: jieguo-coder <1193249232@qq.com> * style: format metadata.go and report_instance.go to pass CI Signed-off-by: jieguo-coder <1193249232@qq.com> * fix(metadata): resolve data races pointed out in code review Signed-off-by: jieguo-coder <1193249232@qq.com> * test: fix failing tests and improve unit test coverage Signed-off-by: jieguo-coder <1193249232@qq.com> * test: fix testifylint error by avoiding require in goroutines Signed-off-by: jieguo-coder <1193249232@qq.com> * fix(metadata): deep copy ServiceInfo to prevent write-on-read data race and reduce lock granularity in report creation Signed-off-by: jieguo-coder <1193249232@qq.com> * chore: trigger CI re-run for codecov gpg issue * refactor(metadata): implement no-lock helper to fix data race and avoid deadlocks in ReplaceExportedServices Signed-off-by: jieguo-coder <1193249232@qq.com> * test: fix compilation error in instance changed listener tests by adding missing registryId argument Signed-off-by: jieguo-coder <1193249232@qq.com> * fix(metadata): add RLock around global map lookup in RemoveService and RemoveSubscribeURL to prevent concurrent map read/write Signed-off-by: jieguo-coder <1193249232@qq.com> --------- Signed-off-by: jieguo-coder <1193249232@qq.com>
1 parent 706090f commit c745999

10 files changed

Lines changed: 216 additions & 10 deletions

‎metadata/info/metadata_info.go‎

Lines changed: 63 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import (
2424
"sort"
2525
"strconv"
2626
"strings"
27+
"sync"
2728
)
2829

2930
import (
@@ -65,6 +66,7 @@ type MetadataInfo struct {
6566
Services map[string]*ServiceInfo `json:"services,omitempty" hessian:"services"`
6667
exportedServiceURLs map[string][]*common.URL `hessian:"-"` // server exported service urls
6768
subscribedServiceURLs map[string][]*common.URL `hessian:"-"` // client subscribed service urls
69+
mu sync.RWMutex `json:"-" hessian:"-"`
6870
}
6971

7072
func NewAppMetadataInfo(app string) *MetadataInfo {
@@ -97,6 +99,15 @@ func (info *MetadataInfo) JavaClassName() string {
9799

98100
// AddService add provider service info to MetadataInfo
99101
func (info *MetadataInfo) AddService(url *common.URL) {
102+
info.mu.Lock()
103+
defer info.mu.Unlock()
104+
105+
info.addServiceWithoutLock(url)
106+
}
107+
108+
// addServiceWithoutLock adds a service URL without acquiring the lock.
109+
// The caller must hold info.mu.Lock() before calling this method.
110+
func (info *MetadataInfo) addServiceWithoutLock(url *common.URL) {
100111
service := NewServiceInfoWithURL(url)
101112
info.Services[service.GetMatchKey()] = service
102113
addUrl(info.exportedServiceURLs, url)
@@ -136,6 +147,9 @@ func deleteItem(slice []*common.URL, index int) []*common.URL {
136147
}
137148

138149
func (info *MetadataInfo) RemoveService(url *common.URL) {
150+
info.mu.Lock()
151+
defer info.mu.Unlock()
152+
139153
service := NewServiceInfoWithURL(url)
140154
removeUrl(info.exportedServiceURLs, url)
141155
if replacement := info.findExportedServiceURL(service.GetMatchKey()); replacement != nil {
@@ -147,15 +161,24 @@ func (info *MetadataInfo) RemoveService(url *common.URL) {
147161

148162
// AddSubscribeURL client subscribe a service url
149163
func (info *MetadataInfo) AddSubscribeURL(url *common.URL) {
164+
info.mu.Lock()
165+
defer info.mu.Unlock()
166+
150167
addUrl(info.subscribedServiceURLs, url)
151168
}
152169

153170
// RemoveSubscribeURL client unsubscribe a service url
154171
func (info *MetadataInfo) RemoveSubscribeURL(url *common.URL) {
172+
info.mu.Lock()
173+
defer info.mu.Unlock()
174+
155175
removeUrl(info.subscribedServiceURLs, url)
156176
}
157177

158178
func (info *MetadataInfo) GetExportedServiceURLs() []*common.URL {
179+
info.mu.RLock()
180+
defer info.mu.RUnlock()
181+
159182
res := make([]*common.URL, 0)
160183
for _, urls := range info.exportedServiceURLs {
161184
res = append(res, urls...)
@@ -164,18 +187,37 @@ func (info *MetadataInfo) GetExportedServiceURLs() []*common.URL {
164187
}
165188

166189
func (info *MetadataInfo) GetSubscribedURLs() []*common.URL {
190+
info.mu.RLock()
191+
defer info.mu.RUnlock()
192+
167193
res := make([]*common.URL, 0)
168194
for _, urls := range info.subscribedServiceURLs {
169195
res = append(res, urls...)
170196
}
171197
return res
172198
}
173199

200+
// GetServices returns a deep copy of the Services map for safe iteration by external callers.
201+
// Each ServiceInfo is fully copied with lazy fields eagerly populated to prevent write-on-read races.
202+
func (info *MetadataInfo) GetServices() map[string]*ServiceInfo {
203+
info.mu.Lock()
204+
defer info.mu.Unlock()
205+
206+
cp := make(map[string]*ServiceInfo, len(info.Services))
207+
for k, v := range info.Services {
208+
cp[k] = v.DeepCopy()
209+
}
210+
return cp
211+
}
212+
174213
func (info *MetadataInfo) ReplaceExportedServices(urls []*common.URL) {
214+
info.mu.Lock()
215+
defer info.mu.Unlock()
216+
175217
info.Services = make(map[string]*ServiceInfo)
176218
info.exportedServiceURLs = make(map[string][]*common.URL)
177219
for _, serviceURL := range urls {
178-
info.AddService(serviceURL)
220+
info.addServiceWithoutLock(serviceURL)
179221
}
180222
}
181223

@@ -289,6 +331,26 @@ func (si *ServiceInfo) GetServiceKey() string {
289331
return si.ServiceKey
290332
}
291333

334+
// DeepCopy returns a fully independent copy of ServiceInfo with lazy fields eagerly populated.
335+
func (si *ServiceInfo) DeepCopy() *ServiceInfo {
336+
params := make(map[string]string, len(si.Params))
337+
for k, v := range si.Params {
338+
params[k] = v
339+
}
340+
return &ServiceInfo{
341+
Name: si.Name,
342+
Group: si.Group,
343+
Version: si.Version,
344+
Protocol: si.Protocol,
345+
Port: si.Port,
346+
Path: si.Path,
347+
Params: params,
348+
ServiceKey: si.GetServiceKey(),
349+
MatchKey: si.GetMatchKey(),
350+
URL: si.URL,
351+
}
352+
}
353+
292354
// toDescString returns a deterministic string representation of ServiceInfo
293355
// for revision calculation. Aligned with Java dubbo ServiceInfo.toDescString().
294356
//

‎metadata/info/metadata_info_test.go‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -205,6 +205,27 @@ func TestServiceInfoGetMatchKey(t *testing.T) {
205205
assert.NotEmpty(t, si.GetMatchKey())
206206
}
207207

208+
func TestMetadataInfoGetServices(t *testing.T) {
209+
metadataInfo := &MetadataInfo{
210+
Services: make(map[string]*ServiceInfo),
211+
exportedServiceURLs: make(map[string][]*common.URL),
212+
subscribedServiceURLs: make(map[string][]*common.URL),
213+
}
214+
url, _ := common.NewURL("dubbo://127.0.0.1:20000?application=foo&category=providers&check=false&dubbo=dubbo-go+v1.5.0&interface=com.foo.Bar&methods=GetPetByID%2CGetPetTypes&organization=Apache&owner=foo&revision=1.0.0&side=provider&version=1.0.0")
215+
metadataInfo.AddService(url)
216+
217+
services := metadataInfo.GetServices()
218+
require.Len(t, services, 1)
219+
assert.NotEmpty(t, services)
220+
221+
// GetServices returns a copy: modifying the original does not affect the snapshot
222+
metadataInfo.RemoveService(url)
223+
assert.Len(t, services, 1)
224+
225+
// A fresh call reflects the removal
226+
assert.Empty(t, metadataInfo.GetServices())
227+
}
228+
208229
func TestServiceInfoJavaClassName(t *testing.T) {
209230
assert.Equalf(t, "org.apache.dubbo.metadata.MetadataInfo", NewAppMetadataInfo("dubbo").JavaClassName(), "JavaClassName()")
210231
}

‎metadata/metadata.go‎

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,19 @@
1818
// Package metadata collects and exposes information of all services for service discovery purpose.
1919
package metadata
2020

21+
import (
22+
"sync"
23+
)
24+
2125
import (
2226
"dubbo.apache.org/dubbo-go/v3/common"
2327
"dubbo.apache.org/dubbo-go/v3/common/constant"
2428
"dubbo.apache.org/dubbo-go/v3/metadata/info"
2529
)
2630

2731
var (
28-
registryMetadataInfo = make(map[string]*info.MetadataInfo)
32+
registryMetadataInfo = make(map[string]*info.MetadataInfo)
33+
registryMetadataLock sync.RWMutex
2934
metadataService MetadataService = &DefaultMetadataService{metadataMap: registryMetadataInfo}
3035
)
3136

@@ -34,39 +39,55 @@ func GetMetadataService() MetadataService {
3439
}
3540

3641
func GetMetadataInfo(registryId string) *info.MetadataInfo {
42+
registryMetadataLock.RLock()
43+
defer registryMetadataLock.RUnlock()
3744
return registryMetadataInfo[registryId]
3845
}
3946

4047
func AddService(registryId string, url *common.URL) {
48+
registryMetadataLock.Lock()
4149
if _, exist := registryMetadataInfo[registryId]; !exist {
4250
registryMetadataInfo[registryId] = info.NewMetadataInfo(
4351
url.GetParam(constant.ApplicationKey, ""),
4452
url.GetParam(constant.ApplicationTagKey, ""),
4553
)
4654
}
47-
registryMetadataInfo[registryId].AddService(url)
55+
metaInfo := registryMetadataInfo[registryId]
56+
registryMetadataLock.Unlock()
57+
58+
metaInfo.AddService(url)
4859
}
4960

5061
func AddSubscribeURL(registryId string, url *common.URL) {
62+
registryMetadataLock.Lock()
5163
if _, exist := registryMetadataInfo[registryId]; !exist {
5264
registryMetadataInfo[registryId] = info.NewMetadataInfo(
5365
url.GetParam(constant.ApplicationKey, ""),
5466
url.GetParam(constant.ApplicationTagKey, ""),
5567
)
5668
}
57-
registryMetadataInfo[registryId].AddSubscribeURL(url)
69+
metaInfo := registryMetadataInfo[registryId]
70+
registryMetadataLock.Unlock()
71+
72+
metaInfo.AddSubscribeURL(url)
5873
}
5974

6075
func RemoveService(registryId string, url *common.URL) {
76+
registryMetadataLock.RLock()
6177
metadataInfo, exist := registryMetadataInfo[registryId]
78+
registryMetadataLock.RUnlock()
79+
6280
if !exist {
6381
return
6482
}
6583
metadataInfo.RemoveService(url)
6684
}
6785

6886
func RemoveSubscribeURL(registryId string, url *common.URL) {
87+
registryMetadataLock.RLock()
6988
metadataInfo, exist := registryMetadataInfo[registryId]
89+
registryMetadataLock.RUnlock()
90+
7091
if !exist {
7192
return
7293
}

‎metadata/metadata_service.go‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,8 @@ func (mts *DefaultMetadataService) GetMetadataInfo(revision string) (*info.Metad
9595
if revision == "" {
9696
return nil, nil
9797
}
98+
registryMetadataLock.RLock()
99+
defer registryMetadataLock.RUnlock()
98100
for _, metadataInfo := range mts.metadataMap {
99101
if metadataInfo.Revision == revision {
100102
return metadataInfo, nil
@@ -106,6 +108,8 @@ func (mts *DefaultMetadataService) GetMetadataInfo(revision string) (*info.Metad
106108

107109
// GetExportedServiceURLs get exported service urls
108110
func (mts *DefaultMetadataService) GetExportedServiceURLs() ([]*common.URL, error) {
111+
registryMetadataLock.RLock()
112+
defer registryMetadataLock.RUnlock()
109113
urls := make([]*common.URL, 0)
110114
for _, metadataInfo := range mts.metadataMap {
111115
urls = append(urls, metadataInfo.GetExportedServiceURLs()...)
@@ -124,6 +128,8 @@ func (mts *DefaultMetadataService) GetMetadataServiceURL() (*common.URL, error)
124128
}
125129

126130
func (mts *DefaultMetadataService) GetSubscribedURLs() ([]*common.URL, error) {
131+
registryMetadataLock.RLock()
132+
defer registryMetadataLock.RUnlock()
127133
urls := make([]*common.URL, 0)
128134
for _, metadataInfo := range mts.metadataMap {
129135
urls = append(urls, metadataInfo.GetSubscribedURLs()...)
@@ -257,7 +263,7 @@ func (mtsV2 *MetadataServiceV2) GetMetadataInfo(ctx context.Context, req *triple
257263
return &tripleapi.MetadataInfoV2{
258264
App: metadataInfo.App,
259265
Version: metadataInfo.Revision,
260-
Services: convertV2(metadataInfo.Services),
266+
Services: convertV2(metadataInfo.GetServices()),
261267
Tag: metadataInfo.Tag,
262268
}, err
263269
}

‎metadata/metadata_service_test.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ package metadata
2020
import (
2121
"context"
2222
"strconv"
23+
"sync"
2324
"testing"
2425
)
2526

@@ -391,3 +392,35 @@ func Test_serviceExporterExport(t *testing.T) {
391392
require.NoError(t, err)
392393
})
393394
}
395+
396+
func TestDefaultMetadataServiceConcurrentReadAccess(t *testing.T) {
397+
mts := &DefaultMetadataService{
398+
metadataMap: newMetadataMap(),
399+
}
400+
401+
var wg sync.WaitGroup
402+
for i := 0; i < 10; i++ {
403+
wg.Add(1)
404+
go func() {
405+
defer wg.Done()
406+
urls, err := mts.GetExportedServiceURLs()
407+
assert.NoError(t, err)
408+
assert.NotEmpty(t, urls)
409+
}()
410+
wg.Add(1)
411+
go func() {
412+
defer wg.Done()
413+
urls, err := mts.GetSubscribedURLs()
414+
assert.NoError(t, err)
415+
assert.NotEmpty(t, urls)
416+
}()
417+
wg.Add(1)
418+
go func() {
419+
defer wg.Done()
420+
info, err := mts.GetMetadataInfo("1")
421+
assert.NoError(t, err)
422+
assert.NotNil(t, info)
423+
}()
424+
}
425+
wg.Wait()
426+
}

‎metadata/metadata_test.go‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
package metadata
1919

2020
import (
21+
"sync"
2122
"testing"
2223
)
2324

@@ -179,3 +180,36 @@ func TestGetMetadataService(t *testing.T) {
179180
})
180181
}
181182
}
183+
184+
func TestAddServiceConcurrent(t *testing.T) {
185+
var wg sync.WaitGroup
186+
for i := 0; i < 20; i++ {
187+
registryId := "concurrent-reg"
188+
wg.Add(1)
189+
go func(idx int) {
190+
defer wg.Done()
191+
url := common.NewURLWithOptions(
192+
common.WithProtocol("dubbo"),
193+
common.WithParamsValue(constant.ApplicationKey, "dubbo"),
194+
common.WithParamsValue(constant.ApplicationTagKey, "v1"),
195+
)
196+
AddService(registryId, url)
197+
}(i)
198+
wg.Add(1)
199+
go func(idx int) {
200+
defer wg.Done()
201+
url := common.NewURLWithOptions(
202+
common.WithProtocol("dubbo"),
203+
common.WithParamsValue(constant.ApplicationKey, "dubbo"),
204+
common.WithParamsValue(constant.ApplicationTagKey, "v1"),
205+
)
206+
AddSubscribeURL(registryId, url)
207+
}(i)
208+
wg.Add(1)
209+
go func(idx int) {
210+
defer wg.Done()
211+
_ = GetMetadataInfo(registryId)
212+
}(i)
213+
}
214+
wg.Wait()
215+
}

‎metadata/report/nacos/report_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,7 @@ func newNacosMetadataReport(f fields) *nacosMetadataReport {
171171
}
172172

173173
func Test_nacosMetadataReport_GetAppMetadata(t *testing.T) {
174-
mi := info.MetadataInfo{
174+
mi := &info.MetadataInfo{
175175
App: "GetAppMetadata",
176176
}
177177
data, _ := json.Marshal(mi)
@@ -198,7 +198,7 @@ func Test_nacosMetadataReport_GetAppMetadata(t *testing.T) {
198198
application: "dubbo",
199199
revision: "revision",
200200
},
201-
want: &mi,
201+
want: mi,
202202
wantErr: false,
203203
},
204204
}

0 commit comments

Comments
 (0)