diff --git a/lcnode/const.go b/lcnode/const.go index c6ce073a5..5ca3f2995 100644 --- a/lcnode/const.go +++ b/lcnode/const.go @@ -22,13 +22,12 @@ import ( ) const ( - configListen = proto.ListenPort - configMasterAddr = proto.MasterAddr - configBatchExpirationGetNumStr = "batchExpirationGetNum" - configScanCheckIntervalStr = "scanCheckInterval" - configLcScanRoutineNumPerTaskStr = "lcScanRoutineNumPerTask" - configLcScanLimitPerSecondStr = "lcScanLimitPerSecond" - + configListen = proto.ListenPort + configMasterAddr = proto.MasterAddr + configSimpleQueueInitCapacityStr = "simpleQueueInitCapacity" + configScanCheckIntervalStr = "scanCheckInterval" + configLcScanRoutineNumPerTaskStr = "lcScanRoutineNumPerTask" + configLcScanLimitPerSecondStr = "lcScanLimitPerSecond" configSnapshotRoutineNumPerTaskStr = "snapshotRoutineNumPerTask" configLcNodeTaskCountLimit = "lcNodeTaskCountLimit" configDelayDelMinute = "delayDelMinute" @@ -37,19 +36,16 @@ const ( // Default of configuration value const ( - defaultListen = "80" - ModuleName = "lcNode" - defaultBatchExpirationGetNum = 100 - maxBatchExpirationGetNum = 10000 - defaultScanCheckInterval = 60 - defaultLcScanRoutineNumPerTask = 100 - defaultLcScanLimitPerSecond = rate.Inf - defaultLcScanLimitBurst = 1000 - - maxLcScanRoutineNumPerTask = 5000 - defaultReadDirLimit = 1000 - + defaultListen = "80" + ModuleName = "lcNode" + defaultReadDirLimit = 1000 + defaultScanCheckInterval = 60 + defaultLcScanRoutineNumPerTask = 20 + maxLcScanRoutineNumPerTask = 500 + defaultLcScanLimitPerSecond = rate.Inf + defaultLcScanLimitBurst = 1000 defaultUnboundedChanInitCapacity = 10000 + defaultSimpleQueueInitCapacity = 1000000 defaultLcNodeTaskCountLimit = 1 maxLcNodeTaskCountLimit = 20 defaultDelayDelMinute = 1440 // default retention min(1 day) of old eks after migration @@ -59,12 +55,11 @@ const ( var ( // Regular expression used to verify the configuration of the service listening port. // A valid service listening port configuration is a string containing only numbers. - regexpListen = regexp.MustCompile(`^(\d)+$`) - batchExpirationGetNum int - scanCheckInterval int64 - lcScanRoutineNumPerTask int - lcScanLimitPerSecond rate.Limit - + regexpListen = regexp.MustCompile(`^(\d)+$`) + simpleQueueInitCapacity int + scanCheckInterval int64 + lcScanRoutineNumPerTask int + lcScanLimitPerSecond rate.Limit snapshotRoutineNumPerTask int lcNodeTaskCountLimit int maxDirChanNum = 1000000 diff --git a/lcnode/lc_scanner.go b/lcnode/lc_scanner.go index 265c4ef66..559c713fa 100644 --- a/lcnode/lc_scanner.go +++ b/lcnode/lc_scanner.go @@ -49,10 +49,9 @@ type LcScanner struct { adminTask *proto.AdminTask rule *proto.Rule dirChan *unboundedchan.UnboundedChan - fileChan *unboundedchan.UnboundedChan + fileChan chan interface{} dirRPool *routinepool.RoutinePool fileRPool *routinepool.RoutinePool - batchDentries *proto.BatchDentries currentStat *proto.LcNodeRuleTaskStatistics limiter *rate.Limiter now time.Time @@ -79,22 +78,21 @@ func NewS3Scanner(adminTask *proto.AdminTask, l *LcNode) (*LcScanner, error) { } scanner := &LcScanner{ - ID: scanTask.Id, - Volume: scanTask.VolName, - lcnode: l, - mw: metaWrapper, - adminTask: adminTask, - rule: scanTask.Rule, - dirChan: unboundedchan.NewUnboundedChan(defaultUnboundedChanInitCapacity), - fileChan: unboundedchan.NewUnboundedChan(defaultUnboundedChanInitCapacity), - dirRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), - fileRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), - batchDentries: proto.NewBatchDentries(), - currentStat: &proto.LcNodeRuleTaskStatistics{}, - limiter: rate.NewLimiter(lcScanLimitPerSecond, defaultLcScanLimitBurst), - now: time.Now(), - receiveStopC: make(chan bool), - stopC: make(chan bool), + ID: scanTask.Id, + Volume: scanTask.VolName, + lcnode: l, + mw: metaWrapper, + adminTask: adminTask, + rule: scanTask.Rule, + dirChan: unboundedchan.NewUnboundedChan(defaultUnboundedChanInitCapacity), + fileChan: make(chan interface{}, simpleQueueInitCapacity), + dirRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), + fileRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), + currentStat: &proto.LcNodeRuleTaskStatistics{}, + limiter: rate.NewLimiter(lcScanLimitPerSecond, defaultLcScanLimitBurst), + now: time.Now(), + receiveStopC: make(chan bool), + stopC: make(chan bool), } var ebsClient *blobstore.BlobStoreClient @@ -230,7 +228,8 @@ func (s *LcScanner) Start() (err error) { return } - go s.scan() + go s.handleFileChan() + go s.handleDirChan() var currentPath string if len(prefixDirs) > 0 { @@ -313,10 +312,10 @@ func (s *LcScanner) FindPrefixInode() (inode uint64, prefixDirs []string, err er return } -func (s *LcScanner) scan() { - log.LogInfof("Enter scan %+v", s) +func (s *LcScanner) handleFileChan() { + log.LogInfof("Enter handleFileChan, %+v", s) defer func() { - log.LogInfof("Exit scan %+v", s) + log.LogInfof("Exit handleFileChan, %+v", s) }() prefix := s.rule.GetPrefix() @@ -324,11 +323,11 @@ func (s *LcScanner) scan() { for { select { case <-s.stopC: - log.LogInfof("receive stop, stop scan %v", s.ID) + log.LogInfof("receive stop, stop handleFileChan %v", s.ID) return - case val, ok := <-s.fileChan.Out: + case val, ok := <-s.fileChan: if !ok { - log.LogErrorf("fileChan closed") + log.LogWarnf("fileChan closed, id(%v)", s.ID) return } dentry := val.(*proto.ScanDentry) @@ -341,71 +340,76 @@ func (s *LcScanner) scan() { } _, err := s.fileRPool.Submit(job) if err != nil { - log.LogErrorf("fileRPool closed") + log.LogWarnf("fileRPool.Submit err(%v), id(%v)", err, s.ID) } - default: - select { - case <-s.stopC: - log.LogInfof("receive stop, stop scan %v", s.ID) - return - case val, ok := <-s.fileChan.Out: - if !ok { - log.LogErrorf("fileChan closed") - return - } - dentry := val.(*proto.ScanDentry) - if !strings.HasPrefix(dentry.Path, prefix) { - continue - } + } + } +} - job := func() { - s.handleFile(dentry) +func (s *LcScanner) handleDirChan() { + log.LogInfof("Enter handleDirChan, %+v", s) + defer func() { + log.LogInfof("Exit handleDirChan, %+v", s) + }() + + for { + select { + case <-s.stopC: + log.LogInfof("receive stop, stop handleDirChan %v", s.ID) + return + case val, ok := <-s.dirChan.Out: + if !ok { + log.LogWarnf("dirChan closed, id(%v)", s.ID) + return + } + dentry := val.(*proto.ScanDentry) + + var job func() + if s.dirChan.Len() > maxDirChanNum { + job = func() { + s.handleDirLimitDepthFirst(dentry) } - _, err := s.fileRPool.Submit(job) - if err != nil { - log.LogErrorf("fileRPool closed") - } - case val, ok := <-s.dirChan.Out: - if !ok { - log.LogErrorf("dirChan closed") - return - } - dentry := val.(*proto.ScanDentry) - var job func() - if s.dirChan.Len() > maxDirChanNum { - job = func() { - s.handleDirLimitDepthFirst(dentry) - } - } else { - job = func() { - s.handleDirLimitBreadthFirst(dentry) - } - } - _, err := s.dirRPool.Submit(job) - if err != nil { - log.LogErrorf("handleDir failed, err(%v)", err) + } else { + job = func() { + s.handleDirLimitBreadthFirst(dentry) } } + _, err := s.dirRPool.Submit(job) + if err != nil { + log.LogWarnf("dirRPool.Submit err(%v), id(%v)", err, s.ID) + } } } } func (s *LcScanner) handleFile(dentry *proto.ScanDentry) { - log.LogDebugf("handleFile: %v, fileChan: %v", dentry, s.fileChan.Len()) - op := dentry.Op - if op != "" { - s.limiter.Wait(context.Background()) - log.LogDebugf("handleFile: %v, start", dentry) - } + log.LogInfof("handleFile: %v, fileChan: %v", dentry, len(s.fileChan)) + atomic.AddInt64(&s.currentStat.TotalFileScannedNum, 1) - var err error + s.limiter.Wait(context.Background()) start := time.Now() - if op != "" { - defer func() { - auditlog.LogLcNodeOp(op, s.Volume, dentry.Name, dentry.Path, dentry.ParentId, dentry.Inode, dentry.Size, dentry.WriteGen, - dentry.HasMek, dentry.StorageClass, proto.OpTypeToStorageType(op), time.Since(start).Milliseconds(), err) - }() + + info, err := s.mw.InodeGet_ll(dentry.Inode) + if err != nil { + log.LogWarnf("handleFile InodeGet_ll err: %v, dentry: %+v", err, dentry) + return } + op := s.inodeExpired(info, s.rule.Expiration, s.rule.Transitions) + if op == "" { + return + } + dentry.Op = op + dentry.Size = info.Size + dentry.StorageClass = info.StorageClass + dentry.WriteGen = info.WriteGen + dentry.HasMek = info.HasMigrationEk + atomic.AddInt64(&s.currentStat.TotalFileExpiredNum, 1) + log.LogInfof("handleFile: %v, is expired", dentry) + + defer func() { + auditlog.LogLcNodeOp(op, s.Volume, dentry.Name, dentry.Path, dentry.ParentId, dentry.Inode, dentry.Size, dentry.WriteGen, + dentry.HasMek, dentry.StorageClass, proto.OpTypeToStorageType(op), time.Since(start).Milliseconds(), err) + }() switch op { case proto.OpTypeDelete: @@ -492,11 +496,7 @@ func (s *LcScanner) handleFile(dentry *proto.ScanDentry) { atomic.AddInt64(&s.currentStat.ExpiredMToBlobstoreBytes, int64(dentry.Size)) default: - atomic.AddInt64(&s.currentStat.TotalFileScannedNum, 1) - s.batchDentries.Append(dentry) - if s.batchDentries.Len() >= batchExpirationGetNum { - s.batchHandleFile() - } + log.LogWarnf("invalid op: %v", dentry) } } @@ -516,25 +516,6 @@ func isSkipErr(err error) bool { return false } -func (s *LcScanner) batchHandleFile() { - dentries, inodes := s.batchDentries.BatchGetAndClear() - inodesInfo := s.mw.BatchInodeGet(inodes) - for _, info := range inodesInfo { - if op := s.inodeExpired(info, s.rule.Expiration, s.rule.Transitions); op != "" { - d := dentries[info.Inode] - if d != nil { - d.Op = op - d.Size = info.Size - d.StorageClass = info.StorageClass - d.WriteGen = info.WriteGen - d.HasMek = info.HasMigrationEk - s.fileChan.In <- d - atomic.AddInt64(&s.currentStat.TotalFileExpiredNum, 1) - } - } - } -} - func (s *LcScanner) inodeExpired(inode *proto.InodeInfo, condE *proto.Expiration, condT []*proto.Transition) (op string) { if inode == nil { return @@ -656,7 +637,7 @@ func (s *LcScanner) handleDirLimitDepthFirst(dentry *proto.ScanDentry) { } for _, file := range files { - s.fileChan.In <- file + s.fileChan <- file } for _, dir := range dirs { s.handleDirLimitDepthFirst(dir) @@ -719,7 +700,7 @@ func (s *LcScanner) handleDirLimitBreadthFirst(dentry *proto.ScanDentry) { Type: child.Type, } if !os.FileMode(childDentry.Type).IsDir() { - s.fileChan.In <- childDentry + s.fileChan <- childDentry } else { s.dirChan.In <- childDentry } @@ -782,44 +763,39 @@ func (s *LcScanner) checkScanning() { return case <-taskCheckTimer.C: if s.DoneScanning() { - if s.batchDentries.Len() > 0 { - log.LogInfof("checkScanning last batchDentries") - s.batchHandleFile() - } else { - log.LogInfof("checkScanning completed for task(%v)", s.adminTask) - taskCheckTimer.Stop() - t := time.Now() - response := s.adminTask.Response.(*proto.LcNodeRuleTaskResponse) - response.EndTime = &t - response.Status = proto.TaskSucceeds - response.Done = true - response.ID = s.ID - response.LcNode = s.lcnode.localServerAddr - response.Volume = s.Volume - response.Rule = s.rule - response.ExpiredDeleteNum = s.currentStat.ExpiredDeleteNum - response.ExpiredMToHddNum = s.currentStat.ExpiredMToHddNum - response.ExpiredMToBlobstoreNum = s.currentStat.ExpiredMToBlobstoreNum - response.ExpiredMToHddBytes = s.currentStat.ExpiredMToHddBytes - response.ExpiredMToBlobstoreBytes = s.currentStat.ExpiredMToBlobstoreBytes - response.ExpiredSkipNum = s.currentStat.ExpiredSkipNum - response.TotalFileScannedNum = s.currentStat.TotalFileScannedNum - response.TotalFileExpiredNum = s.currentStat.TotalFileExpiredNum - response.TotalDirScannedNum = s.currentStat.TotalDirScannedNum - response.ErrorDeleteNum = s.currentStat.ErrorDeleteNum - response.ErrorMToHddNum = s.currentStat.ErrorMToHddNum - response.ErrorMToBlobstoreNum = s.currentStat.ErrorMToBlobstoreNum - response.ErrorReadDirNum = s.currentStat.ErrorReadDirNum - log.LogInfof("checkScanning completed response(%+v)", response) + log.LogInfof("checkScanning completed for task(%v)", s.adminTask) + taskCheckTimer.Stop() + t := time.Now() + response := s.adminTask.Response.(*proto.LcNodeRuleTaskResponse) + response.EndTime = &t + response.Status = proto.TaskSucceeds + response.Done = true + response.ID = s.ID + response.LcNode = s.lcnode.localServerAddr + response.Volume = s.Volume + response.Rule = s.rule + response.ExpiredDeleteNum = s.currentStat.ExpiredDeleteNum + response.ExpiredMToHddNum = s.currentStat.ExpiredMToHddNum + response.ExpiredMToBlobstoreNum = s.currentStat.ExpiredMToBlobstoreNum + response.ExpiredMToHddBytes = s.currentStat.ExpiredMToHddBytes + response.ExpiredMToBlobstoreBytes = s.currentStat.ExpiredMToBlobstoreBytes + response.ExpiredSkipNum = s.currentStat.ExpiredSkipNum + response.TotalFileScannedNum = s.currentStat.TotalFileScannedNum + response.TotalFileExpiredNum = s.currentStat.TotalFileExpiredNum + response.TotalDirScannedNum = s.currentStat.TotalDirScannedNum + response.ErrorDeleteNum = s.currentStat.ErrorDeleteNum + response.ErrorMToHddNum = s.currentStat.ErrorMToHddNum + response.ErrorMToBlobstoreNum = s.currentStat.ErrorMToBlobstoreNum + response.ErrorReadDirNum = s.currentStat.ErrorReadDirNum + log.LogInfof("checkScanning completed response(%+v)", response) - s.lcnode.scannerMutex.Lock() - s.Stop() - delete(s.lcnode.lcScanners, s.ID) - s.lcnode.scannerMutex.Unlock() + s.lcnode.scannerMutex.Lock() + s.Stop() + delete(s.lcnode.lcScanners, s.ID) + s.lcnode.scannerMutex.Unlock() - s.lcnode.respondToMaster(s.adminTask) - return - } + s.lcnode.respondToMaster(s.adminTask) + return } taskCheckTimer.Reset(dur) } @@ -828,18 +804,32 @@ func (s *LcScanner) checkScanning() { func (s *LcScanner) DoneScanning() bool { log.LogInfof("dirChan.Len(%v) fileChan.Len(%v) fileRPool.RunningNum(%v) dirRPool.RunningNum(%v)", - s.dirChan.Len(), s.fileChan.Len(), s.fileRPool.RunningNum(), s.dirRPool.RunningNum()) - return s.dirChan.Len() == 0 && s.fileChan.Len() == 0 && s.fileRPool.RunningNum() == 0 && s.dirRPool.RunningNum() == 0 + s.dirChan.Len(), len(s.fileChan), s.fileRPool.RunningNum(), s.dirRPool.RunningNum()) + return s.dirChan.Len() == 0 && len(s.fileChan) == 0 && s.fileRPool.RunningNum() == 0 && s.dirRPool.RunningNum() == 0 } func (s *LcScanner) Stop() { close(s.stopC) + s.clearFileChan() // clear fileChan avoid blocking dirRPool s.fileRPool.WaitAndClose() s.dirRPool.WaitAndClose() close(s.dirChan.In) - close(s.fileChan.In) + close(s.fileChan) s.mw.Close() s.transitionMgr.ec.Close() s.transitionMgr.ecForW.Close() - log.LogInfof("scanner(%v) stopped", s.ID) + log.LogInfof("stop: scanner(%v) stopped", s.ID) +} + +func (s *LcScanner) clearFileChan() { + var num int + for { + select { + case <-s.fileChan: + num++ + default: + log.LogInfof("stop: clearFileChan clear num(%v)", num) + return + } + } } diff --git a/lcnode/lc_scanner_test.go b/lcnode/lc_scanner_test.go index ebe740aa5..11aa36103 100644 --- a/lcnode/lc_scanner_test.go +++ b/lcnode/lc_scanner_test.go @@ -26,6 +26,7 @@ import ( ) func TestLcScanner(t *testing.T) { + // log.InitLog("", "", log.InfoLevel, nil, 0) lcScanRoutineNumPerTask = 1 maxDirChanNum = 0 scanCheckInterval = 1 @@ -59,15 +60,14 @@ func TestLcScanner(t *testing.T) { Days: &days3, }, }, - dirChan: unboundedchan.NewUnboundedChan(10), - fileChan: unboundedchan.NewUnboundedChan(10), - dirRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), - fileRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), - batchDentries: proto.NewBatchDentries(), - currentStat: &proto.LcNodeRuleTaskStatistics{}, - limiter: rate.NewLimiter(defaultLcScanLimitPerSecond, defaultLcScanLimitBurst), - now: time.Now(), - stopC: make(chan bool), + dirChan: unboundedchan.NewUnboundedChan(10), + fileChan: make(chan interface{}), + dirRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), + fileRPool: routinepool.NewRoutinePool(lcScanRoutineNumPerTask), + currentStat: &proto.LcNodeRuleTaskStatistics{}, + limiter: rate.NewLimiter(defaultLcScanLimitPerSecond, defaultLcScanLimitBurst), + now: time.Now(), + stopC: make(chan bool), } err := scanner.Start() require.NoError(t, err) diff --git a/lcnode/meta.go b/lcnode/meta.go index c1c9d0b09..311ec9a98 100644 --- a/lcnode/meta.go +++ b/lcnode/meta.go @@ -20,7 +20,7 @@ type MetaWrapper interface { ReadDirLimitForSnapShotClean(parentID uint64, from string, limit uint64, verSeq uint64, isDir bool) ([]proto.Dentry, error) Delete_Ver_ll(parentID uint64, name string, isDir bool, verSeq uint64, fullPath string) (*proto.InodeInfo, error) Lookup_ll(parentID uint64, name string) (inode uint64, mode uint32, err error) - BatchInodeGet(inodes []uint64) []*proto.InodeInfo + InodeGet_ll(inode uint64) (*proto.InodeInfo, error) DeleteWithCond_ll(parentID, cond uint64, name string, isDir bool, fullPath string) (inode *proto.InodeInfo, err error) Evict(inode uint64, fullPath string) error UpdateExtentKeyAfterMigration(inode uint64, storageType uint32, extentKeys []proto.ObjExtentKey, writeGen uint64, delayDelMinute uint64, fullPath string) error diff --git a/lcnode/meta_test.go b/lcnode/meta_test.go index 9676bd40c..4ce6ea9fa 100644 --- a/lcnode/meta_test.go +++ b/lcnode/meta_test.go @@ -39,28 +39,33 @@ func (*MockMetaWrapper) Lookup_ll(parentID uint64, name string) (inode uint64, m return } -func (*MockMetaWrapper) BatchInodeGet(inodes []uint64) []*proto.InodeInfo { - return []*proto.InodeInfo{ - { +func (*MockMetaWrapper) InodeGet_ll(inode uint64) (*proto.InodeInfo, error) { + switch inode { + case 1: + return &proto.InodeInfo{ Inode: 1, AccessTime: time.Now().AddDate(0, 0, -2), Size: 100, - }, - { + }, nil + case 2: + return &proto.InodeInfo{ Inode: 2, AccessTime: time.Now().AddDate(0, 0, -3), Size: 200, - }, - { + }, nil + case 3: + return &proto.InodeInfo{ Inode: 3, AccessTime: time.Now().AddDate(0, 0, -4), - }, - { + }, nil + case 6: + return &proto.InodeInfo{ Inode: 6, AccessTime: time.Now().AddDate(0, 0, -4), ForbiddenLc: true, - }, + }, nil } + return nil, nil } func (*MockMetaWrapper) DeleteWithCond_ll(parentID, cond uint64, name string, isDir bool, fullPath string) (*proto.InodeInfo, error) { diff --git a/lcnode/server.go b/lcnode/server.go index bd7818fd6..6beaa4594 100644 --- a/lcnode/server.go +++ b/lcnode/server.go @@ -142,13 +142,6 @@ func (l *LcNode) parseConfig(cfg *config.Config) (err error) { l.masters = masters l.mc = master.NewMasterClient(masters, false) - // parse batchExpirationGetNum - batchExpirationGetNum = cfg.GetInt(configBatchExpirationGetNumStr) - if batchExpirationGetNum <= 0 || batchExpirationGetNum > maxBatchExpirationGetNum { - batchExpirationGetNum = defaultBatchExpirationGetNum - } - log.LogInfof("loadConfig: setup config: %v(%v)", configBatchExpirationGetNumStr, batchExpirationGetNum) - // parse scanCheckInterval scanCheckInterval = cfg.GetInt64(configScanCheckIntervalStr) if scanCheckInterval <= 0 { @@ -163,6 +156,13 @@ func (l *LcNode) parseConfig(cfg *config.Config) (err error) { } log.LogInfof("loadConfig: setup config: %v(%v)", configLcScanRoutineNumPerTaskStr, lcScanRoutineNumPerTask) + // parse simpleQueueInitCapacity + simpleQueueInitCapacity = cfg.GetInt(configSimpleQueueInitCapacityStr) + if simpleQueueInitCapacity <= lcScanRoutineNumPerTask*1000 { + simpleQueueInitCapacity = defaultSimpleQueueInitCapacity + } + log.LogInfof("loadConfig: setup config: %v(%v)", configSimpleQueueInitCapacityStr, simpleQueueInitCapacity) + // parse snapshotRoutineNumPerTask snapshotRoutineNumPerTask = cfg.GetInt(configSnapshotRoutineNumPerTaskStr) if snapshotRoutineNumPerTask <= 0 || snapshotRoutineNumPerTask > maxLcScanRoutineNumPerTask { diff --git a/proto/lifecycle.go b/proto/lifecycle.go index 2be726091..2827e8bce 100644 --- a/proto/lifecycle.go +++ b/proto/lifecycle.go @@ -19,7 +19,6 @@ import ( "fmt" "regexp" "strings" - "sync" "time" "github.com/cubefs/cubefs/util/log" @@ -400,38 +399,3 @@ type ScanDentry struct { WriteGen uint64 `json:"gen"` // for migrate: used to determine whether a file is modified HasMek bool `json:"mek"` // for migrate: if HasMek, call DeleteMigrationExtentKey instead of migrating } - -type BatchDentries struct { - sync.RWMutex - dentries map[uint64]*ScanDentry -} - -func NewBatchDentries() *BatchDentries { - return &BatchDentries{ - dentries: make(map[uint64]*ScanDentry), - } -} - -func (f *BatchDentries) Append(dentry *ScanDentry) { - f.Lock() - defer f.Unlock() - f.dentries[dentry.Inode] = dentry -} - -func (f *BatchDentries) Len() int { - f.RLock() - defer f.RUnlock() - return len(f.dentries) -} - -func (f *BatchDentries) BatchGetAndClear() (map[uint64]*ScanDentry, []uint64) { - f.Lock() - defer f.Unlock() - dentries := f.dentries - var inodes []uint64 - for i := range f.dentries { - inodes = append(inodes, i) - } - f.dentries = make(map[uint64]*ScanDentry) - return dentries, inodes -}