更新

antlabs · Feb 10, 2024 · f3e81a4 · f3e81a4
1 parent 764be37
commit f3e81a4
Show file tree

Hide file tree

Showing 2 changed files with 76 additions and 30 deletions.
diff --git a/min_heap_node.go b/min_heap_node.go
@@ -18,10 +18,7 @@ func (m *minHeapNode) Stop() {
 	m.root.removeTimeNode(m)
 }
 func (m *minHeapNode) Reset(d time.Duration) {
-	// m.root.removeTimeNode(m)
-	// m.userExpire = d
-	// m.absExpire = time.Now().Add(d)
-	// // m.root.addTimeNode(m)
+
 }
 
 func (m *minHeapNode) Next(now time.Time) time.Time {

diff --git a/min_heap_test.go b/min_heap_test.go
@@ -1,6 +1,8 @@
 package timer
 
 import (
+	"log"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -10,52 +12,94 @@ import (
 func Test_MinHeap_AfterFunc_Run(t *testing.T) {
 	t.Run("1ms", func(t *testing.T) {
 		tm := NewTimer(WithMinHeap())
-		now := time.Now()
+
 		go tm.Run()
 		count := int32(0)
 
 		tc := make(chan time.Duration, 2)
 
-		tm.AfterFunc(time.Millisecond, func() {
+		var mu sync.Mutex
+		isClose := false
+		now := time.Now()
+		node1 := tm.AfterFunc(time.Millisecond, func() {
+
+			mu.Lock()
 			atomic.AddInt32(&count, 1)
-			tc <- time.Since(now)
+			if atomic.LoadInt32(&count) <= 2 && !isClose {
+				tc <- time.Since(now)
+			}
+			mu.Unlock()
 		})
-		tm.AfterFunc(time.Millisecond, func() {
+
+		node2 := tm.AfterFunc(time.Millisecond, func() {
+			mu.Lock()
 			atomic.AddInt32(&count, 1)
-			tc <- time.Since(now)
+			if atomic.LoadInt32(&count) <= 2 && !isClose {
+				tc <- time.Since(now)
+			}
+			mu.Unlock()
 		})
 
-		time.Sleep(time.Millisecond * 2)
+		time.Sleep(time.Millisecond * 3)
+		mu.Lock()
+		isClose = true
 		close(tc)
+		node1.Stop()
+		node2.Stop()
+		mu.Unlock()
 		for tv := range tc {
 			if tv < time.Millisecond || tv > 2*time.Millisecond {
 				t.Errorf("tc < time.Millisecond tc > 2*time.Millisecond")
 
 			}
 		}
 		if atomic.LoadInt32(&count) != 2 {
-			t.Errorf("count != 2")
+			t.Errorf("count:%d != 2", atomic.LoadInt32(&count))
 		}
 
 	})
 
 	t.Run("10ms", func(t *testing.T) {
 		tm := NewTimer(WithMinHeap())
-		now := time.Now()
-		go tm.Run()
+
+		go tm.Run() // 运行事件循环
 		count := int32(0)
 		tc := make(chan time.Duration, 2)
-		tm.AfterFunc(time.Millisecond*10, func() {
+
+		var mu sync.Mutex
+		isClosed := false
+		now := time.Now()
+		node1 := tm.AfterFunc(time.Millisecond*10, func() {
+			now2 := time.Now()
+			mu.Lock()
 			atomic.AddInt32(&count, 1)
-			tc <- time.Since(now)
+			if atomic.LoadInt32(&count) <= 2 && !isClosed {
+				tc <- time.Since(now)
+			}
+			mu.Unlock()
+			log.Printf("node1.Lock:%v\n", time.Since(now2))
 		})
-		tm.AfterFunc(time.Millisecond*10, func() {
+		node2 := tm.AfterFunc(time.Millisecond*10, func() {
+			now2 := time.Now()
+			mu.Lock()
 			atomic.AddInt32(&count, 1)
-			tc <- time.Since(now)
+			if atomic.LoadInt32(&count) <= 2 && !isClosed {
+				tc <- time.Since(now)
+			}
+			mu.Unlock()
+			log.Printf("node2.Lock:%v\n", time.Since(now2))
 		})
 
-		time.Sleep(time.Millisecond * 20)
+		time.Sleep(time.Millisecond * 24)
+		now3 := time.Now()
+		mu.Lock()
+		node1.Stop()
+		node2.Stop()
+		isClosed = true
 		close(tc)
+		mu.Unlock()
+
+		log.Printf("node1.Stop:%v\n", time.Since(now3))
 		cnt := 1
 		for tv := range tc {
 			left := time.Millisecond * 10 * time.Duration(cnt)
@@ -66,7 +110,7 @@ func Test_MinHeap_AfterFunc_Run(t *testing.T) {
 			// cnt++
 		}
 		if atomic.LoadInt32(&count) != 2 {
-			t.Errorf("count != 2")
+			t.Errorf("count:%d != 2", atomic.LoadInt32(&count))
 		}
 
 	})
@@ -108,7 +152,7 @@ func Test_MinHeap_ScheduleFunc_Run(t *testing.T) {
 
 		time.Sleep(time.Millisecond * 5)
 		if atomic.LoadInt32(&count) != 2 {
-			t.Errorf("count != 2")
+			t.Errorf("count:%d != 2", atomic.LoadInt32(&count))
 		}
 
 	})
@@ -117,24 +161,29 @@ func Test_MinHeap_ScheduleFunc_Run(t *testing.T) {
 		tm := NewTimer(WithMinHeap())
 		go tm.Run()
 		count := int32(0)
-		c := make(chan bool, 1)
 		tc := make(chan time.Duration, 2)
+		var mu sync.Mutex
+		isClosed := false
 		now := time.Now()
+
 		node := tm.ScheduleFunc(time.Millisecond*10, func() {
+			mu.Lock()
 			atomic.AddInt32(&count, 1)
-			tc <- time.Since(now)
-			if atomic.LoadInt32(&count) >= 2 {
-				c <- true
+
+			if atomic.LoadInt32(&count) <= 2 && !isClosed {
+				tc <- time.Since(now)
 			}
+			mu.Unlock()
 		})
 
-		go func() {
-			<-c
-			node.Stop()
-		}()
+		time.Sleep(time.Millisecond * 25)
 
-		time.Sleep(time.Millisecond * 40)
+		mu.Lock()
 		close(tc)
+		isClosed = true
+		node.Stop()
+		mu.Unlock()
+
 		cnt := 1
 		for tv := range tc {
 			left := time.Millisecond * 10 * time.Duration(cnt)
@@ -170,7 +219,7 @@ func Test_MinHeap_ScheduleFunc_Run(t *testing.T) {
 
 		time.Sleep(time.Millisecond * 70)
 		if atomic.LoadInt32(&count) != 2 {
-			t.Errorf("count != 2")
+			t.Errorf("count:%d != 2", atomic.LoadInt32(&count))
 		}
 
 	})