diff --git a/client/fs/dcache.go b/client/fs/dcache.go index 87e0b773a..82dd4bbfa 100644 --- a/client/fs/dcache.go +++ b/client/fs/dcache.go @@ -22,15 +22,17 @@ import ( // DentryCache defines the dentry cache. type DentryCache struct { sync.Mutex - cache map[string]uint64 - expiration time.Time + cache map[string]uint64 + expiration time.Time + acceleration bool } // NewDentryCache returns a new dentry cache. -func NewDentryCache() *DentryCache { +func NewDentryCache(acceleration bool) *DentryCache { return &DentryCache{ - cache: make(map[string]uint64), - expiration: time.Now().Add(DentryValidDuration), + cache: make(map[string]uint64), + expiration: time.Now().Add(DentryValidDuration), + acceleration: acceleration, } } @@ -41,6 +43,9 @@ func (dc *DentryCache) Put(name string, ino uint64) { } dc.Lock() defer dc.Unlock() + if dc.cache == nil { + dc.cache = make(map[string]uint64) + } dc.cache[name] = ino dc.expiration = time.Now().Add(DentryValidDuration) } @@ -53,7 +58,7 @@ func (dc *DentryCache) Get(name string) (uint64, bool) { dc.Lock() defer dc.Unlock() - if dc.expiration.Before(time.Now()) { + if dc.expiration.Before(time.Now()) && !dc.acceleration { dc.cache = make(map[string]uint64) return 0, false } diff --git a/client/fs/dir.go b/client/fs/dir.go index bd8d65924..993370df1 100644 --- a/client/fs/dir.go +++ b/client/fs/dir.go @@ -165,7 +165,7 @@ func (d *Dir) Release(ctx context.Context, req *fuse.ReleaseRequest) (err error) bgTime := stat.BeginStat() defer func() { stat.EndStat("Release:dir", nil, bgTime, 1) - log.LogDebugf("TRACE Release exit: ino(%v) name(%v)", d.info.Inode, d.name) + log.LogDebugf("TRACE DirRelease exit: ino(%v) name(%v)", d.info.Inode, d.name) }() if !d.super.metaCacheAcceleration { @@ -346,9 +346,14 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo stat.EndStat("Lookup", err, bgTime, 1) d.super.runningMonitor.SubClientOp(runningStat, err) }() - + if d.super.metaCacheAcceleration && (req.Name == ".AppleDouble" || strings.HasPrefix(req.Name, "._")) { + err = syscall.ENOENT + return nil, ParseError(err) + } log.LogDebugf("TRACE Lookup: parent(%v) req(%v)", d.info.Inode, req) - log.LogDebugf("TRACE Lookup: parent(%v) path(%v) d.super.bcacheDir(%v)", d.info.Inode, d.getCwd(), d.super.bcacheDir) + if log.EnableDebug() { + log.LogDebugf("TRACE Lookup: parent(%v) path(%v) d.super.bcacheDir(%v) miss(%v)", d.info.Inode, path.Join(d.getCwd(), req.Name), d.super.bcacheDir, atomic.LoadUint32(&d.missCount)) + } if d.needDentrycache() { dcachev2 = true @@ -381,6 +386,7 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo } else { cino, ok := d.dcache.Get(req.Name) if !ok { + log.LogDebugf("Lookup %v from parent %v miss, try to get from meta", path.Join(d.getCwd(), req.Name), d.info.Inode) cino, _, err = d.super.mw.Lookup_ll(d.info.Inode, req.Name) if err != nil { if err != syscall.ENOENT { @@ -420,14 +426,18 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo mode := proto.OsMode(info.Mode) if mode.IsDir() { d.super.mw.AddInoInfoCache(info.Inode, d.info.Inode, req.Name) - } else if mode.IsRegular() && missCache && d.super.metaCacheAcceleration && d.info.Nlink >= uint32(d.super.minimumNlinkReadDir) { + } + + if missCache && d.super.metaCacheAcceleration { now := timeutil.GetCurrentTime() if atomic.AddUint32(&d.missCount, 1) > 5 && (atomic.LoadInt64(&d.lastTime) == 0 || now.Sub(time.Unix(d.lastTime, 0)) >= 5*time.Minute) { + log.LogDebugf("trigger ReadDirAll for missCache %v Nlink %v missCount %v metaCacheAcceleration %v ino(%v) name(%v)", + missCache, d.info.Nlink, atomic.LoadUint32(&d.missCount), d.super.metaCacheAcceleration, d.info.Inode, d.getCwd()) atomic.StoreInt64(&d.lastTime, now.Unix()) atomic.StoreUint32(&d.missCount, 0) meta.GetExtetnsPool.Run(func() { log.LogDebugf("trigger ReadDirAll for ino(%v) name(%v)", d.info.Inode, d.getCwd()) - auditlog.LogClientOp("TriggerReadDirAll", d.getCwd(), "", err, time.Since(*bgTime).Microseconds(), ino, 0) + auditlog.LogClientOp("TriggerReadDirAllParent", d.getCwd(), "", err, time.Since(*bgTime).Microseconds(), ino, 0) d.ReadDirAll(context.Background()) }) } @@ -460,22 +470,25 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo } } d.super.fslock.Unlock() - - if d.dcache == nil { - d.dcache = NewDentryCache() + // maybe some dir never called ReadDir + if d.super.metaCacheAcceleration { + if d.dcache == nil { + d.dcache = NewDentryCache(d.super.metaCacheAcceleration) + } + log.LogDebugf("Lookup store %v %v to cache ", path.Join(d.getCwd(), req.Name), ino) + d.dcache.Put(req.Name, ino) } - - d.dcache.Put(req.Name, ino) // If subdirectories may be the directories where training data is located, // the ReadDirAll operation will be triggered when metaCacheAcceleration is enabled if mode.IsDir() && child.(*Dir).info.Nlink >= uint32(child.(*Dir).super.minimumNlinkReadDir) && child.(*Dir).super.metaCacheAcceleration { now := timeutil.GetCurrentTime() - if atomic.LoadInt64(&child.(*Dir).lastTime) == 0 || now.Sub(time.Unix(child.(*Dir).lastTime, 0)) >= 5*time.Minute { + if atomic.LoadInt64(&child.(*Dir).lastTime) == 0 || (now.Sub(time.Unix(child.(*Dir).lastTime, 0)) >= 2*time.Minute && atomic.LoadUint32(&d.missCount) > 5) { + log.LogDebugf("trigger ReadDirAll for lastTime %v Nlink %v metaCacheAcceleration %v ino(%v) name(%v)", + atomic.LoadInt64(&child.(*Dir).lastTime), child.(*Dir).info.Nlink, child.(*Dir).super.metaCacheAcceleration, child.(*Dir).info.Inode, child.(*Dir).getCwd()) atomic.StoreInt64(&child.(*Dir).lastTime, now.Unix()) - meta.GetExtetnsPool.Run(func() { log.LogDebugf("trigger ReadDirAll for ino(%v) name(%v)", child.(*Dir).info.Inode, child.(*Dir).getCwd()) - auditlog.LogClientOp("TriggerReadDirAll", child.(*Dir).getCwd(), "", err, time.Since(*bgTime).Microseconds(), ino, 0) + auditlog.LogClientOp("TriggerReadDirAllSub", child.(*Dir).getCwd(), "", err, time.Since(*bgTime).Microseconds(), ino, 0) child.(*Dir).ReadDirAll(context.Background()) }) } @@ -483,7 +496,7 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo resp.EntryValid = LookupValidDuration - log.LogDebugf("TRACE Lookup exit: parent(%v) req(%v) cost (%d)", d.info.Inode, req, time.Since(*bgTime).Microseconds()) + log.LogDebugf("TRACE Lookup exit: parent(%v) req(%v) cost (%v)", d.info.Inode, req, time.Since(*bgTime).String()) return child, nil } @@ -566,7 +579,7 @@ func (d *Dir) ReadDir(ctx context.Context, req *fuse.ReadRequest, resp *fuse.Rea log.LogDebugf("Readdir ino(%v) path(%v) d.super.bcacheDir(%v)", d.info.Inode, d.getCwd(), d.super.bcacheDir) var dcache *DentryCache if !d.super.disableDcache { - dcache = NewDentryCache() + dcache = NewDentryCache(d.super.metaCacheAcceleration) } if d.super.metaCacheAcceleration && d.dcache != nil { @@ -623,9 +636,6 @@ func (d *Dir) ReadDir(ctx context.Context, req *fuse.ReadRequest, resp *fuse.Rea elapsed := time.Since(start) log.LogDebugf("TRACE ReadDir exit: ino(%v) name(%v) dcache(%v) (%v)ns %v", d.info.Inode, d.name, d.dcache.Len(), elapsed.Nanoseconds(), req) - if d.super.metaCacheAcceleration { - atomic.StoreInt64(&d.lastTime, timeutil.GetCurrentTimeUnix()) - } return dirents, err } @@ -638,6 +648,7 @@ func (d *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) { defer func() { stat.EndStat("ReadDirAll", err, bgTime, 1) metric.SetWithLabels(err, map[string]string{exporter.Vol: d.super.volname}) + auditlog.LogClientOp("ReadDirAllComplete", d.getCwd(), "", err, time.Since(*bgTime).Microseconds(), d.info.Inode, 0) }() // transform ReadDirAll to ReadDirLimit_ll @@ -669,7 +680,7 @@ func (d *Dir) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) { log.LogDebugf("Readdir ino(%v) path(%v) d.super.bcacheDir(%v)", d.info.Inode, d.getCwd(), d.super.bcacheDir) var dcache *DentryCache if !d.super.disableDcache { - dcache = NewDentryCache() + dcache = NewDentryCache(d.super.metaCacheAcceleration) } var dcachev2 bool diff --git a/client/fs/file.go b/client/fs/file.go index 75c672970..5e4cf5e01 100644 --- a/client/fs/file.go +++ b/client/fs/file.go @@ -386,7 +386,7 @@ func (f *File) Release(ctx context.Context, req *fuse.ReleaseRequest) (err error return ParseError(err) } elapsed := time.Since(start) - log.LogDebugf("TRACE Release: ino(%v) req(%v) name(%v)(%v)ns", ino, req, path.Join(f.getParentPath(), f.name), elapsed.Nanoseconds()) + log.LogDebugf("TRACE FileRelease: ino(%v) req(%v) name(%v)(%v)ns", ino, req, path.Join(f.getParentPath(), f.name), elapsed.Nanoseconds()) return nil } diff --git a/client/fs/icache.go b/client/fs/icache.go index c48dacfea..16ab5f9de 100644 --- a/client/fs/icache.go +++ b/client/fs/icache.go @@ -92,7 +92,7 @@ func (ic *InodeCache) Get(ino uint64) *proto.InodeInfo { } info := element.Value.(*proto.InodeInfo) - if inodeExpired(info) && DisableMetaCache { + if inodeExpired(info) && DisableMetaCache && !ic.acceleration { ic.RUnlock() log.LogDebugf("Inode Cache %v expired", ino) return nil diff --git a/client/fs/super.go b/client/fs/super.go index e33a98c44..f77575919 100644 --- a/client/fs/super.go +++ b/client/fs/super.go @@ -162,14 +162,12 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) { } s.keepCache = opt.KeepCache + s.ic = NewInodeCache(inodeExpiration, int(opt.InodeLruLimit), s.metaCacheAcceleration) + s.inodeLruLimit = opt.InodeLruLimit if opt.MaxStreamerLimit > 0 || !opt.StopWarmMeta { - s.ic = NewInodeCache(inodeExpiration, int(opt.InodeLruLimit), s.metaCacheAcceleration) s.dc = NewDcache(inodeExpiration, MaxInodeCache) - s.inodeLruLimit = opt.InodeLruLimit } else { - s.ic = NewInodeCache(inodeExpiration, DefaultMaxInodeCache, s.metaCacheAcceleration) s.dc = NewDcache(inodeExpiration, DefaultMaxInodeCache) - s.inodeLruLimit = DefaultMaxInodeCache } s.orphan = NewOrphanInodeList() s.nodeCache = make(map[uint64]fs.Node) @@ -230,6 +228,7 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) { s.maxWarmUpConcurrency = opt.MaxWarmUpConcurrency s.stopWarmMeta = opt.StopWarmMeta s.metaCacheAcceleration = opt.MetaCacheAcceleration + s.minimumNlinkReadDir = opt.MinimumNlinkReadDir extentConfig := &stream.ExtentConfig{ Volume: opt.Volname, @@ -336,7 +335,9 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) { stat.PrintModuleStat = func(writer *bufio.Writer) { fmt.Fprintf(writer, "ic:%d dc:%d nodecache:%d dircache:%d\n", s.ic.lruList.Len(), s.dc.lruList.Len(), len(s.nodeCache), s.mw.DirCacheLen()) } - go s.loopSyncMeta() + if !s.metaCacheAcceleration { + go s.loopSyncMeta() + } // Start warm up meta paths goroutine go s.loopWarmUpMetaPaths() diff --git a/client/fuse.go b/client/fuse.go index f21f7d091..f6a87d3f2 100644 --- a/client/fuse.go +++ b/client/fuse.go @@ -406,7 +406,7 @@ func main() { stream.AheadReadConnPool.SetPoolArgs(int64(opt.TcpAliveTime), 0) stream.AheadReadConnPool.SetUseCostPool(true) stream.StreamWriteConnPool.SetPoolArgs(int64(opt.TcpAliveTime), 0) - fmt.Println(fmt.Sprintf("set TcpAliveTime %v", opt.TcpAliveTime)) + fmt.Printf("set TcpAliveTime %v\n", opt.TcpAliveTime) } level := parseLogLevel(opt.Loglvl) diff --git a/sdk/data/stream/extent_client.go b/sdk/data/stream/extent_client.go index b4a299374..d849d8953 100644 --- a/sdk/data/stream/extent_client.go +++ b/sdk/data/stream/extent_client.go @@ -646,7 +646,7 @@ func (client *ExtentClient) RefreshExtentsWithCache(inode *proto.InodeInfo) erro return nil } - s.extents.update(inode.Generation, inode.Size, false, inode.Extents.Extents) + s.extents.update(inode.Extents.Generation, inode.Extents.Size, false, inode.Extents.Extents) return nil } diff --git a/sdk/meta/api.go b/sdk/meta/api.go index d61f81478..c23dce4cd 100644 --- a/sdk/meta/api.go +++ b/sdk/meta/api.go @@ -63,9 +63,7 @@ const ( ForceUpdateRWMP = "ForceUpdateRWMP" ) -var ( - GetExtetnsPool = taskpool.New(50, 100) -) +var GetExtetnsPool = taskpool.New(50, 100) func (mw *MetaWrapper) GetRootIno(subdir string) (uint64, error) { rootIno, err := mw.LookupPath(subdir) @@ -395,10 +393,9 @@ func (mw *MetaWrapper) BatchGetExpiredMultipart(prefix string, days int) (expire } func (mw *MetaWrapper) InodeGetExt_ll(inode uint64) (*proto.InodeInfo, error) { - start := time.Now() defer func() { - log.LogDebugf("InodeGetExt_ll: time cost (%v), ino(%d)", time.Since(start), inode) + log.LogDebugf("InodeGetExt_ll: time cost (%v), ino(%v)", time.Since(start), inode) }() wg := sync.WaitGroup{} @@ -414,7 +411,6 @@ func (mw *MetaWrapper) InodeGetExt_ll(inode uint64) (*proto.InodeInfo, error) { log.LogErrorf("InodeGetExt_ll: get inode fail: ino(%v) err(%v)", inode, getErr) return } - }() var resp *proto.GetExtentsResponse diff --git a/sdk/meta/operation.go b/sdk/meta/operation.go index 8c2cd1182..8bbfea73d 100644 --- a/sdk/meta/operation.go +++ b/sdk/meta/operation.go @@ -1268,6 +1268,7 @@ func (mw *MetaWrapper) getExtents(mp *MetaPartition, inode uint64, isCache bool, bgTime := stat.BeginStat() defer func() { stat.EndStat("getExtents", err, bgTime, 1) + log.LogDebugf("getExtents for inode(%v) cost(%v)", inode, time.Since(*bgTime).String()) }() req := &proto.GetExtentsRequest{ diff --git a/util/conn_pool.go b/util/conn_pool.go index 78fe8be71..2954eb908 100644 --- a/util/conn_pool.go +++ b/util/conn_pool.go @@ -142,7 +142,6 @@ func (cp *ConnectPool) PutConnect(c *net.TCPConn, forceClose bool) { } func (cp *ConnectPool) PutConnectV2(c *net.TCPConn, forceClose bool, addr string, cost int64) { - if c == nil { return } @@ -367,7 +366,6 @@ func NewPoolWithCost(min, max int, timeout, connectTimeout int64, target string, } func (p *PoolWithCost) PutConnectObjectToPool(o *Object) { - if log.EnableDebug() { log.LogDebugf("PoolWithCost PutConnectObjectToPool start") } @@ -393,11 +391,9 @@ func (p *PoolWithCost) PutConnectObjectToPool(o *Object) { } p.conns.PushBack(o) - } func (p *PoolWithCost) autoRelease() { - if log.EnableDebug() { log.LogDebugf("PoolWithCost autoRelease start") } @@ -444,11 +440,9 @@ func (p *PoolWithCost) ReleaseAll() { o.conn.Close() p.conns.Remove(e) } - } func (p *PoolWithCost) GetConnectFromPool() (c *net.TCPConn, err error) { - if log.EnableDebug() { log.LogDebugf("PoolWithCost GetConnectFromPool start") }