# VarMQ Worker Pool Implementation: Key Files and Architecture

> Explore VarMQ's worker pool implementation. Discover the five key files like worker.go and pool.go that orchestrate its efficient architecture and manage job channels for peak performance.

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

---

**VarMQ's worker pool implementation relies on five critical files: [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go) orchestrates the event loop and state machine, [`internal/pool/pool.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go) manages the lock-free node container, [`internal/pool/node.go`](https://github.com/goptics/varmq/blob/main/internal/pool/node.go) handles per-goroutine job channels, [`worker_binder.go`](https://github.com/goptics/varmq/blob/main/worker_binder.go) provides the fluent queue-binding API, and [`internal/pool/pool_test.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool_test.go) ensures concurrency correctness.**

The `goptics/varmq` repository delivers a high-performance job queue system centered on a dynamic, self-tuning worker pool implementation. This architecture enables efficient concurrent job processing through goroutine reuse, automatic load-based scaling, and intelligent idle-worker eviction. Understanding these core files reveals how VarMQ minimizes resource overhead while maintaining high throughput across diverse queue types.

## Core Architecture Components

### Worker Orchestration in [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go)

The [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go) file defines the generic `worker[T, JobType]` struct that serves as the central orchestrator for the entire pool. It manages the event loop, pool state, job lifecycle, and exposes the public API for queue interactions.

```go
type worker[T any, JobType iJob[T]] struct {
    workerFunc      func(j JobType)
    pool            *pool.Pool[JobType]
    queues          *queueManager
    eventLoopSignal chan struct{}
    status          atomic.Uint32
    concurrency     atomic.Uint32
}

```

The event loop, implemented in `goEventLoop()`, wakes whenever a node becomes available or new work arrives. It continuously processes jobs while respecting the configured concurrency limits:

```go
func (w *worker[T, JobType]) goEventLoop() {
    go func(signal <-chan struct{}) {
        for range signal {
            for w.IsRunning() && w.curProcessing.Load() < w.concurrency.Load() && w.queues.Len() > 0 {
                if err := w.processNextJob(); err != nil {
                    w.sendError(err)
                }
            }
        }
    }(w.eventLoopSignal)
}

```

### Pool Container in [`internal/pool/pool.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go)

The pool container implements a lock-free doubly-linked list that stores reusable worker nodes. Built on the library's own linked-list implementation, it provides O(1) node acquisition and return operations with minimal contention.

```go
type Pool[T any] struct {
    *linkedlist.List[Node[T]]
    Cache sync.Pool
}

```

Key methods include `PushNode` for returning idle nodes to the pool and `PopBack` for acquiring available workers. The `sync.Pool` integration caches list node allocations to minimize GC pressure during high-throughput scenarios.

### Pool Node Lifecycle in [`internal/pool/node.go`](https://github.com/goptics/varmq/blob/main/internal/pool/node.go)

Each worker goroutine is wrapped by a `Node[T]` struct that encapsulates the job channel and lifecycle management. The node maintains a `lastUsed` timestamp to support idle-worker eviction.

```go
type Node[T any] struct {
    ch       chan Payload[T]
    lastUsed atomic.Value
}

```

The node implements three critical operations: `Send` delivers jobs through the buffered channel, `Serve` runs the processing loop in a dedicated goroutine, and `Stop` signals termination via a poison-pill message. Timestamp helpers `UpdateLastUsed` and `GetLastUsed` enable the pool to identify and remove stagnant workers during periodic cleanup cycles.

### Queue Binding Interface in [`worker_binder.go`](https://github.com/goptics/varmq/blob/main/worker_binder.go)

