AlexStocks commented on issue #3246:
URL: https://github.com/apache/dubbo-go/issues/3246#issuecomment-4059646934
## Concrete Fix Proposals (with code)
Below are precise, copy-paste-ready fixes for each leak point. Ordered by
priority.
---
### Fix 1 (P0): `registryProtocol.Destroy()` — clean `overrideListeners` and
`serviceConfigurationListeners`
**Current code** (`registry/protocol/protocol.go:457-496`):
```go
func (proto *registryProtocol) Destroy() {
proto.bounds.Range(func(key, value any) bool {
exporter := value.(*exporterChangeableWrapper)
// ... UnRegister + async UnExport ...
return true
})
proto.bounds = &sync.Map{}
proto.registries = &sync.Map{}
// overrideListeners and serviceConfigurationListeners are NEVER cleaned
}
```
**Fix** — add cleanup at the end of `Destroy()`:
```go
func (proto *registryProtocol) Destroy() {
// ... existing bounds cleanup ...
// Clean override listeners to release originInvoker references
proto.overrideListeners.Range(func(key, value any) bool {
proto.overrideListeners.Delete(key)
return true
})
// Clean service configuration listeners to release providerUrl
references
proto.serviceConfigurationListeners.Range(func(key, value any) bool {
proto.serviceConfigurationListeners.Delete(key)
return true
})
proto.bounds = &sync.Map{}
proto.registries = &sync.Map{}
}
```
Also fix `reExport()` — delete old listener before storing new one:
```go
// registry/protocol/protocol.go, in reExport() around L380-400
// Before storing new listener, delete old one:
func (proto *registryProtocol) reExport(invoker base.Invoker, newUrl
*common.URL) {
key := getCacheKey(invoker)
// Delete old override listener if exists
if oldListener, loaded :=
proto.overrideListeners.LoadAndDelete(getProviderUrl(invoker)); loaded {
_ = oldListener // allow GC
}
// ... rest of reExport ...
}
```
---
### Fix 2 (P1): Add `DeleteAttribute` and `ClearAttributes` to URL
**Current code** (`common/url.go:583-597`):
```go
func (c *URL) SetAttribute(key string, value any) {
c.attributesLock.Lock()
defer c.attributesLock.Unlock()
if c.attributes == nil {
c.attributes = make(map[string]any)
}
c.attributes[key] = value
}
func (c *URL) GetAttribute(key string) (any, bool) {
c.attributesLock.RLock()
defer c.attributesLock.RUnlock()
v, ok := c.attributes[key]
return v, ok
}
```
**Fix** — add two new methods right after `GetAttribute`:
```go
// DeleteAttribute removes the attribute with the given key.
// Use this in Destroy() paths to release references held by URL.
func (c *URL) DeleteAttribute(key string) {
c.attributesLock.Lock()
defer c.attributesLock.Unlock()
delete(c.attributes, key)
}
// ClearAttributes removes all attributes.
// Use this when the URL is being destroyed and all references should be
released.
func (c *URL) ClearAttributes() {
c.attributesLock.Lock()
defer c.attributesLock.Unlock()
c.attributes = nil
}
```
**Where to call them** — in every `Destroy()` that holds a URL:
```go
// protocol/base/protocol.go, in BaseProtocol.Destroy() or each protocol's
Destroy:
func (ivk *BaseInvoker) Destroy() {
// ... existing cleanup ...
if ivk.url != nil {
ivk.url.ClearAttributes()
}
}
// registry/directory/directory.go, in RegistryDirectory.Destroy():
func (dir *RegistryDirectory) Destroy() {
dir.DoDestroy(func() {
// ... existing cleanup ...
invokers := dir.cacheInvokers
dir.cacheInvokers = []protocolbase.Invoker{}
for _, ivk := range invokers {
ivk.GetURL().ClearAttributes() // release attribute references
before Destroy
ivk.Destroy()
}
})
}
```
---
### Fix 3 (P1): `RegistryDirectory.Destroy()` — clean `cacheInvokersMap`
**Current code** (`registry/directory/directory.go:549-573`):
```go
func (dir *RegistryDirectory) Destroy() {
dir.DoDestroy(func() {
// ...
invokers := dir.cacheInvokers
dir.cacheInvokers = []protocolbase.Invoker{}
for _, ivk := range invokers {
ivk.Destroy()
}
// cacheInvokersMap (sync.Map) is NEVER cleaned!
})
}
```
**Fix** — add cleanup right after the invoker loop:
```go
func (dir *RegistryDirectory) Destroy() {
dir.DoDestroy(func() {
// ...
invokers := dir.cacheInvokers
dir.cacheInvokers = []protocolbase.Invoker{}
for _, ivk := range invokers {
ivk.Destroy()
}
// Clean cacheInvokersMap to release all invoker references
dir.cacheInvokersMap.Range(func(key, _ any) bool {
dir.cacheInvokersMap.Delete(key)
return true
})
// Also nil out configurators to release Configurator -> URL
references
dir.configurators = nil
})
}
```
---
### Fix 4 (P1): `invokerBlackList` key mismatch — unify key generation
**Root cause**: `SetInvokerUnhealthyStatus` uses `invoker.GetURL().Key()`,
but `RemoveUrlKeyUnhealthyStatus` receives keys from `GetCacheInvokerMapKey()`
which has a different format.
**Current code** (`protocol/base/rpc_status.go:212-243`):
```go
func SetInvokerUnhealthyStatus(invoker Invoker) {
invokerBlackList.Store(invoker.GetURL().Key(), invoker) // Key() format
}
func RemoveUrlKeyUnhealthyStatus(key string) {
invokerBlackList.Delete(key) // key comes from GetCacheInvokerMapKey()
format
}
```
**Caller** (`registry/directory/directory.go:457`):
```go
func (dir *RegistryDirectory) uncacheInvokerWithKey(key string) {
// key = ServiceEvent.Key() = URL.GetCacheInvokerMapKey()
protocolbase.RemoveUrlKeyUnhealthyStatus(key)
}
```
**Two-part fix:**
Option A — Store only key string, not invoker object; unify key format:
```go
// rpc_status.go
// Change blacklist to store URL.Key() -> struct{} (not full invoker)
// and add a separate method that accepts URL directly
func SetInvokerUnhealthyStatus(invoker Invoker) {
key := invoker.GetURL().Key()
invokerBlackList.Store(key, struct{}{}) // don't store full invoker
blackListCacheDirty.Store(true)
logger.Infof("%.10s add to black list", key)
}
func RemoveInvokerUnhealthyStatus(invoker Invoker) {
key := invoker.GetURL().Key()
invokerBlackList.Delete(key)
blackListCacheDirty.Store(true)
logger.Infof("%.10s removed from black list", key)
}
// Keep RemoveUrlKeyUnhealthyStatus but convert key format
func RemoveUrlKeyUnhealthyStatus(cacheKey string) {
// Also try to find and delete by iterating (since key formats differ)
invokerBlackList.Range(func(k, _ any) bool {
if strings.Contains(k.(string), extractServiceKey(cacheKey)) {
invokerBlackList.Delete(k)
blackListCacheDirty.Store(true)
return false // found, stop
}
return true
})
}
```
Option B (simpler, recommended) — make `uncacheInvokerWithKey` pass the
correct key:
```go
// registry/directory/directory.go
func (dir *RegistryDirectory) uncacheInvokerWithKey(key string) {
logger.Debugf("%.10s will be uncached", key)
if cacheInvoker, ok := dir.cacheInvokersMap.Load(key); ok {
dir.cacheInvokersMap.Delete(key)
invoker := cacheInvoker.(protocolbase.Invoker)
// Use the invoker's own URL.Key() to remove from blacklist
protocolbase.RemoveUrlKeyUnhealthyStatus(invoker.GetURL().Key())
invoker.Destroy()
}
}
```
---
### Fix 5 (P1): `GetParams()` — return copy instead of internal reference
**Current code** (`common/url.go:657-659`):
```go
func (c *URL) GetParams() url.Values {
return c.params
}
```
**Fix**:
```go
// GetParams returns a deep copy of the URL parameters.
// Callers can safely modify the returned map without affecting the URL.
func (c *URL) GetParams() url.Values {
c.paramsLock.RLock()
defer c.paramsLock.RUnlock()
if c.params == nil {
return url.Values{}
}
copy := make(url.Values, len(c.params))
for k, vs := range c.params {
vsCopy := make([]string, len(vs))
for i, v := range vs {
vsCopy[i] = v
}
copy[k] = vsCopy
}
return copy
}
```
**Impact**: Callers that relied on mutating the returned map to affect the
URL will break. Audit these callers:
```go
// registry/base_configuration_listener.go:111 — currently mutates returned
map
override := url.GetParams()
delete(override, constant.AnyhostKey) // after fix: only deletes from copy,
URL unchanged
// FIX: if intent is to modify URL, use url.DelParam(constant.AnyhostKey)
instead
// config_center/configurator/override.go:88 — uses
SetParams(configUrl.GetParams())
// After fix: this still works correctly since SetParams deep-copies its
input
```
---
### Fix 6 (P1): `configurators` — replace instead of append
**Current code** (`registry/directory/directory.go:375-386`):
```go
func (dir *RegistryDirectory) convertUrl(res *registry.ServiceEvent)
*common.URL {
ret := res.Service
if ret.Protocol == constant.OverrideProtocol ||
ret.GetParam(constant.CategoryKey, constant.DefaultCategory) ==
constant.ConfiguratorsCategory {
dir.configurators = append(dir.configurators,
extension.GetDefaultConfigurator(ret))
ret = nil
}
// ...
}
```
**Fix** — follow the same pattern as `BaseConfigurationListener.Process()`
(L88), which replaces the entire slice:
```go
func (dir *RegistryDirectory) convertUrl(res *registry.ServiceEvent)
*common.URL {
ret := res.Service
if ret.Protocol == constant.OverrideProtocol ||
ret.GetParam(constant.CategoryKey, constant.DefaultCategory) ==
constant.ConfiguratorsCategory {
// Replace configurator for this service key, not append
newConfigurator := extension.GetDefaultConfigurator(ret)
dir.replaceConfigurator(newConfigurator)
ret = nil
}
// ...
}
// replaceConfigurator replaces existing configurator for the same URL key,
// or appends if no matching configurator exists.
func (dir *RegistryDirectory) replaceConfigurator(newConf
config_center.Configurator) {
newKey := newConf.GetUrl().Key()
for i, existing := range dir.configurators {
if existing.GetUrl().Key() == newKey {
dir.configurators[i] = newConf
return
}
}
dir.configurators = append(dir.configurators, newConf)
}
```
---
### Fix 7 (P2): `MergeURL` — add lock when reading `anotherUrl.attributes`
**Current code** (`common/url.go:879-891`):
```go
// merge attributes
if mergedURL.attributes == nil {
mergedURL.attributes = make(map[string]any, len(anotherUrl.attributes))
}
for attrK, attrV := range anotherUrl.attributes { // NO LOCK on anotherUrl!
if _, ok := mergedURL.GetAttribute(attrK); !ok {
mergedURL.attributes[attrK] = attrV
}
}
```
**Fix**:
```go
// merge attributes
anotherUrl.attributesLock.RLock()
anotherAttrs := make(map[string]any, len(anotherUrl.attributes))
for k, v := range anotherUrl.attributes {
anotherAttrs[k] = v
}
anotherUrl.attributesLock.RUnlock()
if len(anotherAttrs) > 0 {
if mergedURL.attributes == nil {
mergedURL.attributes = make(map[string]any, len(anotherAttrs))
}
for attrK, attrV := range anotherAttrs {
if _, ok := mergedURL.GetAttribute(attrK); !ok {
mergedURL.attributes[attrK] = attrV
}
}
}
```
---
### Fix 8 (P2): `GetCacheInvokerMapKey` — avoid allocating temporary URL
**Current code** (`common/url.go:418-428`):
```go
func (c *URL) GetCacheInvokerMapKey() string {
urlNew, _ := NewURL(c.PrimitiveURL) // allocates entire URL just to
read one param
buildString := urlNew.Protocol + "://" +
urlNew.GetParam(constant.HostnameKey, urlNew.Location) +
"/" + c.GetParam(constant.InterfaceKey, c.Path) +
"?" + constant.TimeStamp + "=" +
c.GetParam(constant.RemoteTimestampKey, "") +
"&" + constant.MeshClusterIDKey + "=" +
c.GetParam(constant.MeshClusterIDKey, "")
return buildString
}
```
**Fix** — read HostnameKey directly from params, no need for NewURL:
```go
func (c *URL) GetCacheInvokerMapKey() string {
hostname := c.GetParam(constant.HostnameKey, c.Location)
return c.Protocol + "://" + hostname +
"/" + c.GetParam(constant.InterfaceKey, c.Path) +
"?" + constant.TimeStamp + "=" +
c.GetParam(constant.RemoteTimestampKey, "") +
"&" + constant.MeshClusterIDKey + "=" +
c.GetParam(constant.MeshClusterIDKey, "")
}
```
If `PrimitiveURL` contains a different hostname than the parsed `Location`,
it means the URL parsing was inconsistent. In that case,
`c.GetParam(constant.HostnameKey, c.Location)` is the safer fallback anyway.
---
### Fix 9 (P2): `isMatched` — use `SetParam` instead of `AddParam`
**Current code** (`registry/protocol/protocol.go:400`):
```go
func isMatched(party *common.URL, url *common.URL) bool {
// ...
providerUrl.AddParam(constant.CategoryKey,
constant.ConfiguratorsCategory) // appends, accumulates
// ...
}
```
**Fix**:
```go
providerUrl.SetParam(constant.CategoryKey, constant.ConfiguratorsCategory)
// overwrites, no accumulation
```
---
### Fix 10 (P2): `registryProtocol.Destroy()` — add else branch for async
cleanup
**Current code** (`registry/protocol/protocol.go:468-491`):
```go
go func() {
if configShutdown := config.GetShutDown(); configShutdown != nil {
<-time.After(configShutdown.GetStepTimeout() +
configShutdown.GetConsumerUpdateWaitTime())
exporter.UnExport()
proto.bounds.Delete(key)
return
}
if shutdownConfRaw, ok :=
exporter.registerUrl.GetAttribute(constant.ShutdownConfigPrefix); ok {
if shutdownConf, ok := shutdownConfRaw.(*global.ShutdownConfig); ok {
// ... same pattern ...
return
}
}
// BUG: if both conditions fail, nothing happens — exporter leaks
}()
```
**Fix** — add else branch:
```go
go func() {
if configShutdown := config.GetShutDown(); configShutdown != nil {
<-time.After(configShutdown.GetStepTimeout() +
configShutdown.GetConsumerUpdateWaitTime())
exporter.UnExport()
proto.bounds.Delete(key)
return
}
if shutdownConfRaw, ok :=
exporter.registerUrl.GetAttribute(constant.ShutdownConfigPrefix); ok {
if shutdownConf, ok := shutdownConfRaw.(*global.ShutdownConfig); ok {
// ... same pattern ...
return
}
}
// Fallback: no shutdown config found, clean up immediately
logger.Warnf("No shutdown config found for exporter %s, cleaning up
immediately", key)
exporter.UnExport()
proto.bounds.Delete(key)
}()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
