gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(datanode): Fix the enhance the persistence process of extentId #23118077

Browse Source 
Signed-off-by: leonrayang <chl696@sina.com>
This commit is contained in:
leonrayang 2025-04-27 10:19:51 +08:00 committed by zhumingze1108
parent 8c35095b29
commit efeeea3dcd
2 changed files with 33 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
datanode/storage/extent_store.go
View File

@ -1390,6 +1390,7 @@ func (s *ExtentStore) GetHasDeleteTinyRecords() (extentDes []ExtentDeleted, err
// This function can only be called by the leader.
func (s *ExtentStore) NextExtentID() (extentID uint64, err error) {
extentID = atomic.AddUint64(&s.baseExtentID, 1)
s.PreAllocSpaceOnVerfiyFile()
err = s.PersistenceBaseExtentID(extentID)
return
}
@ -1411,11 +1412,11 @@ func (s *ExtentStore) UpdateBaseExtentID(id uint64) (err error) {
if IsTinyExtent(id) {
return
}
if id > atomic.LoadUint64(&s.baseExtentID) {
needPer := s.PreAllocSpaceOnVerfiyFile()
if id > atomic.LoadUint64(&s.baseExtentID) || needPer {
atomic.StoreUint64(&s.baseExtentID, id)
err = s.PersistenceBaseExtentID(atomic.LoadUint64(&s.baseExtentID))
}
s.PreAllocSpaceOnVerfiyFile(atomic.LoadUint64(&s.baseExtentID))
return
}

66
datanode/storage/persistence_crc.go
View File

@ -213,11 +213,6 @@ func (s *ExtentStore) GetPersistenceBaseExtentID() (extentID uint64, err error)
}
func (s *ExtentStore) WriteExtentIDOnVerifyFile(baseExtentID uint64, preAllocExtentID uint64) (err error) {
if preAllocExtentID != 0 && baseExtentID > preAllocExtentID {
err = fmt.Errorf("baseExtentID %v larger than preAllocExtentID %v", baseExtentID, preAllocExtentID)
log.LogErrorf("action[WriteExtentIDOnVerifyFile] partition %v err %v", s.partitionID, err)
return
}
value := make([]byte, 20)
binary.BigEndian.PutUint64(value[:8], baseExtentID)
@ -283,41 +278,40 @@ func (s *ExtentStore) PreAllocSpaceOnVerfiyFileForAppend(idx int) {
}
}
func (s *ExtentStore) PreAllocSpaceOnVerfiyFile(currExtentID uint64) {
func (s *ExtentStore) PreAllocSpaceOnVerfiyFile() (needPersist bool) {
if !proto.IsNormalDp(s.partitionType) {
return
}
if currExtentID > atomic.LoadUint64(&s.hasAllocSpaceExtentIDOnVerfiyFile) {
s.hasAllocSpaceExtentIDOnVerfiyFile = currExtentID
prevAllocSpaceExtentID := int64(atomic.LoadUint64(&s.hasAllocSpaceExtentIDOnVerfiyFile))
endAllocSpaceExtentID := int64(prevAllocSpaceExtentID + 1000)
size := int64(1000 * util.BlockHeaderSize)
err := fallocate(int(s.verifyExtentFp.Fd()), 1, prevAllocSpaceExtentID*util.BlockHeaderSize, size)
if err != nil {
return
}
for id, fp := range s.verifyExtentFpAppend {
stat, _ := fp.Stat()
log.LogDebugf("PreAllocSpaceOnVerfiyFile. id %v name %v size %v", id, fp.Name(), stat.Size())
err = fallocate(int(fp.Fd()), 1, prevAllocSpaceExtentID*util.BlockHeaderSize, size)
if err != nil {
log.LogErrorf("PreAllocSpaceOnVerfiyFile. id %v name %v err %v", id, fp.Name(), err)
return
}
}
if err = s.WritePreAllocSpaceExtentIDOnVerifyFile(uint64(endAllocSpaceExtentID)); err != nil {
return
}
atomic.StoreUint64(&s.hasAllocSpaceExtentIDOnVerfiyFile, uint64(endAllocSpaceExtentID))
log.LogInfof("Action(PreAllocSpaceOnVerifyFile) PartitionID(%v) currentExtent(%v)"+
"PrevAllocSpaceExtentIDOnVerifyFile(%v) EndAllocSpaceExtentIDOnVerifyFile(%v)"+
" has allocSpaceOnVerifyFile to (%v)", s.partitionID, currExtentID, prevAllocSpaceExtentID, endAllocSpaceExtentID,
prevAllocSpaceExtentID*util.BlockHeaderSize+size)
currExtentID := atomic.LoadUint64(&s.baseExtentID)
if currExtentID <= atomic.LoadUint64(&s.hasAllocSpaceExtentIDOnVerfiyFile) {
return
}
s.hasAllocSpaceExtentIDOnVerfiyFile = currExtentID
prevAllocSpaceExtentID := int64(atomic.LoadUint64(&s.hasAllocSpaceExtentIDOnVerfiyFile))
endAllocSpaceExtentID := int64(prevAllocSpaceExtentID + 1000)
size := int64(1000 * util.BlockHeaderSize)
err := fallocate(int(s.verifyExtentFp.Fd()), 1, prevAllocSpaceExtentID*util.BlockHeaderSize, size)
if err != nil {
return
}
for id, fp := range s.verifyExtentFpAppend {
stat, _ := fp.Stat()
log.LogDebugf("PreAllocSpaceOnVerfiyFile. id %v name %v size %v", id, fp.Name(), stat.Size())
err = fallocate(int(fp.Fd()), 1, prevAllocSpaceExtentID*util.BlockHeaderSize, size)
if err != nil {
log.LogErrorf("PreAllocSpaceOnVerfiyFile. id %v name %v err %v", id, fp.Name(), err)
return
}
}
atomic.StoreUint64(&s.hasAllocSpaceExtentIDOnVerfiyFile, uint64(endAllocSpaceExtentID))
log.LogInfof("Action(PreAllocSpaceOnVerifyFile) PartitionID(%v) currentExtent(%v)"+
"PrevAllocSpaceExtentIDOnVerifyFile(%v) EndAllocSpaceExtentIDOnVerifyFile(%v)"+
" has allocSpaceOnVerifyFile to (%v)", s.partitionID, currExtentID, prevAllocSpaceExtentID, endAllocSpaceExtentID,
prevAllocSpaceExtentID*util.BlockHeaderSize+size)
return true
}
func (s *ExtentStore) PersistenceHasDeleteExtent(extentID uint64) (err error) {