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(client):if directIO is enabled, wait for flush before write

Browse Source 
close:#1000511680

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2025-12-04 10:21:25 +08:00
parent e485333226
commit 109e4f1998
3 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
sdk/data/stream/extent_cache.go
View File

@ -248,7 +248,7 @@ func (cache *ExtentCache) SplitExtentKey(inodeID uint64, ekPivot *proto.ExtentKe
// Append appends an extent key.
func (cache *ExtentCache) Append(ek *proto.ExtentKey, sync bool) (discardExtents []proto.ExtentKey) {
log.LogDebugf("action[ExtentCache.Append] ek %v", ek)
log.LogDebugf("ExtentCache Append:ino(%v) new ek %v sync %v", cache.inode, ek, sync)
ekEnd := ek.FileOffset + uint64(ek.Size)
lower := &proto.ExtentKey{FileOffset: ek.FileOffset}
upper := &proto.ExtentKey{FileOffset: ekEnd}
@ -269,17 +269,18 @@ func (cache *ExtentCache) Append(ek *proto.ExtentKey, sync bool) (discardExtents
cache.root.AscendRange(lower, upper, func(i btree.Item) bool {
found := i.(*proto.ExtentKey)
discard = append(discard, found)
log.LogDebugf("ExtentCache Append: ino(%v) add discard ek(%v) for new ek(%v)", cache.inode, discard, ek)
return true
})
// After deleting the data between lower and upper, we will do the append
for _, key := range discard {
cache.root.Delete(key)
log.LogDebugf("ExtentCache del: ino(%v) ek(%v) ", cache.inode, key)
log.LogDebugf("ExtentCache Append: ino(%v) del discard ek(%v) for new ek(%v)", cache.inode, key, ek)
if key.PartitionId != 0 && key.ExtentId != 0 && (key.PartitionId != ek.PartitionId || key.ExtentId != ek.ExtentId || ek.ExtentOffset != key.ExtentOffset) {
if sync || (ek.PartitionId == 0 && ek.ExtentId == 0) {
cache.discard.ReplaceOrInsert(key)
// log.LogDebugf("ExtentCache Append add to discard: ino(%v) ek(%v) discard(%v)", cache.inode, ek, key)
log.LogDebugf("ExtentCache Append: ino(%v) add to discard ek(%v) discard(%v)", cache.inode, ek, key)
}
}
}

9
sdk/data/stream/extent_client.go
View File

@ -789,8 +789,9 @@ func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int
}
// errGetExtents = s.GetExtents(isMigration)
if log.EnableDebug() {
log.LogDebugf("Read: ino(%v) offset(%v) size(%v) storageClass(%v) isMigration(%v) errGetExtents(%v)",
inode, offset, size, storageClass, isMigration, errGetExtents)
log.LogDebugf("Read: ino(%v) offset(%v) size(%v) storageClass(%v) isMigration(%v) errGetExtents(%v) "+
"rdonly(%v) dirty(%v)",
inode, offset, size, storageClass, isMigration, errGetExtents, s.rdonly, s.dirty)
}
})
if errGetExtents != nil {
@ -798,7 +799,9 @@ func (client *ExtentClient) Read(inode uint64, data []byte, offset int, size int
log.LogErrorf("Read: ino(%v) offset(%v) size(%v): %v", inode, offset, size, err)
return 0, err
}
log.LogDebugf("Read: ino(%v) offset(%v) size(%v) storageClass(%v) isMigration(%v) errGetExtents(%v) "+
"rdonly(%v) dirty(%v) waitForFlush(%v)",
inode, offset, size, storageClass, isMigration, errGetExtents, s.rdonly, s.dirty, s.waitForFlush)
if !s.rdonly || s.dirty {
err = s.IssueFlushRequest()
if err != nil {

2
sdk/data/stream/stream_writer.go
View File

@ -385,7 +385,7 @@ begin:
isChecked := false
// Must flush before doing overwrite
for _, req := range requests {
if req.ExtentKey == nil && !req.CreateNewEk {
if !s.waitForFlush && req.ExtentKey == nil && !req.CreateNewEk {
continue
}
err = s.flush(true, uuid.New().String())