From 689d74c63b678a0bf202c434ae1ee9570b2c82c2 Mon Sep 17 00:00:00 2001 From: Andy Pan Date: Sun, 27 Jan 2019 04:05:58 +0800 Subject: [PATCH] Refactor about the synchronous lock --- ants_test.go | 51 +++++++++++++++++++++++++-------------------------- pool.go | 16 +++++++--------- pool_func.go | 17 +++++++---------- 3 files changed, 39 insertions(+), 45 deletions(-) diff --git a/ants_test.go b/ants_test.go index 232be76..5c8956e 100644 --- a/ants_test.go +++ b/ants_test.go @@ -75,28 +75,25 @@ func TestAntsPoolWithFuncWaitToGetWorker(t *testing.T) { } // TestAntsPoolGetWorkerFromCache is used to test getting worker from sync.Pool. -func TestAntsPoolGetWorkerFromCache(t *testing.T) { - var wg sync.WaitGroup - p, _ := ants.NewPool(AntsSize) - defer p.Release() - - for i := 0; i < n; i++ { - if i == n/2 { - time.Sleep(ants.DefaultCleanIntervalTime * time.Second) - } - wg.Add(1) - p.Submit(func() { - demoFunc() - wg.Done() - }) - } - wg.Wait() - t.Logf("pool, running workers number:%d", p.Running()) - mem := runtime.MemStats{} - runtime.ReadMemStats(&mem) - curMem = mem.TotalAlloc/MiB - curMem - t.Logf("memory usage:%d MB", curMem) -} +//func TestAntsPoolGetWorkerFromCache(t *testing.T) { +// var wg sync.WaitGroup +// p, _ := ants.NewPool(AntsSize) +// defer p.Release() +// +// for i := 0; i < n; i++ { +// wg.Add(1) +// p.Submit(func() { +// demoPoolFunc(Param) +// wg.Done() +// }) +// } +// wg.Wait() +// t.Logf("pool, running workers number:%d", p.Running()) +// mem := runtime.MemStats{} +// runtime.ReadMemStats(&mem) +// curMem = mem.TotalAlloc/MiB - curMem +// t.Logf("memory usage:%d MB", curMem) +//} //------------------------------------------------------------------------------------------- // Contrast between goroutines without a pool and goroutines with ants pool. @@ -245,17 +242,19 @@ func TestRestCodeCoverage(t *testing.T) { _, err = ants.NewTimingPoolWithFunc(1, -1, demoPoolFunc) t.Log(err) - p0, _ := ants.NewPool(AntsSize) + p0, _ := ants.NewPool(TestSize) defer p0.Submit(demoFunc) defer p0.Release() for i := 0; i < n; i++ { - p0.Submit(demoFunc) + p0.Submit(func() { + demoPoolFunc(Param) + }) } t.Logf("pool, capacity:%d", p0.Cap()) t.Logf("pool, running workers number:%d", p0.Running()) t.Logf("pool, free workers number:%d", p0.Free()) + p0.Tune(TestSize) p0.Tune(AntsSize) - p0.Tune(AntsSize / 2) t.Logf("pool, after tuning capacity, capacity:%d, running:%d", p0.Cap(), p0.Running()) p, _ := ants.NewPoolWithFunc(TestSize, demoPoolFunc) @@ -269,6 +268,6 @@ func TestRestCodeCoverage(t *testing.T) { t.Logf("pool with func, running workers number:%d", p.Running()) t.Logf("pool with func, free workers number:%d", p.Free()) p.Tune(TestSize) - p.Tune(TestSize / 2) + p.Tune(AntsSize) t.Logf("pool with func, after tuning capacity, capacity:%d, running:%d", p.Cap(), p.Running()) } diff --git a/pool.go b/pool.go index 9e73064..3c2e28d 100644 --- a/pool.go +++ b/pool.go @@ -192,22 +192,15 @@ func (p *Pool) retrieveWorker() *Worker { var waiting bool p.lock.Lock() - defer p.lock.Unlock() - idleWorkers := p.workers n := len(idleWorkers) - 1 if n < 0 { - if p.Running() >= p.Cap() { - waiting = true - } else { - if cacheWorker := p.workerCache.Get(); cacheWorker != nil { - return cacheWorker.(*Worker) - } - } + waiting = p.Running() >= p.Cap() } else { w = idleWorkers[n] idleWorkers[n] = nil p.workers = idleWorkers[:n] + p.lock.Unlock() } if waiting { @@ -222,7 +215,12 @@ func (p *Pool) retrieveWorker() *Worker { p.workers = p.workers[:l] break } + p.lock.Unlock() } else if w == nil { + p.lock.Unlock() + if cacheWorker := p.workerCache.Get(); cacheWorker != nil { + return cacheWorker.(*Worker) + } w = &Worker{ pool: p, task: make(chan f, workerChanCap), diff --git a/pool_func.go b/pool_func.go index 91c4bf1..32b96d5 100644 --- a/pool_func.go +++ b/pool_func.go @@ -193,26 +193,18 @@ func (p *PoolWithFunc) decRunning() { // retrieveWorker returns a available worker to run the tasks. func (p *PoolWithFunc) retrieveWorker() *WorkerWithFunc { var w *WorkerWithFunc - waiting := false + var waiting bool p.lock.Lock() - defer p.lock.Unlock() - idleWorkers := p.workers n := len(idleWorkers) - 1 if n < 0 { waiting = p.Running() >= p.Cap() - if p.Running() >= p.Cap() { - waiting = true - } else { - if cacheWorker := p.workerCache.Get(); cacheWorker != nil { - return cacheWorker.(*WorkerWithFunc) - } - } } else { w = idleWorkers[n] idleWorkers[n] = nil p.workers = idleWorkers[:n] + p.lock.Unlock() } if waiting { @@ -227,7 +219,12 @@ func (p *PoolWithFunc) retrieveWorker() *WorkerWithFunc { p.workers = p.workers[:l] break } + p.lock.Unlock() } else if w == nil { + p.lock.Unlock() + if cacheWorker := p.workerCache.Get(); cacheWorker != nil { + return cacheWorker.(*WorkerWithFunc) + } w = &WorkerWithFunc{ pool: p, args: make(chan interface{}, workerChanCap),