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): add minAheadReadSize

Browse Source 
close:#1000517044

Signed-off-by: chihe <chihe@oppo.com>
(cherry picked from commit 6abbbe772e)
This commit is contained in:
chihe 2025-12-18 10:57:38 +08:00
parent a02b6e578b
commit 85d119ca04
5 changed files with 22 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
client/fs/super.go
View File

@ -266,6 +266,7 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
AheadReadTotalMem: opt.AheadReadTotalMem,
AheadReadBlockTimeOut: opt.AheadReadBlockTimeOut,
AheadReadWindowCnt: opt.AheadReadWindowCnt,
MinReadAheadSize: int(opt.MinReadAheadSize),
NeedRemoteCache: true,
ForceRemoteCache: opt.ForceRemoteCache,
EnableAsyncFlush: opt.EnableAsyncFlush,

2
client/fuse.go
View File

@ -1045,6 +1045,8 @@ func parseMountOption(cfg *config.Config) (*proto.MountOptions, error) {
fmt.Printf("available ahead read mem: %v\n", available)
}
}
// minimum file size (bytes) to trigger ahead read
opt.MinReadAheadSize = GlobalMountOptions[proto.MinReadAheadSize].GetInt64()
opt.ReadDirLimit = GlobalMountOptions[proto.ReadDirLimit].GetInt64()
opt.MaxWarmUpConcurrency = GlobalMountOptions[proto.MaxWarmUpConcurrency].GetInt64()
opt.StopWarmMeta = GlobalMountOptions[proto.StopWarmMeta].GetBool()

4
proto/mount_options.go
View File

@ -83,6 +83,8 @@ const (
AheadReadTotalMemGB
AheadReadBlockTimeOut
AheadReadWindowCnt
// min read ahead size
MinReadAheadSize
ReqChanCnt
// remotecache
ForceRemoteCache
@ -196,6 +198,7 @@ func InitMountOptions(opts []MountOption) {
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(128)}
opts[MinReadAheadSize] = MountOption{"minReadAheadSize", "minimum file size to trigger ahead read (bytes)", "", int64(1048576)} // default 1MB
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}
@ -397,4 +400,5 @@ type MountOptions struct {
MinimumNlinkReadDir int64
InodeLruLimit int64
FuseServeThreads int64
MinReadAheadSize int64
}

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

@ -163,6 +163,7 @@ type ExtentConfig struct {
AheadReadTotalMem int64
AheadReadBlockTimeOut int
AheadReadWindowCnt int
MinReadAheadSize int
// remoteCache
NeedRemoteCache bool
ForceRemoteCache bool
@ -903,6 +904,7 @@ func (client *ExtentClient) GetStreamer(inode uint64) *Streamer {
go s.server()
go s.asyncBlockCache()
go s.asyncFlushManager()
if client.AheadRead != nil && s.aheadReadWindow != nil {
go s.aheadReadWindow.backgroundAheadReadTask()
}

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

@ -85,6 +85,8 @@ type Streamer struct {
aheadReadBlockSize uint32
waitForFlush bool
// minimum file size to trigger ahead read (bytes)
minReadAheadSize int
}
type bcacheKey struct {
@ -131,8 +133,13 @@ func NewStreamer(client *ExtentClient, inode uint64, openForWrite, isCache bool,
}
if client.AheadRead != nil {
s.aheadReadEnable = client.AheadRead.enable
s.aheadReadWindow = NewAheadReadWindow(client.AheadRead, s)
s.aheadReadBlockSize = util.CacheReadBlockSize
// set min read ahead size from config, default 1MB when zero
if client.extentConfig != nil && client.extentConfig.MinReadAheadSize > 0 {
s.minReadAheadSize = client.extentConfig.MinReadAheadSize
} else {
s.minReadAheadSize = util.MB // 1MB default
}
}
go s.server()
go s.asyncBlockCache()
@ -264,7 +271,11 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
} else {
log.LogDebugf("Stream read: ino(%v) req(%v) s.needBCache(%v) s.client.bcacheEnable(%v) aheadReadEnable(%v) aheadReadBlockSize(%v) %p",
s.inode, req, s.needBCache, s.client.bcacheEnable, s.aheadReadEnable, s.aheadReadBlockSize, s)
if s.aheadReadEnable {
if s.aheadReadEnable && filesize > s.minReadAheadSize {
// Lazily initialize ahead read window when threshold is satisfied
if s.aheadReadWindow == nil && s.client.AheadRead != nil {
s.aheadReadWindow = NewAheadReadWindow(s.client.AheadRead, s)
}
bgTime := stat.BeginStat()
readBytes, err = s.aheadRead(req, storageClass)
if err == nil && readBytes == req.Size {