# How Jobs Are Distributed to Workers via Channels in VarMQ

> Discover how VarMQ distributes jobs to workers using channels for O(1) dispatch and dynamic scaling. Explore its efficient, lock-free linked list and sync.Pool architecture.

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

---

**VarMQ routes every enqueued job through a channel‑based pool of worker nodes, achieving O(1) dispatch and dynamic scaling via lock‑free linked lists and `sync.Pool` reuse.**

The `goptics/varmq` repository implements a high‑throughput job queue for Go that relies on channels—not mutexes—to coordinate work between producers and consumers. Understanding how jobs are distributed to workers via channels in VarMQ reveals a design optimized for constant‑time dispatch and elastic concurrency. This article traces the exact path from `Enqueue` to execution using the actual source code.

## The Four‑Stage Channel Distribution Pipeline

VarMQ’s distribution logic breaks down into four discrete stages that move a job from the shared queue into a worker goroutine.

### Stage 1: Queue Selection via queueManager

When a worker’s event loop signals readiness, it queries the `queueManager` to select the next target queue. Implemented in [`queue_manager.go`](https://github.com/goptics/varmq/blob/main/queue_manager.go) (lines 38‑50), the manager applies configurable strategies—round‑robin, max‑len, min‑len, or priority—to determine which pending job batch to process.

### Stage 2: Job Extraction and Type Casting

Once a queue is selected, `processNextJob` in [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go) (lines 33‑58) dequeues the raw payload. The function handles type conversion, casting the raw data into the concrete `JobType` or unmarshaling `[]byte` into the expected structure before dispatch.

### Stage 3: Channel Dispatch to Pool Nodes

The critical hand‑off occurs in `sendToNextChannel` ([`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go), lines 13‑22). Here, the job enters the channel‑based distribution layer:

- If an idle node exists in the pool, `PopBack` retrieves it and the job is sent directly to that node’s channel.
- If no idle nodes exist, `initPoolNode` ([`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go), lines 26‑42) creates a new node, initializes its channel, and immediately dispatches the job.

### Stage 4: Node Execution and Lifecycle Management

Each `pool.Node` runs a dedicated goroutine executing the `Serve` method ([`internal/pool/node.go`](https://github.com/goptics/varmq/blob/main/internal/pool/node.go), lines 33‑40). This goroutine blocks on its channel, executes the user‑provided worker function when a job arrives, then marks the job finished. The node returns to the pool via `freePoolNode` or is stopped if idle‑worker eviction applies ([`internal/pool/node.go`](https://github.com/goptics/varmq/blob/main/internal/pool/node.go), lines 44‑46).

## Lock‑Free Pool Architecture and O(1) Dispatch

Underpinning the channel distribution is a lock‑free linked list backed by `sync.Pool` for node reuse, defined in [`internal/pool/pool.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go) (lines 9‑23). This architecture eliminates mutex contention during node retrieval and return.

When `freePoolNode` ([`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go), lines 94‑108) executes, it either pushes the node back onto the pool or stops it if the pool exceeds the current concurrency limit. This ensures that channel dispatch remains **O(1)**—constant time regardless of pool depth—because operations manipulate the linked list head without global locks.

## Dynamic Scaling and Back‑Pressure Handling

VarMQ’s channel‑based model supports elastic concurrency through two mechanisms:

**Pool Growth**: When jobs arrive faster than workers process them, `sendToNextChannel` creates new nodes dynamically, ensuring zero job drops due to saturation.

**Idle Worker Eviction**: The `goRemoveIdleWorkers` routine monitors node activity. If a node remains idle beyond the configured threshold, `Stop` is invoked ([`internal/pool/node.go`](https://github.com/goptics/varmq/blob/main/internal/pool/node.go), lines 44‑46), closing its channel and releasing resources while the `sync.Pool` retains the node object for future reuse.

Users can tune concurrency at runtime via `TunePool`, which adjusts the target pool size and triggers the eviction logic to match new requirements.

## Practical Implementation Example

The following runnable example demonstrates the channel‑based distribution flow:

```go
package main

import (
    "fmt"
    "log"
    "github.com/goptics/varmq"
)

func main() {
    // 1️⃣ Create a worker that prints the job payload.
    w := varmq.NewWorker(func(j varmq.Job[int]) {
        fmt.Println("processed:", j.Data())
    })

    // 2️⃣ Start the worker (defaults to 1 concurrent goroutine).
    if err := w.Start(); err != nil {
        log.Fatal(err)
    }

    // 3️⃣ Enqueue jobs; each travels through the channel‑based pipeline.
    for i := 0; i < 10; i++ {
        _ = w.Enqueue(i) // Returns error only on full queue.
    }

    // 4️⃣ Increase concurrency; pool allocates additional channels.
    if err := w.TunePool(4); err != nil {
        log.Fatal(err)
    }

    // 5️⃣ Wait for completion and stop.
    w.WaitUntilFinished()
    w.Stop()
}

```

In this flow, `Enqueue` submits jobs to the internal queue, `sendToNextChannel` distributes them to `pool.Node` channels, and the closure executes inside `Node.Serve` goroutines.

## Summary

- VarMQ uses **channels**, not mutexes, to hand off jobs from the dispatcher to worker goroutines.
- Distribution occurs in four stages: queue selection ([`queue_manager.go`](https://github.com/goptics/varmq/blob/main/queue_manager.go)), job casting (`processNextJob`), channel dispatch (`sendToNextChannel`), and node execution (`pool.Node.Serve`).
- The **lock‑free pool** ([`internal/pool/pool.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go)) backed by `sync.Pool` enables **O(1)** dispatch complexity.
- **Dynamic scaling** is achieved by creating nodes when load spikes and evicting idle workers via `goRemoveIdleWorkers` and `Node.Stop`.
- Back‑pressure is handled by immediate node creation rather than queue dropping.

## Frequently Asked Questions

### What happens if all worker channels are busy when a job is enqueued?

If no idle nodes exist in the pool, `sendToNextChannel` invokes `initPoolNode` to create a new node with its own channel, ensuring the job is dispatched immediately without blocking or dropping.

### How does VarMQ achieve O(1) job dispatch complexity?

The system uses a lock‑free linked list for the node pool ([`internal/pool/pool.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go)) and channel sends, which are constant‑time operations. Unlike mutex‑based work stealing, retrieving or returning a node requires only atomic pointer swaps at the list head.

### Can I adjust the number of worker channels at runtime?

Yes. Calling `TunePool(n)` dynamically adjusts the target concurrency. The system will either allocate new nodes with channels for increased capacity or trigger idle‑worker eviction via `freePoolNode` to reduce resource usage.

### How are idle worker nodes cleaned up?

The `goRemoveIdleWorkers` background routine monitors node activity. When a node exceeds the idle timeout, `Stop` is called ([`internal/pool/node.go`](https://github.com/goptics/varmq/blob/main/internal/pool/node.go) lines 44‑46), which closes the node’s channel and removes it from active duty, though the node object remains in `sync.Pool` for reuse.