From d9e7a851095b2516d7978ccce32287bb8ab847fe Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Tue, 18 Aug 2026 09:20:14 +0330 Subject: [PATCH 1/8] fix(CubeMaster/pkg/base/localcache/localcache_test.go): fixing the data races in localcache Signed-off-by: Dwin Gharibi --- CubeMaster/pkg/base/localcache/localcache.go | 64 ++++++++++--------- .../pkg/base/localcache/localcache_test.go | 8 +-- 2 files changed, 39 insertions(+), 33 deletions(-) diff --git a/CubeMaster/pkg/base/localcache/localcache.go b/CubeMaster/pkg/base/localcache/localcache.go index 6ef3123d4..2fedfd3af 100644 --- a/CubeMaster/pkg/base/localcache/localcache.go +++ b/CubeMaster/pkg/base/localcache/localcache.go @@ -65,6 +65,8 @@ type LocalCache struct { localCacheConfig *LocalCacheConfig sharedCalls util.SharedCalls consecutiveFailNum int64 + expiredUse atomic.Bool + destroyOnce sync.Once } func NewCache(name string, loader LoaderFunc, localCacheConfig *LocalCacheConfig) *LocalCache { @@ -77,6 +79,7 @@ func NewCache(name string, loader LoaderFunc, localCacheConfig *LocalCacheConfig localCache.loadFile(localCacheConfig.LoadFileName) } localCache.localCacheConfig = localCache.SetupConfig(localCacheConfig) + localCache.expiredUse.Store(localCache.localCacheConfig.ExpiredUse) localCache.chShrinkCache = make(chan bool, 1) localCache.chCacheExit = make(chan bool) @@ -99,36 +102,39 @@ func NewCache(name string, loader LoaderFunc, localCacheConfig *LocalCacheConfig } func (localCache *LocalCache) Destroy() { - if localCache != nil { + if localCache == nil { + return + } + localCache.destroyOnce.Do(func() { CubeLog.Infof("LruCache(%s) Destroy", localCache.name) - if localCache.chCacheExit != nil { - if localCache.localCacheConfig.OpenCacheFile { - localCache.saveFile(localCache.localCacheConfig.LoadFileName) - } - localCache.cache.Flush() - close(localCache.chCacheExit) - localCache.waitGroup.Wait() - localCache.chCacheExit = nil + if localCache.chCacheExit == nil { + return } - } + if localCache.localCacheConfig.OpenCacheFile { + localCache.saveFile(localCache.localCacheConfig.LoadFileName) + } + localCache.cache.Flush() + close(localCache.chCacheExit) + localCache.waitGroup.Wait() + }) } func (localCache *LocalCache) Get(ctx context.Context, key string) (interface{}, bool, error) { item, found := localCache.cache.Get(key) if found { element := item.(*list.Element) + + localCache.Lock() itm := element.Value.(*util.CacheValue) + localCache.valueList.MoveToBack(element) + localCache.Unlock() if time.Now().Add(-itm.Expired).After(time.Unix(itm.LastAccess, 0)) { - if !localCache.localCacheConfig.ExpiredUse { + if !localCache.expiredUse.Load() { r, f, err := localCache.loadAndRefresh(ctx, key) if err != nil && localCache.localCacheConfig.DemotionExpiredUse { - - localCache.Lock() - localCache.valueList.MoveToBack(element) - localCache.Unlock() return itm.Value, true, nil } @@ -142,9 +148,6 @@ func (localCache *LocalCache) Get(ctx context.Context, key string) (interface{}, localCache.asyncRefresh(ctx, key) } - localCache.Lock() - localCache.valueList.MoveToBack(element) - localCache.Unlock() return itm.Value, true, nil } @@ -159,14 +162,17 @@ func (localCache *LocalCache) put(key string, val interface{}, expired time.Dura if item, found := localCache.cache.Get(key); found { element := item.(*list.Element) localCache.Lock() + prev := element.Value.(*util.CacheValue) + next := &util.CacheValue{ + Key: prev.Key, + Value: val, + LastAccess: time.Now().Unix(), + Expired: expired} + element.Value = next localCache.valueList.MoveToBack(element) localCache.Unlock() - itm := element.Value.(*util.CacheValue) - atomic.AddInt64(&localCache.curCacheSize, -itm.Size()) - itm.Value = val - itm.Expired = expired - itm.LastAccess = time.Now().Unix() - atomic.AddInt64(&localCache.curCacheSize, itm.Size()) + atomic.AddInt64(&localCache.curCacheSize, -prev.Size()) + atomic.AddInt64(&localCache.curCacheSize, next.Size()) } else { itm := &util.CacheValue{ Key: key, @@ -180,7 +186,7 @@ func (localCache *LocalCache) put(key string, val interface{}, expired time.Dura localCache.cache.Set(key, element, -1) } - if localCache.curCacheSize >= localCache.localCacheConfig.HighCacheSize { + if atomic.LoadInt64(&localCache.curCacheSize) >= localCache.localCacheConfig.HighCacheSize { localCache.chShrinkCache <- true } } @@ -198,7 +204,7 @@ func (localCache *LocalCache) loadAndRefresh(ctx context.Context, key string) (i CubeLog.Errorf("Cache LoadAndRefresh Error:%s, %v, %s", key, found, err) return nil, err } else { - localCache.consecutiveFailNum = 0 + atomic.StoreInt64(&localCache.consecutiveFailNum, 0) } if !found { @@ -305,12 +311,12 @@ func (localCache *LocalCache) errStrategy() { int64(localCache.localCacheConfig.MaxConsecutiveFailNum) { if localCache.localCacheConfig.DemotionExpiredUse { - localCache.localCacheConfig.ExpiredUse = true + localCache.expiredUse.Store(true) } } else { - if localCache.localCacheConfig.DemotionExpiredUse && localCache.localCacheConfig.ExpiredUse { - localCache.localCacheConfig.ExpiredUse = false + if localCache.localCacheConfig.DemotionExpiredUse && localCache.expiredUse.Load() { + localCache.expiredUse.Store(false) } } } diff --git a/CubeMaster/pkg/base/localcache/localcache_test.go b/CubeMaster/pkg/base/localcache/localcache_test.go index b0d2c6de7..0cdd9b8fe 100644 --- a/CubeMaster/pkg/base/localcache/localcache_test.go +++ b/CubeMaster/pkg/base/localcache/localcache_test.go @@ -45,13 +45,13 @@ func TestLocalCache(t *testing.T) { for i := 0; i < 2700; i++ { wg.Add(1) - ctx = context.WithValue(ctx, ctxKey, i) - go func() { + iterCtx := context.WithValue(ctx, ctxKey, i) + go func(c context.Context) { defer wg.Done() start := time.Now() - localCache.Get(ctx, RandString(8)) + localCache.Get(c, RandString(8)) fmt.Printf("=====%d====\n", time.Since(start).Milliseconds()) - }() + }(iterCtx) } wg.Wait() From 721576203a05ffe55810512b5ffa8379e520b7f8 Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Tue, 18 Aug 2026 09:20:36 +0330 Subject: [PATCH 2/8] test(CubeMaster/pkg/base/localcache/localcache_race_test.go): adding some proper tests for fixing the data race Signed-off-by: Dwin Gharibi --- .../base/localcache/localcache_race_test.go | 90 +++++++++++++++++++ 1 file changed, 90 insertions(+) create mode 100644 CubeMaster/pkg/base/localcache/localcache_race_test.go diff --git a/CubeMaster/pkg/base/localcache/localcache_race_test.go b/CubeMaster/pkg/base/localcache/localcache_race_test.go new file mode 100644 index 000000000..bec76386a --- /dev/null +++ b/CubeMaster/pkg/base/localcache/localcache_race_test.go @@ -0,0 +1,90 @@ +// Copyright (c) 2024 Tencent Inc. +// SPDX-License-Identifier: Apache-2.0 +// + +package localcache + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestConcurrentGetAndRefreshOnSameKey(t *testing.T) { + var loads int64 + localCache := NewCache("race-probe", + func(ctx context.Context, key string) (interface{}, bool, error) { + atomic.AddInt64(&loads, 1) + return RandString(16), true, nil + }, + &LocalCacheConfig{ + LowCacheSize: 1000000, + HighCacheSize: 2000000, + Expired: time.Millisecond, + AsyncRefreshBefore: time.Millisecond, + MaxAsyncRefreshNum: 100, + ExpiredUse: true, + }) + defer localCache.Destroy() + + ctx := context.Background() + const readers = 32 + const iterations = 300 + + var wg sync.WaitGroup + wg.Add(readers) + for i := 0; i < readers; i++ { + go func() { + defer wg.Done() + for j := 0; j < iterations; j++ { + v, found, err := localCache.Get(ctx, "hot-key") + if err != nil { + t.Errorf("Get: %v", err) + return + } + if found { + if _, ok := v.(string); !ok { + t.Errorf("Get returned %T, want string; a torn interface read", v) + return + } + } + } + }() + } + wg.Wait() + + if atomic.LoadInt64(&loads) == 0 { + t.Fatal("loader never ran; the test did not exercise the refresh path") + } +} + +func TestDestroyIsIdempotent(t *testing.T) { + localCache := NewCache("destroy-probe", + func(ctx context.Context, key string) (interface{}, bool, error) { + return "v", true, nil + }, + &LocalCacheConfig{LowCacheSize: 1000, HighCacheSize: 2000, Expired: time.Minute}) + + localCache.Destroy() + localCache.Destroy() +} + +func TestConcurrentDestroyDoesNotRaceWithBackgroundLoops(t *testing.T) { + localCache := NewCache("destroy-race-probe", + func(ctx context.Context, key string) (interface{}, bool, error) { + return "v", true, nil + }, + &LocalCacheConfig{LowCacheSize: 1000, HighCacheSize: 2000, Expired: time.Minute}) + + var wg sync.WaitGroup + wg.Add(4) + for i := 0; i < 4; i++ { + go func() { + defer wg.Done() + localCache.Destroy() + }() + } + wg.Wait() +} From 076549930542e10b6ca0f625986148b34c9e133e Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Tue, 18 Aug 2026 10:35:19 +0330 Subject: [PATCH 3/8] fix(CubeMaster/pkg/base/localcache/localcache.go): lock the saveFile snapshot and read curCacheSize atomically Signed-off-by: Dwin Gharibi --- CubeMaster/pkg/base/localcache/localcache.go | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/CubeMaster/pkg/base/localcache/localcache.go b/CubeMaster/pkg/base/localcache/localcache.go index 2fedfd3af..32f48a518 100644 --- a/CubeMaster/pkg/base/localcache/localcache.go +++ b/CubeMaster/pkg/base/localcache/localcache.go @@ -246,10 +246,10 @@ func (localCache *LocalCache) shrinkCache() { select { case <-localCache.chShrinkCache: curTime := time.Now() - curCacheSize := localCache.curCacheSize + curCacheSize := atomic.LoadInt64(&localCache.curCacheSize) var shrinkNum, shrinkSize, size int64 for { - if localCache.curCacheSize > localCache.localCacheConfig.LowCacheSize { + if atomic.LoadInt64(&localCache.curCacheSize) > localCache.localCacheConfig.LowCacheSize { localCache.Lock() element := localCache.valueList.Front() if element == nil { @@ -330,10 +330,13 @@ func (localCache *LocalCache) saveFile(file string) { var itm *util.CacheValue switch value := item.Object.(type) { case *list.Element: + localCache.Lock() + stored := value.Value + localCache.Unlock() var ok bool - itm, ok = value.Value.(*util.CacheValue) + itm, ok = stored.(*util.CacheValue) if !ok { - CubeLog.Errorf("Cache(%s) cannot persist key %s: list element contains %T", localCache.name, key, value.Value) + CubeLog.Errorf("Cache(%s) cannot persist key %s: list element contains %T", localCache.name, key, stored) continue } case *util.CacheValue: From 389d18b1f6a40761f53478d4977a96b6ce4836ff Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Tue, 18 Aug 2026 10:35:19 +0330 Subject: [PATCH 4/8] test(CubeMaster/pkg/base/localcache/localcache_race_test.go): overlap Destroy with an in-flight refresh Signed-off-by: Dwin Gharibi --- .../base/localcache/localcache_race_test.go | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/CubeMaster/pkg/base/localcache/localcache_race_test.go b/CubeMaster/pkg/base/localcache/localcache_race_test.go index bec76386a..45ef9d329 100644 --- a/CubeMaster/pkg/base/localcache/localcache_race_test.go +++ b/CubeMaster/pkg/base/localcache/localcache_race_test.go @@ -6,6 +6,8 @@ package localcache import ( "context" + "path/filepath" + "strconv" "sync" "sync/atomic" "testing" @@ -88,3 +90,48 @@ func TestConcurrentDestroyDoesNotRaceWithBackgroundLoops(t *testing.T) { } wg.Wait() } + +func TestDestroySnapshotDoesNotRaceWithInFlightRefresh(t *testing.T) { + localCache := NewCache("destroy-snapshot-probe", + func(ctx context.Context, key string) (interface{}, bool, error) { + return RandString(16), true, nil + }, + &LocalCacheConfig{ + LowCacheSize: 1000000, + HighCacheSize: 2000000, + Expired: time.Millisecond, + AsyncRefreshBefore: time.Millisecond, + MaxAsyncRefreshNum: 100, + ExpiredUse: true, + OpenCacheFile: true, + LoadFileName: filepath.Join(t.TempDir(), "cache.gob"), + }) + + ctx := context.Background() + const keys = 64 + for i := 0; i < keys; i++ { + if _, _, err := localCache.Get(ctx, "k"+strconv.Itoa(i)); err != nil { + t.Fatalf("seed Get: %v", err) + } + } + + var stop atomic.Bool + var wg sync.WaitGroup + wg.Add(8) + for r := 0; r < 8; r++ { + go func(r int) { + defer wg.Done() + for i := 0; !stop.Load(); i++ { + if _, _, err := localCache.Get(ctx, "k"+strconv.Itoa((i+r)%keys)); err != nil { + t.Errorf("Get: %v", err) + return + } + } + }(r) + } + + time.Sleep(50 * time.Millisecond) + localCache.Destroy() + stop.Store(true) + wg.Wait() +} From 9f269a2578d793a792127b3182c200d1e0faabb2 Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Tue, 18 Aug 2026 12:20:07 +0330 Subject: [PATCH 5/8] fix(CubeMaster/pkg/base/localcache/localcache.go): keep master's LRU promotion points and guard a nil config Signed-off-by: Dwin Gharibi --- CubeMaster/pkg/base/localcache/localcache.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/CubeMaster/pkg/base/localcache/localcache.go b/CubeMaster/pkg/base/localcache/localcache.go index 32f48a518..97b626971 100644 --- a/CubeMaster/pkg/base/localcache/localcache.go +++ b/CubeMaster/pkg/base/localcache/localcache.go @@ -79,7 +79,7 @@ func NewCache(name string, loader LoaderFunc, localCacheConfig *LocalCacheConfig localCache.loadFile(localCacheConfig.LoadFileName) } localCache.localCacheConfig = localCache.SetupConfig(localCacheConfig) - localCache.expiredUse.Store(localCache.localCacheConfig.ExpiredUse) + localCache.expiredUse.Store(localCache.localCacheConfig != nil && localCache.localCacheConfig.ExpiredUse) localCache.chShrinkCache = make(chan bool, 1) localCache.chCacheExit = make(chan bool) @@ -126,7 +126,6 @@ func (localCache *LocalCache) Get(ctx context.Context, key string) (interface{}, localCache.Lock() itm := element.Value.(*util.CacheValue) - localCache.valueList.MoveToBack(element) localCache.Unlock() if time.Now().Add(-itm.Expired).After(time.Unix(itm.LastAccess, 0)) { @@ -135,6 +134,10 @@ func (localCache *LocalCache) Get(ctx context.Context, key string) (interface{}, r, f, err := localCache.loadAndRefresh(ctx, key) if err != nil && localCache.localCacheConfig.DemotionExpiredUse { + + localCache.Lock() + localCache.valueList.MoveToBack(element) + localCache.Unlock() return itm.Value, true, nil } @@ -148,6 +151,9 @@ func (localCache *LocalCache) Get(ctx context.Context, key string) (interface{}, localCache.asyncRefresh(ctx, key) } + localCache.Lock() + localCache.valueList.MoveToBack(element) + localCache.Unlock() return itm.Value, true, nil } From 9c1ecd531724de4e125273af221d4be73bed10e5 Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Tue, 18 Aug 2026 12:20:07 +0330 Subject: [PATCH 6/8] test(CubeMaster/pkg/base/localcache/localcache_race_test.go): pin that a failing refresh does not promote Signed-off-by: Dwin Gharibi --- .../base/localcache/localcache_race_test.go | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/CubeMaster/pkg/base/localcache/localcache_race_test.go b/CubeMaster/pkg/base/localcache/localcache_race_test.go index 45ef9d329..182071963 100644 --- a/CubeMaster/pkg/base/localcache/localcache_race_test.go +++ b/CubeMaster/pkg/base/localcache/localcache_race_test.go @@ -6,12 +6,15 @@ package localcache import ( "context" + "errors" "path/filepath" "strconv" "sync" "sync/atomic" "testing" "time" + + "github.com/tencentcloud/CubeSandbox/CubeMaster/pkg/base/localcache/util" ) func TestConcurrentGetAndRefreshOnSameKey(t *testing.T) { @@ -135,3 +138,83 @@ func TestDestroySnapshotDoesNotRaceWithInFlightRefresh(t *testing.T) { stop.Store(true) wg.Wait() } + +func frontKey(t *testing.T, c *LocalCache) string { + t.Helper() + c.Lock() + defer c.Unlock() + front := c.valueList.Front() + if front == nil { + t.Fatal("value list is empty") + } + return front.Value.(*util.CacheValue).Key +} + +func TestFailingRefreshDoesNotPromoteTheEntry(t *testing.T) { + var fail atomic.Bool + localCache := NewCache("lru-demotion-probe", + func(ctx context.Context, key string) (interface{}, bool, error) { + if fail.Load() && key == "a" { + return nil, false, errors.New("loader is down") + } + return "v-" + key, true, nil + }, + &LocalCacheConfig{ + LowCacheSize: 1000000, + HighCacheSize: 2000000, + Expired: time.Millisecond, + ExpiredUse: false, + DemotionExpiredUse: false, + }) + defer localCache.Destroy() + + ctx := context.Background() + for _, k := range []string{"a", "b"} { + if _, _, err := localCache.Get(ctx, k); err != nil { + t.Fatalf("seed Get(%s): %v", k, err) + } + } + if got := frontKey(t, localCache); got != "a" { + t.Fatalf("front is %q before the probe, want a", got) + } + + fail.Store(true) + time.Sleep(5 * time.Millisecond) + + if _, _, err := localCache.Get(ctx, "a"); err == nil { + t.Fatal("Get(a) succeeded while the loader was failing") + } + if got := frontKey(t, localCache); got != "a" { + t.Fatalf("a failing entry was promoted: front is %q, want a to stay evictable", got) + } +} + +func TestSuccessfulHitPromotesTheEntry(t *testing.T) { + localCache := NewCache("lru-promote-probe", + func(ctx context.Context, key string) (interface{}, bool, error) { + return "v-" + key, true, nil + }, + &LocalCacheConfig{ + LowCacheSize: 1000000, + HighCacheSize: 2000000, + Expired: time.Hour, + }) + defer localCache.Destroy() + + ctx := context.Background() + for _, k := range []string{"a", "b"} { + if _, _, err := localCache.Get(ctx, k); err != nil { + t.Fatalf("seed Get(%s): %v", k, err) + } + } + if got := frontKey(t, localCache); got != "a" { + t.Fatalf("front is %q before the probe, want a", got) + } + + if _, _, err := localCache.Get(ctx, "a"); err != nil { + t.Fatalf("Get(a): %v", err) + } + if got := frontKey(t, localCache); got != "b" { + t.Fatalf("a fresh hit did not promote: front is %q, want b", got) + } +} From 90ad2c518296381d2e942b4df97c18bbc849959c Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Wed, 19 Aug 2026 11:38:59 +0330 Subject: [PATCH 7/8] fix(CubeMaster/pkg/base/localcache/localcache.go): drop the dead Destroy guard and stop put blocking after shutdown Signed-off-by: Dwin Gharibi --- CubeMaster/pkg/base/localcache/localcache.go | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/CubeMaster/pkg/base/localcache/localcache.go b/CubeMaster/pkg/base/localcache/localcache.go index 97b626971..bb37ab3bd 100644 --- a/CubeMaster/pkg/base/localcache/localcache.go +++ b/CubeMaster/pkg/base/localcache/localcache.go @@ -105,11 +105,11 @@ func (localCache *LocalCache) Destroy() { if localCache == nil { return } + if localCache.chCacheExit == nil { + return + } localCache.destroyOnce.Do(func() { CubeLog.Infof("LruCache(%s) Destroy", localCache.name) - if localCache.chCacheExit == nil { - return - } if localCache.localCacheConfig.OpenCacheFile { localCache.saveFile(localCache.localCacheConfig.LoadFileName) } @@ -193,7 +193,10 @@ func (localCache *LocalCache) put(key string, val interface{}, expired time.Dura } if atomic.LoadInt64(&localCache.curCacheSize) >= localCache.localCacheConfig.HighCacheSize { - localCache.chShrinkCache <- true + select { + case localCache.chShrinkCache <- true: + default: + } } } From 35cba98c40572a75191814c13ba68abb820fcc00 Mon Sep 17 00:00:00 2001 From: Dwin Gharibi Date: Wed, 19 Aug 2026 11:38:59 +0330 Subject: [PATCH 8/8] test(CubeMaster/pkg/base/localcache/localcache_race_test.go): cover put after Destroy and require a real refresh Signed-off-by: Dwin Gharibi --- .../base/localcache/localcache_race_test.go | 29 +++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/CubeMaster/pkg/base/localcache/localcache_race_test.go b/CubeMaster/pkg/base/localcache/localcache_race_test.go index 182071963..3581e94fa 100644 --- a/CubeMaster/pkg/base/localcache/localcache_race_test.go +++ b/CubeMaster/pkg/base/localcache/localcache_race_test.go @@ -60,8 +60,8 @@ func TestConcurrentGetAndRefreshOnSameKey(t *testing.T) { } wg.Wait() - if atomic.LoadInt64(&loads) == 0 { - t.Fatal("loader never ran; the test did not exercise the refresh path") + if got := atomic.LoadInt64(&loads); got < 2 { + t.Fatalf("loader ran %d time(s); only the initial miss was exercised, not the refresh path", got) } } @@ -218,3 +218,28 @@ func TestSuccessfulHitPromotesTheEntry(t *testing.T) { t.Fatalf("a fresh hit did not promote: front is %q, want b", got) } } + +func TestPutAfterDestroyDoesNotBlockForever(t *testing.T) { + localCache := NewCache("post-destroy-put-probe", + func(ctx context.Context, key string) (interface{}, bool, error) { + return "v", true, nil + }, + &LocalCacheConfig{LowCacheSize: 1, HighCacheSize: 1, Expired: time.Hour}) + + localCache.put("seed-a", RandString(16), time.Hour) + localCache.Destroy() + + done := make(chan struct{}) + go func() { + defer close(done) + localCache.put("after-destroy-1", RandString(16), time.Hour) + localCache.put("after-destroy-2", RandString(16), time.Hour) + localCache.put("after-destroy-3", RandString(16), time.Hour) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("put blocked after Destroy: the shrink signal has no reader once shrinkCache has exited") + } +}