From 288d85f411009ff4fd1def2dfd10d7c6ef4c80d2 Mon Sep 17 00:00:00 2001
From: georgehao <haohongfan@gmail.com>
Date: Tue, 4 Feb 2025 15:20:28 +0800
Subject: [PATCH] fix syncx dead lock

---
 internal/syncx/mutex.go      | 68 ++++++++++++++++++++++---------
 internal/syncx/mutex_test.go | 79 ++++++++++++++++++++++++++++++++++++
 2 files changed, 127 insertions(+), 20 deletions(-)
 create mode 100644 internal/syncx/mutex_test.go

diff --git a/internal/syncx/mutex.go b/internal/syncx/mutex.go
index 96a21986c6..812af4b225 100644
--- a/internal/syncx/mutex.go
+++ b/internal/syncx/mutex.go
@@ -17,36 +17,62 @@
 // Package syncx contains exotic synchronization primitives.
 package syncx
 
-// ClosableMutex is a mutex that can also be closed.
-// Once closed, it can never be taken again.
+import (
+	"sync"
+)
+
+// ClosableMutex is a mutex that can be closed. Once closed, it cannot be locked again.
 type ClosableMutex struct {
-	ch chan struct{}
+	mu     sync.Mutex // Protects the following fields
+	closed bool
+	ch     chan struct{}
 }
 
+// NewClosableMutex creates a new closable mutex.
 func NewClosableMutex() *ClosableMutex {
 	ch := make(chan struct{}, 1)
 	ch <- struct{}{}
-	return &ClosableMutex{ch}
+	return &ClosableMutex{ch: ch}
 }
 
-// TryLock attempts to lock cm.
-// If the mutex is closed, TryLock returns false.
+// TryLock attempts to acquire the lock. Returns true if successful, false if the lock is closed or unavailable.
 func (cm *ClosableMutex) TryLock() bool {
-	_, ok := <-cm.ch
-	return ok
-}
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	if cm.closed {
+		return false
+	}
 
-// MustLock locks cm.
-// If the mutex is closed, MustLock panics.
-func (cm *ClosableMutex) MustLock() {
-	_, ok := <-cm.ch
-	if !ok {
-		panic("mutex closed")
+	select {
+	case <-cm.ch:
+		return true
+	default:
+		return false
 	}
 }
 
-// Unlock unlocks cm.
+// MustLock acquires the lock. Panics if the lock is already closed.
+func (cm *ClosableMutex) MustLock() {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	if cm.closed {
+		panic("mutex closed")
+	}
+	select {
+	case <-cm.ch:
+		return
+	default:
+		panic("mutex is already locked")
+	}
+}
+
+// Unlock releases the lock. Panics if the lock is already closed or if called without holding the lock.
 func (cm *ClosableMutex) Unlock() {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	if cm.closed {
+		panic("Unlock after Close")
+	}
 	select {
 	case cm.ch <- struct{}{}:
 	default:
@@ -54,11 +80,13 @@ func (cm *ClosableMutex) Unlock() {
 	}
 }
 
-// Close locks the mutex, then closes it.
+// Close closes the mutex, preventing further lock operations. Panics if called on an already-closed mutex.
 func (cm *ClosableMutex) Close() {
-	_, ok := <-cm.ch
-	if !ok {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	if cm.closed {
 		panic("Close of already-closed ClosableMutex")
 	}
-	close(cm.ch)
+	cm.closed = true
+	close(cm.ch) // Closing the channel will cause subsequent send operations to panic
 }
diff --git a/internal/syncx/mutex_test.go b/internal/syncx/mutex_test.go
new file mode 100644
index 0000000000..7fe10bc7ac
--- /dev/null
+++ b/internal/syncx/mutex_test.go
@@ -0,0 +1,79 @@
+package syncx
+
+import (
+	"testing"
+	"time"
+)
+
+func TestClosableMutex_TryLock(t *testing.T) {
+	cm := NewClosableMutex()
+	if !cm.TryLock() {
+		t.Fatal("expected TryLock to succeed")
+	}
+	if cm.TryLock() {
+		t.Fatal("expected TryLock to fail when already locked")
+	}
+	cm.Unlock()
+	if !cm.TryLock() {
+		t.Fatal("expected TryLock to succeed after unlock")
+	}
+	cm.Close()
+	if cm.TryLock() {
+		t.Fatal("expected TryLock to fail after close")
+	}
+}
+
+func TestClosableMutex_MustLock(t *testing.T) {
+	cm := NewClosableMutex()
+	cm.MustLock()
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatal("expected MustLock to panic when already locked")
+		}
+	}()
+	cm.MustLock()
+}
+
+func TestClosableMutex_Unlock(t *testing.T) {
+	cm := NewClosableMutex()
+	cm.MustLock()
+	cm.Unlock()
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatal("expected Unlock to panic when already unlocked")
+		}
+	}()
+	cm.Unlock()
+}
+
+func TestClosableMutex_Close(t *testing.T) {
+	cm := NewClosableMutex()
+	cm.MustLock()
+	cm.Close()
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatal("expected Close to panic when already closed")
+		}
+	}()
+	cm.Close()
+}
+
+func TestClosableMutex_Concurrent(t *testing.T) {
+	cm := NewClosableMutex()
+	done := make(chan struct{})
+	go func() {
+		cm.MustLock()
+		time.Sleep(100 * time.Millisecond)
+		cm.Unlock()
+		close(done)
+	}()
+	time.Sleep(50 * time.Millisecond)
+	if cm.TryLock() {
+		t.Fatal("expected TryLock to fail when locked by another goroutine")
+	}
+	<-done
+	if !cm.TryLock() {
+		t.Fatal("expected TryLock to succeed after other goroutine unlocks")
+	}
+	cm.Unlock()
+}