reuse workers

This commit is contained in:
Andy Pan 2018-05-19 12:57:01 +08:00
parent 7bf85c2a66
commit b7b2b27596
3 changed files with 14 additions and 12 deletions

View File

@ -11,7 +11,7 @@ func Push(task f) error {
} }
func Size() int { func Size() int {
return int(defaultPool.Size()) return int(defaultPool.Running())
} }
func Cap() int { func Cap() int {

20
pool.go
View File

@ -13,7 +13,7 @@ type f func()
type Pool struct { type Pool struct {
capacity int32 capacity int32
length int32 free int32
tasks chan f tasks chan f
workers chan *Worker workers chan *Worker
destroy chan sig destroy chan sig
@ -23,6 +23,7 @@ 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),
@ -32,7 +33,6 @@ func NewPool(size int) *Pool {
return p return p
} }
//------------------------------------------------------------------------- //-------------------------------------------------------------------------
func (p *Pool) loop() { func (p *Pool) loop() {
@ -57,8 +57,12 @@ func (p *Pool) Push(task f) error {
p.tasks <- task p.tasks <- task
return nil return nil
} }
func (p *Pool) Size() int32 { func (p *Pool) Running() int32 {
return atomic.LoadInt32(&p.length) return atomic.LoadInt32(&p.capacity) - atomic.LoadInt32(&p.free)
}
func (p *Pool) Free() int32 {
return atomic.LoadInt32(&p.free)
} }
func (p *Pool) Cap() int32 { func (p *Pool) Cap() int32 {
@ -74,11 +78,10 @@ func (p *Pool) Destroy() error {
return nil return nil
} }
//------------------------------------------------------------------------- //-------------------------------------------------------------------------
func (p *Pool) reachLimit() bool { func (p *Pool) reachLimit() bool {
return p.Size() >= p.Cap() return p.Running() >= p.Cap()
} }
func (p *Pool) newWorker() *Worker { func (p *Pool) newWorker() *Worker {
@ -88,16 +91,15 @@ func (p *Pool) newWorker() *Worker {
exit: make(chan sig), exit: make(chan sig),
} }
worker.run() worker.run()
atomic.AddInt32(&p.length, 1) atomic.AddInt32(&p.free, -1)
return worker return worker
} }
func (p *Pool) getWorker() *Worker { func (p *Pool) getWorker() *Worker {
var worker *Worker var worker *Worker
if p.reachLimit() { if p.reachLimit() || p.Free() > 0 {
worker = <-p.workers worker = <-p.workers
} }
worker = p.newWorker() worker = p.newWorker()
return worker return worker
} }

View File

@ -15,8 +15,8 @@ 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)
case <-w.exit: case <-w.exit:
atomic.AddInt32(&w.pool.length, -1)
return return
} }
} }