Skip to content

Commit 23c4f48

Browse files
GlebRadchenkopanjf2000
GlebRadchenko
authored and
panjf2000
committed
fix: exit ticktock goroutine when pool is closed
1 parent 3fbd956 commit 23c4f48

File tree

2 files changed

+92
-38
lines changed

2 files changed

+92
-38
lines changed

‎pool.go

+46-19
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,9 @@ type Pool struct {
6262
heartbeatDone int32
6363
stopHeartbeat context.CancelFunc
6464

65+
ticktockDone int32
66+
stopTicktock context.CancelFunc
67+
6568
now atomic.Value
6669

6770
options *Options
@@ -111,15 +114,44 @@ func (p *Pool) purgeStaleWorkers(ctx context.Context) {
111114
}
112115

113116
// ticktock is a goroutine that updates the current time in the pool regularly.
114-
func (p *Pool) ticktock() {
117+
func (p *Pool) ticktock(ctx context.Context) {
115118
ticker := time.NewTicker(nowTimeUpdateInterval)
116-
defer ticker.Stop()
119+
defer func() {
120+
ticker.Stop()
121+
atomic.StoreInt32(&p.ticktockDone, 1)
122+
}()
123+
124+
for {
125+
select {
126+
case <-ctx.Done():
127+
return
128+
case <-ticker.C:
129+
}
130+
131+
if p.IsClosed() {
132+
break
133+
}
117134

118-
for range ticker.C {
119135
p.now.Store(time.Now())
120136
}
121137
}
122138

139+
func (p *Pool) startHeartbeat() {
140+
// Start a goroutine to clean up expired workers periodically.
141+
var ctx context.Context
142+
ctx, p.stopHeartbeat = context.WithCancel(context.Background())
143+
if !p.options.DisablePurge {
144+
go p.purgeStaleWorkers(ctx)
145+
}
146+
}
147+
148+
func (p *Pool) startTicktock() {
149+
p.now.Store(time.Now())
150+
var ctx context.Context
151+
ctx, p.stopTicktock = context.WithCancel(context.Background())
152+
go p.ticktock(ctx)
153+
}
154+
123155
func (p *Pool) nowTime() time.Time {
124156
return p.now.Load().(time.Time)
125157
}
@@ -166,15 +198,8 @@ func NewPool(size int, options ...Option) (*Pool, error) {
166198

167199
p.cond = sync.NewCond(p.lock)
168200

169-
// Start a goroutine to clean up expired workers periodically.
170-
var ctx context.Context
171-
ctx, p.stopHeartbeat = context.WithCancel(context.Background())
172-
if !p.options.DisablePurge {
173-
go p.purgeStaleWorkers(ctx)
174-
}
175-
176-
p.now.Store(time.Now())
177-
go p.ticktock()
201+
p.startHeartbeat()
202+
p.startTicktock()
178203

179204
return p, nil
180205
}
@@ -259,17 +284,21 @@ func (p *Pool) Release() {
259284

260285
// ReleaseTimeout is like Release but with a timeout, it waits all workers to exit before timing out.
261286
func (p *Pool) ReleaseTimeout(timeout time.Duration) error {
262-
if p.IsClosed() || p.stopHeartbeat == nil {
287+
if p.IsClosed() || p.stopHeartbeat == nil || p.stopTicktock == nil {
263288
return ErrPoolClosed
264289
}
265290

266291
p.stopHeartbeat()
267292
p.stopHeartbeat = nil
293+
p.stopTicktock()
294+
p.stopTicktock = nil
268295
p.Release()
269296

270297
endTime := time.Now().Add(timeout)
271298
for time.Now().Before(endTime) {
272-
if p.Running() == 0 && (p.options.DisablePurge || atomic.LoadInt32(&p.heartbeatDone) == 1) {
299+
if p.Running() == 0 &&
300+
(p.options.DisablePurge || atomic.LoadInt32(&p.heartbeatDone) == 1) &&
301+
atomic.LoadInt32(&p.ticktockDone) == 1 {
273302
return nil
274303
}
275304
time.Sleep(10 * time.Millisecond)
@@ -281,11 +310,9 @@ func (p *Pool) ReleaseTimeout(timeout time.Duration) error {
281310
func (p *Pool) Reboot() {
282311
if atomic.CompareAndSwapInt32(&p.state, CLOSED, OPENED) {
283312
atomic.StoreInt32(&p.heartbeatDone, 0)
284-
var ctx context.Context
285-
ctx, p.stopHeartbeat = context.WithCancel(context.Background())
286-
if !p.options.DisablePurge {
287-
go p.purgeStaleWorkers(ctx)
288-
}
313+
p.startHeartbeat()
314+
atomic.StoreInt32(&p.ticktockDone, 0)
315+
p.startTicktock()
289316
}
290317
}
291318

‎pool_func.go

+46-19
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,9 @@ type PoolWithFunc struct {
6464
heartbeatDone int32
6565
stopHeartbeat context.CancelFunc
6666

67+
ticktockDone int32
68+
stopTicktock context.CancelFunc
69+
6770
now atomic.Value
6871

6972
options *Options
@@ -134,15 +137,44 @@ func (p *PoolWithFunc) purgeStaleWorkers(ctx context.Context) {
134137
}
135138

136139
// ticktock is a goroutine that updates the current time in the pool regularly.
137-
func (p *PoolWithFunc) ticktock() {
140+
func (p *PoolWithFunc) ticktock(ctx context.Context) {
138141
ticker := time.NewTicker(nowTimeUpdateInterval)
139-
defer ticker.Stop()
142+
defer func() {
143+
ticker.Stop()
144+
atomic.StoreInt32(&p.ticktockDone, 1)
145+
}()
146+
147+
for {
148+
select {
149+
case <-ctx.Done():
150+
return
151+
case <-ticker.C:
152+
}
153+
154+
if p.IsClosed() {
155+
break
156+
}
140157

141-
for range ticker.C {
142158
p.now.Store(time.Now())
143159
}
144160
}
145161

162+
func (p *PoolWithFunc) startHeartbeat() {
163+
// Start a goroutine to clean up expired workers periodically.
164+
var ctx context.Context
165+
ctx, p.stopHeartbeat = context.WithCancel(context.Background())
166+
if !p.options.DisablePurge {
167+
go p.purgeStaleWorkers(ctx)
168+
}
169+
}
170+
171+
func (p *PoolWithFunc) startTicktock() {
172+
p.now.Store(time.Now())
173+
var ctx context.Context
174+
ctx, p.stopTicktock = context.WithCancel(context.Background())
175+
go p.ticktock(ctx)
176+
}
177+
146178
func (p *PoolWithFunc) nowTime() time.Time {
147179
return p.now.Load().(time.Time)
148180
}
@@ -191,15 +223,8 @@ func NewPoolWithFunc(size int, pf func(interface{}), options ...Option) (*PoolWi
191223
}
192224
p.cond = sync.NewCond(p.lock)
193225

194-
// Start a goroutine to clean up expired workers periodically.
195-
var ctx context.Context
196-
ctx, p.stopHeartbeat = context.WithCancel(context.Background())
197-
if !p.options.DisablePurge {
198-
go p.purgeStaleWorkers(ctx)
199-
}
200-
201-
p.now.Store(time.Now())
202-
go p.ticktock()
226+
p.startHeartbeat()
227+
p.startTicktock()
203228

204229
return p, nil
205230
}
@@ -288,17 +313,21 @@ func (p *PoolWithFunc) Release() {
288313

289314
// ReleaseTimeout is like Release but with a timeout, it waits all workers to exit before timing out.
290315
func (p *PoolWithFunc) ReleaseTimeout(timeout time.Duration) error {
291-
if p.IsClosed() || p.stopHeartbeat == nil {
316+
if p.IsClosed() || p.stopHeartbeat == nil || p.stopTicktock == nil {
292317
return ErrPoolClosed
293318
}
294319

295320
p.stopHeartbeat()
296321
p.stopHeartbeat = nil
322+
p.stopTicktock()
323+
p.stopTicktock = nil
297324
p.Release()
298325

299326
endTime := time.Now().Add(timeout)
300327
for time.Now().Before(endTime) {
301-
if p.Running() == 0 && (p.options.DisablePurge || atomic.LoadInt32(&p.heartbeatDone) == 1) {
328+
if p.Running() == 0 &&
329+
(p.options.DisablePurge || atomic.LoadInt32(&p.heartbeatDone) == 1) &&
330+
atomic.LoadInt32(&p.ticktockDone) == 1 {
302331
return nil
303332
}
304333
time.Sleep(10 * time.Millisecond)
@@ -310,11 +339,9 @@ func (p *PoolWithFunc) ReleaseTimeout(timeout time.Duration) error {
310339
func (p *PoolWithFunc) Reboot() {
311340
if atomic.CompareAndSwapInt32(&p.state, CLOSED, OPENED) {
312341
atomic.StoreInt32(&p.heartbeatDone, 0)
313-
var ctx context.Context
314-
ctx, p.stopHeartbeat = context.WithCancel(context.Background())
315-
if !p.options.DisablePurge {
316-
go p.purgeStaleWorkers(ctx)
317-
}
342+
p.startHeartbeat()
343+
atomic.StoreInt32(&p.ticktockDone, 0)
344+
p.startTicktock()
318345
}
319346
}
320347

0 commit comments

Comments
 (0)