AlexStocks commented on code in PR #3601:
URL: https://github.com/apache/dubbo-go/pull/3601#discussion_r3791023162
##########
registry/servicediscovery/store/cache_manager_test.go:
##########
@@ -150,3 +155,229 @@ 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)
Review Comment:
[P1] StopDump 测试没有覆盖本 PR 要修的锁互等窗口
这里把 `dumpInterval` 固定为 `time.Hour`，所以调用 `StopDump` 前 dump goroutine 不会进入
`dumpCache`。我在当前 Head 的隔离副本中恢复旧实现的 `enableDump` 状态，并恢复“持有 `cm.lock` 时向无缓冲
`stop` 发送”的逻辑后，这个测试在 `go test -race -count=20` 下仍全部通过。原死锁需要 dump 已选中 ticker、正等待
`cm.lock`，同时 `StopDump` 持有该锁等待 goroutine 接收 stop；当前用例只证明 64
个调用者的幂等性，不能检出该核心回退。请增加确定性同步点，让 dump 在选中 ticker 后暂停并等待锁，再调用 `StopDump`
并断言能结束；恢复旧持锁发送实现时该测试必须失败。
--
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]
