How VarMQ's Concurrency Control Mechanism Works: Dynamic Lock-Free Worker Pools

VarMQ controls concurrency through a dynamic, lock-free worker pool that uses atomic operations and an event-loop goroutine to enforce limits while allowing runtime scaling via TunePool.

The goptics/varmq repository implements a high-performance job processing system that avoids heavyweight mutex contention through careful use of Go's atomic primitives. This article examines the VarMQ concurrency control mechanism, detailing how the library manages parallel job execution, dynamic pool resizing, and graceful state transitions without blocking the main event loop.

Atomic Concurrency Configuration

VarMQ stores the desired concurrency level in an atomic.Uint32 field to enable lock-free reads and updates across goroutines.

In worker.go at line 45, the worker.concurrency field holds the configured limit. This atomic value is initialized during worker creation through the WithConcurrency option defined in config.go. When no concurrency value is provided, withSafeConcurrency automatically defaults to the number of CPU cores using utils.Cpus() (lines 37-43).

Event Loop Job Distribution

A single dedicated goroutine named goEventLoop manages job distribution without holding locks during the hot path.

Located at lines 21-23 in worker.go, the event loop continuously monitors a signaling channel while the worker status is running. The loop enforces a strict invariant: it only pulls the next job when the current number of processing jobs (curProcessing) is strictly lower than the atomic concurrency value and pending jobs exist. This design ensures that the configured limit is never exceeded while maintaining high throughput through lock-free checks.

Dynamic Pool Scaling with TunePool

VarMQ allows runtime adjustment of concurrency levels through the TunePool method, which atomically updates the worker limit and triggers immediate capacity changes.

Increasing Concurrency

When TunePool(newConcurrency) is called with a higher value (lines 87-99 in worker.go), the method atomically updates worker.concurrency and invokes notifyToPullNextJobs. This signals the event loop to immediately utilize the additional capacity, pulling more jobs up to the new limit.

w := varmq.NewWorker(func(j varmq.Job[int]) {
    // job processing logic
}, varmq.WithConcurrency(8))

// Dynamically increase to 12 workers
err := w.TunePool(12)
if err != nil {
    log.Fatal(err)
}

Decreasing Concurrency

When reducing the pool size, TunePool removes idle worker nodes immediately. For gradual reduction, combine TunePool with WithIdleWorkerExpiryDuration to enable automatic cleanup of excess workers after a specified idle period.

// Create worker with 10 concurrency and 30-second idle expiry
w := varmq.NewWorker(myFunc, 
    varmq.WithConcurrency(10), 
    varmq.WithIdleWorkerExpiryDuration(30*time.Second))

// Shrink pool to 6 workers
if err := w.TunePool(6); err != nil {
    log.Fatal(err)
}

Reusable Node Pool Architecture

Worker goroutines are backed by a pool.Node structure that holds a buffered channel for job distribution. To minimize allocations during pool expansion or contraction, VarMQ maintains these nodes in a sync.Pool accessed via worker.pool.Cache (lines 14-22 in internal/pool/pool.go).

This object pool pattern ensures that scaling operations—whether increasing concurrency during high load or shrinking during quiet periods—do not trigger expensive garbage collection cycles from repeated channel allocations.

State Synchronization Primitives

While the hot path remains lock-free, VarMQ protects worker state transitions using lightweight synchronization primitives.

The worker.mx field (a sync.RWMutex) and the worker.waiters condition variable (lines 50-52 and 83-85 in worker.go) coordinate shutdown sequences and pause/resume operations. These primitives block only during administrative state changes, leaving the event loop unblocked for job processing.

Pause and Resume Operations

Pausing does not alter the configured worker.concurrency value; it merely prevents the event loop from pulling new jobs until Resume restores the running status.

w.Pause()   // Sets status to paused
w.Resume()  // Restores running status and restarts event loop

Default Configuration Behavior

When users omit explicit concurrency settings, VarMQ automatically configures the pool size to match the host machine's CPU count. The withSafeConcurrency helper (lines 37-43 in config.go) calls utils.Cpus() to determine the optimal default, ensuring that the worker pool aligns with available hardware parallelism without manual tuning.

Summary

• VarMQ uses an atomic.Uint32 in worker.go (line 45) to store the concurrency limit, enabling lock-free access patterns.
• The goEventLoop goroutine enforces limits by comparing curProcessing against the atomic concurrency value before pulling jobs (lines 21-23).
• TunePool provides atomic runtime scaling, immediately utilizing increased capacity or removing idle nodes when shrinking (lines 87-99).
• A sync.Pool in internal/pool/pool.go (lines 14-22) caches worker nodes to minimize allocation overhead during dynamic scaling.
• sync.RWMutex and condition variables protect state transitions (pause, resume, shutdown) without blocking the hot path event loop.
• Default concurrency falls back to CPU core count via utils.Cpus() when not explicitly configured.

Frequently Asked Questions

How does VarMQ prevent race conditions when changing concurrency at runtime?

VarMQ prevents race conditions by storing the concurrency limit in an atomic.Uint32 (worker.concurrency) updated through the TunePool method. The event loop reads this value atomically when checking whether curProcessing is below the limit, ensuring consistent state without mutex contention during the hot path.

Can I change VarMQ concurrency without stopping existing jobs?

Yes. TunePool adjusts the concurrency limit atomically without interrupting currently processing jobs. When increasing concurrency, the event loop immediately pulls additional jobs up to the new limit. When decreasing, idle workers are removed first, while active workers continue processing until completion.

What happens to excess workers when I reduce concurrency?

When reducing concurrency via TunePool, VarMQ removes idle worker nodes immediately. If you configure an idle-worker expiry duration using WithIdleWorkerExpiryDuration, excess workers that become idle are automatically cleaned up by a background remover after the specified timeout, gradually shrinking the pool without forced termination of active goroutines.

Why does VarMQ default to the number of CPU cores?

The default concurrency matches CPU core count (via utils.Cpus() in config.go lines 37-43) to optimize throughput for CPU-bound workloads. This default prevents oversubscription while maximizing hardware utilization, though you can override it with WithConcurrency based on your specific I/O or processing requirements.

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