From 7c026a4daf4cdf960396b227eb7dabf2924be039 Mon Sep 17 00:00:00 2001
From: chihe <chihe@oppo.com>
Date: Fri, 21 Mar 2025 09:25:12 +0800
Subject: [PATCH] fix(flashnode): delete pre-allocated size when return from
 InitForCacheRead

close:#1000012153

Signed-off-by: chihe <chihe@oppo.com>
---
 flashnode/cachengine/block.go | 11 ++++++++---
 flashnode/flashnode_op.go     |  6 ++++--
 util/limit_io.go              | 12 +++++++-----
 util/limit_io_test.go         |  2 +-
 4 files changed, 20 insertions(+), 11 deletions(-)

diff --git a/flashnode/cachengine/block.go b/flashnode/cachengine/block.go
index af92d3dcb..fb54dd65c 100644
--- a/flashnode/cachengine/block.go
+++ b/flashnode/cachengine/block.go
@@ -561,9 +561,9 @@ func (cb *CacheBlock) GetRootPath() string {
 	return cb.rootPath
 }
 
-func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto.DataSource) {
+func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto.DataSource, done chan struct{}) {
 	cb.initOnce.Do(func() {
-		cb.InitForCacheRead(sources, engine.readDataNodeTimeout)
+		cb.InitForCacheRead(sources, engine.readDataNodeTimeout, done)
 		select {
 		case <-cb.closeCh:
 			engine.deleteCacheBlock(cb.blockKey)
@@ -573,7 +573,7 @@ func (cb *CacheBlock) InitOnceForCacheRead(engine *CacheEngine, sources []*proto
 	})
 }
 
-func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNodeTimeout int) {
+func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNodeTimeout int, done chan struct{}) {
 	var err error
 	var file *os.File
 	bgTime := stat.BeginStat()
@@ -586,8 +586,13 @@ func (cb *CacheBlock) InitForCacheRead(sources []*proto.DataSource, readDataNode
 			}
 			cb.notifyClose()
 		}
+		if value, ok := cb.cacheEngine.lruCacheMap.Load(cb.rootPath); ok {
+			cacheItem := value.(*lruCacheItem)
+			cacheItem.lruCache.FreePreAllocatedSize(cb.blockKey)
+		}
 		stat.EndStat("MissCacheRead:InitForCacheRead", err, bgTime, 1)
 		metric.SetWithLabels(err, map[string]string{exporter.Vol: cb.volume})
+		close(done)
 	}()
 	sb := strings.Builder{}
 	for _, s := range sources {
diff --git a/flashnode/flashnode_op.go b/flashnode/flashnode_op.go
index 2a78b1a73..3fde4f978 100644
--- a/flashnode/flashnode_op.go
+++ b/flashnode/flashnode_op.go
@@ -185,14 +185,16 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
 			}
 		}
 		bgTime2 := stat.BeginStat()
+		taskDone := make(chan struct{})
 		missCacheMetric := exporter.NewTPCnt("MissCacheRead")
 		// try to cache more miss data, but reply to client more quickly
 		if writable := f.limitWrite.TryRunWithContext(ctx, int(req.Size_), func() {
 			if block2, err := f.cacheEngine.CreateBlock(cr, conn.RemoteAddr().String(), false); err != nil {
 				log.LogWarnf("opCacheRead: CreateBlock failed, req(%v) err(%v)", req, err)
+				close(taskDone)
 				return
 			} else {
-				block2.InitOnceForCacheRead(f.cacheEngine, cr.Sources)
+				block2.InitOnceForCacheRead(f.cacheEngine, cr.Sources, taskDone)
 			}
 		}); !writable {
 			err = fmt.Errorf("create block cache limited")
@@ -205,7 +207,7 @@ func (f *FlashNode) opCacheRead(conn net.Conn, p *proto.Packet) (err error) {
 			stat.EndStat("MissCacheReadCancel", ctx.Err(), bgTime2, 1)
 			missCacheMetric.SetWithLabels(err, map[string]string{exporter.Vol: volume})
 			return ctx.Err()
-		default:
+		case <-taskDone:
 			block, err = f.cacheEngine.GetCacheBlockForRead(volume, cr.Inode, cr.FixedFileOffset, cr.Version, req.Size_)
 		}
 		stat.EndStat("MissCacheRead", err, bgTime2, 1)
diff --git a/util/limit_io.go b/util/limit_io.go
index e2e2aa409..ca4e32413 100644
--- a/util/limit_io.go
+++ b/util/limit_io.go
@@ -109,14 +109,14 @@ func (l *IoLimiter) Run(size int, allowHang bool, taskFn func()) (err error) {
 func (l *IoLimiter) RunNoWait(size int, allowHang bool, taskFn func()) (err error) {
 	if size > 0 && l.limit > 0 {
 		if !l.flow.AllowN(time.Now(), size) {
-			return fmt.Errorf("run limited")
+			return fmt.Errorf("flow limited")
 		}
 	}
 	return l.getIO().Run(taskFn, allowHang)
 }
 
 func (l *IoLimiter) TryRun(size int, taskFn func()) bool {
-	if ok := l.getIO().TryRun(taskFn); !ok {
+	if ok := l.getIO().TryRun(taskFn, false); !ok {
 		return false
 	}
 	if size > 0 {
@@ -135,7 +135,7 @@ func (l *IoLimiter) TryRunWithContext(ctx context.Context, size int, taskFn func
 			return false
 		}
 	}
-	if ok := l.getIO().TryRun(taskFn); !ok {
+	if ok := l.getIO().TryRun(taskFn, true); !ok {
 		return false
 	}
 	return true
@@ -286,7 +286,7 @@ func (q *ioQueue) Run(taskFn func(), allowHang bool) (err error) {
 	return
 }
 
-func (q *ioQueue) TryRun(taskFn func()) bool {
+func (q *ioQueue) TryRun(taskFn func(), async bool) bool {
 	if q.concurrency <= 0 {
 		taskFn()
 		return true
@@ -305,7 +305,9 @@ func (q *ioQueue) TryRun(taskFn func()) bool {
 		taskFn()
 		return true
 	case q.queue <- task:
-		<-task.done
+		if !async {
+			<-task.done
+		}
 		return true
 	default:
 		return false
diff --git a/util/limit_io_test.go b/util/limit_io_test.go
index 177c34e8d..2e299e14d 100644
--- a/util/limit_io_test.go
+++ b/util/limit_io_test.go
@@ -74,7 +74,7 @@ func TestLimitIOBase(t *testing.T) {
 		q := l.getIO()
 		l.Close()
 		q.Run(f, true)
-		require.True(t, q.TryRun(f))
+		require.True(t, q.TryRun(f, false))
 		t.Logf("closed status: %+v", q.Status())
 	}
 }