This is an automated email from the ASF dual-hosted git repository.
Alanxtl 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 c79f6166f fix:cache manager race #3569 (#3601)
c79f6166f is described below
commit c79f6166fda32c3047cb1444d9ccff4476f85630
Author: xiaobaicai66695 <[email protected]>
AuthorDate: Thu Aug 20 13:26:05 2026 +0800
fix:cache manager race #3569 (#3601)
* fix: guard service discovery cache manager access
* test: verify cache manager dump lifecycle
* style: modernize cache manager concurrency tests
* test: reduce cache manager test complexity
* test: verify cache manager get all snapshot
* test: cover cache manager stop dump deadlock
---
registry/servicediscovery/store/cache_manager.go | 69 +++--
.../servicediscovery/store/cache_manager_test.go | 283 +++++++++++++++++++++
2 files changed, 334 insertions(+), 18 deletions(-)
diff --git a/registry/servicediscovery/store/cache_manager.go
b/registry/servicediscovery/store/cache_manager.go
index 2655a2772..0be57997e 100644
--- a/registry/servicediscovery/store/cache_manager.go
+++ b/registry/servicediscovery/store/cache_manager.go
@@ -35,9 +35,10 @@ type CacheManager struct {
cacheFile string // The file path where the cache is stored
dumpInterval time.Duration // The duration after which the cache dump
stop chan struct{} // Channel used to stop the cache expiration
routine
+ done chan struct{} // Channel closed after the cache expiration
routine stops
cache *lru.Cache // The LRU cache implementation
lock sync.Mutex
- enableDump bool
+ stopOnce sync.Once
}
type Item struct {
@@ -45,6 +46,8 @@ type Item struct {
Value any
}
+var cacheManagerBeforeDumpCache func()
+
// NewCacheManager creates a new CacheManager instance.
// It initializes the cache manager with the provided parameters and starts a
routine for cache dumping.
func NewCacheManager(name, cacheFile string, dumpInterval time.Duration,
maxCacheSize int, enableDump bool) (*CacheManager, error) {
@@ -53,7 +56,7 @@ func NewCacheManager(name, cacheFile string, dumpInterval
time.Duration, maxCach
cacheFile: cacheFile,
dumpInterval: dumpInterval,
stop: make(chan struct{}),
- enableDump: enableDump,
+ done: make(chan struct{}),
}
cache, err := lru.New(maxCacheSize)
if err != nil {
@@ -70,6 +73,8 @@ func NewCacheManager(name, cacheFile string, dumpInterval
time.Duration, maxCach
if enableDump {
cm.runDumpTask()
+ } else {
+ close(cm.done)
}
return cm, nil
@@ -77,31 +82,56 @@ func NewCacheManager(name, cacheFile string, dumpInterval
time.Duration, maxCach
// Get retrieves the value associated with the given key from the cache.
func (cm *CacheManager) Get(key string) (any, bool) {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
return cm.cache.Get(key)
}
// Set sets the value associated with the given key in the cache.
func (cm *CacheManager) Set(key string, value any) {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
cm.cache.Add(key, value)
}
// Delete removes the value associated with the given key from the cache.
func (cm *CacheManager) Delete(key string) {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
cm.cache.Remove(key)
}
// GetAll returns all the key-value pairs in the cache.
func (cm *CacheManager) GetAll() map[string]any {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
+
+ return cm.getAllLocked()
+}
+
+func (cm *CacheManager) getAllLocked() map[string]any {
keys := cm.cache.Keys()
- result := make(map[string]any)
+ result := make(map[string]any, len(keys))
for _, k := range keys {
- result[k.(string)], _ = cm.cache.Get(k)
+ key, ok := k.(string)
+ if !ok {
+ continue
+ }
+ if value, ok := cm.cache.Get(k); ok {
+ result[key] = value
+ }
}
return result
}
+func (cm *CacheManager) len() int {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
+ return cm.cache.Len()
+}
+
// loadCache loads the cache from the cache file.
func (cm *CacheManager) loadCache() error {
cf, err := os.Open(cm.cacheFile)
@@ -121,7 +151,7 @@ func (cm *CacheManager) loadCache() error {
return err
}
// Add the loaded keys to the front of the LRU list
- cm.cache.Add(it.Key, it.Value)
+ cm.Set(it.Key, it.Value)
}
return nil
@@ -129,10 +159,6 @@ func (cm *CacheManager) loadCache() error {
// dumpCache dumps the cache to the cache file.
func (cm *CacheManager) dumpCache() error {
-
- cm.lock.Lock()
- defer cm.lock.Unlock()
-
items := cm.GetAll()
file, err := os.Create(cm.cacheFile)
@@ -158,18 +184,24 @@ func (cm *CacheManager) dumpCache() error {
func (cm *CacheManager) runDumpTask() {
go func() {
ticker := time.NewTicker(cm.dumpInterval)
+ defer func() {
+ ticker.Stop()
+ close(cm.done)
+ }()
for {
select {
case <-ticker.C:
+ if cacheManagerBeforeDumpCache != nil {
+ cacheManagerBeforeDumpCache()
+ }
// Dump the cache to the file
if err := cm.dumpCache(); err != nil {
// Handle error
logger.Warnf("[Registry][ServiceDiscovery] failed to dump cache, err=%v", err)
} else {
-
logger.Infof("[Registry][ServiceDiscovery] dumping [%s] caches, latest
entries=%d", cm.name, cm.cache.Len())
+
logger.Infof("[Registry][ServiceDiscovery] dumping [%s] caches, latest
entries=%d", cm.name, cm.len())
}
case <-cm.stop:
- ticker.Stop()
return
}
}
@@ -177,18 +209,19 @@ func (cm *CacheManager) runDumpTask() {
}
func (cm *CacheManager) StopDump() {
- cm.lock.Lock()
- defer cm.lock.Unlock()
- if cm.enableDump {
- cm.stop <- struct{}{} // Stop the cache dump routine
- cm.enableDump = false
- }
+ cm.stopOnce.Do(func() {
+ close(cm.stop)
+ })
+ <-cm.done
}
// destroy stops the cache dump routine, clears the cache and removes the
cache file.
func (cm *CacheManager) destroy() {
- cm.StopDump() // Stop the cache dump routine
+ cm.StopDump() // Stop the cache dump routine
+
+ cm.lock.Lock()
cm.cache.Purge() // Clear the cache
+ cm.lock.Unlock()
// Delete the cache file if it exists
if _, err := os.Stat(cm.cacheFile); err == nil {
diff --git a/registry/servicediscovery/store/cache_manager_test.go
b/registry/servicediscovery/store/cache_manager_test.go
index 36541d7ea..8cd68f1d3 100644
--- a/registry/servicediscovery/store/cache_manager_test.go
+++ b/registry/servicediscovery/store/cache_manager_test.go
@@ -18,6 +18,11 @@
package store
import (
+ "fmt"
+ "os"
+ "path/filepath"
+ "runtime"
+ "sync"
"testing"
"time"
)
@@ -150,3 +155,281 @@ func TestMetaInfoCacheManager(t *testing.T) {
cm2.destroy()
cm.destroy() // clear cache file
}
+
+func TestCacheManagerConcurrentAccess(t *testing.T) {
+ cacheFile := filepath.Join(t.TempDir(), "race_cache")
+ cm, err := NewCacheManager("raceTest", cacheFile, time.Millisecond, 32,
true)
+ if err != nil {
+ t.Fatalf("failed to create cache manager: %v", err)
+ }
+ defer cm.destroy()
+
+ runConcurrentCacheAccess(cm)
+ waitForCacheDump(t, cacheFile)
+ cm.StopDump()
+
+ loaded, err := NewCacheManager("raceTestReloaded", cacheFile,
time.Hour, 32, false)
+ if err != nil {
+ t.Fatalf("failed to reload cache manager: %v", err)
+ }
+ defer loaded.destroy()
+
+ assertCacheEntries(t, loaded.GetAll())
+}
+
+func runConcurrentCacheAccess(cm *CacheManager) {
+ var wg sync.WaitGroup
+ for i := range 16 {
+ wg.Add(1)
+ go func(worker int) {
+ defer wg.Done()
+ exerciseCacheManager(cm, worker)
+ }(i)
+ }
+ wg.Wait()
+}
+
+func exerciseCacheManager(cm *CacheManager, worker int) {
+ for j := range 500 {
+ key := fmt.Sprintf("key-%d", j%64)
+ cm.Set(key, fmt.Sprintf("value-%d-%d", worker, j))
+ cm.Get(key)
+ if j%3 == 0 {
+ cm.Delete(fmt.Sprintf("key-%d", (j+worker)%64))
+ }
+ if j%7 == 0 {
+ cm.GetAll()
+ }
+ }
+}
+
+func waitForCacheDump(t *testing.T, cacheFile string) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for {
+ info, statErr := os.Stat(cacheFile)
+ if statErr == nil && info.Size() > 0 {
+ break
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("cache dump was not created: %v", statErr)
+ }
+ time.Sleep(time.Millisecond)
+ }
+}
+
+func assertCacheEntries(t *testing.T, items map[string]any) {
+ t.Helper()
+ if len(items) == 0 {
+ t.Fatal("reloaded cache dump contained no entries")
+ }
+ for key, value := range items {
+ if value == nil {
+ t.Fatalf("reloaded cache entry %q contained a nil
value", key)
+ }
+ if _, ok := value.(string); !ok {
+ t.Fatalf("reloaded cache entry %q had unexpected type
%T", key, value)
+ }
+ }
+}
+
+func TestCacheManagerGetAllReturnsAtomicSnapshotDuringReplacement(t
*testing.T) {
+ previousGOMAXPROCS := runtime.GOMAXPROCS(4)
+ defer runtime.GOMAXPROCS(previousGOMAXPROCS)
+
+ cm, err := NewCacheManager("snapshotTest", filepath.Join(t.TempDir(),
"snapshot_cache"), time.Hour, 1, false)
+ if err != nil {
+ t.Fatalf("failed to create cache manager: %v", err)
+ }
+ defer cm.destroy()
+
+ cm.Set("a", "value-a")
+
+ stop, errCh, waitReaders := startSnapshotReaders(cm, 8)
+ replaceCapacityOneEntries(cm, 100000)
+ close(stop)
+ waitReaders()
+
+ if msg := snapshotError(errCh); msg != "" {
+ t.Fatal(msg)
+ }
+}
+
+func startSnapshotReaders(cm *CacheManager, readers int) (chan struct{}, chan
string, func()) {
+ start := make(chan struct{})
+ stop := make(chan struct{})
+ errCh := make(chan string, 1)
+
+ var wg sync.WaitGroup
+ for range readers {
+ wg.Go(func() {
+ readSnapshotsUntilStopped(cm, start, stop, errCh)
+ })
+ }
+ close(start)
+
+ return stop, errCh, wg.Wait
+}
+
+func readSnapshotsUntilStopped(cm *CacheManager, start, stop <-chan struct{},
errCh chan<- string) {
+ <-start
+ for !snapshotStopped(stop) {
+ if msg := validateCapacityOneSnapshot(cm.GetAll()); msg != "" {
+ reportSnapshotError(errCh, msg)
+ return
+ }
+ }
+}
+
+func snapshotStopped(stop <-chan struct{}) bool {
+ select {
+ case <-stop:
+ return true
+ default:
+ return false
+ }
+}
+
+func replaceCapacityOneEntries(cm *CacheManager, replacements int) {
+ for i := range replacements {
+ if i%2 == 0 {
+ cm.Set("b", "value-b")
+ } else {
+ cm.Set("a", "value-a")
+ }
+ }
+}
+
+func validateCapacityOneSnapshot(items map[string]any) string {
+ if len(items) != 1 {
+ return fmt.Sprintf("GetAll returned %d entries during
capacity-1 replacement: %#v", len(items), items)
+ }
+
+ if value, ok := items["a"]; ok {
+ return validateSnapshotEntry("a", value, "value-a")
+ }
+ if value, ok := items["b"]; ok {
+ return validateSnapshotEntry("b", value, "value-b")
+ }
+ return fmt.Sprintf("GetAll returned unexpected snapshot: %#v", items)
+}
+
+func validateSnapshotEntry(key string, value, expected any) string {
+ if value == expected {
+ return ""
+ }
+ return fmt.Sprintf("GetAll returned unexpected entry %q=%#v", key,
value)
+}
+
+func snapshotError(errCh <-chan string) string {
+ select {
+ case msg := <-errCh:
+ return msg
+ default:
+ return ""
+ }
+}
+
+func reportSnapshotError(errCh chan<- string, msg string) {
+ select {
+ case errCh <- msg:
+ default:
+ }
+}
+
+func TestCacheManagerConcurrentStopDump(t *testing.T) {
+ tests := []struct {
+ name string
+ enableDump bool
+ }{
+ {name: "enabled", enableDump: true},
+ {name: "disabled", enableDump: false},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ cm, err := NewCacheManager("stopTest",
filepath.Join(t.TempDir(), "stop_cache"), time.Hour, 8, tt.enableDump)
+ if err != nil {
+ t.Fatalf("failed to create cache manager: %v",
err)
+ }
+ defer cm.destroy()
+
+ const callers = 64
+ start := make(chan struct{})
+ var wg sync.WaitGroup
+ for range callers {
+ wg.Go(func() {
+ <-start
+ cm.StopDump()
+ })
+ }
+ close(start)
+
+ done := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("concurrent StopDump calls did not
complete")
+ }
+
+ cm.StopDump()
+ })
+ }
+}
+
+func TestCacheManagerStopDumpCompletesAfterDumpTickSelected(t *testing.T) {
+ dumpSelected := make(chan struct{})
+ releaseDump := make(chan struct{})
+ var selectedOnce sync.Once
+ var releaseOnce sync.Once
+ release := func() {
+ releaseOnce.Do(func() {
+ close(releaseDump)
+ })
+ }
+
+ cacheManagerBeforeDumpCache = func() {
+ selectedOnce.Do(func() {
+ close(dumpSelected)
+ })
+ <-releaseDump
+ }
+ t.Cleanup(func() {
+ release()
+ cacheManagerBeforeDumpCache = nil
+ })
+
+ cm, err := NewCacheManager("stopTickTest", filepath.Join(t.TempDir(),
"stop_tick_cache"), 100*time.Millisecond, 8, true)
+ if err != nil {
+ t.Fatalf("failed to create cache manager: %v", err)
+ }
+ cm.Set("key", "value")
+
+ select {
+ case <-dumpSelected:
+ case <-time.After(2 * time.Second):
+ t.Fatal("dump task did not select ticker")
+ }
+
+ done := make(chan struct{})
+ cm.lock.Lock()
+ go func() {
+ cm.StopDump()
+ close(done)
+ }()
+ time.Sleep(20 * time.Millisecond)
+
+ release()
+ cm.lock.Unlock()
+
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("StopDump did not complete after selected dump was
released")
+ }
+}
