From 4ab0642da0d55dd9a2a032a4e8446025ab7f0e5f Mon Sep 17 00:00:00 2001 From: baihailong Date: Mon, 11 Nov 2024 14:19:32 +0800 Subject: [PATCH] fix(sdk): enable config bcache only for clod data.#22770439 Signed-off-by: baihailong --- client/fs/super.go | 3 +++ client/fuse.go | 3 +++ proto/mount_options.go | 3 +++ sdk/data/stream/extent_client.go | 8 +++++++ sdk/data/stream/stream_reader.go | 40 +++++++++++++++++++++----------- 5 files changed, 44 insertions(+), 13 deletions(-) diff --git a/client/fs/super.go b/client/fs/super.go index 51a567826..2cb30786c 100644 --- a/client/fs/super.go +++ b/client/fs/super.go @@ -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) diff --git a/client/fuse.go b/client/fuse.go index 8ba722426..13a4c8d65 100644 --- a/client/fuse.go +++ b/client/fuse.go @@ -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 { diff --git a/proto/mount_options.go b/proto/mount_options.go index 9b531dbd7..dade726c4 100644 --- a/proto/mount_options.go +++ b/proto/mount_options.go @@ -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 diff --git a/sdk/data/stream/extent_client.go b/sdk/data/stream/extent_client.go index a3a21af96..fd1b5aaf5 100644 --- a/sdk/data/stream/extent_client.go +++ b/sdk/data/stream/extent_client.go @@ -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 diff --git a/sdk/data/stream/stream_reader.go b/sdk/data/stream/stream_reader.go index 7c2bae0c8..351fcd18c 100644 --- a/sdk/data/stream/stream_reader.go +++ b/sdk/data/stream/stream_reader.go @@ -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) {