Skip to content

cubemaster: fix the data races in localcache #1376

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fslongjin merged 8 commits into from
Aug 19, 2026
+294 −31
Merged

cubemaster: fix the data races in localcache #1376

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d9e7a85
fix(CubeMaster/pkg/base/localcache/localcache_test.go): fixing the da…
dwin-gharibi Aug 18, 2026
7215762
test(CubeMaster/pkg/base/localcache/localcache_race_test.go): adding …
dwin-gharibi Aug 18, 2026
0765499
fix(CubeMaster/pkg/base/localcache/localcache.go): lock the saveFile …
dwin-gharibi Aug 18, 2026
389d18b
test(CubeMaster/pkg/base/localcache/localcache_race_test.go): overlap…
dwin-gharibi Aug 18, 2026
9f269a2
fix(CubeMaster/pkg/base/localcache/localcache.go): keep master's LRU …
dwin-gharibi Aug 18, 2026
9c1ecd5
test(CubeMaster/pkg/base/localcache/localcache_race_test.go): pin tha…
dwin-gharibi Aug 18, 2026
90ad2c5
fix(CubeMaster/pkg/base/localcache/localcache.go): drop the dead Dest…
dwin-gharibi Aug 19, 2026
35cba98
test(CubeMaster/pkg/base/localcache/localcache_race_test.go): cover p…
dwin-gharibi Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 45 additions & 27 deletions CubeMaster/pkg/base/localcache/localcache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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 != nil && localCache.localCacheConfig.ExpiredUse)

localCache.chShrinkCache = make(chan bool, 1)
localCache.chCacheExit = make(chan bool)
Expand All @@ -99,29 +102,35 @@ func NewCache(name string, loader LoaderFunc, localCacheConfig *LocalCacheConfig
}

func (localCache *LocalCache) Destroy() {
if localCache != nil {
if localCache == nil {
return
}
if localCache.chCacheExit == 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.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()

@cubesandboxbot cubesandboxbot Bot Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The PR description says Get reads the *CacheValue pointer and does its MoveToBack inside one critical section, but the MoveToBack at the end of Get is still a separate lock acquisition. The separation is actually load-bearing — promoting the entry before the refresh decision would make a failed refresh promote the entry, which TestFailingRefreshDoesNotPromoteTheEntry explicitly forbids — so the code is correct, just worth a comment. Note it also means every hit now acquires the cache-wide mutex twice (snapshot read + MoveToBack), doubling exclusive-lock contention on the hot read path of this shared cache.

itm := element.Value.(*util.CacheValue)
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 {
Expand Down Expand Up @@ -159,14 +168,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

@cubesandboxbot cubesandboxbot Bot Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

put() now writes element.Value = next while holding localCache's mutex, but saveFile() (unchanged in this PR, called from Destroy()) reads the same *list.Element.Value field — value.Value.(*util.CacheValue) at line 328 — without holding that mutex.

This is a newly-introduced data race. Before this PR, element.Value was write-once (assigned at PushBack), so saveFile's unlocked read was safe; now every refresh of an existing key writes the field. The race is reachable: asyncRefresh() spawns goroutines that are not tracked by waitGroup, and concurrent Gets can still be inside loadAndRefresh → put while Destroy() runs saveFile. The new race tests never overlap Destroy with an in-flight put, so the reported green -race run wouldn't catch it.

Fix suggestion: snapshot the *util.CacheValue under localCache.Lock() in saveFile (e.g. read value.Value into a local while holding the lock, then gob-encode after), so all readers of element.Value are serialized.

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,
Expand All @@ -180,8 +192,11 @@ func (localCache *LocalCache) put(key string, val interface{}, expired time.Dura
localCache.cache.Set(key, element, -1)
}

if localCache.curCacheSize >= localCache.localCacheConfig.HighCacheSize {
localCache.chShrinkCache <- true
if atomic.LoadInt64(&localCache.curCacheSize) >= localCache.localCacheConfig.HighCacheSize {
select {
case localCache.chShrinkCache <- true:
default:
}
}
}

Expand All @@ -198,7 +213,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 {
Expand Down Expand Up @@ -240,10 +255,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 {
Expand Down Expand Up @@ -305,12 +320,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)
}
}
}
Expand All @@ -324,10 +339,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:
Expand Down
Loading
Loading