github/client-go
1
0
Fork 0
mirror of https://github.com/kubernetes/client-go.git synced 2025-07-16 00:06:55 +00:00
Code Issues Projects Releases Wiki Activity

Stop() for Ticker to enable leak-free code

Browse Source 
Kubernetes-commit: 1f393cdef96fe6e4ddcbf93825d65a9980463406
This commit is contained in:
Mikhail Mazurskiy 2018-03-31 19:41:43 +11:00 committed by Kubernetes Publisher
parent 9cabba1998
commit 9446f4ef9d
2 changed files with 5 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
util/workqueue/delaying_queue.go
View File

@ -45,7 +45,7 @@ func newDelayingQueue(clock clock.Clock, name string) DelayingInterface {
ret := &delayingType{ ret := &delayingType{
Interface: NewNamed(name), Interface: NewNamed(name),
clock: clock, clock: clock,
heartbeat: clock.Tick(maxWait), heartbeat: clock.NewTicker(maxWait),
stopCh: make(chan struct{}), stopCh: make(chan struct{}),
waitingForAddCh: make(chan *waitFor, 1000), waitingForAddCh: make(chan *waitFor, 1000),
metrics: newRetryMetrics(name), metrics: newRetryMetrics(name),
@ -67,10 +67,7 @@ type delayingType struct {
stopCh chan struct{} stopCh chan struct{}
// heartbeat ensures we wait no more than maxWait before firing // heartbeat ensures we wait no more than maxWait before firing
// heartbeat clock.Ticker
// TODO: replace with Ticker (and add to clock) so this can be cleaned up.
// clock.Tick will leak.
heartbeat <-chan time.Time
// waitingForAddCh is a buffered channel that feeds waitingForAdd // waitingForAddCh is a buffered channel that feeds waitingForAdd
waitingForAddCh chan *waitFor waitingForAddCh chan *waitFor
@ -138,6 +135,7 @@ func (pq waitForPriorityQueue) Peek() interface{} {
func (q *delayingType) ShutDown() { func (q *delayingType) ShutDown() {
q.Interface.ShutDown() q.Interface.ShutDown()
close(q.stopCh) close(q.stopCh)
q.heartbeat.Stop()
} }
// AddAfter adds the given item to the work queue after the given delay // AddAfter adds the given item to the work queue after the given delay
@ -209,7 +207,7 @@ func (q *delayingType) waitingLoop() {
case <-q.stopCh: case <-q.stopCh:
return return
case <-q.heartbeat: case <-q.heartbeat.C():
// continue the loop, which will add ready items // continue the loop, which will add ready items
case <-nextReadyAt: case <-nextReadyAt:

2
util/workqueue/rate_limitting_queue_test.go
View File

@ -30,7 +30,7 @@ func TestRateLimitingQueue(t *testing.T) {
delayingQueue := &delayingType{ delayingQueue := &delayingType{
Interface: New(), Interface: New(),
clock: fakeClock, clock: fakeClock,
heartbeat: fakeClock.Tick(maxWait), heartbeat: fakeClock.NewTicker(maxWait),
stopCh: make(chan struct{}), stopCh: make(chan struct{}),
waitingForAddCh: make(chan *waitFor, 1000), waitingForAddCh: make(chan *waitFor, 1000),
metrics: newRetryMetrics(""), metrics: newRetryMetrics(""),