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): init load files using a separate coroutine for handling #23092015

Browse Source 
Signed-off-by: clinx <chenlin1@oppo.com>
This commit is contained in:
clinx 2025-03-11 11:15:07 +08:00 committed by zhumingze1108
parent 1801c729a3
commit 28d975ce10
5 changed files with 91 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
flashnode/cachengine/engine.go
View File

@ -16,7 +16,6 @@ package cachengine
import (
"fmt"
syslog "log"
"math"
"os"
"path"
@ -29,12 +28,12 @@ import (
"syscall"
"time"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/util/atomicutil"
syslog "log"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/tmpfs"
@ -77,6 +76,13 @@ type cacheLoadTask struct {
dataPath string
}
type cacheLoadFile struct {
volume string
dataPath string
fullPath string
fileName string
}
type lruCacheItem struct {
lruCache LruCache
config CacheConfig
@ -107,6 +113,7 @@ type CacheEngine struct {
creatingCacheBlockMap sync.Map
cachePrepareTaskCh chan cachePrepareTask
cacheLoadWorkerNum int
cacheEvictWorkerNum int
lruCacheMap sync.Map
lruFhCache LruCache
readSourceFunc ReadExtentData
@ -126,7 +133,7 @@ type (
)
func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64, disks []*Disk,
capacity int, fhCapacity int, diskUnavailableCbErrorCount int64, cacheLoadWorkerNum int, mc *master.MasterClient, expireTime time.Duration, readFunc ReadExtentData, enableTmpfs bool, localAddr string,
capacity int, fhCapacity int, diskUnavailableCbErrorCount int64, cacheLoadWorkerNum int, cacheEvictWorkerNum int, mc *master.MasterClient, expireTime time.Duration, readFunc ReadExtentData, enableTmpfs bool, localAddr string,
) (s *CacheEngine, err error) {
s = new(CacheEngine)
s.enableTmpfs = enableTmpfs
@ -139,6 +146,7 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
s.closeCh = make(chan struct{})
s.fhCapacity = fhCapacity
s.cacheLoadWorkerNum = cacheLoadWorkerNum
s.cacheEvictWorkerNum = cacheEvictWorkerNum
s.localAddr = localAddr
if s.enableTmpfs {
fullPath := path.Join(memDataDir, DefaultCacheDirName)
@ -296,6 +304,7 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
var (
dirEntryWg sync.WaitGroup
workerWg sync.WaitGroup
fileLoadWg sync.WaitGroup
cbNum atomicutil.Int64
errorCbNum atomicutil.Int64
)
@ -305,7 +314,15 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
syslog.Print(msg)
log.LogInfo(msg)
}()
filePathChan := make(chan cacheLoadFile, c.cacheLoadWorkerNum*2)
for i := 0; i < c.cacheLoadWorkerNum*2; i++ {
go func() {
for fileInfo := range filePathChan {
c.handlerFile(&fileInfo, &cbNum, &errorCbNum)
fileLoadWg.Done()
}
}()
}
cacheLoadTaskCh := make(chan cacheLoadTask, 1024)
chs := make([]chan struct{}, c.cacheLoadWorkerNum)
workerWg.Add(c.cacheLoadWorkerNum)
@ -333,23 +350,8 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
log.LogWarnf("[LoadDisk] find invalid cacheBlock file[%v] on dataPath(%v)", filename, fullPath)
continue
}
inode, offset, version, err := unmarshalCacheBlockName(filename)
if err != nil {
log.LogErrorf("action[LoadDisk] unmarshal cacheBlockName(%v) from dataPath(%v) volume(%v) err(%v) ",
filename, fullPath, volume, err.Error())
continue
}
log.LogDebugf("acton[LoadDisk] dataPath(%v) cacheBlockName(%v) volume(%v) inode(%v) offset(%v) version(%v).",
fullPath, fileInfo.Name(), volume, inode, offset, version)
if _, err := c.createCacheBlockFromExist(dataPath, volume, inode, offset, version, 0, ""); err != nil {
c.deleteCacheBlock(GenCacheBlockKey(volume, inode, offset, version))
log.LogInfof("action[LoadDisk] createCacheBlock(%v) from dataPath(%v) volume(%v) err(%v) ",
filename, fullPath, volume, err.Error())
errorCbNum.Add(1)
continue
}
cbNum.Add(1)
fileLoadWg.Add(1)
filePathChan <- cacheLoadFile{volume: volume, dataPath: diskPath, fullPath: fullPath, fileName: filename}
}
dirEntryWg.Done()
case <-closeCh:
@ -365,6 +367,11 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
log.LogDebugf("action[LoadDisk] load cacheBlock from path(%v).", diskPath)
entries, err := os.ReadDir(diskPath)
if err != nil {
log.LogErrorf("action[LoadDisk] read dir(%v) err(%v).", diskPath, err)
for _, ch := range chs {
close(ch)
}
close(filePathChan)
return
}
dirEntryWg.Add(len(entries))
@ -379,9 +386,31 @@ func (c *CacheEngine) LoadDisk(diskPath string) (err error) {
}
workerWg.Wait()
close(cacheLoadTaskCh)
fileLoadWg.Wait()
close(filePathChan)
return
}
func (c *CacheEngine) handlerFile(file *cacheLoadFile, cbNum *atomicutil.Int64, errorCbNum *atomicutil.Int64) {
inode, offset, version, err := unmarshalCacheBlockName(file.fileName)
if err != nil {
log.LogErrorf("action[LoadDisk] unmarshal cacheBlockName(%v) from dataPath(%v) volume(%v) err(%v) ",
file.fileName, file.fullPath, file.volume, err.Error())
return
}
log.LogDebugf("acton[LoadDisk] dataPath(%v) cacheBlockName(%v) volume(%v) inode(%v) offset(%v) version(%v).",
file.fullPath, file.fileName, file.volume, inode, offset, version)
if _, err := c.createCacheBlockFromExist(file.dataPath, file.volume, inode, offset, version, 0, ""); err != nil {
c.deleteCacheBlock(GenCacheBlockKey(file.volume, inode, offset, version))
log.LogInfof("action[LoadDisk] createCacheBlock(%v) from dataPath(%v) volume(%v) err(%v) ",
file.fileName, file.fullPath, file.volume, err.Error())
errorCbNum.Add(1)
return
}
cbNum.Add(1)
}
func (c *CacheEngine) Start() (err error) {
c.startCachePrepareWorkers()
if !c.enableTmpfs {
@ -836,7 +865,7 @@ func (c *CacheEngine) EvictCacheAll() {
wg.Add(1)
go func(item *lruCacheItem) {
defer wg.Done()
item.lruCache.EvictAll()
item.lruCache.EvictAll(c.cacheEvictWorkerNum)
}(cacheItem)
return true
})
@ -903,7 +932,7 @@ func (c *CacheEngine) DoInactiveDisk(dataPath string) {
log.LogWarnf(msg)
atomic.StoreInt32(&cacheItem.status, proto.Unavailable)
go func() {
cacheItem.lruCache.EvictAll()
cacheItem.lruCache.EvictAll(c.cacheEvictWorkerNum)
}()
var keysToDelete []interface{}
@ -949,7 +978,7 @@ func (c *CacheEngine) triggerCacheError(key string, dataPath string) {
log.LogWarnf(msg)
atomic.StoreInt32(&cacheItem.status, proto.Unavailable)
go func() {
cacheItem.lruCache.EvictAll()
cacheItem.lruCache.EvictAll(c.cacheEvictWorkerNum)
}()
var keysToDelete []interface{}

16
flashnode/cachengine/engine_test.go
View File

@ -54,9 +54,9 @@ func TestEngineNew(t *testing.T) {
if !enabledTmpfs() {
disks := make([]*Disk, 0)
disks = append(disks, &Disk{Path: testTmpFS, TotalSpace: 200 * util.MB, Capacity: 1024})
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, 1024, 1024, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, 1024, 1024, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, 200*util.MB, DefaultCacheMaxUsedRatio, nil, 1024, 1024, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine(testTmpFS, 200*util.MB, DefaultCacheMaxUsedRatio, nil, 1024, 1024, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
@ -80,9 +80,9 @@ func TestEngineOverFlow(t *testing.T) {
if !enabledTmpfs() {
disks := make([]*Disk, 0)
disks = append(disks, &Disk{Path: testTmpFS, TotalSpace: util.GB, Capacity: 1024})
ce, err = NewCacheEngine("", 0, 1.1, disks, 1024, 1024, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine("", 0, 1.1, disks, 1024, 1024, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, util.GB, 1.1, nil, 1024, 1024, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine(testTmpFS, util.GB, 1.1, nil, 1024, 1024, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
defer func() { require.NoError(t, ce.Stop()) }()
@ -152,9 +152,9 @@ func TestEngineTTL(t *testing.T) {
if !enabledTmpfs() {
disks := make([]*Disk, 0)
disks = append(disks, &Disk{Path: testTmpFS, TotalSpace: util.GB, Capacity: 1024})
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, lruCap, lruCap, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, lruCap, lruCap, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
defer func() { require.NoError(t, ce.Stop()) }()
@ -213,9 +213,9 @@ func TestEngineLru(t *testing.T) {
if !enabledTmpfs() {
disks := make([]*Disk, 0)
disks = append(disks, &Disk{Path: testTmpFS, TotalSpace: util.GB, Capacity: lruCap})
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, lruCap, lruCap, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, lruCap, lruCap, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, 10, 10, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
ce.Start()

23
flashnode/cachengine/lru.go
View File

@ -31,7 +31,7 @@ type LruCache interface {
Peek(key interface{}) (interface{}, bool)
Set(key interface{}, value interface{}, expiration time.Duration) (int, error)
Evict(key interface{}) bool
EvictAll()
EvictAll(cacheEvictWorkerNum int)
Close() error
Status() *Status
StatusAll() *Status
@ -350,19 +350,28 @@ func (c *fCache) Peek(key interface{}) (interface{}, bool) {
}
// EvictAll is used to completely clear the cache.
func (c *fCache) EvictAll() {
func (c *fCache) EvictAll(cacheEvictWorkerNum int) {
c.lock.Lock()
toEvicts := make([]interface{}, 0, len(c.items))
var wg sync.WaitGroup
toEvicts := make(chan interface{}, cacheEvictWorkerNum)
for i := 0; i < cacheEvictWorkerNum; i++ {
go func() {
for e := range toEvicts {
_ = c.onDelete(e, "execute evictAll operation")
wg.Done()
}
}()
}
for _, ent := range c.items {
if c.cacheType == LRUCacheBlockCacheType {
c.DeleteKeyFromPreAllocatedKeyMap(ent.Value.(*entry).key)
}
toEvicts = append(toEvicts, c.deleteElement(ent))
wg.Add(1)
toEvicts <- c.deleteElement(ent)
}
wg.Wait()
close(toEvicts)
c.lock.Unlock()
for _, e := range toEvicts {
_ = c.onDelete(e, "execute evictAll operation")
}
}
func (c *fCache) Evict(key interface{}) bool {

2
flashnode/cachengine/lru_test.go
View File

@ -100,6 +100,6 @@ func TestLRUExpired(t *testing.T) {
require.Error(t, err)
require.Equal(t, 1, c.Len())
c.EvictAll()
c.EvictAll(2)
require.Equal(t, 0, c.Len())
}

11
flashnode/flashnode.go
View File

@ -51,6 +51,7 @@ const (
_defaultLRUFhCapacity = 10000
_defaultDiskUnavailableCbErrorCount = 3
_defaultCacheLoadWorkerNum = 8
_defaultCacheEvictWorkerNum = 8
_tcpServerTimeoutSec = 60 * 5
_connPoolIdleTimeout = 60 // 60s
_extentReadMaxRetry = 3
@ -67,6 +68,7 @@ const (
cfgLruFhCapacity = "lruFileHandleCapacity"
cfgDiskUnavailableCbErrorCount = "diskUnavailableCbErrorCount"
cfgCacheLoadWorkerNum = "cacheLoadWorkerNum"
cfgCacheEvictWorkerNum = "cacheEvictWorkerNum"
cfgZoneName = "zoneName"
cfgReadRps = "readRps"
cfgLowerHitRate = "lowerHitRate"
@ -86,6 +88,7 @@ type FlashNode struct {
lruFhCapacity int // file handle capacity
diskUnavailableCbErrorCount int64
cacheLoadWorkerNum int
cacheEvictWorkerNum int
memDataPath string
disks []*cachengine.Disk
mc *master.MasterClient
@ -301,6 +304,11 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) {
cacheLoadWorkerNum = _defaultCacheLoadWorkerNum
}
f.cacheLoadWorkerNum = cacheLoadWorkerNum
cacheEvictWorkerNum := cfg.GetInt(cfgCacheEvictWorkerNum)
if cacheEvictWorkerNum <= 0 || cacheEvictWorkerNum > 100 {
cacheEvictWorkerNum = _defaultCacheEvictWorkerNum
}
f.cacheEvictWorkerNum = cacheEvictWorkerNum
f.lowerHitRate = cfg.GetFloat(cfgLowerHitRate)
log.LogInfof("[parseConfig] load listen[%s].", f.listen)
@ -310,6 +318,7 @@ func (f *FlashNode) parseConfig(cfg *config.Config) (err error) {
log.LogInfof("[parseConfig] load lruFileHandleCapacity[%d]", f.lruFhCapacity)
log.LogInfof("[parseConfig] load diskUnavailableCbErrorCount[%d]", f.diskUnavailableCbErrorCount)
log.LogInfof("[parseConfig] load cacheLoadWorkerNum[%d]", f.cacheLoadWorkerNum)
log.LogInfof("[parseConfig] load cacheEvictWorkerNum[%d]", f.cacheEvictWorkerNum)
log.LogInfof("[parseConfig] load readRps[%d].", f.readRps)
log.LogInfof("[parseConfig] load lowerHitRate[%.2f].", f.lowerHitRate)
log.LogInfof("[parseConfig] load enableTmpfs[%v].", f.enableTmpfs)
@ -335,7 +344,7 @@ func (f *FlashNode) stopCacheEngine() {
func (f *FlashNode) startCacheEngine() (err error) {
if f.cacheEngine, err = cachengine.NewCacheEngine(f.memDataPath, int64(f.memTotal),
0, f.disks, f.lruCapacity, f.lruFhCapacity, f.diskUnavailableCbErrorCount, f.cacheLoadWorkerNum, f.mc, time.Hour, ReadExtentData, f.enableTmpfs, f.localAddr); err != nil {
0, f.disks, f.lruCapacity, f.lruFhCapacity, f.diskUnavailableCbErrorCount, f.cacheLoadWorkerNum, f.cacheEvictWorkerNum, f.mc, time.Hour, ReadExtentData, f.enableTmpfs, f.localAddr); err != nil {
log.LogErrorf("startCacheEngine failed:%v", err)
return
}