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

fix(client): update inode expiration time and enhance logging for extents. #1000402826

Browse Source 
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2025-10-31 15:19:07 +08:00 committed by chihe
parent 4ae784e64b
commit 360f72e9db
7 changed files with 139 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
client/fs/const.go
View File

@ -32,7 +32,7 @@ const (
)
const (
DefaultInodeExpiration = 120 * time.Second
DefaultInodeExpiration = 300 * time.Second
MaxInodeCache = 10000000 // in terms of the number of items
DefaultMaxInodeCache = 2000000
)

25
client/fs/dir.go
View File

@ -164,10 +164,9 @@ func (d *Dir) Release(ctx context.Context, req *fuse.ReleaseRequest) (err error)
log.LogDebugf("TRACE Release exit: ino(%v) name(%v)", d.info.Inode, d.name)
}()
// d.dctx.Clear()
d.dctx.Remove(req.Handle)
d.dcache.Clear()
ino := d.info.Inode
d.super.ic.Delete(ino)
// d.dcache.Clear()
// ino := d.info.Inode
// d.super.ic.Delete(ino)
return nil
}
@ -418,8 +417,8 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo
child = NewDir(d.super, info, d.info.Inode, req.Name)
} else {
child = NewFile(d.super, info, DefaultFlag, d.info.Inode, req.Name)
log.LogDebugf("Lookup: new file nodeCache parent(%v) name(%v) ino(%v) storageClass(%v) fullPath(%v)",
d.info.Inode, req.Name, ino, child.(*File).info.StorageClass, fullPath)
log.LogDebugf("Lookup: new file nodeCache parent(%v) name(%v) ino(%v) storageClass(%v) fullPath(%v), hasExtents(%v)",
d.info.Inode, req.Name, ino, child.(*File).info.StorageClass, fullPath, info.HasExtents())
}
d.super.nodeCache[ino] = child
} else {
@ -432,8 +431,8 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo
if child.(*File).info.StorageClass != info.StorageClass {
child = NewFile(d.super, info, DefaultFlag, d.info.Inode, req.Name)
}
log.LogDebugf("Lookup: update nodeCache parent(%v) name(%v) ino(%v) storageClass(%v)",
d.info.Inode, req.Name, ino, child.(*File).info.StorageClass)
log.LogDebugf("Lookup: update nodeCache parent(%v) name(%v) ino(%v) storageClass(%v), hasExtents(%v)",
d.info.Inode, req.Name, ino, child.(*File).info.StorageClass, info.HasExtents())
d.super.nodeCache[ino] = child
}
}
@ -504,6 +503,8 @@ func (d *Dir) ReadDir(ctx context.Context, req *fuse.ReadRequest, resp *fuse.Rea
// skip the first one, which is already accessed
childrenNr := uint64(len(children))
if childrenNr == 0 || (dirCtx.Name != "" && childrenNr == 1) {
log.LogDebugf("Readdir no more children: ino(%v) path(%v) d.super.bcacheDir(%v) childrenNr(%v) dirCtx.Name(%v)",
d.info.Inode, d.getCwd(), d.super.bcacheDir, childrenNr, dirCtx.Name)
return make([]fuse.Dirent, 0), io.EOF
} else if childrenNr < limit {
err = io.EOF
@ -525,6 +526,10 @@ func (d *Dir) ReadDir(ctx context.Context, req *fuse.ReadRequest, resp *fuse.Rea
dcache = NewDentryCache()
}
if d.dcache != nil {
dcache = d.dcache
}
var dcachev2 bool
if d.needDentrycache() {
dcachev2 = true
@ -550,7 +555,7 @@ func (d *Dir) ReadDir(ctx context.Context, req *fuse.ReadRequest, resp *fuse.Rea
}
}
infos := d.super.mw.BatchInodeGet(inodes)
infos := d.super.mw.BatchInodeGetExtents(inodes)
for _, info := range infos {
cacheInfo := d.super.ic.Get(info.Inode)
if cacheInfo != nil {
@ -640,7 +645,7 @@ func (d *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
}
}
infos := d.super.mw.BatchInodeGet(inodes)
infos := d.super.mw.BatchInodeGetExtents(inodes)
for _, info := range infos {
d.super.ic.Put(info)
}

65
client/fs/file.go
View File

@ -187,6 +187,13 @@ func (f *File) Attr(ctx context.Context, a *fuse.Attr) error {
fillAttr(info, a)
a.ParentIno = f.parentIno
// var fileSize int64
// var gen uint64
// if info.Extents != nil {
// a.Size = uint64(info.Extents.Size)
// }
fileSize, gen := f.fileSizeVersion2(ino)
log.LogDebugf("Attr: ino(%v) fileSize(%v) gen(%v) inode.gen(%v)", ino, fileSize, gen, info.Generation)
if gen >= info.Generation {
@ -277,7 +284,11 @@ func (f *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenR
}
log.LogDebugf("TRACE open ino(%v) f.super.bcacheDir(%v) needBCache(%v)", ino, f.super.bcacheDir, needBCache)
f.super.ec.RefreshExtentsCache(ino)
if f.info.Extents != nil {
f.super.ec.RefreshExtentsWithCache(f.info)
} else {
f.super.ec.RefreshExtentsCache(ino)
}
if f.super.keepCache && resp != nil {
resp.Flags |= fuse.OpenKeepCache
@ -336,27 +347,27 @@ func (f *File) Release(ctx context.Context, req *fuse.ReleaseRequest) (err error
stat.EndStat("Release:file", err, bgTime, 1)
log.LogInfof("action[Release] %v", f.fWriter)
f.fWriter.FreeCache()
if f.super.ec.RefCnt(ino) == 0 {
// keep nodeCache hold the latest inode info
f.super.fslock.Lock()
delete(f.super.nodeCache, ino)
f.super.fslock.Unlock()
if DisableMetaCache {
f.super.ic.Delete(ino)
}
f.super.fslock.Lock()
delete(f.super.nodeCache, ino)
node, ok := f.super.nodeCache[f.parentIno]
if ok {
parent, ok := node.(*Dir)
if ok {
parent.dcache.Delete(f.name)
log.LogDebugf("TRACE Release exit: ino(%v) name(%v) decache(%v)",
parent.info.Inode, parent.name, parent.dcache.Len())
}
}
f.super.fslock.Unlock()
}
// if f.super.ec.RefCnt(ino) == 0 {
// // keep nodeCache hold the latest inode info
// f.super.fslock.Lock()
// delete(f.super.nodeCache, ino)
// f.super.fslock.Unlock()
// if DisableMetaCache {
// f.super.ic.Delete(ino)
// }
// f.super.fslock.Lock()
// delete(f.super.nodeCache, ino)
// node, ok := f.super.nodeCache[f.parentIno]
// if ok {
// parent, ok := node.(*Dir)
// if ok {
// parent.dcache.Delete(f.name)
// log.LogDebugf("TRACE Release exit: ino(%v) name(%v) decache(%v)",
// parent.info.Inode, parent.name, parent.dcache.Len())
// }
// }
// f.super.fslock.Unlock()
// }
f.super.runningMonitor.SubClientOp(runningStat, err)
}()
@ -631,7 +642,15 @@ func (f *File) Flush(ctx context.Context, req *fuse.FlushRequest) (err error) {
}
if DisableMetaCache {
f.super.ic.Delete(f.info.Inode)
openForWrite := false
if req.Flags&0x0f != syscall.O_RDONLY {
openForWrite = true
}
if openForWrite {
f.super.ic.Delete(f.info.Inode)
}
}
elapsed := time.Since(start)

9
client/fs/icache.go
View File

@ -73,8 +73,8 @@ func (ic *InodeCache) Put(info *proto.InodeInfo) {
element := ic.lruList.PushFront(info)
ic.cache[info.Inode] = element
ic.Unlock()
log.LogDebugf("InodeCache put inode: inode(%v) expire(%v)",
info.Inode, info.Expiration())
log.LogDebugf("InodeCache put inode: inode(%v) expire(%v), hasExtents(%v)",
info.Inode, info.Expiration(), info.HasExtents())
}
// Get returns the inode info based on the given inode number.
@ -96,9 +96,10 @@ func (ic *InodeCache) Get(ino uint64) *proto.InodeInfo {
ic.RUnlock()
if info != nil {
log.LogDebugf("Inode Cache found ino(%v) storageClass(%v)",
ino, info.StorageClass)
log.LogDebugf("Inode Cache found ino(%v) storageClass(%v), expiration(%v)",
ino, info.StorageClass, info.Expiration())
}
info.SetExpiration(time.Now().Add(ic.expiration).UnixNano())
return info
}

17
proto/fs_proto.go
View File

@ -97,12 +97,17 @@ type InodeInfo struct {
expiration int64
PersistAccessTime time.Time `json:"pat"`
StorageClass uint32 `json:"storageClass"`
LeaseExpireTime uint64 `json:"leaseExpireTime"`
ForbiddenLc bool `json:"forbiddenLc"`
MigrationStorageClass uint32 `json:"migrationStorageClass"`
HasMigrationEk bool `json:"hasMigrationEk"`
MigrationExtentKeyExpiredTime time.Time `json:"mekExpiredTime"`
StorageClass uint32 `json:"storageClass"`
LeaseExpireTime uint64 `json:"leaseExpireTime"`
ForbiddenLc bool `json:"forbiddenLc"`
MigrationStorageClass uint32 `json:"migrationStorageClass"`
HasMigrationEk bool `json:"hasMigrationEk"`
MigrationExtentKeyExpiredTime time.Time `json:"mekExpiredTime"`
Extents *GetExtentsResponse `json:"eks"`
}
func (info *InodeInfo) HasExtents() bool {
return info.Extents != nil
}
type SimpleExtInfo struct {

17
sdk/data/stream/extent_client.go
View File

@ -637,6 +637,16 @@ func (client *ExtentClient) EvictStream(inode uint64) error {
return nil
}
func (client *ExtentClient) RefreshExtentsWithCache(inode *proto.InodeInfo) error {
s := client.GetStreamer(inode.Inode)
if s == nil {
return nil
}
s.extents.update(inode.Generation, inode.Size, false, inode.Extents.Extents)
return nil
}
// RefreshExtentsCache refreshes the extent cache.
func (client *ExtentClient) RefreshExtentsCache(inode uint64) error {
s := client.GetStreamer(inode)
@ -769,7 +779,12 @@ func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int
var errGetExtents error
s.once.Do(func() {
errGetExtents = s.GetExtents(isMigration)
if s.extents.gen == 0 {
errGetExtents = s.GetExtents(isMigration)
} else {
errGetExtents = nil
}
// errGetExtents = s.GetExtents(isMigration)
if log.EnableDebug() {
log.LogDebugf("Read: ino(%v) offset(%v) size(%v) storageClass(%v) isMigration(%v) errGetExtents(%v)",
inode, offset, size, storageClass, isMigration, errGetExtents)

49
sdk/meta/api.go
View File

@ -27,10 +27,12 @@ import (
"syscall"
"time"
"github.com/cubefs/cubefs/blobstore/util/taskpool"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/log"
"github.com/cubefs/cubefs/util/stat"
)
// Low-level API, i.e. work with inode
@ -61,6 +63,10 @@ const (
ForceUpdateRWMP = "ForceUpdateRWMP"
)
var (
getExtetnsPool = taskpool.New(50, 100)
)
func (mw *MetaWrapper) GetRootIno(subdir string) (uint64, error) {
rootIno, err := mw.LookupPath(subdir)
if err != nil {
@ -388,6 +394,11 @@ func (mw *MetaWrapper) BatchGetExpiredMultipart(prefix string, days int) (expire
return
}
// func (mw *MetaWrapper) InodeGet_llExt(inode uint64) (*proto.InodeInfo, error) {
// wg := sync.WaitGroup{}
// }
func (mw *MetaWrapper) InodeGet_ll(inode uint64) (*proto.InodeInfo, error) {
mp := mw.getPartitionByInode(inode)
if mp == nil {
@ -434,6 +445,44 @@ func (mw *MetaWrapper) doInodeGet(inode uint64) (*proto.InodeInfo, error) {
return info, nil
}
func (mw *MetaWrapper) BatchInodeGetExtents(inodes []uint64) []*proto.InodeInfo {
begin := stat.BeginStat()
defer stat.EndStat("BatchInodeGetExtents", nil, begin, 1)
infos := mw.BatchInodeGet(inodes)
wg := sync.WaitGroup{}
for _, info := range infos {
if !proto.IsRegular(info.Mode) {
continue
}
tmpInfo := info
wg.Add(1)
getExtetnsPool.Run(func() {
defer wg.Done()
mp := mw.getPartitionByInode(tmpInfo.Inode)
if mp == nil {
log.LogErrorf("BatchInodeGetExtents: no partition: ino(%v)", tmpInfo.Inode)
return
}
resp, err := mw.getExtents(mp, tmpInfo.Inode, false, false, false)
if err != nil {
log.LogErrorf("BatchInodeGetExtents: get extents fail: ino(%v) err(%v)", tmpInfo.Inode, err)
return
}
tmpInfo.Extents = resp
})
}
wg.Wait()
return infos
}
func (mw *MetaWrapper) BatchInodeGet(inodes []uint64) []*proto.InodeInfo {
var wg sync.WaitGroup