From 30b29217a15a251ddea89ece56d54f4b385b4c4f Mon Sep 17 00:00:00 2001 From: Wu Huocheng Date: Thu, 17 Apr 2025 17:39:04 +0800 Subject: [PATCH] feat(datanode): evict the extent cache by ttl.#1000005968 Signed-off-by: Wu Huocheng --- datanode/partition.go | 7 +++++++ datanode/server.go | 8 ++++++++ datanode/space_manager.go | 17 ++++++++++++++++ datanode/stat_log.go | 11 ++++++++--- datanode/storage/extent.go | 8 ++++---- datanode/storage/extent_cache.go | 34 ++++++++++++++++++++++++++++++++ datanode/storage/extent_store.go | 4 ++++ 7 files changed, 82 insertions(+), 7 deletions(-) diff --git a/datanode/partition.go b/datanode/partition.go index 03240291e..b3e58f91f 100644 --- a/datanode/partition.go +++ b/datanode/partition.go @@ -1743,3 +1743,10 @@ func (dp *DataPartition) setChangeMemberWaiting() bool { func (dp *DataPartition) setRestoreReplicaFinish() bool { return atomic.CompareAndSwapUint32(&dp.responseStatus, responseWait, responseInitial) } + +func (dp *DataPartition) EvictExtentCache() { + err := dp.extentStore.DoExtentCacheTtl(dp.dataNode.ExtentCacheTtlByHour) + if err != nil { + log.LogWarnf("[EvictExtentCache] DoExtentCacheTtl err: %s", err.Error()) + } +} diff --git a/datanode/server.go b/datanode/server.go index 60959376f..812243ce0 100644 --- a/datanode/server.go +++ b/datanode/server.go @@ -71,6 +71,7 @@ const ( DefaultDiskUnavailableErrorCount = 5 DefaultDiskUnavailablePartitionErrorCount = 3 DefaultGOGCValue = 100 + DefaultExtentCacheTtlByHour = 1 ) const ( @@ -136,6 +137,7 @@ const ( // disk status becomes unavailable if disk error partition count reaches this value ConfigKeyDiskUnavailablePartitionErrorCount = "diskUnavailablePartitionErrorCount" ConfigKeyCacheCap = "cacheCap" + ConfigExtentCacheTtlByHour = "extentCacheTtlByHour" // storage device media type, for hybrid cloud, in string: SDD or HDD ConfigMediaType = "mediaType" @@ -220,6 +222,7 @@ type DataNode struct { nodeForbidWriteOpOfProtoVer0 bool // whether forbid by node granularity, VolsForbidWriteOpOfProtoVer0 map[string]struct{} // whether forbid by volume granularity, DirectReadVols map[string]struct{} + ExtentCacheTtlByHour int } type verOp2Phase struct { @@ -474,6 +477,8 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) { } s.mediaType = mediaType + s.ExtentCacheTtlByHour = cfg.GetIntWithDefault(ConfigExtentCacheTtlByHour, DefaultExtentCacheTtlByHour) + log.LogDebugf("action[parseConfig] load masterAddrs(%v).", MasterClient.Nodes()) log.LogDebugf("action[parseConfig] load port(%v).", s.port) log.LogDebugf("action[parseConfig] load zoneName(%v).", s.zoneName) @@ -666,6 +671,9 @@ func (s *DataNode) startSpaceManager(cfg *config.Config) (err error) { s.space.StartDiskSample() s.updateQosLimit() // load from config s.markAllDiskLoaded() + + go s.space.StartEvictExtentCache() + return nil } diff --git a/datanode/space_manager.go b/datanode/space_manager.go index a899acfbb..3a10f69b3 100644 --- a/datanode/space_manager.go +++ b/datanode/space_manager.go @@ -960,3 +960,20 @@ func (manager *SpaceManager) getPartitions() []*DataPartition { } return partitions } + +func (manager *SpaceManager) StartEvictExtentCache() { + ticker := time.NewTicker(time.Hour) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + partitions := manager.getPartitions() + for _, partition := range partitions { + partition.EvictExtentCache() + } + case <-manager.stopC: + return + } + } +} diff --git a/datanode/stat_log.go b/datanode/stat_log.go index 90ba488d1..5e6b1ae2a 100644 --- a/datanode/stat_log.go +++ b/datanode/stat_log.go @@ -15,12 +15,17 @@ import ( func (d *DataNode) startStat(cfg *config.Config) { logDir := cfg.GetString(ConfigKeyLogDir) var err error + var logLeftSpaceLimitRatio float64 logLeftSpaceLimitRatioStr := cfg.GetString("logLeftSpaceLimitRatio") - logLeftSpaceLimitRatio, err := strconv.ParseFloat(logLeftSpaceLimitRatioStr, 64) - if err != nil { - log.LogWarnf("get log limt ratio failed, err %s", err.Error()) + if logLeftSpaceLimitRatioStr == "" { logLeftSpaceLimitRatio = log.DefaultLogLeftSpaceLimitRatio + } else { + logLeftSpaceLimitRatio, err = strconv.ParseFloat(logLeftSpaceLimitRatioStr, 64) + if err != nil { + log.LogWarnf("get log limt ratio failed, err %s", err.Error()) + logLeftSpaceLimitRatio = log.DefaultLogLeftSpaceLimitRatio + } } stat.DpStat, err = stat.NewOpLogger(logDir, "dp_op.log", stat.DefaultMaxOps, stat.DefaultDuration, logLeftSpaceLimitRatio) diff --git a/datanode/storage/extent.go b/datanode/storage/extent.go index 717095b7b..f13654edf 100644 --- a/datanode/storage/extent.go +++ b/datanode/storage/extent.go @@ -35,6 +35,7 @@ import ( "github.com/cubefs/cubefs/util" "github.com/cubefs/cubefs/util/atomicutil" "github.com/cubefs/cubefs/util/log" + "github.com/cubefs/cubefs/util/timeutil" ) const ( @@ -344,8 +345,8 @@ func (e *Extent) InitToFS() (err error) { e.dataSize = 0 return } - atomic.StoreInt64(&e.modifyTime, time.Now().Unix()) - atomic.StoreInt64(&e.accessTime, time.Now().Unix()) + atomic.StoreInt64(&e.modifyTime, timeutil.GetCurrentTimeUnix()) + atomic.StoreInt64(&e.accessTime, timeutil.GetCurrentTimeUnix()) e.dataSize = 0 return } @@ -451,8 +452,7 @@ func (e *Extent) RestoreFromFS() (err error) { atomic.StoreInt64(&e.modifyTime, info.ModTime().Unix()) - ts := info.Sys().(*syscall.Stat_t) - atomic.StoreInt64(&e.accessTime, time.Unix(int64(ts.Atim.Sec), int64(ts.Atim.Nsec)).Unix()) + atomic.StoreInt64(&e.accessTime, timeutil.GetCurrentTimeUnix()) return } diff --git a/datanode/storage/extent_cache.go b/datanode/storage/extent_cache.go index 75cf8f085..50ab0dbff 100644 --- a/datanode/storage/extent_cache.go +++ b/datanode/storage/extent_cache.go @@ -17,9 +17,11 @@ package storage import ( "container/list" "sync" + "sync/atomic" "time" "github.com/cubefs/cubefs/util/log" + "github.com/cubefs/cubefs/util/timeutil" ) const ( @@ -91,6 +93,7 @@ func (cache *ExtentCache) Get(extentID uint64) (e *Extent, ok bool) { cache.extentList.MoveToBack(item.element) } e = item.e + atomic.StoreInt64(&e.accessTime, timeutil.GetCurrentTimeUnix()) } return } @@ -219,3 +222,34 @@ func (cache *ExtentCache) Flush() { item.e.Flush() } } + +func (cache *ExtentCache) EvictTTL(ttl int) error { + cmpTime := timeutil.GetCurrentTimeUnix() - int64(ttl)*int64(time.Hour) + if cache.capacity <= 0 || cmpTime <= 0 { + return nil + } + + cache.lock.Lock() + defer cache.lock.Unlock() + + delList := make([]*list.Element, 0, cache.extentList.Len()) + + for e := cache.extentList.Front(); e != nil; e = e.Next() { + item := e.Value.(*Extent) + if IsTinyExtent(item.extentID) { + continue + } + if item.accessTime < cmpTime { + element := e + delList = append(delList, element) + delete(cache.extentMap, item.extentID) + item.Close() + } + } + + for _, element := range delList { + cache.extentList.Remove(element) + } + + return nil +} diff --git a/datanode/storage/extent_store.go b/datanode/storage/extent_store.go index 33d64f10a..8867e0993 100644 --- a/datanode/storage/extent_store.go +++ b/datanode/storage/extent_store.go @@ -1836,3 +1836,7 @@ func (s *ExtentStore) ExtentBatchUnlockNormalExtent(ext []*proto.ExtentKey) { func (s *ExtentStore) GetExtentCountWithoutLock() (count int) { return len(s.extentInfoMap) } + +func (s *ExtentStore) DoExtentCacheTtl(ttl int) error { + return s.cache.EvictTTL(ttl) +}