gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(flashnode): if the disk is umounted, the cacheblock can be detected and stopped timely.

Browse Source 
close:#23021760

Signed-off-by: shuqiang-zheng <zhengshuqiang@oppo.com>
This commit is contained in:
shuqiang-zheng 2025-02-24 10:07:13 +08:00 committed by zhumingze1108
parent c3a505fb1e
commit 1ebe5b6dfd
2 changed files with 44 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
flashnode/cachengine/engine.go
View File

@ -44,6 +44,7 @@ const (
DefaultExpireTime = 60 * 60
InitFileName = "flash.init"
DefaultCacheDirName = "cache"
DefaultCacheMaxUsedRatio = 0.9
DefaultEnableTmpfs = true
@ -125,7 +126,6 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
capacity int, fhCapacity int, diskUnavailableCbErrorCount int64, mc *master.MasterClient, expireTime time.Duration, readFunc ReadExtentData, enableTmpfs bool, localAddr string,
) (s *CacheEngine, err error) {
s = new(CacheEngine)
s.memDataPath = memDataDir
s.enableTmpfs = enableTmpfs
if maxUseRatio < 1e-1 {
maxUseRatio = DefaultCacheMaxUsedRatio
@ -137,15 +137,17 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
s.fhCapacity = fhCapacity
s.localAddr = localAddr
if s.enableTmpfs {
fullPath := path.Join(memDataDir, DefaultCacheDirName)
memCacheConfig := CacheConfig{
Medium: "memory",
Path: memDataDir,
Path: fullPath,
MaxAlloc: int64(float64(totalMemSize) * maxUseRatio),
Total: totalMemSize,
Capacity: capacity,
DiskUnavailableCbErrorCount: diskUnavailableCbErrorCount,
}
s.memDataPath = fullPath
s.cachePrepareTaskCh = make(chan cachePrepareTask, 1024)
s.cacheLoadTaskCh = make(chan cacheLoadTask, 1024)
cache := NewCache(LRUCacheBlockCacheType, memCacheConfig.Capacity, memCacheConfig.MaxAlloc, expireTime,
@ -157,7 +159,7 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
cb := v.(*CacheBlock)
return cb.Close()
})
s.lruCacheMap.Store(memDataDir, &lruCacheItem{lruCache: cache, config: memCacheConfig, status: proto.ReadWrite})
s.lruCacheMap.Store(fullPath, &lruCacheItem{lruCache: cache, config: memCacheConfig, status: proto.ReadWrite})
s.totalCacheNum = 1
s.lruFhCache = NewCache(LRUFileHandleCacheType, fhCapacity, -1, expireTime,
func(v interface{}) error {
@ -173,14 +175,26 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
if err = s.doMount(); err != nil {
return
}
if _, err = os.Stat(fullPath); err != nil {
if !os.IsNotExist(err.(*os.PathError)) {
return
}
if err = os.Mkdir(fullPath, 0o755); err != nil {
if !os.IsExist(err) {
return
}
}
}
return
}
for _, d := range disks {
fullPath := path.Join(d.Path, DefaultCacheDirName)
diskCacheConfig := CacheConfig{
Medium: "disk",
Path: d.Path,
Path: fullPath,
MaxAlloc: int64(float64(d.TotalSpace) * maxUseRatio),
Total: int64(d.TotalSpace),
Capacity: d.Capacity,
@ -199,8 +213,19 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
cb := v.(*CacheBlock)
return cb.Close()
})
s.lruCacheMap.Store(d.Path, &lruCacheItem{lruCache: cache, config: diskCacheConfig, status: proto.ReadWrite})
s.lruCacheMap.Store(fullPath, &lruCacheItem{lruCache: cache, config: diskCacheConfig, status: proto.ReadWrite})
s.totalCacheNum++
if _, err = os.Stat(fullPath); err != nil {
if !os.IsNotExist(err.(*os.PathError)) {
return
}
if err = os.Mkdir(fullPath, 0o755); err != nil {
if !os.IsExist(err) {
return
}
}
}
}
s.lruFhCache = NewCache(LRUFileHandleCacheType, fhCapacity, -1, expireTime,
func(v interface{}) error {
@ -825,7 +850,10 @@ func (c *CacheEngine) DoInactiveDisk(dataPath string) {
cacheItem := value.(*lruCacheItem)
if atomic.LoadInt32(&cacheItem.status) == proto.ReadWrite {
atomic.StoreInt32(&cacheItem.status, proto.Unavailable)
cacheItem.lruCache.EvictAll()
go func() {
cacheItem.lruCache.EvictAll()
}()
var keysToDelete []interface{}
c.keyToDiskMap.Range(func(key, value interface{}) bool {
item := value.(*lruCacheItem)
@ -865,7 +893,10 @@ func (c *CacheEngine) triggerCacheError(key string, dataPath string) {
cacheItem.config.Path, cacheErrCnt, cacheErrCbCnt, cacheItem.config.DiskUnavailableCbErrorCount)
log.LogWarnf(msg)
atomic.StoreInt32(&cacheItem.status, proto.Unavailable)
cacheItem.lruCache.EvictAll()
go func() {
cacheItem.lruCache.EvictAll()
}()
var keysToDelete []interface{}
c.keyToDiskMap.Range(func(key, value interface{}) bool {
item := value.(*lruCacheItem)

7
flashnode/cachengine/engine_test.go
View File

@ -18,6 +18,7 @@ import (
"crypto/rand"
"fmt"
"os"
"path"
"sync"
"sync/atomic"
"testing"
@ -192,7 +193,8 @@ func TestEngineTTL(t *testing.T) {
t.Logf("%+v", status)
}
_, err = ce.GetCacheBlockForRead(fmt.Sprintf("%s_%d", t.Name(), lruCap-1), inode, fixedOffset, version, 0)
value, _ := ce.lruCacheMap.Load(testTmpFS)
fullPath := path.Join(testTmpFS, DefaultCacheDirName)
value, _ := ce.lruCacheMap.Load(fullPath)
lruCacheLen := value.(*lruCacheItem).lruCache.Len()
require.Error(t, err, fmt.Sprintf("test[%s] expect get cacheBlock[%s] fail, but success, lruCacheCap(%d) lruCacheLen(%d)",
t.Name(), GenCacheBlockKey(fmt.Sprintf("%s_%d", t.Name(), lruCap-1), inode, fixedOffset, version), lruCap, lruCacheLen))
@ -232,7 +234,8 @@ func TestEngineLru(t *testing.T) {
}
offset += 1024
}
value, _ := ce.lruCacheMap.Load(testTmpFS)
fullPath := path.Join(testTmpFS, DefaultCacheDirName)
value, _ := ce.lruCacheMap.Load(fullPath)
lruCacheLen := value.(*lruCacheItem).lruCache.Len()
require.LessOrEqual(t, lruCacheLen, lruCap)
}