add codes for reusing workers

This commit is contained in:
Andy Pan 2018-05-19 13:09:44 +08:00
parent b7b2b27596
commit 8ff88950b8
2 changed files with 13 additions and 9 deletions

20
pool.go
View File

@ -13,7 +13,7 @@ type f func()
type Pool struct { type Pool struct {
capacity int32 capacity int32
free int32 running int32
tasks chan f tasks chan f
workers chan *Worker workers chan *Worker
destroy chan sig destroy chan sig
@ -23,7 +23,6 @@ type Pool struct {
func NewPool(size int) *Pool { func NewPool(size int) *Pool {
p := &Pool{ p := &Pool{
capacity: int32(size), capacity: int32(size),
free: int32(size),
tasks: make(chan f, math.MaxInt32), tasks: make(chan f, math.MaxInt32),
//workers: &sync.Pool{New: func() interface{} { return &Worker{} }}, //workers: &sync.Pool{New: func() interface{} { return &Worker{} }},
workers: make(chan *Worker, size), workers: make(chan *Worker, size),
@ -58,11 +57,11 @@ func (p *Pool) Push(task f) error {
return nil return nil
} }
func (p *Pool) Running() int32 { func (p *Pool) Running() int32 {
return atomic.LoadInt32(&p.capacity) - atomic.LoadInt32(&p.free) return atomic.LoadInt32(&p.running)
} }
func (p *Pool) Free() int32 { func (p *Pool) Free() int32 {
return atomic.LoadInt32(&p.free) return atomic.LoadInt32(&p.capacity) - atomic.LoadInt32(&p.running)
} }
func (p *Pool) Cap() int32 { func (p *Pool) Cap() int32 {
@ -91,15 +90,20 @@ func (p *Pool) newWorker() *Worker {
exit: make(chan sig), exit: make(chan sig),
} }
worker.run() worker.run()
atomic.AddInt32(&p.free, -1) atomic.AddInt32(&p.running, 1)
return worker return worker
} }
func (p *Pool) getWorker() *Worker { func (p *Pool) getWorker() *Worker {
var worker *Worker var worker *Worker
if p.reachLimit() || p.Free() > 0 { if p.reachLimit() {
worker = <-p.workers select {
} case worker = <-p.workers:
return worker
default:
worker = p.newWorker() worker = p.newWorker()
}
}
worker = <-p.workers
return worker return worker
} }

View File

@ -15,7 +15,7 @@ func (w *Worker) run() {
case f := <-w.task: case f := <-w.task:
f() f()
w.pool.workers <- w w.pool.workers <- w
atomic.AddInt32(&w.pool.free, 1) atomic.AddInt32(&w.pool.running, -1)
case <-w.exit: case <-w.exit:
return return
} }