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 rightOffset to make window for aheadRead move forward

Browse Source 
@formatter:off

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-10-30 15:22:36 +08:00 committed by 曾雪伟
parent d4ecea0527
commit dcaeee8970
2 changed files with 11 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
proto/mount_options.go
View File

@ -191,7 +191,7 @@ func InitMountOptions(opts []MountOption) {
opts[AheadReadEnable] = MountOption{"aheadReadEnable", "enable ahead read", "", false}
opts[AheadReadTotalMemGB] = MountOption{"aheadReadTotalMemGB", "ahead read total mem(GB)", "", int64(30)}
opts[AheadReadBlockTimeOut] = MountOption{"aheadReadBlockTimeOut", "ahead read block expiration time", "", int64(3)}
opts[AheadReadWindowCnt] = MountOption{"aheadReadWindowCnt", "ahead read window block count", "", int64(8)}
opts[AheadReadWindowCnt] = MountOption{"aheadReadWindowCnt", "ahead read window block count", "", int64(128)}
opts[ForceRemoteCache] = MountOption{"forceRemoteCache", "All read requests are handled by the remote cache.", "", false}
opts[DebugCluster] = MountOption{"debugCluster", "display cluster name", "", ""}
opts[EnableAsyncFlush] = MountOption{"enableAsyncFlush", "async flush extent handler", "", true}

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

@ -509,6 +509,7 @@ func (client *ExtentClient) OpenStream(inode uint64, openForWrite, isCache bool,
if !ok {
s = NewStreamer(client, inode, openForWrite, isCache, fullPath)
client.streamers[inode] = s
log.LogDebugf("action[OpenStream] create new streamer for ino(%v) %p", inode, s)
} else {
// If you open a file in write mode first and then open the same file
// in read mode without modifying any attributes, maintaining the file's immutability status.
@ -516,6 +517,11 @@ func (client *ExtentClient) OpenStream(inode uint64, openForWrite, isCache bool,
s.openForWrite = openForWrite
}
// TODO: update isCache?
log.LogDebugf("action[OpenStream] reuse streamer for ino(%v)%p", inode, s)
// update rightOffset to make window for aheadRead move forward
if s.aheadReadWindow != nil {
s.aheadReadWindow.rightOffset = 0
}
}
return s.IssueOpenRequest()
}
@ -554,6 +560,10 @@ func (client *ExtentClient) OpenStreamWithCache(inode uint64, needBCache, openFo
if !client.disableMetaCache && needBCache {
client.streamerList.PushFront(inode)
}
} else {
if s.aheadReadWindow != nil {
s.aheadReadWindow.rightOffset = 0
}
}
s.needBCache = needBCache
if !s.isOpen && !client.disableMetaCache {