This commit is contained in:
Andy Pan 2019-01-26 21:45:49 +08:00
parent f8d71bb276
commit 90f672b9d5
4 changed files with 25 additions and 6 deletions

12
pool.go
View File

@ -56,6 +56,8 @@ type Pool struct {
once sync.Once once sync.Once
cachePool sync.Pool
// PanicHandler is used to handle panics from each worker goroutine. // PanicHandler is used to handle panics from each worker goroutine.
// if nil, panics will be thrown out again from worker goroutines. // if nil, panics will be thrown out again from worker goroutines.
PanicHandler func(interface{}) PanicHandler func(interface{})
@ -185,7 +187,7 @@ func (p *Pool) decRunning() {
// getWorker returns a available worker to run the tasks. // getWorker returns a available worker to run the tasks.
func (p *Pool) getWorker() *Worker { func (p *Pool) getWorker() *Worker {
var w *Worker var w *Worker
waiting := false var waiting bool
p.lock.Lock() p.lock.Lock()
defer p.lock.Unlock() defer p.lock.Unlock()
@ -193,7 +195,13 @@ func (p *Pool) getWorker() *Worker {
idleWorkers := p.workers idleWorkers := p.workers
n := len(idleWorkers) - 1 n := len(idleWorkers) - 1
if n < 0 { if n < 0 {
waiting = p.Running() >= p.Cap() if p.Running() >= p.Cap() {
waiting = true
} else {
if cacheWorker := p.cachePool.Get(); cacheWorker != nil {
return cacheWorker.(*Worker)
}
}
} else { } else {
w = idleWorkers[n] w = idleWorkers[n]
idleWorkers[n] = nil idleWorkers[n] = nil

View File

@ -59,6 +59,8 @@ type PoolWithFunc struct {
once sync.Once once sync.Once
cachePool sync.Pool
// PanicHandler is used to handle panics from each worker goroutine. // PanicHandler is used to handle panics from each worker goroutine.
// if nil, panics will be thrown out again from worker goroutines. // if nil, panics will be thrown out again from worker goroutines.
PanicHandler func(interface{}) PanicHandler func(interface{})
@ -198,6 +200,13 @@ func (p *PoolWithFunc) getWorker() *WorkerWithFunc {
n := len(idleWorkers) - 1 n := len(idleWorkers) - 1
if n < 0 { if n < 0 {
waiting = p.Running() >= p.Cap() waiting = p.Running() >= p.Cap()
if p.Running() >= p.Cap() {
waiting = true
} else {
if cacheWorker := p.cachePool.Get(); cacheWorker != nil {
return cacheWorker.(*WorkerWithFunc)
}
}
} else { } else {
w = idleWorkers[n] w = idleWorkers[n]
idleWorkers[n] = nil idleWorkers[n] = nil

View File

@ -54,14 +54,15 @@ func (w *Worker) run() {
} }
} }
}() }()
for f := range w.task { for f := range w.task {
if f == nil { if f == nil {
w.pool.decRunning() w.pool.decRunning()
w.pool.cachePool.Put(w)
return return
} }
f() f()
w.pool.putWorker(w) w.pool.putWorker(w)
} }
}() }()
} }

View File

@ -54,14 +54,15 @@ func (w *WorkerWithFunc) run() {
} }
} }
}() }()
for args := range w.args { for args := range w.args {
if args == nil { if args == nil {
w.pool.decRunning() w.pool.decRunning()
w.pool.cachePool.Put(w)
return return
} }
w.pool.poolFunc(args) w.pool.poolFunc(args)
w.pool.putWorker(w) w.pool.putWorker(w)
} }
}() }()
} }