Concurrency

A worker’s concurrency is the most jobs it runs at the same time. This page shows how to set the limit and how to change it while the worker runs. It also covers the options that decide how many idle goroutines the worker keeps. How it works shows the pool behind it.

Setting the limit

The default is 1. A plain int after the worker function sets it. WithConcurrency sets it as an option. 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

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 goroutines while jobs wait, up to the limit, and they 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:

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 blocks every job inside the worker function until the program releases it. It adds three jobs, one at a time, and waits for each to start before it adds the next.

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()
}

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.

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 gives 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 Idle or Running. TunePool only works on an active worker. In other states it returns the error for the worker’s state:

  • ErrNotRunningWorker before Start.
  • ErrWorkerPausing or ErrWorkerPaused while pausing or paused.
  • ErrWorkerStopping or ErrWorkerStopped while stopping or stopped.

  The lifecycle page has the states.

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. There, one goroutine can take up to 32 jobs at once from an in-memory FIFO queue. Raising the limit does not spread out the jobs it already holds. How it works explains batches.

The program below starts a worker with a limit of 1 and blocks every job until the program releases it. 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.

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()
}

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

Nothing running is interrupted. NumProcessing can stay above NumConcurrency until the busy goroutines finish the jobs they already hold. Those jobs still start after the change. The worker removes idle goroutines it no longer needs at once, or on the expiry timer if you set one (see below).

Idle goroutines

A worker with default settings keeps one idle goroutine in its pool. A goroutine that finishes a job stays in the pool in two cases:

• More jobs wait than the idle goroutines can take.
• The pool has 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 stay idle for up to about twice the duration before it exits. 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.

Idle goroutines let the next burst of jobs start without the cost of starting 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.

Next: Adding many jobs.