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): clears the cache on flashNode when removing it.

Browse Source 
close:#23003554

Signed-off-by: shuqiang-zheng <zhengshuqiang@oppo.com>
This commit is contained in:
shuqiang-zheng 2025-02-08 16:04:35 +08:00 committed by zhumingze1108
parent 08639d8b52
commit e0880fa4d0
8 changed files with 77 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cli/cmd/flashgroup.go
View File

@ -309,14 +309,14 @@ func newCmdFlashGroupList(client *master.MasterClient) *cobra.Command {
func newCmdFlashGroupClient(client *master.MasterClient) *cobra.Command {
return &cobra.Command{
Use: "client",
Short: "show client response",
Short: "show flash group response passed from master to client",
Args: cobra.MinimumNArgs(0),
RunE: func(cmd *cobra.Command, args []string) (err error) {
fgv, err := client.AdminAPI().ClientFlashGroups()
if err != nil {
return
}
stdoutln("Client Response:")
stdoutln("Flash Group Response:")
stdoutln(formatIndent(fgv))
return
},

60
flashnode/cachengine/engine.go
View File

@ -30,8 +30,6 @@ import (
"time"
"github.com/cubefs/cubefs/sdk/master"
"github.com/cubefs/cubefs/util/ump"
"github.com/cubefs/cubefs/util/atomicutil"
"github.com/cubefs/cubefs/proto"
@ -112,6 +110,7 @@ type CacheEngine struct {
closeCh chan struct{}
enableTmpfs bool // for testing in docker
localAddr string
readDataNodeTimeout int
}
@ -122,7 +121,7 @@ type (
)
func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64, disks []*Disk,
capacity int, fhCapacity int, diskUnavailableCbErrorCount int64, mc *master.MasterClient, expireTime time.Duration, readFunc ReadExtentData, enableTmpfs bool,
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
@ -135,6 +134,7 @@ func NewCacheEngine(memDataDir string, totalMemSize int64, maxUseRatio float64,
s.readSourceFunc = readFunc
s.closeCh = make(chan struct{})
s.fhCapacity = fhCapacity
s.localAddr = localAddr
if s.enableTmpfs {
memCacheConfig := CacheConfig{
Medium: "memory",
@ -430,6 +430,10 @@ func (c *CacheEngine) deleteCacheBlock(key string) {
if ok {
cacheItem := value.(*lruCacheItem)
cacheItem.lruCache.Evict(key)
_, getErr := c.lruFhCache.Get(key)
if getErr == nil {
c.lruFhCache.Evict(key)
}
c.keyToDiskMap.Delete(key)
}
}
@ -469,8 +473,8 @@ func (c *CacheEngine) PeekCacheBlock(key string) (block *CacheBlock, err error)
}
func (c *CacheEngine) selectAvailableLruCache() (cacheItem *lruCacheItem, err error) {
var maxExpectedLeftSpace int64 = 0
var maxRealLeftSpace int64 = 0
var maxLeftSpace int64 = 0
var leftSpace int64 = 0
c.lruCacheMap.Range(func(key, value interface{}) bool {
item := value.(*lruCacheItem)
if atomic.LoadInt32(&item.status) == proto.ReadWrite {
@ -480,15 +484,23 @@ func (c *CacheEngine) selectAvailableLruCache() (cacheItem *lruCacheItem, err er
return true
}
realLeftSpace := int64(fs.Bavail * uint64(fs.Bsize))
if realLeftSpace >= maxRealLeftSpace && expectedLeftSpace >= maxExpectedLeftSpace {
maxExpectedLeftSpace = expectedLeftSpace
maxRealLeftSpace = realLeftSpace
if realLeftSpace < expectedLeftSpace {
leftSpace = realLeftSpace
} else {
leftSpace = expectedLeftSpace
}
if leftSpace >= maxLeftSpace {
maxLeftSpace = leftSpace
cacheItem = item
}
}
return true
})
if cacheItem != nil {
log.LogInfof("select disk(%v) success", cacheItem.config.Path)
return
}
return nil, errors.NewErrorf("no available disk can select")
@ -520,16 +532,20 @@ func (c *CacheEngine) createCacheBlockFromExist(dataPath string, volume string,
c.keyToDiskMap.Store(key, cacheItem)
if _, err = cacheItem.lruCache.Set(key, block, time.Duration(block.ttl)*time.Second); err != nil {
if err = block.initFilePath(true); err != nil {
block.Delete()
return
}
err = block.initFilePath(true)
if _, err = cacheItem.lruCache.Set(key, block, time.Duration(block.ttl)*time.Second); err != nil {
block.Delete()
return
}
return
}
func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, version uint32, ttl int64, allocSize uint64, clientIP string) (block *CacheBlock, err error) {
func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64, version uint32, ttl int64, allocSize uint64, clientIP string, isPrepare bool) (block *CacheBlock, err error) {
if allocSize == 0 {
return nil, fmt.Errorf("alloc size is zero")
}
@ -581,7 +597,9 @@ func (c *CacheEngine) createCacheBlock(volume string, inode, fixedOffset uint64,
if _, err = cacheItem.lruCache.Set(key, block, time.Duration(ttl)*time.Second); err != nil {
return
}
if !isPrepare {
cacheItem.lruCache.AddMisses()
}
err = block.initFilePath(false)
}
@ -634,7 +652,7 @@ func (c *CacheEngine) startCachePrepareWorkers() {
}
func (c *CacheEngine) PrepareCache(reqID int64, req *proto.CacheRequest, clientIP string) (err error) {
if _, err = c.CreateBlock(req, clientIP); err != nil {
if _, err = c.CreateBlock(req, clientIP, true); err != nil {
return
}
select {
@ -645,11 +663,11 @@ func (c *CacheEngine) PrepareCache(reqID int64, req *proto.CacheRequest, clientI
return
}
func (c *CacheEngine) CreateBlock(req *proto.CacheRequest, clientIP string) (block *CacheBlock, err error) {
func (c *CacheEngine) CreateBlock(req *proto.CacheRequest, clientIP string, isPrepare bool) (block *CacheBlock, err error) {
if len(req.Sources) == 0 {
return nil, fmt.Errorf("no source data")
}
if block, err = c.createCacheBlock(req.Volume, req.Inode, req.FixedFileOffset, req.Version, req.TTL, computeAllocSize(req.Sources), clientIP); err != nil {
if block, err = c.createCacheBlock(req.Volume, req.Inode, req.FixedFileOffset, req.Version, req.TTL, computeAllocSize(req.Sources), clientIP, isPrepare); err != nil {
c.deleteCacheBlock(GenCacheBlockKey(req.Volume, req.Inode, req.FixedFileOffset, req.Version))
return nil, err
}
@ -760,7 +778,7 @@ func (c *CacheEngine) GetHeartBeatCacheStat() []*proto.CacheStatus {
Total: cacheItem.config.Total,
MaxAlloc: cacheItem.config.MaxAlloc,
HasAlloc: cacheItem.lruCache.GetAllocated(),
HitRate: cacheItem.lruCache.GetRateStat().HitRate,
HitRate: math.Trunc(cacheItem.lruCache.GetRateStat().HitRate*1e4+0.5) * 1e-4,
Evicts: int(cacheItem.lruCache.GetRateStat().Evicts),
Num: cacheItem.lruCache.Len(),
Status: int(atomic.LoadInt32(&cacheItem.status)),
@ -775,7 +793,7 @@ func (c *CacheEngine) GetHitRate() map[string]float64 {
result := make(map[string]float64)
c.lruCacheMap.Range(func(key, value interface{}) bool {
cacheItem := value.(*lruCacheItem)
result[cacheItem.config.Path] = cacheItem.lruCache.GetRateStat().HitRate
result[cacheItem.config.Path] = math.Trunc(cacheItem.lruCache.GetRateStat().HitRate*1e4+0.5) * 1e-4
return true
})
return result
@ -792,11 +810,7 @@ func (c *CacheEngine) GetEvictCount() map[string]int {
}
func (c *CacheEngine) doInactiveFLashNode() (err error) {
var addr string
if addr, err = ump.GetLocalIpAddr(); err != nil {
return err
}
return c.mc.NodeAPI().SetFlashNode(addr, false)
return c.mc.NodeAPI().SetFlashNode(c.localAddr, false)
}
func (c *CacheEngine) triggerCacheError(key string, dataPath string) {
@ -804,7 +818,7 @@ func (c *CacheEngine) triggerCacheError(key string, dataPath string) {
cacheItem := value.(*lruCacheItem)
if atomic.LoadInt32(&cacheItem.status) == proto.ReadWrite {
cacheItem.cacheErrCbSet.Store(key, struct{}{})
cacheErrCnt := atomic.LoadUint64(&cacheItem.cacheErrCnt)
cacheErrCnt := atomic.AddUint64(&cacheItem.cacheErrCnt, 1)
cacheErrCbList := make([]string, 0)
cacheItem.cacheErrCbSet.Range(func(key, value interface{}) bool {
cacheErrCbList = append(cacheErrCbList, key.(string))

24
flashnode/cachengine/engine_test.go
View File

@ -53,16 +53,16 @@ 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, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, 1024, 1024, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, 200*util.MB, DefaultCacheMaxUsedRatio, nil, 1024, 1024, 0, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine(testTmpFS, 200*util.MB, DefaultCacheMaxUsedRatio, nil, 1024, 1024, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
defer func() { require.NoError(t, ce.Stop()) }()
var cb *CacheBlock
inode, fixedOffset, version := uint64(1), uint64(1024), uint32(112358796)
cb, err = ce.createCacheBlock(t.Name(), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, "")
cb, err = ce.createCacheBlock(t.Name(), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, "", false)
require.NoError(t, err)
require.NoError(t, cb.WriteAt(bytesCommon, 0, 1024))
}
@ -79,9 +79,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, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine("", 0, 1.1, disks, 1024, 1024, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, util.GB, 1.1, nil, 1024, 1024, 0, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine(testTmpFS, util.GB, 1.1, nil, 1024, 1024, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
defer func() { require.NoError(t, ce.Stop()) }()
@ -100,7 +100,7 @@ func TestEngineOverFlow(t *testing.T) {
inode, fixedOffset, version := uint64(1), uint64(1024), uint32(112358796)
cb, err1 := ce.createCacheBlock(fmt.Sprintf("%s_%d_%d", t.Name(), round, thread),
inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, "")
inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, "", false)
if err1 != nil {
isErr.Store(true)
return
@ -151,9 +151,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, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, lruCap, lruCap, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
defer func() { require.NoError(t, ce.Stop()) }()
@ -164,7 +164,7 @@ func TestEngineTTL(t *testing.T) {
wg.Add(1)
go func(index int) {
defer wg.Done()
cb, err := ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), index), inode, fixedOffset, version, ttl, proto.CACHE_BLOCK_SIZE, "")
cb, err := ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), index), inode, fixedOffset, version, ttl, proto.CACHE_BLOCK_SIZE, "", false)
require.NoError(t, err)
var offset int64
for {
@ -211,9 +211,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, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine("", 0, DefaultCacheMaxUsedRatio, disks, lruCap, lruCap, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
} else {
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, nil, DefaultExpireTime, nil, enabledTmpfs())
ce, err = NewCacheEngine(testTmpFS, util.GB, DefaultCacheMaxUsedRatio, nil, lruCap, lruCap, 0, nil, DefaultExpireTime, nil, enabledTmpfs(), "")
}
require.NoError(t, err)
ce.Start()
@ -223,7 +223,7 @@ func TestEngineLru(t *testing.T) {
var cb *CacheBlock
var offset int64
inode, fixedOffset, version := uint64(1), uint64(1024), uint32(112358796)
cb, err = ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), j), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, "")
cb, err = ce.createCacheBlock(fmt.Sprintf("%s_%d", t.Name(), j), inode, fixedOffset, version, DefaultExpireTime, proto.CACHE_BLOCK_SIZE, "", false)
require.NoError(t, err)
for {
err = cb.WriteAt(bytesCommon, offset, 1024)

8
flashnode/cachengine/lru.go
View File

@ -39,6 +39,7 @@ type LruCache interface {
GetRateStat() RateStat
GetAllocated() int64
GetExpiredTime(key interface{}) (time.Time, bool)
AddMisses()
}
type Status struct {
@ -205,7 +206,6 @@ func (c *fCache) Set(key, value interface{}, expiration time.Duration) (n int, e
if c.cacheType == LRUCacheBlockCacheType {
newCb := value.(*CacheBlock)
cbSize = newCb.getAllocSize()
atomic.AddInt64(&c.allocated, cbSize)
fs := syscall.Statfs_t{}
if err = syscall.Statfs(newCb.rootPath, &fs); err != nil {
@ -214,6 +214,8 @@ func (c *fCache) Set(key, value interface{}, expiration time.Duration) (n int, e
}
diskSpaceLeft = int64(fs.Bavail * uint64(fs.Bsize))
diskSpaceLeft -= cbSize
atomic.AddInt64(&c.allocated, cbSize)
}
c.items[key] = c.lru.PushFront(&entry{
@ -366,3 +368,7 @@ func (c *fCache) GetExpiredTime(key interface{}) (time.Time, bool) {
}
return time.Time{}, false
}
func (c *fCache) AddMisses() {
atomic.AddInt32(&c.misses, 1)
}

2
flashnode/flashnode.go
View File

@ -326,7 +326,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.mc, time.Hour, ReadExtentData, f.enableTmpfs); err != nil {
0, f.disks, f.lruCapacity, f.lruFhCapacity, f.diskUnavailableCbErrorCount, f.mc, time.Hour, ReadExtentData, f.enableTmpfs, f.localAddr); err != nil {
log.LogErrorf("startCacheEngine failed:%v", err)
return
}

4
flashnode/flashnode_op.go
View File

@ -153,7 +153,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
cr := req.CacheRequest
block, err := f.cacheEngine.GetCacheBlockForRead(volume, cr.Inode, cr.FixedFileOffset, cr.Version, req.Size_)
if err != nil {
log.LogWarnf("opCacheRead: GetCacheBlockForRead failed, req(%v) err(%v)", req, err)
log.LogInfof("opCacheRead: GetCacheBlockForRead failed, req(%v) err(%v)", req, err)
hitRateMap := f.cacheEngine.GetHitRate()
for dataPath, hitRate := range hitRateMap {
if hitRate < f.lowerHitRate {
@ -162,7 +162,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
errMetric.AddWithLabels(1, map[string]string{exporter.FlashNode: f.localAddr, exporter.Disk: dataPath, exporter.Err: "LowerHitRate"})
}
}
if block, err = f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String()); err != nil {
if block, err = f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String(), false); err != nil {
log.LogErrorf("opCacheRead: CreateBlock failed, req(%v) err(%v)", req, err)
return err
}

1
master/flash_group.go
View File

@ -489,6 +489,7 @@ func (c *Cluster) setFlashNodeToUnused(addr string, flashGroupID uint64) (flashN
log.LogErrorf("flashNode[%v] evict all failed, err:%v", flashNode.Addr, err)
return
}
return
}()
return

15
master/flash_node.go
View File

@ -19,9 +19,12 @@ import (
"fmt"
"net/http"
"strconv"
"strings"
"sync"
"time"
"github.com/cubefs/cubefs/sdk/httpclient"
"github.com/cubefs/cubefs/cmd/common"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
@ -317,6 +320,18 @@ func (c *Cluster) removeFlashNode(flashNode *FlashNode) (err error) {
flashGroup.removeFlashNode(flashNode.Addr)
c.flashNodeTopo.updateClientCache()
}
go func() {
arr := strings.SplitN(flashNode.Addr, ":", 2)
p, _ := strconv.ParseUint(arr[1], 10, 64)
addr := fmt.Sprintf("%s:%d", arr[0], p+1)
if err = httpclient.New().Addr(addr).FlashNode().EvictAll(); err != nil {
log.LogErrorf("flashNode[%v] evict all failed, err:%v", flashNode.Addr, err)
return
}
return
}()
log.LogInfof("action[removeFlashNode], clusterID[%s] node[%s] flashGroupID[%d] offline success",
c.Name, flashNode.Addr, flashGroupID)
return