diff --git a/lcnode/lc_scanner.go b/lcnode/lc_scanner.go index a9fa5a21c..502a6146f 100644 --- a/lcnode/lc_scanner.go +++ b/lcnode/lc_scanner.go @@ -402,7 +402,7 @@ func (s *LcScanner) handleFile(dentry *proto.ScanDentry) { dentry.Op = op dentry.Size = info.Size dentry.StorageClass = info.StorageClass - dentry.WriteGen = info.WriteGen + dentry.WriteGen = info.LeaseExpireTime dentry.HasMek = info.HasMigrationEk if op == "" { log.LogInfof("handleFile: %+v, ctime(%v), atime(%v), is not expired", dentry, info.CreateTime, info.AccessTime) @@ -528,7 +528,7 @@ func (s *LcScanner) inodeExpired(inode *proto.InodeInfo, condE *proto.Expiration } if inode.ForbiddenLc { - log.LogWarnf("ForbiddenLc, lease is occupied, inode: %+v, WriteGen(%v)", inode, inode.WriteGen) + log.LogWarnf("ForbiddenLc, lease is occupied, inode: %+v, WriteGen(%v)", inode, inode.LeaseExpireTime) return } @@ -565,7 +565,7 @@ func (s *LcScanner) inodeExpired(inode *proto.InodeInfo, condE *proto.Expiration func expired(inode *proto.InodeInfo, now int64, days *int, date *time.Time) bool { if days != nil && *days > 0 { if inode.AccessTime.Before(inode.CreateTime) { - log.LogWarnf("AccessTime before CreateTime, skip, inode: %+v, WriteGen(%v), AccessTime(%v), CreateTime(%v)", inode, inode.WriteGen, inode.AccessTime, inode.CreateTime) + log.LogWarnf("AccessTime before CreateTime, skip, inode: %+v, WriteGen(%v), AccessTime(%v), CreateTime(%v)", inode, inode.LeaseExpireTime, inode.AccessTime, inode.CreateTime) return false } inodeTime := inode.AccessTime.Unix() diff --git a/master/client_info.go b/master/client_info.go index 7eef132eb..29e4784d5 100644 --- a/master/client_info.go +++ b/master/client_info.go @@ -46,6 +46,10 @@ func (cm *ClientMgr) PutItem(ip, host, vol, version, role, enableBcache string) role = defaultRole } + if enableBcache == "" { + enableBcache = "false" + } + key := fmt.Sprintf("_%s_%s_%s_%s_%s_enableBcache-%s", vol, version, role, ip, host, enableBcache) if len(cm.clients) > maxClientCnt { diff --git a/metanode/const.go b/metanode/const.go index 13a51373b..748825597 100644 --- a/metanode/const.go +++ b/metanode/const.go @@ -103,8 +103,6 @@ type ( SetCreateTimeRequest = proto.SetCreateTimeRequest DeleteMigrationExtentKeyRequest = proto.DeleteMigrationExtentKeyRequest - - ForbiddenMigrationRequest = proto.ForbiddenMigrationRequest ) // op code should be fixed, order change will cause raft fsm log apply fail diff --git a/metanode/forbidden_migration_list.go b/metanode/forbidden_migration_list.go deleted file mode 100644 index a253891db..000000000 --- a/metanode/forbidden_migration_list.go +++ /dev/null @@ -1,100 +0,0 @@ -package metanode - -import ( - "container/list" - "sync" - "time" - - "github.com/cubefs/cubefs/util/log" -) - -const ( - BgEvictionInterval = 2 * time.Minute -) - -type forbiddenMigrationList struct { - sync.RWMutex - index map[uint64]*list.Element - list *list.List - expiration time.Duration -} - -type forbiddenInodeInfo struct { - ino uint64 - expiration int64 -} - -func newForbiddenMigrationList(exp time.Duration) *forbiddenMigrationList { - fmList := &forbiddenMigrationList{ - index: make(map[uint64]*list.Element), - list: list.New(), - expiration: exp, - } - return fmList -} - -func (fmList *forbiddenMigrationList) Put(ino uint64) { - fmList.Lock() - old, ok := fmList.index[ino] - if ok { - fmList.list.Remove(old) - delete(fmList.index, ino) - } - expiration := time.Now().Add(fmList.expiration).Unix() - info := &forbiddenInodeInfo{ - ino: ino, - expiration: expiration, - } - element := fmList.list.PushFront(info) - fmList.index[ino] = element - fmList.Unlock() -} - -func (fmList *forbiddenMigrationList) Delete(ino uint64) { - fmList.Lock() - element, ok := fmList.index[ino] - if ok { - fmList.list.Remove(element) - delete(fmList.index, ino) - } - fmList.Unlock() -} - -func (fmList *forbiddenMigrationList) getExpiredForbiddenMigrationInodes(id uint64) []uint64 { - fmList.Lock() - defer fmList.Unlock() - var expiredInos []uint64 - currentTime := time.Now().Unix() - log.LogDebugf("[getExpiredForbiddenMigrationInodes] mp(%v) len(%v)", id, fmList.list.Len()) - for e := fmList.list.Back(); e != nil; { - info := e.Value.(*forbiddenInodeInfo) - // the first one that has not expired - if info.expiration > currentTime { - log.LogDebugf("[getExpiredForbiddenMigrationInodes] mp(%v) ino %v is not expired:%v", id, info.ino, info.expiration) - return expiredInos - } - // reset - expiredInos = append(expiredInos, info.ino) - next := e.Prev() - fmList.list.Remove(e) - delete(fmList.index, info.ino) - log.LogDebugf("[getExpiredForbiddenMigrationInodes] mp(%v) remove expired ino %v[%v]", id, info.ino, info.expiration) - e = next - } - return expiredInos -} - -func (fmList *forbiddenMigrationList) getAllForbiddenMigrationInodes(mpId uint64) []uint64 { - fmList.RLock() - defer fmList.RUnlock() - var allInos []uint64 - log.LogDebugf("[getAllForbiddenMigrationInodes] mp %v len %v:", mpId, fmList.list.Len()) - for e := fmList.list.Back(); e != nil; e = e.Prev() { - if info, ok := e.Value.(*forbiddenInodeInfo); ok { - allInos = append(allInos, info.ino) - } else { - log.LogWarnf("[getAllForbiddenMigrationInodes] mp %v value %v", mpId, e.Value) - } - } - return allInos -} diff --git a/metanode/forbidden_migration_list_test.go b/metanode/forbidden_migration_list_test.go deleted file mode 100644 index 8de879f9c..000000000 --- a/metanode/forbidden_migration_list_test.go +++ /dev/null @@ -1,49 +0,0 @@ -package metanode - -import ( - "testing" - "time" -) - -func TestForbiddenMigrationList(t *testing.T) { - expiration := time.Second * 5 - fmList := newForbiddenMigrationList(expiration) - - // Test Put and getAllForbiddenMigrationInodes methods - fmList.Put(1) - fmList.Put(2) - allInos := fmList.getAllForbiddenMigrationInodes(2) - expectedInos := []uint64{1, 2} - if !equalSlices(allInos, expectedInos) { - t.Errorf("getAllForbiddenMigrationInodes: expected %v, got %v", expectedInos, allInos) - } - - // Test Delete and getAllForbiddenMigrationInodes methods - fmList.Delete(1) - allInos = fmList.getAllForbiddenMigrationInodes(2) - expectedInos = []uint64{2} - if !equalSlices(allInos, expectedInos) { - t.Errorf("getAllForbiddenMigrationInodes: expected %v, got %v", expectedInos, allInos) - } - - // Test getExpiredForbiddenMigrationInodes method - time.Sleep(expiration) // Wait for expiration - expiredInos := fmList.getExpiredForbiddenMigrationInodes(2) - expectedExpiredInos := []uint64{2} - if !equalSlices(expiredInos, expectedExpiredInos) { - t.Errorf("getExpiredForbiddenMigrationInodes: expected %v, got %v", expectedExpiredInos, expiredInos) - } -} - -// Helper function to check if two slices are equal -func equalSlices(a, b []uint64) bool { - if len(a) != len(b) { - return false - } - for i := range a { - if a[i] != b[i] { - return false - } - } - return true -} diff --git a/metanode/inode.go b/metanode/inode.go index 3ee4227ef..7640d4c2d 100644 --- a/metanode/inode.go +++ b/metanode/inode.go @@ -24,7 +24,6 @@ import ( "math" "runtime/debug" "sync" - "sync/atomic" "time" "github.com/cubefs/cubefs/util/errors" @@ -101,8 +100,12 @@ type Inode struct { StorageClass uint32 HybridCloudExtents *SortedHybridCloudExtents HybridCloudExtentsMigration *SortedHybridCloudExtentsMigration - ForbiddenMigration uint32 - WriteGeneration uint64 + ClientID uint32 + LeaseExpireTime uint64 +} + +func (i *Inode) LeaseNotExpire() bool { + return i.LeaseExpireTime >= uint64(timeutil.GetCurrentTimeUnix()) } func (i *Inode) GetMultiVerString() string { @@ -452,8 +455,8 @@ func (i *Inode) String() string { if i.HybridCloudExtentsMigration != nil { buff.WriteString(fmt.Sprintf("MigrationExtents[%s]", i.HybridCloudExtentsMigration)) } - buff.WriteString(fmt.Sprintf("ForbiddenMigration[%v]", i.ForbiddenMigration)) - buff.WriteString(fmt.Sprintf("WriteGeneration[%v]", i.WriteGeneration)) + buff.WriteString(fmt.Sprintf("ClientID[%v]", i.ClientID)) + buff.WriteString(fmt.Sprintf("LeaseExpireTime[%v]", i.LeaseExpireTime)) buff.WriteString("}") return buff.String() } @@ -475,8 +478,8 @@ func NewInode(ino uint64, t uint32) *Inode { multiSnap: nil, StorageClass: proto.StorageClass_Unspecified, HybridCloudExtents: NewSortedHybridCloudExtents(), - ForbiddenMigration: ApproverToMigration, - WriteGeneration: 0, + LeaseExpireTime: 0, + ClientID: 0, HybridCloudExtentsMigration: NewSortedHybridCloudExtentsMigration(), } if proto.IsDir(t) { @@ -512,8 +515,8 @@ func (i *Inode) Copy() BtreeItem { newIno.Reserved = i.Reserved newIno.StorageClass = i.StorageClass newIno.Extents = i.Extents.Clone() - newIno.WriteGeneration = i.WriteGeneration - newIno.ForbiddenMigration = i.ForbiddenMigration + newIno.LeaseExpireTime = i.LeaseExpireTime + newIno.ClientID = i.ClientID // newIno.ObjExtents = i.ObjExtents.Clone() if i.multiSnap != nil { newIno.multiSnap = &InodeMultiSnap{ @@ -569,8 +572,8 @@ func (i *Inode) CopyDirectly() BtreeItem { newIno.Reserved = i.Reserved newIno.StorageClass = i.StorageClass newIno.Extents = i.Extents.Clone() - newIno.WriteGeneration = i.WriteGeneration - newIno.ForbiddenMigration = i.ForbiddenMigration + newIno.LeaseExpireTime = i.LeaseExpireTime + newIno.ClientID = i.ClientID // newIno.ObjExtents = i.ObjExtents.Clone() if i.HybridCloudExtents.sortedEks != nil { if proto.IsStorageClassReplica(i.StorageClass) { @@ -654,6 +657,7 @@ func (i *Inode) Unmarshal(raw []byte) (err error) { if err != nil { err = errors.NewErrorf("[Unmarshal] inode(%v) UnmarshalValue: %s", i.Inode, err.Error()) } + return } @@ -805,8 +809,8 @@ func (i *Inode) MarshalInodeValue(buff *bytes.Buffer) { log.LogDebugf("MarshalInodeValue ino(%v) V4MigrationExtentsFlag", i.Inode) } - log.LogDebugf("MarshalInodeValue ino(%v) storageClass(%v) Reserved(%v) ForbiddenMigration(%v) WriteGeneration(%v)", - i.Inode, i.StorageClass, i.Reserved, i.ForbiddenMigration, i.WriteGeneration) + log.LogDebugf("MarshalInodeValue ino(%v) storageClass(%v) Reserved(%v) ClientID(%v) LeaseExpireTime(%v)", + i.Inode, i.StorageClass, i.Reserved, i.ClientID, i.LeaseExpireTime) if err = binary.Write(buff, binary.BigEndian, &i.Reserved); err != nil { panic(err) } @@ -864,10 +868,10 @@ func (i *Inode) MarshalInodeValue(buff *bytes.Buffer) { if err = binary.Write(buff, binary.BigEndian, &i.StorageClass); err != nil { panic(err) } - if err = binary.Write(buff, binary.BigEndian, &i.ForbiddenMigration); err != nil { + if err = binary.Write(buff, binary.BigEndian, &i.ClientID); err != nil { panic(err) } - if err = binary.Write(buff, binary.BigEndian, &i.WriteGeneration); err != nil { + if err = binary.Write(buff, binary.BigEndian, &i.LeaseExpireTime); err != nil { panic(err) } @@ -1155,11 +1159,11 @@ func (i *Inode) UnmarshalInodeValue(buff *bytes.Buffer) (err error) { err = UnmarshalInodeFiledError("StorageClass(v4)", err) return } - if err = binary.Read(buff, binary.BigEndian, &i.ForbiddenMigration); err != nil { + if err = binary.Read(buff, binary.BigEndian, &i.ClientID); err != nil { err = UnmarshalInodeFiledError("ForbiddenMigration(v4)", err) return } - if err = binary.Read(buff, binary.BigEndian, &i.WriteGeneration); err != nil { + if err = binary.Read(buff, binary.BigEndian, &i.LeaseExpireTime); err != nil { err = UnmarshalInodeFiledError("WriteGeneration(v4)", err) return } @@ -1253,7 +1257,7 @@ func (i *Inode) UnmarshalValue(val []byte) (err error) { return } - if i.Reserved&V3EnableSnapInodeFlag > 0 { + if i.Reserved&V3EnableSnapInodeFlag > 0 && clusterEnableSnapshot { var verCnt int32 if err = binary.Read(buff, binary.BigEndian, &verCnt); err != nil { log.LogErrorf("[UnmarshalValue] inode[%v] newSeq[%v], get ver cnt err: %v", i.Inode, i.getVer(), err.Error()) @@ -2404,8 +2408,8 @@ func (i *Inode) UpdateHybridCloudParams(paramIno *Inode) { defer i.Unlock() i.StorageClass = paramIno.StorageClass i.HybridCloudExtents.sortedEks = paramIno.HybridCloudExtents.sortedEks - i.WriteGeneration = atomic.LoadUint64(&(paramIno.WriteGeneration)) - i.ForbiddenMigration = atomic.LoadUint32(&(paramIno.ForbiddenMigration)) + i.LeaseExpireTime = paramIno.LeaseExpireTime + i.ClientID = paramIno.ClientID i.HybridCloudExtentsMigration.storageClass = paramIno.HybridCloudExtentsMigration.storageClass i.HybridCloudExtentsMigration.sortedEks = paramIno.HybridCloudExtentsMigration.sortedEks i.HybridCloudExtentsMigration.expiredTime = paramIno.HybridCloudExtentsMigration.expiredTime diff --git a/metanode/manager.go b/metanode/manager.go index 82d7ffbdc..13862e970 100644 --- a/metanode/manager.go +++ b/metanode/manager.go @@ -391,8 +391,6 @@ func (m *metadataManager) HandleMetadataOperation(conn net.Conn, p *Packet, remo err = m.opMetaUpdateExtentKeyAfterMigration(conn, p, remoteAddr) case proto.OpDeleteMigrationExtentKey: err = m.opDeleteMigrationExtentKey(conn, p, remoteAddr) - case proto.OpMetaForbiddenMigration: - err = m.opMetaForbiddenMigration(conn, p, remoteAddr) default: err = fmt.Errorf("%s unknown Opcode: %d, reqId: %d", remoteAddr, p.Opcode, p.GetReqID()) diff --git a/metanode/manager_op.go b/metanode/manager_op.go index 18ec3a875..dadc5306f 100644 --- a/metanode/manager_op.go +++ b/metanode/manager_op.go @@ -3009,35 +3009,3 @@ func (m *metadataManager) opDeleteMigrationExtentKey(conn net.Conn, p *Packet, "resp body: %s", remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data) return } - -func (m *metadataManager) opMetaForbiddenMigration(conn net.Conn, p *Packet, - remoteAddr string, -) (err error) { - req := &ForbiddenMigrationRequest{} - if err = json.Unmarshal(p.Data, req); err != nil { - p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) - m.respondToClientWithVer(conn, p) - err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error()) - return - } - mp, err := m.getPartition(req.PartitionID) - if err != nil { - p.PacketErrorWithBody(proto.OpErr, ([]byte)(err.Error())) - m.respondToClientWithVer(conn, p) - err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error()) - return - } - if !m.serveProxy(conn, mp, p) { - return - } - if err = m.checkMultiVersionStatus(mp, p); err != nil { - err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data)) - return - } - mp.ForbiddenMigration(req, p, remoteAddr) - m.updatePackRspSeq(mp, p) - m.respondToClientWithVer(conn, p) - log.LogDebugf("%s [opMetaForbiddenMigration] req: %d - %v, resp body: %v, "+ - "resp body: %s", remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data) - return -} diff --git a/metanode/multi_ver_test.go b/metanode/multi_ver_test.go index a695aec83..068c4280d 100644 --- a/metanode/multi_ver_test.go +++ b/metanode/multi_ver_test.go @@ -64,10 +64,6 @@ var cfgJSON = `{ }` var tlog *testing.T -func tLogf(format string, args ...interface{}) { - tlog.Log(fmt.Sprintf(format, args...)) -} - func newPartition(conf *MetaPartitionConfig, manager *metadataManager) (mp *metaPartition) { mp = &metaPartition{ config: conf, @@ -86,7 +82,6 @@ func newPartition(conf *MetaPartitionConfig, manager *metadataManager) (mp *meta verSeq: conf.VerSeq, statByStorageClass: make([]*proto.StatOfStorageClass, 0), statByMigrateStorageClass: make([]*proto.StatOfStorageClass, 0), - fmList: newForbiddenMigrationList(proto.ForbiddenMigrationRenewalPeriod), } mp.config.Cursor = 0 mp.config.End = 100000 diff --git a/metanode/partition.go b/metanode/partition.go index 162dc2ed6..e512724fb 100644 --- a/metanode/partition.go +++ b/metanode/partition.go @@ -157,7 +157,6 @@ type OpInode interface { InodeGetWithEk(req *InodeGetReq, p *Packet) (err error) SetCreateTime(req *SetCreateTimeRequest, reqData []byte, p *Packet) (err error) // for debugging DeleteMigrationExtentKey(req *proto.DeleteMigrationExtentKeyRequest, p *Packet, remoteAddr string) (err error) - ForbiddenMigration(req *proto.ForbiddenMigrationRequest, p *Packet, remoteAddr string) (err error) } type OpExtend interface { @@ -598,7 +597,6 @@ type metaPartition struct { accessTimeValidInterval uint64 statByStorageClass []*proto.StatOfStorageClass statByMigrateStorageClass []*proto.StatOfStorageClass - fmList *forbiddenMigrationList syncAtimeCh chan uint64 } @@ -1071,8 +1069,6 @@ func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaP }, enableAuditLog: true, - - fmList: newForbiddenMigrationList(proto.ForbiddenMigrationRenewalPeriod), } if manager != nil { mp.config.ForbidWriteOpOfProtoVer0 = manager.isVolForbidWriteOpOfProtoVer0(mp.config.VolName) diff --git a/metanode/partition_forbidden_migration_list.go b/metanode/partition_forbidden_migration_list.go deleted file mode 100644 index 0599535da..000000000 --- a/metanode/partition_forbidden_migration_list.go +++ /dev/null @@ -1,95 +0,0 @@ -package metanode - -import ( - "time" - - "github.com/cubefs/cubefs/proto" - "github.com/cubefs/cubefs/util/log" -) - -const ( - ApproverToMigration uint32 = 0 - ForbiddenToMigration uint32 = 1 -) - -func (mp *metaPartition) startManageForbiddenMigrationList() { - go mp.checkForbiddenMigrationWorker() -} - -func (mp *metaPartition) checkForbiddenMigrationWorker() { - var ( - t = time.NewTicker(proto.ForbiddenMigrationRenewalPeriod / 2) - isLeader bool - ) - defer t.Stop() - log.LogDebugf("[checkForbiddenMigrationWorker] mp %v run", mp.config.PartitionId) - for { - select { - case <-mp.stopC: - log.LogDebugf("[metaPartition] checkForbiddenMigrationWorker stop partition: %v", - mp.config.PartitionId) - return - case <-t.C: - if _, isLeader = mp.IsLeader(); !isLeader { - log.LogDebugf("[checkForbiddenMigrationWorker] mp %v nodeId not leader, quit", - mp.config.PartitionId) - return - } - log.LogDebugf("[checkForbiddenMigrationWorker] mp %v getExpiredForbiddenMigrationInodes begin", mp.config.PartitionId) - inos := mp.fmList.getExpiredForbiddenMigrationInodes(mp.config.PartitionId) - log.LogDebugf("[checkForbiddenMigrationWorker] mp %v getExpiredForbiddenMigrationInodes end", mp.config.PartitionId) - // - freeInodes := make([]*Inode, 0) - for _, ino := range inos { - ref := &Inode{Inode: ino} - inode, ok := mp.inodeTree.Get(ref).(*Inode) - if !ok { - log.LogDebugf("[checkForbiddenMigrationWorker] . mp %v inode [%v] not found", - mp.config.PartitionId, ino) - continue - } - freeInodes = append(freeInodes, inode) - } - - batchSize := 1024 - totalInodes := len(freeInodes) - bufSlice := make([]byte, 0, 8*batchSize) - - for i := 0; i < totalInodes; i += batchSize { - end := i + batchSize - if end > totalInodes { - end = totalInodes - } - batch := freeInodes[i:end] - for _, inode := range batch { - bufSlice = append(bufSlice, inode.MarshalKey()...) - } - log.LogDebugf("[checkForbiddenMigrationWorker] mp %v sync %v to follower", - mp.config.PartitionId, len(batch)) - err := mp.syncToRaftFollowersFreeForbiddenMigrationInode(bufSlice) - if err != nil { - log.LogWarnf("[checkForbiddenMigrationWorker] raft commit inode list: %v, "+ - "response %s", len(batch), err.Error()) - } - for _, inode := range batch { - if err == nil { - mp.freeForbiddenMigrationInode(inode) - } else { - mp.fmList.Put(inode.Inode) - } - } - bufSlice = bufSlice[:0] - } - } - } -} - -func (mp *metaPartition) refreshForbiddenMigrationList() { - log.LogDebugf("[refreshForbiddenMigrationList] pid: %v HandleLeaderChange become leader "+ - " nodeId: %v ", mp.config.PartitionId, mp.config.NodeId) - allInos := mp.fmList.getAllForbiddenMigrationInodes(mp.config.PartitionId) - for _, ino := range allInos { - mp.fmList.Put(ino) - } - mp.startManageForbiddenMigrationList() -} diff --git a/metanode/partition_fsm.go b/metanode/partition_fsm.go index 9575f5ea2..5f44d86c3 100644 --- a/metanode/partition_fsm.go +++ b/metanode/partition_fsm.go @@ -473,14 +473,6 @@ func (mp *metaPartition) Apply(command []byte, index uint64) (resp interface{}, err = mp.fsmUniqCheckerEvict(req) case opFSMVersionOp: err = mp.fsmVersionOp(msg.V) - case opFSMForbiddenMigrationInode: - ino := NewInode(0, 0) - if err = ino.Unmarshal(msg.V); err != nil { - return - } - resp = mp.fsmForbiddenInodeMigration(ino) - case opFSMInternalFreeForbiddenMigrationInode: - err = mp.internalFreeForbiddenMigrationInode(msg.V) case opFSMRenewalForbiddenMigration: ino := NewInode(0, 0) if err = ino.Unmarshal(msg.V); err != nil { @@ -1006,8 +998,6 @@ func (mp *metaPartition) HandleLeaderChange(leader uint64) { ino.StorageClass = mp.GetVolStorageClass() go mp.initInode(ino) } - // refresh forbidden migration list - mp.refreshForbiddenMigrationList() } // Put puts the given key-value pair (operation key and operation request) into the raft store. diff --git a/metanode/partition_fsmop_inode.go b/metanode/partition_fsmop_inode.go index 4fdeb9770..95c94a3a0 100644 --- a/metanode/partition_fsmop_inode.go +++ b/metanode/partition_fsmop_inode.go @@ -20,7 +20,6 @@ import ( "encoding/json" "fmt" "io" - "sync/atomic" "time" "github.com/cubefs/cubefs/datanode/storage" @@ -772,10 +771,6 @@ func (mp *metaPartition) checkAndInsertFreeList(ino *Inode) { } else if ino.ShouldDeleteMigrationExtentKey(true) { mp.freeHybridList.Push(ino.Inode) } - if atomic.LoadUint32(&ino.ForbiddenMigration) == ForbiddenToMigration { - mp.fmList.Put(ino.Inode) - log.LogDebugf("action[checkAndInsertFreeList] put ino %v to forbidden migration check list", ino.Inode) - } } func (mp *metaPartition) fsmSetAttr(req *SetattrRequest) (err error) { @@ -1070,59 +1065,6 @@ func (mp *metaPartition) fsmBatchSyncInodeAccessTime(bufSlice []byte) (status ui return } -func (mp *metaPartition) internalFreeForbiddenMigrationInode(val []byte) (err error) { - if len(val) == 0 { - return - } - buf := bytes.NewBuffer(val) - ino := NewInode(0, 0) - for { - err = binary.Read(buf, binary.BigEndian, &ino.Inode) - if err != nil { - if err == io.EOF { - err = nil - return - } - return - } - log.LogDebugf("action[internalFreeForbiddenMigration]: received internal free forbidden migration"+ - ": partitionID(%v) inode(%v)", - mp.config.PartitionId, ino.Inode) - item := mp.inodeTree.CopyGet(ino) - if item == nil { - log.LogDebugf("action[internalFreeForbiddenMigration]: cannot find partitionID(%v) inode(%v)", mp.config.PartitionId, ino.Inode) - continue - } - inode := item.(*Inode) - mp.freeForbiddenMigrationInode(inode) - - } -} - -func (mp *metaPartition) freeForbiddenMigrationInode(ino *Inode) { - log.LogDebugf("action[freeForbiddenMigrationInode] ino %v really be freed", ino) - atomic.StoreUint32(&ino.ForbiddenMigration, ApproverToMigration) - mp.fmList.Delete(ino.Inode) - return -} - -func (mp *metaPartition) fsmForbiddenInodeMigration(ino *Inode) (resp *InodeResponse) { - resp = NewInodeResponse() - resp.Status = proto.OpOk - item := mp.inodeTree.CopyGet(ino) - if item == nil { - resp.Status = proto.OpNotExistErr - return - } - i := item.(*Inode) - atomic.StoreUint32(&i.ForbiddenMigration, ForbiddenToMigration) - atomic.AddUint64(&i.WriteGeneration, 1) - log.LogDebugf("action[fsmForbiddenInodeMigration] inode [%v] forbiddenMigration %v writeGen %v", i.Inode, - atomic.LoadUint32(&i.ForbiddenMigration), atomic.LoadUint64(&i.WriteGeneration)) - mp.fmList.Put(i.Inode) - return -} - func (mp *metaPartition) fsmRenewalInodeForbiddenMigration(ino *Inode) (resp *InodeResponse) { resp = NewInodeResponse() resp.Status = proto.OpOk @@ -1132,8 +1074,8 @@ func (mp *metaPartition) fsmRenewalInodeForbiddenMigration(ino *Inode) (resp *In return } i := item.(*Inode) - mp.fmList.Put(i.Inode) - log.LogDebugf("action[fsmRenewalInodeForbiddenMigration] inode %v is renewal", i.Inode) + i.LeaseExpireTime = ino.LeaseExpireTime + log.LogDebugf("action[fsmRenewalInodeForbiddenMigration] inode %v is renewal, expireTime %d", i.Inode, ino.LeaseExpireTime) return } @@ -1154,9 +1096,9 @@ func (mp *metaPartition) fsmUpdateExtentKeyAfterMigration(inoParam *Inode) (resp return } - if i.WriteGeneration > inoParam.WriteGeneration || i.ForbiddenMigration == ForbiddenToMigration { + if i.LeaseExpireTime != inoParam.LeaseExpireTime { log.LogErrorf("fsmUpdateExtentKeyAfterMigration: inode is forbidden to migrate. gen %d, reqGen %d, ino %d", - i.WriteGeneration, inoParam.WriteGeneration, i.Inode) + i.LeaseExpireTime, inoParam.LeaseExpireTime, i.Inode) resp.Status = proto.OpLeaseOccupiedByOthers return } diff --git a/metanode/partition_op_extent.go b/metanode/partition_op_extent.go index 3d2692257..3a9a4a240 100644 --- a/metanode/partition_op_extent.go +++ b/metanode/partition_op_extent.go @@ -461,7 +461,7 @@ func (mp *metaPartition) ExtentsList(req *proto.GetExtentsRequest, p *Packet) (e log.LogInfof("action[ExtentsList] inode[%v] request verseq [%v] ino ver [%v] extent size %v ino.Size %v ino[%v] hist len %v", req.Inode, req.VerSeq, ino.getVer(), len(ino.Extents.eks), ino.Size, ino, ino.getLayerLen()) - resp.WriteGeneration = ino.WriteGeneration + resp.LeaseExpireTime = ino.LeaseExpireTime if req.VerSeq > 0 && ino.getVer() > 0 && (req.VerSeq < ino.getVer() || isInitSnapVer(req.VerSeq)) { mp.GetExtentByVer(ino, req, resp) vIno := ino.Copy().(*Inode) diff --git a/metanode/partition_op_inode.go b/metanode/partition_op_inode.go index 2b04a0593..45e9d82d9 100644 --- a/metanode/partition_op_inode.go +++ b/metanode/partition_op_inode.go @@ -18,7 +18,6 @@ import ( "encoding/binary" "encoding/json" "fmt" - "sync/atomic" "time" "github.com/cubefs/cubefs/depends/tiglabs/raft" @@ -26,6 +25,7 @@ import ( "github.com/cubefs/cubefs/util/auditlog" "github.com/cubefs/cubefs/util/errors" "github.com/cubefs/cubefs/util/log" + "github.com/cubefs/cubefs/util/timeutil" ) func replyInfoNoCheck(info *proto.InodeInfo, ino *Inode) bool { @@ -49,10 +49,8 @@ func replyInfoNoCheck(info *proto.InodeInfo, ino *Inode) bool { info.ModifyTime = time.Unix(ino.ModifyTime, 0) info.StorageClass = ino.StorageClass info.MigrationStorageClass = ino.HybridCloudExtentsMigration.storageClass - info.WriteGen = atomic.LoadUint64(&ino.WriteGeneration) - if atomic.LoadUint32(&ino.ForbiddenMigration) == ForbiddenToMigration { - info.ForbiddenLc = true - } + info.LeaseExpireTime = ino.LeaseExpireTime + info.ForbiddenLc = ino.LeaseNotExpire() return true } @@ -79,10 +77,9 @@ func replyInfo(info *proto.InodeInfo, ino *Inode, quotaInfos map[uint32]*proto.M info.ModifyTime = time.Unix(ino.ModifyTime, 0) info.QuotaInfos = quotaInfos info.StorageClass = ino.StorageClass - info.WriteGen = atomic.LoadUint64(&ino.WriteGeneration) - if atomic.LoadUint32(&ino.ForbiddenMigration) == ForbiddenToMigration { - info.ForbiddenLc = true - } + info.LeaseExpireTime = ino.LeaseExpireTime + info.ForbiddenLc = ino.LeaseNotExpire() + info.MigrationStorageClass = ino.HybridCloudExtentsMigration.storageClass if ino.HybridCloudExtentsMigration.sortedEks != nil { info.HasMigrationEk = true @@ -106,13 +103,11 @@ func txReplyInfo(inode *Inode, txInfo *proto.TransactionInfo, quotaInfos map[uin QuotaInfos: quotaInfos, Target: nil, StorageClass: inode.StorageClass, - WriteGen: atomic.LoadUint64(&inode.WriteGeneration), + LeaseExpireTime: inode.LeaseExpireTime, MigrationStorageClass: inode.HybridCloudExtentsMigration.storageClass, } - if atomic.LoadUint32(&inode.ForbiddenMigration) == ForbiddenToMigration { - inoInfo.ForbiddenLc = true - } + inoInfo.ForbiddenLc = inode.LeaseNotExpire() if inode.HybridCloudExtentsMigration.sortedEks != nil { inoInfo.HasMigrationEk = true @@ -1120,6 +1115,12 @@ func (mp *metaPartition) RenewalForbiddenMigration(req *proto.RenewalForbiddenMi } else { ino.UpdateHybridCloudParams(item.(*Inode)) } + + newExpireTime := timeutil.GetCurrentTimeUnix() + proto.ForbiddenMigrationRenewalSeonds + if newExpireTime > int64(ino.LeaseExpireTime) { + ino.LeaseExpireTime = uint64(newExpireTime) + } + val, err := ino.Marshal() if err != nil { p.PacketErrorWithBody(proto.OpErr, []byte(err.Error())) @@ -1164,15 +1165,15 @@ func (mp *metaPartition) UpdateExtentKeyAfterMigration(req *proto.UpdateExtentKe return } - if inoParm.ForbiddenMigration == ForbiddenToMigration { + if inoParm.LeaseNotExpire() { err = fmt.Errorf("mp(%v) inode(%v) is forbidden to migration for lease is occupied by others", mp.config.PartitionId, inoParm.Inode) log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err) p.PacketErrorWithBody(proto.OpLeaseOccupiedByOthers, []byte(err.Error())) return } - writeGen := inoParm.WriteGeneration - if writeGen > req.WriteGen { + writeGen := inoParm.LeaseExpireTime + if writeGen != req.WriteGen { err = fmt.Errorf("mp(%v) inode(%v) write generation not match, curent(%v) request(%v)", mp.config.PartitionId, inoParm.Inode, writeGen, req.WriteGen) log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err) @@ -1421,38 +1422,3 @@ func (mp *metaPartition) DeleteMigrationExtentKey(req *proto.DeleteMigrationExte p.PacketErrorWithBody(msg.Status, nil) return } - -func (mp *metaPartition) ForbiddenMigration(req *proto.ForbiddenMigrationRequest, - p *Packet, remoteAddr string, -) (err error) { - log.LogDebugf("action[ForbiddenMigration] inode[%v] ", req.Inode) - - // note:don't need set reqSeq, extents get be done in next step - ino := NewInode(req.Inode, 0) - if item := mp.inodeTree.Get(ino); item == nil { - err = fmt.Errorf("mp %v inode %v reqeust cann't found", mp.config.PartitionId, ino) - log.LogErrorf("action[ForbiddenMigration] %v", err) - p.PacketErrorWithBody(proto.OpNotExistErr, []byte(err.Error())) - return - } else { - ino.UpdateHybridCloudParams(item.(*Inode)) - } - var ( - val []byte - reply []byte - status = proto.OpOk - ) - val, err = ino.Marshal() - if err != nil { - p.PacketErrorWithBody(proto.OpErr, []byte(err.Error())) - return - } - _, err = mp.submit(opFSMForbiddenMigrationInode, val) - if err != nil { - status = proto.OpErr - reply = []byte(err.Error()) - } - // mark inode as ForbiddenMigration - p.PacketErrorWithBody(status, reply) - return -} diff --git a/metanode/partition_store.go b/metanode/partition_store.go index 32fa3252c..37bcbdc72 100644 --- a/metanode/partition_store.go +++ b/metanode/partition_store.go @@ -198,6 +198,9 @@ func (mp *metaPartition) loadInode(rootDir string, crc uint32) (err error) { err = errors.NewErrorf("[loadInode] Unmarshal: %s", err.Error()) return } + if ino.LeaseExpireTime == 0 { + ino.LeaseExpireTime = uint64(ino.ModifyTime) + proto.ForbiddenMigrationRenewalSeonds + } mp.acucumUidSizeByLoad(ino) // data crc if _, err = crcCheck.Write(inoBuf); err != nil { diff --git a/proto/admin_proto.go b/proto/admin_proto.go index 5c887220f..c8d7c8089 100644 --- a/proto/admin_proto.go +++ b/proto/admin_proto.go @@ -1529,6 +1529,9 @@ func GetStorageClassByMediaType(mediaType uint32) (storageClass uint32) { return } -const ForbiddenMigrationRenewalPeriod = 1 * time.Hour +const ( + ForbiddenMigrationRenewalPeriod = 1 * time.Hour + ForbiddenMigrationRenewalSeonds = 3600 +) // const ForbiddenMigrationRenewalPeriod = 10 * time.Second // for debug diff --git a/proto/fs_proto.go b/proto/fs_proto.go index dc7007104..ee1609ae1 100644 --- a/proto/fs_proto.go +++ b/proto/fs_proto.go @@ -98,7 +98,7 @@ type InodeInfo struct { PersistAccessTime time.Time `json:"pat"` StorageClass uint32 `json:"storageClass"` - WriteGen uint64 `json:"writeGen"` + LeaseExpireTime uint64 `json:"leaseExpireTime"` ForbiddenLc bool `json:"forbiddenLc"` MigrationStorageClass uint32 `json:"migrationStorageClass"` HasMigrationEk bool `json:"hasMigrationEk"` @@ -724,7 +724,7 @@ type GetExtentsResponse struct { Extents []ExtentKey `json:"eks"` LayerInfo []LayerInfo `json:"layer"` Status int - WriteGeneration uint64 `json:"writeGeneration"` + LeaseExpireTime uint64 `json:"leaseExpireTime"` } // TruncateRequest defines the request to truncate. @@ -1121,8 +1121,3 @@ type DeleteMigrationExtentKeyRequest struct { Inode uint64 `json:"ino"` RequestExtend } - -type ForbiddenMigrationRequest struct { - PartitionID uint64 `json:"pid"` - Inode uint64 `json:"ino"` -} diff --git a/proto/packet.go b/proto/packet.go index c655957a1..ea48c0bea 100644 --- a/proto/packet.go +++ b/proto/packet.go @@ -278,7 +278,6 @@ const ( OpLeaseOccupiedByOthers uint8 = 0x86 OpLeaseGenerationNotMatch uint8 = 0x87 OpWriteOpOfProtoVerForbidden uint8 = 0x88 - OpMetaForbiddenMigration uint8 = 0x89 ) const ( @@ -688,8 +687,6 @@ func (p *Packet) GetOpMsg() (m string) { m = "OpMetaUpdateExtentKeyAfterMigration" case OpDeleteMigrationExtentKey: m = "OpDeleteMigrationExtentKey" - case OpMetaForbiddenMigration: - m = "OpMetaForbiddenMigration" default: m = fmt.Sprintf("op:%v not found", p.Opcode) } diff --git a/sdk/meta/api.go b/sdk/meta/api.go index 5804983d0..45ae84199 100644 --- a/sdk/meta/api.go +++ b/sdk/meta/api.go @@ -2421,7 +2421,8 @@ func (mw *MetaWrapper) UpdateSummary_ll(parentIno uint64, filesHddInc int64, fil } func (mw *MetaWrapper) UpdateAccessFileInfo_ll(parentIno uint64, valueCountSsd string, valueSizeSsd string, - valueCountHdd string, valueSizeHdd string, valueCountBlobStore string, valueSizeBlobStore string) { + valueCountHdd string, valueSizeHdd string, valueCountBlobStore string, valueSizeBlobStore string, +) { log.LogDebugf("UpdateAccessFileInfo_ll: ino(%v) valueCountSsd(%v) valueSizeSsd(%v) valueCountHdd(%v) valueSizeHdd(%v) valueCountBlobStore(%v) valueSizeBlobStore(%v)", parentIno, valueCountSsd, valueSizeSsd, valueCountHdd, valueSizeHdd, valueCountBlobStore, valueSizeBlobStore) @@ -2759,7 +2760,8 @@ func updateLocalSummary(inodeInfos []*proto.InodeInfo, splits []string, timeUnit } func (mw *MetaWrapper) refreshSummary(parentIno uint64, errCh chan<- error, wg *sync.WaitGroup, currentGoroutineNum *int32, newGoroutine bool, - goroutineNum int32, accessTimeCfg AccessTimeConfig) { + goroutineNum int32, accessTimeCfg AccessTimeConfig, +) { defer func() { if newGoroutine { atomic.AddInt32(currentGoroutineNum, -1) @@ -3206,7 +3208,8 @@ func (mw *MetaWrapper) GetAccessFileInfo(parentPath string, parentIno uint64, ma } func (mw *MetaWrapper) getDirAccessFileInfo(parentPath string, parentIno uint64, maxDepth int32, currentDepth *int32, info *[]AccessFileInfo, - errCh chan<- error, wg *sync.WaitGroup, currentGoroutineNum *int32, newGoroutine bool, goroutineNum int32) { + errCh chan<- error, wg *sync.WaitGroup, currentGoroutineNum *int32, newGoroutine bool, goroutineNum int32, +) { defer func() { if newGoroutine { atomic.AddInt32(currentGoroutineNum, -1) @@ -3329,7 +3332,8 @@ func (mw *MetaWrapper) getDirAccessFileInfo(parentPath string, parentIno uint64, } func (mw *MetaWrapper) getAccessFileInfo(parentPath string, parentIno uint64, errCh chan<- error, accessCountSsd *[]int64, accessSizeSsd *[]uint64, - accessCountHdd *[]int64, accessSizeHdd *[]uint64, accessCountBlobStore *[]int64, accessSizeBlobStore *[]uint64) { + accessCountHdd *[]int64, accessSizeHdd *[]uint64, accessCountBlobStore *[]int64, accessSizeBlobStore *[]uint64, +) { log.LogDebugf("getAccessFileInfo: parentPath(%v) parentIno(%v)", parentPath, parentIno) xattrInfo, err := mw.XAttrGet_ll(parentIno, AccessFileCountSsdKey) if err != nil { diff --git a/sdk/meta/operation.go b/sdk/meta/operation.go index e0b21b0bf..b9ba9c23f 100644 --- a/sdk/meta/operation.go +++ b/sdk/meta/operation.go @@ -3142,43 +3142,5 @@ func (mw *MetaWrapper) deleteMigrationExtentKey(mp *MetaPartition, inode uint64, } func (mw *MetaWrapper) forbiddenMigration(mp *MetaPartition, inode uint64) (status int, err error) { - bgTime := stat.BeginStat() - defer func() { - stat.EndStat("forbiddenMigration", err, bgTime, 1) - }() - - req := &proto.ForbiddenMigrationRequest{ - PartitionID: mp.PartitionID, - Inode: inode, - } - - packet := proto.NewPacketReqID() - packet.Opcode = proto.OpMetaForbiddenMigration - packet.PartitionID = mp.PartitionID - err = packet.MarshalData(req) - if err != nil { - log.LogErrorf("forbiddenMigration: ino(%v) err(%v)", inode, err) - return - } - - metric := exporter.NewTPCnt(packet.GetOpMsg()) - defer func() { - metric.SetWithLabels(err, map[string]string{exporter.Vol: mw.volname}) - }() - - packet, err = mw.sendToMetaPartition(mp, packet) - if err != nil { - log.LogErrorf("forbiddenMigration: packet(%v) mp(%v) req(%v) err(%v)", packet, mp, *req, err) - return - } - - status = parseStatus(packet.ResultCode) - if status != statusOK { - err = errors.New(packet.GetResultMsg()) - log.LogErrorf("forbiddenMigration: packet(%v) mp(%v) req(%v) result(%v)", packet, mp, *req, packet.GetResultMsg()) - return - } - - log.LogDebugf("forbiddenMigration exit: packet(%v) mp(%v) req(%v)", packet, mp, *req) - return statusOK, nil + return mw.renewalForbiddenMigration(mp, inode) } diff --git a/tool/fsck/cmd/check.go b/tool/fsck/cmd/check.go index add88719f..8f9ec957b 100644 --- a/tool/fsck/cmd/check.go +++ b/tool/fsck/cmd/check.go @@ -581,12 +581,12 @@ func compareInodes(i1 *metanode.Inode, i2 *metanode.Inode) *bytes.Buffer { } } - if i1.ForbiddenMigration != i2.ForbiddenMigration { - buffer.WriteString(fmt.Sprintf("ForbiddenMigration: %v != %v ", i1.ForbiddenMigration, i2.ForbiddenMigration)) + if i1.ClientID != i2.ClientID { + buffer.WriteString(fmt.Sprintf("ClientID: %v != %v ", i1.ClientID, i2.ClientID)) } - if i1.WriteGeneration != i2.WriteGeneration { - buffer.WriteString(fmt.Sprintf("WriteGeneration : %v != %v ", i1.WriteGeneration, i2.WriteGeneration)) + if i1.LeaseExpireTime != i2.LeaseExpireTime { + buffer.WriteString(fmt.Sprintf("LeaseExpireTime : %v != %v ", i1.LeaseExpireTime, i2.LeaseExpireTime)) } return &buffer