How VarMQ Handles Dynamic Pool Tuning at Runtime

VarMQ dynamically adjusts worker pool concurrency at runtime through the TunePool method, which validates worker state, sanitizes concurrency values, and either signals the event loop to spawn new workers or removes idle workers from a thread-safe linked-list pool.

VarMQ is a high-performance job queue library for Go designed for elastic scaling without service restarts. The TunePool method in worker.go serves as the primary interface for dynamic pool tuning, enabling applications to react to load fluctuations by expanding or shrinking worker capacity while maintaining thread safety and minimizing GC overhead.

The TunePool Method: Core Logic for Live Concurrency Adjustments

The TunePool method, located in [worker.go](https://github.com/goptics/varmq/blob/main/worker.go) at lines 82-124, orchestrates all runtime pool modifications. It coordinates state validation, concurrency sanitization, and the actual pool mutation through distinct phases.

State Validation and Safety Checks

Before modifying the pool, TunePool verifies the worker is in the running state. If the worker is not running, it immediately returns ErrNotRunningWorker (lines 82-85). The method then sanitizes the requested concurrency using withSafeConcurrency. If the sanitized value equals the current concurrency, it returns ErrSameConcurrency to avoid unnecessary no-op operations (lines 87-92).

Expanding the Worker Pool

When the requested concurrency exceeds the current value, VarMQ stores the new concurrency atomically and invokes w.notifyToPullNextJobs() (lines 94-100). This notification signals the event loop to create additional idle workers on demand as new jobs arrive. No immediate pool mutation occurs during expansion; workers materialize lazily through the standard job polling mechanism.

Shrinking with Idle-Worker Expiry

When reducing concurrency, VarMQ implements two distinct strategies based on the idleWorkerExpiryDuration configuration (lines 103-124):

• With expiry enabled: The pool size remains unchanged. The background goRemoveIdleWorkers goroutine automatically removes excess idle workers after the configured TTL expires, keeping the TunePool operation lightweight.
• Without expiry: VarMQ actively shrinks the pool by calculating shrinkPoolSize and removing nodes from the tail of the linked-list pool until only minIdleWorkers remain. Each removed node is stopped and returned to the object cache.

Architectural Components Supporting Safe Tuning

VarMQ's dynamic pool tuning relies on lock-free data structures and careful separation of concerns to ensure safety during concurrent access.

Atomic State Management

The worker struct maintains w.status and w.concurrency as atomic.Int32 values. These atomic fields enable lock-free reads and writes across multiple goroutines, ensuring that pool size adjustments never race with job processing status checks.

Doubly-Linked List Pool Structure

Worker nodes reside in a doubly-linked list implemented in [internal/linkedlist/linkedlist.go](https://github.com/goptics/varmq/blob/main/internal/linkedlist/linkedlist.go). This structure provides O(1) insertion and removal operations. During shrinking, VarMQ removes nodes from the tail, which yields the most recently added (and likely idle) workers first, minimizing disruption to active job processing.

Background Idle-Worker Remover

The goRemoveIdleWorkers goroutine periodically evaluates worker idle times against idleWorkerExpiryDuration. When this background removal is active, down-scaling operations delegate cleanup to this goroutine rather than performing immediate removals in the TunePool call path. This design keeps latency-sensitive tuning operations cheap while ensuring eventual consistency with the desired concurrency level.

Object Caching for Allocation-Free Tuning

To prevent GC pressure during rapid pool expansions and contractions, VarMQ utilizes a sync.Pool-based cache defined in [internal/pool/pool.go](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go) (lines 9-22). When shrinking removes worker nodes from the linked list, these node objects return to the cache rather than being garbage collected. Subsequent expansions retrieve nodes from this cache, eliminating allocations during hot tuning cycles.

Implementing Dynamic Tuning in Production

The following example from [examples/tune-worker/main.go](https://github.com/goptics/varmq/blob/main/examples/tune-worker/main.go) demonstrates live pool adjustment based on workload patterns:

package main

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

	"github.com/goptics/varmq"
)

func main() {
	initial := 10
	w := varmq.NewWorker(func(j varmq.Job[int]) {
		// Simulate work
		time.Sleep(time.Duration(rand.Intn(1000)+500) * time.Millisecond)
	}, initial)

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

	// Switch between expanding and shrinking the pool every few seconds
	go func() {
		expand := true
		for range ticker.C {
			if expand {
				initial += 10
			} else {
				initial -= 10
			}
			_ = w.TunePool(initial) // dynamically adjust concurrency
			fmt.Printf("Goroutines: %d, Idle workers: %d, Concurrency: %d, Processing: %d\n",
				runtime.NumGoroutine(), w.NumIdleWorkers(), initial, w.NumProcessing())
			if initial >= 100 {
				expand = false
			}
			if initial <= 10 {
				expand = true
			}
		}
	}()

	// Feed jobs continuously
	for i := 0; i < 1000; i++ {
		q.Add(i)
	}
	w.WaitUntilFinished()
}

This implementation alternates between expansion and contraction every second, demonstrating how TunePool integrates with runtime metrics like runtime.NumGoroutine() and VarMQ's introspection methods NumIdleWorkers() and NumProcessing().

Testing Pool Contraction and Edge Cases

The test suite in [worker_test.go](https://github.com/goptics/varmq/blob/main/worker_test.go) verifies that shrinking operations respect pool constraints and configuration parameters:

func TestTunePoolShrink(t *testing.T) {
	w := varmq.NewWorker(dummyHandler, 5)
	_ = w.Start()
	// ... enqueue jobs, let some workers become idle ...

	// Shrink to 2 workers
	if err := w.TunePool(2); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if got := w.NumIdleWorkers(); got != 2 {
		t.Fatalf("expected 2 idle workers, got %d", got)
	}
}

This test validates that TunePool correctly transitions the pool to the target idle worker count and handles edge cases such as same-concurrency requests, non-running workers, and empty pool scenarios.

Summary

• VarMQ exposes dynamic pool tuning through the TunePool method on the worker type, which validates state via atomic fields before modifying concurrency.
• Expansion occurs lazily by notifying the event loop via notifyToPullNextJobs(), while contraction either uses background expiry or actively removes tail nodes from the linked-list pool.
• Thread safety relies on atomic.Int32 for state fields and a lock-free doubly-linked list (internal/linkedlist) for worker storage.
• Performance optimization comes from a sync.Pool cache in internal/pool/pool.go that recycles node objects, preventing GC stalls during rapid scaling events.
• Operational flexibility allows shrinking to be delegated to a background goroutine when idleWorkerExpiryDuration is configured, reducing the latency impact of down-scaling operations.

Frequently Asked Questions

Can TunePool be called while jobs are actively processing?

Yes. The TunePool method is designed for concurrent use and checks atomic state fields before modifying pool parameters. Active workers continue processing their current jobs regardless of tuning operations. When shrinking without idle expiry, VarMQ removes only idle workers from the tail of the linked list, ensuring in-flight jobs complete normally.

What error does TunePool return if the worker is not running?

If the worker status is not running, TunePool immediately returns ErrNotRunningWorker (defined in worker.go lines 82-85). This prevents pool modifications during initialization or shutdown phases. Additionally, if the requested concurrency equals the current value after sanitization, it returns ErrSameConcurrency to indicate a no-op.

How does VarMQ handle memory allocation during rapid pool scaling?

VarMQ minimizes allocations through the object cache in internal/pool/pool.go (lines 9-22). This sync.Pool implementation recycles worker node objects removed during shrinking operations. When the pool expands again, nodes are retrieved from this cache rather than allocated anew, eliminating GC pressure during frequent tuning cycles common in auto-scaling scenarios.

What is the difference between active shrinking and idle-worker expiry?

Active shrinking occurs immediately within TunePool when idleWorkerExpiryDuration is zero, calculating shrinkPoolSize and removing excess workers from the linked-list tail until minIdleWorkers remain. Idle-worker expiry delegates this responsibility to the goRemoveIdleWorkers background goroutine, which removes workers only after they have been idle for the configured duration. The latter approach keeps TunePool latency minimal while eventually achieving the desired concurrency.

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