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): quit asyncFluashManager when streamer is closed

Browse Source 
@formatter:off

Signed-off-by: chihe <chihe@oppo.com>
(cherry picked from commit 27a5ac6e57)
This commit is contained in:
chihe 2025-11-20 10:03:29 +08:00
parent 3b2ffbcf7a
commit 698b727acb
3 changed files with 9 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
client/fs/dir.go
View File

@ -482,21 +482,6 @@ func (d *Dir) Lookup(ctx context.Context, req *fuse.LookupRequest, resp *fuse.Lo
}
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)) >= 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("TriggerReadDirAllSub", child.(*Dir).getCwd(), "", err, time.Since(*bgTime).Microseconds(), ino, 0)
child.(*Dir).ReadDirAll(context.Background())
})
}
}
resp.EntryValid = LookupValidDuration

8
sdk/data/stream/extent_dirtylist.go
View File

@ -44,7 +44,9 @@ func (dl *DirtyExtentList) Put(eh *ExtentHandler) {
return
}
dl.list.PushBack(eh)
log.LogDebugf("DirtyExtentList: put handler(%v) to dirtyList trace(%v)", eh, string(debug.Stack()))
if log.EnableDebug() {
log.LogDebugf("DirtyExtentList: put handler(%v) to dirtyList trace(%v)", eh, string(debug.Stack()))
}
}
// Get gets the next element in the dirty extent list.
@ -61,7 +63,9 @@ func (dl *DirtyExtentList) Remove(e *list.Element) {
if e != nil {
if eh, ok := e.Value.(*ExtentHandler); ok {
dl.index.Delete(eh.id)
log.LogDebugf("DirtyExtentList: remove handler(%v) to dirtyList trace(%v)", eh, string(debug.Stack()))
if log.EnableDebug() {
log.LogDebugf("DirtyExtentList: remove handler(%v) to dirtyList trace(%v)", eh, string(debug.Stack()))
}
}
}
dl.list.Remove(e)

11
sdk/data/stream/stream_reader.go
View File

@ -475,7 +475,6 @@ func (s *Streamer) UpdateStringPath(fullPath string) {
// asyncFlushManager manages asynchronous flush operations using channel-based producer-consumer pattern
func (s *Streamer) asyncFlushManager() {
log.LogDebugf("asyncFlushManager: started for streamer(%v)", s)
idle := 0
t := time.NewTicker(2 * time.Second)
defer t.Stop()
for {
@ -492,7 +491,6 @@ func (s *Streamer) asyncFlushManager() {
if req == nil {
continue
}
idle = 0
// Check if we should stop processing new requests
select {
case <-s.asyncFlushDone:
@ -528,12 +526,9 @@ func (s *Streamer) asyncFlushManager() {
log.LogDebugf("rdonly stream no need to start asyncFlushManager routine. ino %d", s.inode)
return
}
if atomic.LoadInt32(&s.refcnt) <= 0 {
if idle >= streamWriterIdleTimeoutPeriod && len(s.asyncFlushCh) == 0 {
log.LogDebugf("done asyncFlushManager: no requests for a long time, ino(%v), streamer(%v)", s.inode, s)
return
}
idle++
if !s.isOpen && len(s.asyncFlushCh) == 0 {
log.LogDebugf("asyncFlushManager is done for streamer(%v) closed", s.inode)
return
}
}
}