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

> Discover VarMQ's dynamic lock-free worker pool. Learn how atomic operations and an event-loop goroutine control concurrency and enable runtime scaling.

- Repository: [Goptics/varmq](https://github.com/goptics/varmq)
- Tags: internals
- Published: 2026-03-02

---

**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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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.

```go
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.

```go
// 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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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.

```go
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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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`](https://github.com/goptics/varmq/blob/main/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.