From 109e4f199857d4e8dbafa8938e14540e19eaa1ce Mon Sep 17 00:00:00 2001 From: chihe Date: Thu, 4 Dec 2025 10:21:25 +0800 Subject: [PATCH] fix(client):if directIO is enabled, wait for flush before write close:#1000511680 Signed-off-by: chihe --- sdk/data/stream/extent_cache.go | 7 ++++--- sdk/data/stream/extent_client.go | 9 ++++++--- sdk/data/stream/stream_writer.go | 2 +- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/sdk/data/stream/extent_cache.go b/sdk/data/stream/extent_cache.go index 049760db2..a45e245c9 100644 --- a/sdk/data/stream/extent_cache.go +++ b/sdk/data/stream/extent_cache.go @@ -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) } } } diff --git a/sdk/data/stream/extent_client.go b/sdk/data/stream/extent_client.go index 867a1f7c6..d8c0623dd 100644 --- a/sdk/data/stream/extent_client.go +++ b/sdk/data/stream/extent_client.go @@ -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 { diff --git a/sdk/data/stream/stream_writer.go b/sdk/data/stream/stream_writer.go index 489cd8fa0..6303b75a1 100644 --- a/sdk/data/stream/stream_writer.go +++ b/sdk/data/stream/stream_writer.go @@ -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())