优化清理速度并修复内存泄漏

pool.go

@@ -68,6 +68,7 @@ func (p *Pool) periodicallyPurge() {
 	heartbeat := time.NewTicker(p.expiryDuration)
 	defer heartbeat.Stop()
 
+	var purgeWorkers []*Worker
 	for range heartbeat.C {
 		if CLOSED == atomic.LoadInt32(&p.release) {
 			break
@@ -75,23 +76,29 @@ func (p *Pool) periodicallyPurge() {
 		currentTime := time.Now()
 		p.lock.Lock()
 		idleWorkers := p.workers
-		n := -1
+		n := len(idleWorkers)
-		for i, w := range idleWorkers {
+		i := 0
-			if currentTime.Sub(w.recycleTime) <= p.expiryDuration {
+		for i < n && currentTime.Sub(idleWorkers[i].recycleTime) > p.expiryDuration {
-				break
+			i++
-			}
-			n = i
-			w.task <- nil
-			idleWorkers[i] = nil
 		}
-		if n > -1 {
+		purgeWorkers = append(purgeWorkers[:0], idleWorkers[:i]...)
-			if n >= len(idleWorkers)-1 {
+		if i > 0 {
-				p.workers = idleWorkers[:0]
+			m := copy(idleWorkers, idleWorkers[i:])
-			} else {
+			for i = m; i < n; i++ {
-				p.workers = idleWorkers[n+1:]
+				idleWorkers[i] = nil
 			}
+			p.workers = idleWorkers[:m]
 		}
 		p.lock.Unlock()
+
+		// Notify obsolete workers to stop.
+		// This notification must be outside the p.lock, since w.task
+		// may be blocking and may consume a lot of time if many workers
+		// are located on non-local CPUs.
+		for i, w := range purgeWorkers {
+			w.task <- nil
+			purgeWorkers[i] = nil
+		}
 	}
 }