Fix memory leaks caused by time.After

subscriber.go

@@ -8,9 +8,9 @@ import (
  "sync"
  "time"
 
- "github.com/redis/go-redis/v9"
  "github.com/hibiken/asynq/internal/base"
  "github.com/hibiken/asynq/internal/log"
+ "github.com/redis/go-redis/v9"
 )
 
 type subscriber struct {
@@ -58,12 +58,15 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
  err error
  )
  // Try until successfully connect to Redis.
+ timer := time.NewTimer(s.retryTimeout)
+ defer timer.Stop()
  for {
  pubsub, err = s.broker.CancelationPubSub()
  if err != nil {
  s.logger.Errorf("cannot subscribe to cancelation channel: %v", err)
  select {
- case <-time.After(s.retryTimeout):
+ case <-timer.C:
+ timer.Reset(s.retryTimeout)
  continue
  case <-s.done:
  s.logger.Debug("Subscriber done")

syncer.go

@@ -57,6 +57,8 @@ func (s *syncer) start(wg *sync.WaitGroup) {
  go func() {
  defer wg.Done()
  var requests []*syncRequest
+ timer := time.NewTimer(s.interval)
+ defer timer.Stop()
  for {
  select {
  case <-s.done:
@@ -70,7 +72,7 @@ func (s *syncer) start(wg *sync.WaitGroup) {
  return
  case req := <-s.requestsCh:
  requests = append(requests, req)
- case <-time.After(s.interval):
+ case <-timer.C:
  var temp []*syncRequest
  for _, req := range requests {
  if req.deadline.Before(time.Now()) {
@@ -81,6 +83,7 @@ func (s *syncer) start(wg *sync.WaitGroup) {
  }
  }
  requests = temp
+ timer.Reset(s.interval)
  }
  }
  }()