fix

diff --git a/msgq/Msgq_test.go b/msgq/Msgq_test.go
index ee8ee30b3b44f163cedf5f5c71153dee7c71d029..3d614bd5971036e620d2796657aebd58d689c540 100644 (file)
--- a/msgq/Msgq_test.go
+++ b/msgq/Msgq_test.go
@@ -159,8 +159,9 @@ func Benchmark_1(b *testing.B) {
        mq.Pull_tag_only(`test`, func(a any) (disable bool) {
                return false
        })
+       b.ResetTimer()
        for i := 0; i < b.N; i++ {
-               mq.PushLock_tag(`test`, i)
+               mq.Push_tag(`test`, i)
        }
 }
 
@@ -181,6 +182,17 @@ func Test_RemoveInPush(t *testing.T) {
        }
 }
 
+func Test_3(t *testing.T) {
+       mq := NewTo(time.Millisecond, time.Millisecond*3)
+       go mq.Push_tag(`sss`, nil)
+       mq.Pull_tag(FuncMap{
+               `test`: func(a any) (disable bool) {
+                       return false
+               },
+       })
+       time.Sleep(time.Millisecond * 500)
+}
+
 func Test_Pull_tag_chan(t *testing.T) {
        mq := New()
        ctx, cf := context.WithCancel(context.Background())

diff --git a/sync/RWMutex.go b/sync/RWMutex.go
index f79bd5bb73133f0d8e7c4b0915309a3f3bb91d5b..4943c5dfe24c8b04d326b32d86ff79bff3bf4073 100644 (file)
--- a/sync/RWMutex.go
+++ b/sync/RWMutex.go
@@ -10,19 +10,19 @@ import (
 const (
        lock  = -1
        ulock = 0
+       rlock = 0
 )
 
 type RWMutex struct {
-       rlc       atomic.Int32
-       wantRead  atomic.Bool
-       wantWrite atomic.Bool
+       rlc  atomic.Int32
+       want atomic.Int32
 }
 
 // to[0]: wait lock timeout to[1]: run lock timeout
 //
 // 不要在Rlock内设置变量，有DATA RACE风险
 func (m *RWMutex) RLock(to ...time.Duration) (unlockf func()) {
-       m.wantRead.Store(true)
+       getWant := m.want.CompareAndSwap(ulock, rlock)
        var callC atomic.Bool
        if len(to) > 0 {
                var calls []string
@@ -36,9 +36,9 @@ func (m *RWMutex) RLock(to ...time.Duration) (unlockf func()) {
                        }
                }
                c := time.Now()
-               for m.rlc.Load() < ulock || m.wantWrite.Load() {
+               for m.rlc.Load() < ulock || !getWant && !m.want.CompareAndSwap(ulock, rlock) {
                        if time.Since(c) > to[0] {
-                               panic(fmt.Sprintf("timeout to wait lock while rlocking, rlc:%d", m.rlc.Load()))
+                               panic(fmt.Sprintf("timeout to wait lock while rlocking, rlc:%d, want:%d, getWant:%v", m.rlc.Load(), m.want.Load(), getWant))
                        }
                        runtime.Gosched()
                }
@@ -54,7 +54,7 @@ func (m *RWMutex) RLock(to ...time.Duration) (unlockf func()) {
                        })
                }
        } else {
-               for m.rlc.Load() < ulock || m.wantWrite.Load() {
+               for m.rlc.Load() < ulock || !getWant && m.want.CompareAndSwap(ulock, rlock) {
                        runtime.Gosched()
                }
        }
@@ -64,14 +64,14 @@ func (m *RWMutex) RLock(to ...time.Duration) (unlockf func()) {
                        panic("had unrlock")
                }
                if m.rlc.Add(-1) == ulock {
-                       m.wantRead.Store(false)
+                       m.want.CompareAndSwap(rlock, ulock)
                }
        }
 }
 
 // to[0]: wait lock timeout to[1]: run lock timeout
 func (m *RWMutex) Lock(to ...time.Duration) (unlockf func()) {
-       m.wantWrite.Store(true)
+       getWant := m.want.CompareAndSwap(ulock, lock)
        var callC atomic.Bool
        if len(to) > 0 {
                var calls []string
@@ -85,9 +85,9 @@ func (m *RWMutex) Lock(to ...time.Duration) (unlockf func()) {
                        }
                }
                c := time.Now()
-               for m.rlc.Load() != ulock || m.wantRead.Load() {
+               for m.rlc.Load() != ulock || !getWant && !m.want.CompareAndSwap(ulock, lock) {
                        if time.Since(c) > to[0] {
-                               panic(fmt.Sprintf("timeout to wait rlock while locking, rlc:%d", m.rlc.Load()))
+                               panic(fmt.Sprintf("timeout to wait rlock while locking, rlc:%d, want:%v, getWant:%v", m.rlc.Load(), m.want.Load(), getWant))
                        }
                        runtime.Gosched()
                }
@@ -103,7 +103,7 @@ func (m *RWMutex) Lock(to ...time.Duration) (unlockf func()) {
                        })
                }
        } else {
-               for m.rlc.Load() != ulock || m.wantRead.Load() {
+               for m.rlc.Load() != ulock || !getWant && m.want.CompareAndSwap(ulock, lock) {
                        runtime.Gosched()
                }
        }
@@ -113,7 +113,7 @@ func (m *RWMutex) Lock(to ...time.Duration) (unlockf func()) {
                        panic("had unlock")
                }
                if m.rlc.Add(1) == ulock {
-                       m.wantWrite.Store(false)
+                       m.want.CompareAndSwap(lock, ulock)
                }
        }
 }