Concurrency
A worker’s concurrency is the most jobs it runs at the same time. This page covers how to set it, how to change it while the worker runs, and the options that decide how many idle goroutines the worker keeps. How it works shows the pool behind it.
Setting the limit
Section titled “Setting the limit”The default is 1. A plain int after the worker function sets it, WithConcurrency sets it as an option, and DefaultConcurrency changes the default for workers created after the call:
// A plain int after the function sets the limit.varmq.NewWorker(process, 4)
// The same, as an option.varmq.NewWorker(process, varmq.WithConcurrency(4))
// The default for workers created after this call.varmq.DefaultConcurrency(4)If you pass more than one value to a worker, the last one wins. Worker types lists the other arguments that NewWorker takes.
A value below 1 means one per logical CPU. 0 and -1 both become runtime.NumCPU(), so the number depends on the machine. The same rule applies to DefaultConcurrency and to TunePool, below.
A limit, not a pool size
Section titled “A limit, not a pool size”NumConcurrency() returns the limit. It is an upper bound on the jobs running at once, not a number of goroutines the worker creates. The worker starts a goroutine when jobs are waiting and one more would clear the backlog faster than it costs to start it, up to the limit, and goroutines exit when the backlog is gone. A worker with a limit of 100 and three waiting jobs does not start 100 goroutines. How it works has the details.
Four methods on Worker report what the pool is doing:
| Method | Returns |
|---|---|
NumConcurrency |
the limit |
NumProcessing |
the jobs running right now, counted as busy goroutines |
NumPending |
the jobs that have not started: waiting in a queue, in a batch a goroutine took, or parked by a pause or stop. While a batch runs it can read high: see waiting |
NumIdleWorkers |
the idle goroutines waiting in the pool for a job |
NumProcessing stays at or below the limit, except for a while after TunePool lowers the limit. All four are snapshots: they can change right after you read them.
The program below creates a worker with a limit of 100 and holds every job inside the worker function until the program lets go. It adds three jobs, one at a time, and waits for each to start before it adds the next.
9 collapsed lines
package main
import ( "fmt"
"github.com/goptics/varmq")
func main() { started := make(chan int) release := make(chan struct{})
// Each job says it started, then holds until the program lets go. worker := varmq.NewWorker(func(j varmq.Job[int]) { started <- j.Data() <-release }, 100) queue := worker.BindQueue() fmt.Println("idle goroutines before any job:", worker.NumIdleWorkers())
for i := 1; i <= 3; i++ { queue.Add(i) <-started }
fmt.Println("NumConcurrency:", worker.NumConcurrency()) fmt.Println("NumProcessing:", worker.NumProcessing()) fmt.Println("NumPending:", worker.NumPending()) fmt.Println("NumIdleWorkers:", worker.NumIdleWorkers())
close(release) worker.WaitUntilIdle()}Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.
Before any job, the pool has one idle goroutine. With three jobs running, NumProcessing is 3, not 100. The idle goroutine took a job, so NumIdleWorkers is 0, and nothing is pending.
Move the limit in the scene below and watch the goroutines. The slider calls TunePool on a worker that keeps running: the jobs it already holds are not interrupted.
A simulation of one worker. Each job takes 2.4 seconds. The slider calls TunePool on the running worker. Each block is a live goroutine: a busy one fills while it runs a job and shows +n for the jobs it still holds as a batch, an idle one waits in the pool, and one that runs above a lowered limit has an amber edge. A goroutine that exits disappears. The meter compares the jobs that finish each second with the ceiling, which is the limit divided by 2.4 s, and with the jobs that arrive. With a limit of 1 the one goroutine takes the waiting jobs as a batch, and raising the limit does not spread that batch.
worker := varmq.NewWorker(process, 3)
queue := worker.BindQueue()Each job takes 2.4 s. A block is a live goroutine: it fills while it runs a job, +n is the rest of its batch, and an amber edge means it runs above the limit. The meter compares the jobs done each second with the ceiling, the limit divided by 2.4 s, and with the jobs that arrive.
Changing the limit with TunePool
Section titled “Changing the limit with TunePool”TunePool changes the limit of an active worker, one that is Idle or Running. It returns at once, and the new limit applies to the next job the worker hands out. A value below 1 means one per CPU, as above.
if err := worker.TunePool(8); err != nil { log.Println("could not tune:", err)}It returns an error in two cases:
ErrSameConcurrency: the worker already has that limit.- The worker is not
IdleorRunning.TunePoolonly works on an active worker, so it returns the error for the worker’s state:ErrNotRunningWorkerbeforeStart,ErrWorkerPausingorErrWorkerPausedwhile pausing or paused, andErrWorkerStoppingorErrWorkerStoppedwhile stopping or stopped. The lifecycle page has the states.
Raising the limit
Section titled “Raising the limit”The worker starts jobs on new goroutines as soon as it can. That covers the jobs still in a queue and the jobs you add later.
A job a goroutine has already taken into a batch stays with that goroutine, which runs its batch one job after another. This matters most at concurrency 1, where one goroutine can take up to 32 jobs at once from an in-memory FIFO queue: raising the limit does not spread out jobs it already holds. How it works explains batches.
The program below starts a worker with a limit of 1 and holds every job until the program lets go. With job 1 running, it raises the limit to 3, then adds jobs 2 to 6. It adds them after the change for the reason above.
9 collapsed lines
package main
import ( "fmt"
"github.com/goptics/varmq")
func main() { started := make(chan int, 6) // room for every job, so none blocks here release := make(chan struct{})
worker := varmq.NewWorker(func(j varmq.Job[int]) { started <- j.Data() <-release }) queue := worker.BindQueue()
queue.Add(1) <-started // job 1 is running fmt.Println("TunePool(3):", worker.TunePool(3))
for i := 2; i <= 6; i++ { queue.Add(i) } <-started <-started // jobs 2 and 3 started beside job 1 fmt.Println("processing", worker.NumProcessing(), "pending", worker.NumPending())
fmt.Println("TunePool(3) again:", worker.TunePool(3))
close(release) worker.WaitUntilIdle()}Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.
Jobs 2 and 3 start beside job 1, which is three at once, the new limit. Jobs 4 to 6 wait. The second TunePool(3) returns ErrSameConcurrency, because the worker already has a limit of 3.
Lowering the limit
Section titled “Lowering the limit”Nothing running is interrupted. The worker hands out no new batch until fewer goroutines are busy than the new limit, so NumProcessing can stay above NumConcurrency until those goroutines finish their batches. A goroutine that already holds a batch keeps running it one job after another, so jobs from batches already handed out still start after the change. The worker removes idle goroutines it no longer needs at once, unless you set an expiry duration, in which case the expiry removes them (see below).
Idle goroutines
Section titled “Idle goroutines”A worker with default settings keeps one idle goroutine in its pool. A goroutine that finishes a job stays in the pool if more jobs are waiting than idle goroutines are ready to run them, or if the pool holds fewer idle goroutines than the minimum, which is one. Otherwise it exits. Two options change this.
WithMinIdleWorkerRatio sets the share of the limit to keep idle, as a percentage from 1 to 100. The worker keeps limit * percentage / 100 idle goroutines, rounded down, with a floor of one. With a limit of 4, a ratio of 50 keeps 2 and a ratio of 100 keeps 4. A value outside the range is clamped: 0 becomes 1 and anything above 100 becomes 100. The limit here is the current one, so the minimum follows a TunePool.
WithIdleWorkerExpiryDuration makes the pool shrink on a timer instead. With it set, a goroutine that finishes a job goes back to the pool instead of exiting. Once per duration the worker removes the goroutines that have been idle for longer than the duration, down to the minimum above. Because the check runs once per duration, a goroutine can sit idle for up to about twice the duration before it goes. Without the option there is no timer, and the pool shrinks as goroutines finish.
worker := varmq.NewWorker(process, 100, varmq.WithMinIdleWorkerRatio(10), // keep 10 goroutines idle varmq.WithIdleWorkerExpiryDuration(30*time.Second), // drop the rest after they sit idle)Send a burst below and watch what each setting leaves in the pool.
A simulation of the idle goroutines in one worker pool. Press Add jobs to send one job for each slot of the limit. Each goroutine runs a job for 1.6 seconds, then goes back to the pool or exits. With no options the pool keeps one idle goroutine. A ratio keeps that share of the limit, never fewer than one. An expiry timer keeps every goroutine until a sweep finds it idle for longer than the duration, so a goroutine can wait up to about twice the duration. Changing the limit, the ratio or the timer creates a new worker.
worker := varmq.NewWorker(process, 6)Add jobs sends one job for each slot of the limit. Each job runs for 1.6 s. With no options the pool keeps one idle goroutine. A change to the limit, the ratio or the timer creates a new worker.
Idle goroutines let the next burst of jobs start without paying to start goroutines. Each one holds a goroutine’s memory. DefaultMinIdleWorkerRatio and DefaultIdleWorkerExpiryDuration set both for workers created after the call. Configuration lists every option and its default.
VarMQ is MIT licensed. Benchmark figures come from github.com/goptics/benchmarks.