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(sdk): enable config bcache only for clod data.#22770439

Browse Source 
Signed-off-by: baihailong <baihailong@oppo.com>
This commit is contained in:
baihailong 2024-11-11 14:19:32 +08:00 committed by AmazingChi
parent 531fe6abec
commit 4ab0642da0
5 changed files with 44 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
client/fs/super.go
View File

@ -230,6 +230,7 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
ReadRate: opt.ReadRate,
WriteRate: opt.WriteRate,
BcacheEnable: opt.EnableBcache,
BcacheOnlyForCold: opt.BcacheOnlyForCold,
BcacheDir: opt.BcacheDir,
MaxStreamerLimit: opt.MaxStreamerLimit,
VerReadSeq: opt.VerReadSeq,
@ -250,6 +251,8 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
VolAllowedStorageClass: opt.VolAllowedStorageClass,
VolCacheDpStorageClass: s.cacheDpStorageClass,
OnForbiddenMigration: s.mw.ForbiddenMigration,
OnGetInodeInfo: s.InodeGet,
}
s.ec, err = stream.NewExtentClient(extentConfig)

3
client/fuse.go
View File

@ -532,6 +532,7 @@ func main() {
defer super.Close()
syslog.Printf("enable bcache %v", opt.EnableBcache)
syslog.Printf("bcache only for cold %v", opt.BcacheOnlyForCold)
if cfg.GetString(exporter.ConfigKeyPushAddr) == "" {
pushAddr, err := getPushAddrFromMaster(opt.Master)
@ -945,6 +946,8 @@ func parseMountOption(cfg *config.Config) (*proto.MountOptions, error) {
opt.EnableBcache = true
}
opt.BcacheOnlyForCold = GlobalMountOptions[proto.BcacheOnlyForCold].GetBool()
if opt.Rdonly {
verReadSeq := GlobalMountOptions[proto.SnapshotReadVerSeq].GetInt64()
if verReadSeq == -1 {

3
proto/mount_options.go
View File

@ -78,6 +78,7 @@ const (
DisableMountSubtype
StreamRetryTimeOut
BufferChanSize
BcacheOnlyForCold
MaxMountOption
)
@ -174,6 +175,7 @@ func InitMountOptions(opts []MountOption) {
opts[SnapshotReadVerSeq] = MountOption{"snapshotReadSeq", "Snapshot read seq", "", int64(0)} // default false
opts[DisableMountSubtype] = MountOption{"disableMountSubtype", "Disable Mount Subtype", "", false}
opts[StreamRetryTimeOut] = MountOption{"streamRetryTimeout", "max stream retry timeout, s", "", int64(0)}
opts[BcacheOnlyForCold] = MountOption{"enableBcacheOnlyForCold", "Enable block cache only for cold data", "", false}
for i := 0; i < MaxMountOption; i++ {
flag.StringVar(&opts[i].cmdlineValue, opts[i].keyword, "", opts[i].description)
@ -316,6 +318,7 @@ type MountOptions struct {
CacheThreshold int
EbsBlockSize int
EnableBcache bool
BcacheOnlyForCold bool
BcacheDir string
BcacheFilterFiles string
BcacheCheckIntervalS int64

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

@ -68,6 +68,7 @@ type (
EvictBacheFunc func(key string) error
RenewalForbiddenMigrationFunc func(inode uint64) error
ForbiddenMigrationFunc func(inode uint64) error
GetInodeInfoFunc func(ino uint64) (*proto.InodeInfo, error)
)
const (
@ -128,6 +129,7 @@ type ExtentConfig struct {
ReadRate int64
WriteRate int64
BcacheEnable bool
BcacheOnlyForCold bool
BcacheDir string
MaxStreamerLimit int64
VerReadSeq uint64
@ -150,6 +152,8 @@ type ExtentConfig struct {
VolStorageClass uint32
VolAllowedStorageClass []uint32
VolCacheDpStorageClass uint32
OnGetInodeInfo GetInodeInfoFunc
}
type MultiVerMgr struct {
@ -172,6 +176,7 @@ type ExtentClient struct {
volumeType int
volumeName string
bcacheEnable bool
bcacheOnlyForCold bool
bcacheDir string
BcacheHealth bool
preload bool
@ -192,6 +197,7 @@ type ExtentClient struct {
renewalForbiddenMigration RenewalForbiddenMigrationFunc
forbiddenMigration ForbiddenMigrationFunc
CacheDpStorageClass uint32
getInodeInfo GetInodeInfoFunc
}
func (client *ExtentClient) UidIsLimited(uid uint32) bool {
@ -309,6 +315,7 @@ retry:
client.evictBcache = config.OnEvictBcache
client.volumeName = config.Volume
client.bcacheEnable = config.BcacheEnable
client.bcacheOnlyForCold = config.BcacheOnlyForCold
client.bcacheDir = config.BcacheDir
client.multiVerMgr.verReadSeq = client.dataWrapper.GetReadVerSeq()
client.BcacheHealth = true
@ -317,6 +324,7 @@ retry:
client.renewalForbiddenMigration = config.OnRenewalForbiddenMigration
client.CacheDpStorageClass = config.VolCacheDpStorageClass
client.forbiddenMigration = config.OnForbiddenMigration
client.getInodeInfo = config.OnGetInodeInfo
if config.StreamRetryTimeout <= 0 || config.StreamRetryTimeout >= 600 {
client.streamRetryTimeout = StreamSendMaxTimeout

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

@ -205,21 +205,30 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
}
// skip hole,ek is not nil,read block cache firstly
log.LogDebugf("Stream read: ino(%v) req(%v) s.client.bcacheEnable(%v) s.needBCache(%v)", s.inode, req, s.client.bcacheEnable, s.needBCache)
log.LogDebugf("Stream read: ino(%v) req(%v) s.client.bcacheEnable(%v) s.client.bcacheOnlyForCold(%v) s.needBCache(%v)",
s.inode, req, s.client.bcacheEnable, s.client.bcacheOnlyForCold, s.needBCache)
cacheKey := util.GenerateRepVolKey(s.client.volumeName, s.inode, req.ExtentKey.PartitionId, req.ExtentKey.ExtentId, req.ExtentKey.FileOffset)
if s.client.bcacheEnable && s.needBCache && filesize <= bcache.MaxFileSize {
offset := req.FileOffset - int(req.ExtentKey.FileOffset)
if s.client.loadBcache != nil {
readBytes, err = s.client.loadBcache(cacheKey, req.Data, uint64(offset), uint32(req.Size))
if err == nil && readBytes == req.Size {
total += req.Size
bcacheMetric := exporter.NewCounter("fileReadL1CacheHit")
bcacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
log.LogDebugf("TRACE Stream read. hit blockCache: ino(%v) cacheKey(%v) readBytes(%v) err(%v)", s.inode, cacheKey, readBytes, err)
continue
}
inodeInfo, err := s.client.getInodeInfo(s.inode)
if err != nil {
log.LogErrorf("Streamer read: getInodeInfo failed. ino(%v) req(%v) err(%v)", s.inode, req, err)
return 0, err
}
if !s.client.bcacheOnlyForCold || (s.client.bcacheOnlyForCold && inodeInfo.StorageClass != proto.StorageClass_Replica_SSD) {
offset := req.FileOffset - int(req.ExtentKey.FileOffset)
if s.client.loadBcache != nil {
readBytes, err = s.client.loadBcache(cacheKey, req.Data, uint64(offset), uint32(req.Size))
if err == nil && readBytes == req.Size {
total += req.Size
bcacheMetric := exporter.NewCounter("fileReadL1CacheHit")
bcacheMetric.AddWithLabels(1, map[string]string{exporter.Vol: s.client.volumeName})
log.LogDebugf("TRACE Stream read. hit blockCache: ino(%v) storageClass(%v) cacheKey(%v) readBytes(%v) err(%v)",
s.inode, inodeInfo.StorageClass, cacheKey, readBytes, err)
continue
}
}
log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) loadBcache(%v)", cacheKey, s.client.loadBcache)
}
log.LogDebugf("TRACE Stream read. miss blockCache cacheKey(%v) loadBcache(%v)", cacheKey, s.client.loadBcache)
}
if s.needBCache {
@ -235,10 +244,15 @@ func (s *Streamer) read(data []byte, offset int, size int, storageClass uint32)
}
if s.client.bcacheEnable && s.needBCache && filesize <= bcache.MaxFileSize {
inodeInfo, err := s.client.getInodeInfo(s.inode)
if err != nil {
log.LogErrorf("Streamer read: getInodeInfo failed. ino(%v) req(%v) err(%v)", s.inode, req, err)
return 0, err
}
// limit big block cache
if s.exceedBlockSize(req.ExtentKey.Size) && atomic.LoadInt32(&s.client.inflightL1BigBlock) > 10 {
// do nothing
} else {
} else if !s.client.bcacheOnlyForCold || (s.client.bcacheOnlyForCold && inodeInfo.StorageClass != proto.StorageClass_Replica_SSD) {
select {
case s.pendingCache <- bcacheKey{cacheKey: cacheKey, extentKey: req.ExtentKey, storageClass: storageClass}:
if s.exceedBlockSize(req.ExtentKey.Size) {