From e52c29d081a14f0ec4efb01da11c69215ea40a6a Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Fri, 3 May 2019 13:37:39 +0300 Subject: better worker removal mechanism --- static_pool.go | 20 ++++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) (limited to 'static_pool.go') diff --git a/static_pool.go b/static_pool.go index 1409fc3f..fa4ac027 100644 --- a/static_pool.go +++ b/static_pool.go @@ -42,6 +42,7 @@ type StaticPool struct { workers []*Worker // invalid declares set of workers to be removed from the pool. + mur sync.Mutex remove sync.Map // pool is being destroyed @@ -146,13 +147,13 @@ func (p *StaticPool) Exec(rqs *Payload) (rsp *Payload, err error) { return nil, err } - go p.destroyWorker(w, err) + p.discardWorker(w, err) return nil, err } // worker want's to be terminated if rsp.Body == nil && rsp.Context != nil && string(rsp.Context) == StopRequest { - go p.destroyWorker(w, err) + p.discardWorker(w, err) return p.Exec(rqs) } @@ -172,6 +173,7 @@ func (p *StaticPool) Destroy() { var wg sync.WaitGroup for _, w := range p.Workers() { wg.Add(1) + w.markDisabled() go func(w *Worker) { defer wg.Done() p.destroyWorker(w, nil) @@ -195,7 +197,7 @@ func (p *StaticPool) allocateWorker() (w *Worker, err error) { } if err, remove := p.remove.Load(w); remove { - go p.destroyWorker(w, err) + p.discardWorker(w, err) // get next worker i++ @@ -222,7 +224,7 @@ func (p *StaticPool) allocateWorker() (w *Worker, err error) { } if err, remove := p.remove.Load(w); remove { - go p.destroyWorker(w, err) + p.discardWorker(w, err) // get next worker i++ @@ -243,12 +245,12 @@ func (p *StaticPool) allocateWorker() (w *Worker, err error) { // release releases or replaces the worker. func (p *StaticPool) release(w *Worker) { if p.cfg.MaxJobs != 0 && w.State().NumExecs() >= p.cfg.MaxJobs { - go p.destroyWorker(w, p.cfg.MaxJobs) + p.discardWorker(w, p.cfg.MaxJobs) return } if err, remove := p.remove.Load(w); remove { - go p.destroyWorker(w, err) + p.discardWorker(w, err) return } @@ -279,6 +281,12 @@ func (p *StaticPool) createWorker() (*Worker, error) { return w, nil } +// gentry remove worker +func (p *StaticPool) discardWorker(w *Worker, caused interface{}) { + w.markDisabled() + go p.destroyWorker(w, caused) +} + // destroyWorker destroys workers and removes it from the pool. func (p *StaticPool) destroyWorker(w *Worker, caused interface{}) { go w.Stop() -- cgit v1.2.3