timing_wheel_service_test.go 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181
  1. package service
  2. import (
  3. "errors"
  4. "sync/atomic"
  5. "testing"
  6. "time"
  7. "github.com/zeromicro/go-zero/core/collection"
  8. )
  9. func TestNewTimingWheelService_InitFail_NoPanicAndReturnError(t *testing.T) {
  10. original := newTimingWheel
  11. t.Cleanup(func() { newTimingWheel = original })
  12. newTimingWheel = func(_ time.Duration, _ int, _ collection.Execute) (*collection.TimingWheel, error) {
  13. return nil, errors.New("boom")
  14. }
  15. svc, err := NewTimingWheelService()
  16. if err == nil {
  17. t.Fatalf("期望返回 error,但得到 nil")
  18. }
  19. if svc != nil {
  20. t.Fatalf("期望返回 nil svc,但得到非空")
  21. }
  22. }
  23. func TestNewTimingWheelService_Success(t *testing.T) {
  24. svc, err := NewTimingWheelService()
  25. if err != nil {
  26. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  27. }
  28. if svc == nil {
  29. t.Fatalf("期望 svc 非空,但得到 nil")
  30. }
  31. svc.Stop()
  32. }
  33. func TestNewTimingWheelService_ExecuteCallbackRunsFunc(t *testing.T) {
  34. original := newTimingWheel
  35. t.Cleanup(func() { newTimingWheel = original })
  36. var captured collection.Execute
  37. newTimingWheel = func(interval time.Duration, numSlots int, execute collection.Execute) (*collection.TimingWheel, error) {
  38. captured = execute
  39. return original(interval, numSlots, execute)
  40. }
  41. svc, err := NewTimingWheelService()
  42. if err != nil {
  43. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  44. }
  45. if captured == nil {
  46. t.Fatalf("期望 captured 非空,但得到 nil")
  47. }
  48. called := false
  49. captured("k", func() { called = true })
  50. if !called {
  51. t.Fatalf("期望 execute 回调触发传入函数执行")
  52. }
  53. svc.Stop()
  54. }
  55. func TestTimingWheelService_Schedule_ExecutesOnce(t *testing.T) {
  56. original := newTimingWheel
  57. t.Cleanup(func() { newTimingWheel = original })
  58. newTimingWheel = func(_ time.Duration, _ int, execute collection.Execute) (*collection.TimingWheel, error) {
  59. return original(10*time.Millisecond, 128, execute)
  60. }
  61. svc, err := NewTimingWheelService()
  62. if err != nil {
  63. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  64. }
  65. defer svc.Stop()
  66. ch := make(chan struct{}, 1)
  67. svc.Schedule("once", 30*time.Millisecond, func() { ch <- struct{}{} })
  68. select {
  69. case <-ch:
  70. case <-time.After(500 * time.Millisecond):
  71. t.Fatalf("等待任务执行超时")
  72. }
  73. select {
  74. case <-ch:
  75. t.Fatalf("任务不应重复执行")
  76. case <-time.After(80 * time.Millisecond):
  77. }
  78. }
  79. func TestTimingWheelService_Cancel_PreventsExecution(t *testing.T) {
  80. original := newTimingWheel
  81. t.Cleanup(func() { newTimingWheel = original })
  82. newTimingWheel = func(_ time.Duration, _ int, execute collection.Execute) (*collection.TimingWheel, error) {
  83. return original(10*time.Millisecond, 128, execute)
  84. }
  85. svc, err := NewTimingWheelService()
  86. if err != nil {
  87. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  88. }
  89. defer svc.Stop()
  90. ch := make(chan struct{}, 1)
  91. svc.Schedule("cancel", 80*time.Millisecond, func() { ch <- struct{}{} })
  92. svc.Cancel("cancel")
  93. select {
  94. case <-ch:
  95. t.Fatalf("任务已取消,不应执行")
  96. case <-time.After(200 * time.Millisecond):
  97. }
  98. }
  99. func TestTimingWheelService_Schedule_AfterStop_LogsError(t *testing.T) {
  100. svc, err := NewTimingWheelService()
  101. if err != nil {
  102. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  103. }
  104. svc.Stop()
  105. // Stop 后调用 Schedule 应走 error 日志路径,不应 panic
  106. svc.Schedule("after-stop", 100*time.Millisecond, func() {
  107. t.Fatal("不应被执行")
  108. })
  109. }
  110. func TestTimingWheelService_ScheduleRecurring_AfterStop_LogsError(t *testing.T) {
  111. svc, err := NewTimingWheelService()
  112. if err != nil {
  113. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  114. }
  115. svc.Stop()
  116. // Stop 后调用 ScheduleRecurring 应走 error 日志路径,不应 panic
  117. svc.ScheduleRecurring("after-stop-rec", 100*time.Millisecond, func() {
  118. t.Fatal("不应被执行")
  119. })
  120. }
  121. func TestTimingWheelService_Stop_Idempotent(t *testing.T) {
  122. svc, err := NewTimingWheelService()
  123. if err != nil {
  124. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  125. }
  126. svc.Stop()
  127. svc.Stop() // 第二次调用不应 panic
  128. }
  129. func TestTimingWheelService_ScheduleRecurring_ExecutesMultipleTimes(t *testing.T) {
  130. original := newTimingWheel
  131. t.Cleanup(func() { newTimingWheel = original })
  132. newTimingWheel = func(_ time.Duration, _ int, execute collection.Execute) (*collection.TimingWheel, error) {
  133. return original(10*time.Millisecond, 128, execute)
  134. }
  135. svc, err := NewTimingWheelService()
  136. if err != nil {
  137. t.Fatalf("期望 err 为 nil,但得到: %v", err)
  138. }
  139. defer svc.Stop()
  140. var count int32
  141. svc.ScheduleRecurring("rec", 30*time.Millisecond, func() { atomic.AddInt32(&count, 1) })
  142. deadline := time.Now().Add(500 * time.Millisecond)
  143. for atomic.LoadInt32(&count) < 2 && time.Now().Before(deadline) {
  144. time.Sleep(10 * time.Millisecond)
  145. }
  146. if atomic.LoadInt32(&count) < 2 {
  147. t.Fatalf("期望周期任务至少执行 2 次,但只执行了 %d 次", atomic.LoadInt32(&count))
  148. }
  149. }