From c3c7983f7befdee81ec2911a4b4d3ea7b57936d6 Mon Sep 17 00:00:00 2001
From: j2rong4cn <36783515+j2rong4cn@users.noreply.github.com>
Date: Wed, 1 Oct 2025 18:06:10 +0800
Subject: [PATCH] perf(sync-closers): improve Close method (#1395)

Signed-off-by: j2rong4cn <36783515+j2rong4cn@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
---
 pkg/utils/io.go | 37 ++++++++++++++++++++-----------------
 1 file changed, 20 insertions(+), 17 deletions(-)

diff --git a/pkg/utils/io.go b/pkg/utils/io.go
index 7ce6a912..ff52f843 100644
--- a/pkg/utils/io.go
+++ b/pkg/utils/io.go
@@ -207,31 +207,34 @@ func (c *SyncClosers) AcquireReference() bool {
 		}
 		newRef := ref + 1
 		if atomic.CompareAndSwapInt32(&c.ref, ref, newRef) {
-			log.Debugf("AcquireReference %p: %d", c, newRef)
+			// log.Debugf("AcquireReference %p: %d", c, newRef)
 			return true
 		}
 	}
 }
 
+const closersClosed = math.MinInt16
+
 func (c *SyncClosers) Close() error {
-	for {
-		ref := atomic.LoadInt32(&c.ref)
-		if ref < 0 {
-			return nil
-		}
-		newRef := ref - 1
-		if newRef <= 0 {
-			newRef = math.MinInt16
-		}
-		if atomic.CompareAndSwapInt32(&c.ref, ref, newRef) {
-			log.Debugf("Close %p: %d", c, ref)
-			if newRef > 0 {
-				return nil
-			}
-			break
-		}
+	ref := atomic.AddInt32(&c.ref, -1)
+	if ref > 0 {
+		// log.Debugf("ReleaseReference %p: %d", c, ref)
+		return nil
 	}
 
+	if ref < -1 {
+		atomic.StoreInt32(&c.ref, closersClosed)
+		return nil
+	}
+
+	// Attempt to acquire FinalClose permission.
+	// At this point, ref must be 0 or -1. We try to atomically change it to the closersClosed state.
+	// Only the first successful goroutine gets the cleanup permission.
+	if !atomic.CompareAndSwapInt32(&c.ref, ref, closersClosed) {
+		return nil
+	}
+
+	// log.Debugf("FinalClose %p", c)
 	var errs []error
 	for _, closer := range c.closers {
 		if closer != nil {