Fix a bug that blocks callers infinitely

Fixes #141
This commit is contained in:
Andy Pan 2021-03-16 22:41:49 +08:00
parent e45d13c630
commit 36c4272286
2 changed files with 32 additions and 12 deletions

22
pool.go
View File

@ -67,7 +67,7 @@ func (p *Pool) purgePeriodically() {
defer heartbeat.Stop() defer heartbeat.Stop()
for range heartbeat.C { for range heartbeat.C {
if atomic.LoadInt32(&p.state) == CLOSED { if p.IsClosed() {
break break
} }
@ -143,7 +143,7 @@ func NewPool(size int, options ...Option) (*Pool, error) {
// Submit submits a task to this pool. // Submit submits a task to this pool.
func (p *Pool) Submit(task func()) error { func (p *Pool) Submit(task func()) error {
if atomic.LoadInt32(&p.state) == CLOSED { if p.IsClosed() {
return ErrPoolClosed return ErrPoolClosed
} }
var w *goWorker var w *goWorker
@ -177,12 +177,20 @@ func (p *Pool) Tune(size int) {
atomic.StoreInt32(&p.capacity, int32(size)) atomic.StoreInt32(&p.capacity, int32(size))
} }
// IsClosed indicates whether the pool is closed.
func (p *Pool) IsClosed() bool {
return atomic.LoadInt32(&p.state) == CLOSED
}
// Release Closes this pool. // Release Closes this pool.
func (p *Pool) Release() { func (p *Pool) Release() {
atomic.StoreInt32(&p.state, CLOSED) atomic.StoreInt32(&p.state, CLOSED)
p.lock.Lock() p.lock.Lock()
p.workers.reset() p.workers.reset()
p.lock.Unlock() p.lock.Unlock()
// There might be some callers waiting in retrieveWorker(), so we need to wake them up to prevent
// those callers blocking infinitely.
p.cond.Broadcast()
} }
// Reboot reboots a released pool. // Reboot reboots a released pool.
@ -236,8 +244,10 @@ func (p *Pool) retrieveWorker() (w *goWorker) {
p.cond.Wait() p.cond.Wait()
p.blockingNum-- p.blockingNum--
if p.Running() == 0 { if p.Running() == 0 {
p.lock.Unlock() if !p.IsClosed() {
spawnWorker() p.lock.Unlock()
spawnWorker()
}
return return
} }
@ -253,7 +263,7 @@ func (p *Pool) retrieveWorker() (w *goWorker) {
// revertWorker puts a worker back into free pool, recycling the goroutines. // revertWorker puts a worker back into free pool, recycling the goroutines.
func (p *Pool) revertWorker(worker *goWorker) bool { func (p *Pool) revertWorker(worker *goWorker) bool {
if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || atomic.LoadInt32(&p.state) == CLOSED { if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || p.IsClosed() {
return false return false
} }
worker.recycleTime = time.Now() worker.recycleTime = time.Now()
@ -261,7 +271,7 @@ func (p *Pool) revertWorker(worker *goWorker) bool {
// To avoid memory leaks, add a double check in the lock scope. // To avoid memory leaks, add a double check in the lock scope.
// Issue: https://github.com/panjf2000/ants/issues/113 // Issue: https://github.com/panjf2000/ants/issues/113
if atomic.LoadInt32(&p.state) == CLOSED { if p.IsClosed() {
p.lock.Unlock() p.lock.Unlock()
return false return false
} }

View File

@ -70,7 +70,7 @@ func (p *PoolWithFunc) purgePeriodically() {
var expiredWorkers []*goWorkerWithFunc var expiredWorkers []*goWorkerWithFunc
for range heartbeat.C { for range heartbeat.C {
if atomic.LoadInt32(&p.state) == CLOSED { if p.IsClosed() {
break break
} }
currentTime := time.Now() currentTime := time.Now()
@ -157,7 +157,7 @@ func NewPoolWithFunc(size int, pf func(interface{}), options ...Option) (*PoolWi
// Invoke submits a task to pool. // Invoke submits a task to pool.
func (p *PoolWithFunc) Invoke(args interface{}) error { func (p *PoolWithFunc) Invoke(args interface{}) error {
if atomic.LoadInt32(&p.state) == CLOSED { if p.IsClosed() {
return ErrPoolClosed return ErrPoolClosed
} }
var w *goWorkerWithFunc var w *goWorkerWithFunc
@ -191,6 +191,11 @@ func (p *PoolWithFunc) Tune(size int) {
atomic.StoreInt32(&p.capacity, int32(size)) atomic.StoreInt32(&p.capacity, int32(size))
} }
// IsClosed indicates whether the pool is closed.
func (p *PoolWithFunc) IsClosed() bool {
return atomic.LoadInt32(&p.state) == CLOSED
}
// Release Closes this pool. // Release Closes this pool.
func (p *PoolWithFunc) Release() { func (p *PoolWithFunc) Release() {
atomic.StoreInt32(&p.state, CLOSED) atomic.StoreInt32(&p.state, CLOSED)
@ -201,6 +206,9 @@ func (p *PoolWithFunc) Release() {
} }
p.workers = nil p.workers = nil
p.lock.Unlock() p.lock.Unlock()
// There might be some callers waiting in retrieveWorker(), so we need to wake them up to prevent
// those callers blocking infinitely.
p.cond.Broadcast()
} }
// Reboot reboots a released pool. // Reboot reboots a released pool.
@ -254,8 +262,10 @@ func (p *PoolWithFunc) retrieveWorker() (w *goWorkerWithFunc) {
p.cond.Wait() p.cond.Wait()
p.blockingNum-- p.blockingNum--
if p.Running() == 0 { if p.Running() == 0 {
p.lock.Unlock() if !p.IsClosed() {
spawnWorker() p.lock.Unlock()
spawnWorker()
}
return return
} }
l := len(p.workers) - 1 l := len(p.workers) - 1
@ -272,7 +282,7 @@ func (p *PoolWithFunc) retrieveWorker() (w *goWorkerWithFunc) {
// revertWorker puts a worker back into free pool, recycling the goroutines. // revertWorker puts a worker back into free pool, recycling the goroutines.
func (p *PoolWithFunc) revertWorker(worker *goWorkerWithFunc) bool { func (p *PoolWithFunc) revertWorker(worker *goWorkerWithFunc) bool {
if atomic.LoadInt32(&p.state) == CLOSED || p.Running() > p.Cap() { if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || p.IsClosed() {
return false return false
} }
worker.recycleTime = time.Now() worker.recycleTime = time.Now()
@ -280,7 +290,7 @@ func (p *PoolWithFunc) revertWorker(worker *goWorkerWithFunc) bool {
// To avoid memory leaks, add a double check in the lock scope. // To avoid memory leaks, add a double check in the lock scope.
// Issue: https://github.com/panjf2000/ants/issues/113 // Issue: https://github.com/panjf2000/ants/issues/113
if atomic.LoadInt32(&p.state) == CLOSED { if p.IsClosed() {
p.lock.Unlock() p.lock.Unlock()
return false return false
} }