go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2025-07-04 20:33:45 +08:00
Code Issues Projects Releases Wiki Activity

Fix memory leaks caused by time.After

Browse Source
This commit is contained in:
biningo 2024-09-03 22:41:01 +08:00
parent d04888e748
commit b7cb257b5a
No known key found for this signature in database
GPG Key ID: B508A9E1E355023B
2 changed files with 4 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
subscriber.go
View File

@ -58,12 +58,13 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
err error err error
) )
// Try until successfully connect to Redis. // Try until successfully connect to Redis.
timer := time.NewTimer(s.retryTimeout)
for { for {
pubsub, err = s.broker.CancelationPubSub() pubsub, err = s.broker.CancelationPubSub()
if err != nil { if err != nil {
s.logger.Errorf("cannot subscribe to cancelation channel: %v", err) s.logger.Errorf("cannot subscribe to cancelation channel: %v", err)
select { select {
case <-time.After(s.retryTimeout): case <-timer.C:
continue continue
case <-s.done: case <-s.done:
s.logger.Debug("Subscriber done") s.logger.Debug("Subscriber done")

3
syncer.go
View File

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