How to Tune VarMQ Pool Concurrency at Runtime

VarMQ enables dynamic resizing of worker goroutine pools while jobs are actively processing through the atomic TunePool() method, allowing you to expand or shrink concurrency without restarting the service.

The goptics/varmq library implements a high-performance job queue that maintains a pool of goroutine workers. Unlike static worker pools that require restarts to adjust capacity, VarMQ allows you to tune VarMQ pool concurrency at runtime to match fluctuating workload demands. This capability is particularly valuable for services experiencing variable traffic patterns or those operating under memory constraints that require periodic downsizing.

Understanding VarMQ Worker Pool Architecture

At the heart of VarMQ's concurrency model sits the generic worker[T, JobType] type defined in worker.go. This structure manages a pool of goroutine workers using a thread-safe, linked-list-backed implementation from internal/pool/pool.go.

The Atomic Concurrency Field

The worker stores its current pool size in an atomic uint32 field named concurrency. This atomic storage ensures that read and update operations remain race-free when the tuning method is invoked from external goroutines while workers are actively processing jobs.

Idle Worker Management

When shrinking the pool, VarMQ must decide how to remove excess workers. The behavior depends on your configuration:

• With idle expiry configured: Excess workers are removed by the background idle-worker remover
• Without idle expiry: The TunePool method explicitly pops nodes from the back of the internal pool until the size matches the minimum idle worker threshold (numMinIdleWorkers)

How TunePool Works Under the Hood

The runtime tuning logic resides in worker.go (lines 82-124) within the TunePool(newConcurrency int) method. This implementation follows a precise flow to ensure safe transitions:

1. Validation: Verify the worker is running (returns ErrNotRunningWorker otherwise)
2. Safety Check: Compute safe concurrency via withSafeConcurrency, which defaults to runtime.NumCPU() for non-positive values
3. Early Exit: Return ErrSameConcurrency if the requested size equals the current atomic value
4. Atomic Update: Store the new safe value in the concurrency field
5. Expansion: If growing the pool, call w.notifyToPullNextJobs() to signal the event loop to spin up additional nodes on demand
6. Shrinking: If reducing the pool and no idle-worker expiry is configured, repeatedly pop nodes from the linked-list pool until reaching numMinIdleWorkers

This architecture ensures that increasing concurrency takes effect immediately for new jobs, while decreasing concurrency gracefully removes only idle workers rather than interrupting active processing.

Runtime Tuning Implementation

Basic Dynamic Scaling Example

The repository includes a practical demonstration in examples/tune-worker/main.go that oscillates pool size based on load conditions:

package main

import (
	"fmt"
	"math/rand"
	"runtime"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	initialConcurrency := 10
	tuneType := "expand"

	// Create a worker with an initial concurrency
	w := varmq.NewWorker(func(j varmq.Job[int]) {
		// Simulated work
		randomDuration := time.Duration(rand.Intn(1001)+500) * time.Millisecond
		time.Sleep(randomDuration)
	}, initialConcurrency)

	q := w.BindQueue()
	ticker := time.NewTicker(1 * time.Second)

	// Dynamically adjust the pool size every second
	go func() {
		for range ticker.C {
			if tuneType == "expand" {
				initialConcurrency += 10
			} else {
				initialConcurrency -= 10
			}
			// Runtime tuning point
			_ = w.TunePool(initialConcurrency)

			fmt.Printf(
				"Total Goroutines: %d, Idle Workers: %d\nConcurrency: %d, Processing: %d\nPending Jobs: %d Type: %s\n\n",
				runtime.NumGoroutine(),
				w.NumIdleWorkers(),
				initialConcurrency,
				w.NumProcessing(),
				q.NumPending(),
				tuneType,
			)

			if initialConcurrency >= 100 {
				tuneType = "shrink"
			}
			if initialConcurrency <= 10 {
				tuneType = "expand"
			}
		}
	}()

	// Feed jobs continuously
	for {
		for i := range 1000 {
			q.Add(i)
		}
		fmt.Println("Added jobs")
		w.WaitUntilFinished()
	}
}

Essential API Methods for Concurrency Management

• w.TunePool(30) – Atomically expands the pool to 30 workers; new goroutines spawn as needed when jobs arrive
• w.TunePool(5) – Shrinks the pool to 5 workers; excess idle workers are removed immediately or allowed to expire based on configuration
• w.NumIdleWorkers() – Returns the current count of workers waiting for jobs
• w.NumConcurrency() – Returns the configured concurrency limit (the atomic value set by TunePool)

Configuration Options That Affect Runtime Tuning

When constructing workers in config.go, several functional options influence how TunePool behaves during shrink operations:

w := varmq.NewWorker(
    handler,
    varmq.WithConcurrency(20),                 // Initial pool size
    varmq.WithMinIdleWorkerRatio(20),          // Maintain at least 20% idle workers
    varmq.WithIdleWorkerExpiryDuration(30*time.Second), // Prune idle workers after 30s
)

WithMinIdleWorkerRatio sets numMinIdleWorkers, which acts as a floor during shrinking operations. Even when you call TunePool with a lower value, the pool retains this minimum percentage of idle workers to handle sudden traffic spikes.

WithIdleWorkerExpiryDuration delegates shrinking responsibility to a background goroutine that removes workers idle longer than the specified duration. When this option is active, TunePool updates the atomic target but relies on the expiry mechanism to achieve the desired size gradually rather than immediately popping workers.

Summary

• Atomic safety: VarMQ stores concurrency in an atomic uint32 field, enabling lock-free reads and safe updates from any goroutine.
• Zero-downtime scaling: The TunePool method in worker.go (lines 82-124) validates inputs, applies safety defaults, and transitions the pool size without stopping active workers.
• Flexible expansion: Growing the pool signals the event loop via notifyToPullNextJobs(), creating workers on demand as new jobs arrive.
• Graceful shrinking: Reducing concurrency removes only idle workers, either immediately via linked-list popping or gradually through idle-expiry configuration.
• Production example: The examples/tune-worker program demonstrates oscillating pool sizes between 10 and 100 workers based on real-time metrics.

Frequently Asked Questions

What happens if I call TunePool on a worker that hasn't started?

The method returns ErrNotRunningWorker. You must start the worker via BindQueue() and ensure the event loop is active before attempting runtime tuning.

Does shrinking the pool kill active goroutines?

No. According to the implementation in worker.go, shrinking only removes idle workers from the internal pool linked list. Active workers processing jobs complete their current tasks and return to the pool normally, at which point they may be removed if the pool exceeds the new concurrency limit.

What is the default concurrency if I pass zero or a negative value to TunePool?

The withSafeConcurrency helper converts non-positive values to runtime.NumCPU(), ensuring you never accidentally configure a worker pool with zero capacity.

How does idle worker expiry interact with TunePool shrinking?

When WithIdleWorkerExpiryDuration is configured, TunePool sets the atomic concurrency target but delegates the actual removal to the background expiry goroutine. Without this option, TunePool immediately pops excess nodes from internal/pool/pool.go until reaching the minimum idle worker threshold.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}