gopkg/pond
2
0
Fork 0
Code Pull Requests Activity

Add ability to configure pool growth strategy

Browse Source
This commit is contained in:
alitto
2020-05-22 21:03:43 -03:00
parent 349c5aa212
commit dc67fea72f
16 changed files with 671 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pond.go
+227 -66
View File
@@ -2,6 +2,7 @@ package pond
import (
"fmt"
"math"
"runtime/debug"
"sync"
"sync/atomic"
@@ -19,15 +20,9 @@ func defaultPanicHandler(panic interface{}) {
fmt.Printf("Worker exits from a panic: %v\nStack trace: %s\n", panic, string(debug.Stack()))
}
// linearGrowthFn is a function that determines how many workers to create when backpressure is detected
func linearGrowthFn(workerCount, minWorkers, maxWorkers int) int {
if workerCount < minWorkers {
return minWorkers
}
if workerCount < maxWorkers {
return 1
}
return 0
// ResizingStrategy represents a pool resizing strategy
type ResizingStrategy interface {
Resize(runningWorkers, idleWorkers, minWorkers, maxWorkers, incomingTasks, completedTasks int, delta time.Duration) int
}
// Option represents an option that can be passed when instantiating a worker pool to customize it
@@ -40,13 +35,6 @@ func IdleTimeout(idleTimeout time.Duration) Option {
}
}
// PanicHandler allows to change the panic handler function for a worker pool
func PanicHandler(panicHandler func(interface{})) Option {
return func(pool *WorkerPool) {
pool.panicHandler = panicHandler
}
}
// MinWorkers allows to change the minimum number of workers of a worker pool
func MinWorkers(minWorkers int) Option {
return func(pool *WorkerPool) {
@@ -54,20 +42,42 @@ func MinWorkers(minWorkers int) Option {
}
}
// Strategy allows to change the strategy used to resize the pool
func Strategy(strategy ResizingStrategy) Option {
return func(pool *WorkerPool) {
pool.strategy = strategy
}
}
// PanicHandler allows to change the panic handler function for a worker pool
func PanicHandler(panicHandler func(interface{})) Option {
return func(pool *WorkerPool) {
pool.panicHandler = panicHandler
}
}
// WorkerPool models a pool of workers
type WorkerPool struct {
minWorkers int
maxWorkers int
maxCapacity int
idleTimeout time.Duration
workerCount int32
// Configurable settings
maxWorkers int
maxCapacity int
minWorkers int
idleTimeout time.Duration
strategy ResizingStrategy
panicHandler func(interface{})
// Atomic counters
workerCount int32
idleWorkerCount int32
completedTaskCount uint64
// Private properties
tasks chan func()
dispatchedTasks chan func()
purgerQuit chan struct{}
stopOnce sync.Once
waitGroup sync.WaitGroup
panicHandler func(interface{})
growthFn func(int, int, int) int
// Debug information
debug bool
maxWorkerCount int
}
// New creates a worker pool with that can scale up to the given maximum number of workers (maxWorkers).
@@ -81,9 +91,8 @@ func New(maxWorkers, maxCapacity int, options ...Option) *WorkerPool {
maxWorkers: maxWorkers,
maxCapacity: maxCapacity,
idleTimeout: defaultIdleTimeout,
purgerQuit: make(chan struct{}),
strategy: Balanced,
panicHandler: defaultPanicHandler,
growthFn: linearGrowthFn,
}
// Apply all options
@@ -105,9 +114,10 @@ func New(maxWorkers, maxCapacity int, options ...Option) *WorkerPool {
pool.idleTimeout = defaultIdleTimeout
}
// Create channels
// Create internal channels
pool.tasks = make(chan func(), pool.maxCapacity)
pool.dispatchedTasks = make(chan func(), pool.maxWorkers)
pool.purgerQuit = make(chan struct{})
// Start dispatcher goroutine
pool.waitGroup.Add(1)
@@ -127,7 +137,7 @@ func New(maxWorkers, maxCapacity int, options ...Option) *WorkerPool {
// Start minWorkers workers
if pool.minWorkers > 0 {
pool.startWorkers()
pool.startWorkers(pool.minWorkers, nil)
}
return pool
@@ -138,6 +148,11 @@ func (p *WorkerPool) Running() int {
return int(atomic.LoadInt32(&p.workerCount))
}
// Idle returns the number of idle workers
func (p *WorkerPool) Idle() int {
return int(atomic.LoadInt32(&p.idleWorkerCount))
}
// Submit sends a task to this worker pool for execution. If the queue is full,
// it will wait until the task can be enqueued
func (p *WorkerPool) Submit(task func()) {
@@ -188,8 +203,8 @@ func (p *WorkerPool) SubmitBefore(task func(), deadline time.Duration) {
// Stop causes this pool to stop accepting tasks, without waiting for goroutines to exit
func (p *WorkerPool) Stop() {
p.stopOnce.Do(func() {
// Close the tasks channel to prevent receiving new tasks
close(p.tasks)
// Send signal to stop the purger
close(p.purgerQuit)
})
}
@@ -204,38 +219,120 @@ func (p *WorkerPool) StopAndWait() {
// dispatch represents the work done by the dispatcher goroutine
func (p *WorkerPool) dispatch() {
batchSize := p.maxWorkers
batch := make([]func(), 0)
batchSize := int(math.Max(float64(p.minWorkers), 1000))
var lastCompletedTasks uint64 = 0
var lastCycle time.Time = time.Now()
for task := range p.tasks {
batch = append(batch, task)
// Read up to batchSize - 1 tasks without blocking
BulkReceive:
for i := 0; i < batchSize; i++ {
idleCount := p.Idle()
dispatchedImmediately := 0
// Dispatch up to idleCount tasks without blocking
nextTask := task
ImmediateDispatch:
for i := 0; i < idleCount; i++ {
// Attempt to dispatch
select {
case t := <-p.tasks:
batch = append(batch, t)
case p.dispatchedTasks <- nextTask:
dispatchedImmediately++
default:
break ImmediateDispatch
}
// Attempt to receive another task
select {
case t, ok := <-p.tasks:
if !ok {
// Nothing to dispatch
nextTask = nil
break ImmediateDispatch
}
nextTask = t
default:
nextTask = nil
break ImmediateDispatch
}
}
if nextTask == nil {
continue
}
// Start batching tasks
batch = append(batch, nextTask)
// Read up to batchSize tasks without blocking
BulkReceive:
for i := 0; i < batchSize-1; i++ {
select {
case t, ok := <-p.tasks:
if !ok {
break BulkReceive
}
if t != nil {
batch = append(batch, t)
}
default:
break BulkReceive
}
}
for _, task := range batch {
select {
// Attempt to submit the task to a worker without blocking
case p.dispatchedTasks <- task:
if p.Running() == 0 {
p.startWorkers()
}
default:
// Create a new worker if we haven't reached the limit yet
if p.Running() < p.maxWorkers {
p.startWorkers()
}
// Resize the pool
now := time.Now()
delta := now.Sub(lastCycle)
workload := len(batch)
runningCount := p.Running()
lastCycle = now
currentCompletedTasks := atomic.LoadUint64(&p.completedTaskCount)
completedTasks := int(currentCompletedTasks - lastCompletedTasks)
if completedTasks < 0 {
completedTasks = 0
}
lastCompletedTasks = currentCompletedTasks
targetDelta := p.calculatePoolSizeDelta(runningCount, idleCount, workload+dispatchedImmediately, completedTasks, delta)
// Block until a worker accepts this task
p.dispatchedTasks <- task
// Start up to targetDelta workers
dispatched := 0
if targetDelta > 0 {
p.startWorkers(targetDelta, batch)
dispatched = workload
if targetDelta < workload {
dispatched = targetDelta
}
} else if targetDelta < 0 {
// Kill targetDelta workers
for i := 0; i < -targetDelta; i++ {
p.dispatchedTasks <- nil
}
}
dispatchedBlocking := 0
if workload > dispatched {
for _, task := range batch[dispatched:] {
// Attempt to dispatch the task without blocking
select {
case p.dispatchedTasks <- task:
default:
// Block until a worker accepts this task
p.dispatchedTasks <- task
dispatchedBlocking++
}
}
}
// Adjust batch size
if dispatchedBlocking > 0 {
if batchSize > 1 {
batchSize = 1
}
} else {
maxBatchSize := runningCount + targetDelta
batchSize = batchSize * 2
if batchSize > maxBatchSize {
batchSize = maxBatchSize
}
}
@@ -243,8 +340,8 @@ func (p *WorkerPool) dispatch() {
batch = nil
}
// Send signal to stop the purger
close(p.purgerQuit)
// Send signal to stop all workers
close(p.dispatchedTasks)
}
// purge represents the work done by the purger goroutine
@@ -254,40 +351,81 @@ func (p *WorkerPool) purge() {
for {
select {
// Timed out waiting for any activity to happen, attempt to stop an idle worker
// Timed out waiting for any activity to happen, attempt to resize the pool
case <-ticker.C:
if p.Running() > p.minWorkers {
if p.Idle() > 0 {
select {
case p.dispatchedTasks <- nil:
case p.tasks <- nil:
default:
// If dispatchedTasks channel is full, no need to kill the worker
// If tasks channel is full, there's no need to resize the pool
}
}
// Received the signal to exit
case <-p.purgerQuit:
// Send signal to stop all workers
close(p.dispatchedTasks)
// Close the tasks channel to prevent receiving new tasks
close(p.tasks)
return
}
}
}
// startWorkers launches worker goroutines according to the growth function
func (p *WorkerPool) startWorkers() {
// calculatePoolSizeDelta calculates what's the delta to reach the ideal pool size based on the current size and workload
func (p *WorkerPool) calculatePoolSizeDelta(runningWorkers, idleWorkers,
incomingTasks, completedTasks int, duration time.Duration) int {
count := p.growthFn(p.Running(), p.minWorkers, p.maxWorkers)
delta := p.strategy.Resize(runningWorkers, idleWorkers, p.minWorkers, p.maxWorkers,
incomingTasks, completedTasks, duration)
targetSize := runningWorkers + delta
// Cannot go below minWorkers
if targetSize < p.minWorkers {
targetSize = p.minWorkers
}
// Cannot go above maxWorkers
if targetSize > p.maxWorkers {
targetSize = p.maxWorkers
}
if p.debug {
// Print debugging information
durationSecs := duration.Seconds()
inputRate := float64(incomingTasks) / durationSecs
outputRate := float64(completedTasks) / durationSecs
message := fmt.Sprintf("%d\t%d\t%d\t%d\t\"%f\"\t\"%f\"\t%d\t\"%f\"\n",
runningWorkers, idleWorkers, incomingTasks, completedTasks,
inputRate, outputRate,
delta, durationSecs)
fmt.Printf(message)
}
return targetSize - runningWorkers
}
// startWorkers creates new worker goroutines to run the given tasks
func (p *WorkerPool) startWorkers(count int, firstTasks []func()) {
// Increment worker count
atomic.AddInt32(&p.workerCount, int32(count))
workerCount := atomic.AddInt32(&p.workerCount, int32(count))
// Collect debug information
if p.debug && int(workerCount) > p.maxWorkerCount {
p.maxWorkerCount = int(workerCount)
}
// Increment waiting group semaphore
p.waitGroup.Add(count)
// Launch workers
for i := 0; i < count; i++ {
worker(p.dispatchedTasks, func() {
var firstTask func() = nil
if i < len(firstTasks) {
firstTask = firstTasks[i]
}
worker(firstTask, p.dispatchedTasks, &p.idleWorkerCount, &p.completedTaskCount, func() {
// Decrement worker count
atomic.AddInt32(&p.workerCount, -1)
@@ -307,7 +445,7 @@ func (p *WorkerPool) Group() *TaskGroup {
}
// worker launches a worker goroutine
func worker(tasks chan func(), exitHandler func(), panicHandler func(interface{})) {
func worker(firstTask func(), tasks chan func(), idleWorkerCount *int32, completedTaskCount *uint64, exitHandler func(), panicHandler func(interface{})) {
go func() {
defer func() {
@@ -316,21 +454,44 @@ func worker(tasks chan func(), exitHandler func(), panicHandler func(interface{}
panicHandler(panic)
// Restart goroutine
worker(tasks, exitHandler, panicHandler)
worker(nil, tasks, idleWorkerCount, completedTaskCount, exitHandler, panicHandler)
} else {
// Handle exit
exitHandler()
}
}()
// We have received a task, execute it
func() {
// Increment idle count
defer atomic.AddInt32(idleWorkerCount, 1)
if firstTask != nil {
// Increment completed task count
defer atomic.AddUint64(completedTaskCount, 1)
firstTask()
}
}()
for task := range tasks {
if task == nil {
// We have received a signal to quit
return
}
// Decrement idle count
atomic.AddInt32(idleWorkerCount, -1)
// We have received a task, execute it
task()
func() {
// Increment idle count
defer atomic.AddInt32(idleWorkerCount, 1)
// Increment completed task count
defer atomic.AddUint64(completedTaskCount, 1)
task()
}()
}
}()
}