mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
refactor(meta): refactor client lease logic. #22855673
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319
parent
bd564efdf4
commit
a686e2a394
@ -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()
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
}
|
||||
@ -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
|
||||
}
|
||||
@ -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
|
||||
|
||||
@ -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())
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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()
|
||||
}
|
||||
@ -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.
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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"`
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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 {
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