The [`worker_binder.go`](https://github.com/goptics/varmq/blob/main/worker_binder.go) file exposes a fluent API that connects workers to various queue types including FIFO, priority, persistent, and distributed variants. It creates concrete binder types that lazily start the worker upon first queue attachment.

```go
func (wb *workerBinder[T]) BindQueue(configs ...QueueConfigFunc) Queue[T] {
    return wb.WithQueue(queues.NewQueue[iJob[T]](), configs...)
}

func (wb *workerBinder[T]) WithQueue(q IQueue, configs ...QueueConfigFunc) Queue[T] {
    defer wb.start()
    return newQueue(wb.worker, q, configs...)
}

```

## Dynamic Scaling and Resource Management

### Node Reuse and Recycling

When a job completes, the `freePoolNode` method determines whether to recycle the node back into the pool or terminate the goroutine. This decision balances current queue depth against the configured minimum-idle-worker ratio, ensuring sufficient standby capacity without excessive memory consumption.

### Idle-Worker Eviction

The `goRemoveIdleWorkers` goroutine periodically scans the pool for nodes exceeding `idleWorkerExpiryDuration`. Nodes with stale `lastUsed` timestamps are removed and their goroutines terminated, preventing resource leaks during low-traffic periods.

### Runtime Tuning via `TunePool`

The `TunePool` method enables dynamic scaling by adjusting the target concurrency on-the-fly. When increasing capacity, it spawns new nodes via `initPoolNode`; when decreasing, it allows natural attrition through the existing eviction mechanisms. This runtime configurability allows the VarMQ worker pool to adapt to changing load patterns without process restarts.

## Practical Implementation Example

The following example demonstrates creating a worker pool, binding it to a FIFO queue, and processing jobs:

```go
package main

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

type MyJob struct {
    ID   int
    Data string
}

func process(j varmq.Job[MyJob]) {
    fmt.Printf("processing job %d: %s\n", j.Payload().ID, j.Payload().Data)
    j.ChangeStatus(varmq.Finished)
}

func main() {
    w := varmq.NewWorker[MyJob](process, varmq.WithConcurrency(4))
    q := w.BindQueue()
    
    for i := 1; i <= 10; i++ {
        q.Add(MyJob{ID: i, Data: fmt.Sprintf("msg-%d", i)})
    }
    
    w.WaitUntilFinished()
    fmt.Println("all jobs completed")
}

```

Under the hood, `BindQueue` triggers `wb.start()` to launch the event loop and initialize the first idle node. Each `q.Add` pushes jobs into the internal queue manager, which the event loop drains by acquiring nodes via `processNextJob` and delivering payloads through `node.Value.Send`. When the queue empties, `WaitUntilFinished` blocks until `curProcessing` reaches zero.

## Summary

- **[`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go)** contains the generic `worker[T, JobType]` struct, event loop implementation, and public API for the VarMQ worker pool.
- **[`internal/pool/pool.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go)** provides the lock-free doubly-linked list container with `sync.Pool` integration for efficient node caching.
- **[`internal/pool/node.go`](https://github.com/goptics/varmq/blob/main/internal/pool/node.go)** defines the `Node[T]` wrapper managing per-goroutine job channels, stop signals, and idle-time tracking.
- **[`worker_binder.go`](https://github.com/goptics/varmq/blob/main/worker_binder.go)** implements the fluent binding interface connecting workers to FIFO, priority, persistent, and distributed queues.
- **[`internal/pool/pool_test.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool_test.go)** validates node reuse, idle-worker trimming, and thread-safety guarantees under concurrent access.

## Frequently Asked Questions

### How does VarMQ decide when to create or destroy worker goroutines?

VarMQ uses a dynamic scaling algorithm governed by the `freePoolNode` and `goRemoveIdleWorkers` methods. When jobs complete, `freePoolNode` evaluates queue depth and the minimum-idle ratio to decide between recycling the node or shutting it down. Concurrently, `goRemoveIdleWorkers` scans for nodes exceeding `idleWorkerExpiryDuration` and evicts them to free resources during low-load periods.

### What data structure backs the worker pool in VarMQ?

The pool uses a custom lock-free doubly-linked list implemented in [`internal/pool/pool.go`](https://github.com/goptics/varmq/blob/main/internal/pool/pool.go), layered over `internal/linkedlist`. This structure provides O(1) push and pop operations for node management, supplemented by a `sync.Pool` that caches list node allocations to reduce garbage collection overhead.

### Can the worker pool size be changed at runtime?

Yes. The `TunePool` method exposed in [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go) allows dynamic adjustment of concurrency limits. Increasing concurrency spawns new nodes immediately via `initPoolNode`, while decreasing concurrency allows excess workers to exit naturally through the standard job completion flow and idle-eviction mechanisms.

### How does [`worker_binder.go`](https://github.com/goptics/varmq/blob/main/worker_binder.go) support different queue types?

The binder provides specialized methods like `BindQueue` for standard FIFO queues and `WithQueue` for custom queue implementations. These methods create appropriate façade types (`IWorkerBinder`, `IErrWorkerBinder`, `IResultWorkerBinder`) and lazily initialize the worker pool only when the first queue binding occurs, enabling flexible deployment patterns across priority, persistent, and distributed queue backends.