VarMQ Worker Pool Implementation: Key Files and Architecture

VarMQ's worker pool implementation relies on five critical files: worker.go orchestrates the event loop and state machine, internal/pool/pool.go manages the lock-free node container, internal/pool/node.go handles per-goroutine job channels, worker_binder.go provides the fluent queue-binding API, and 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

The 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.

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:

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

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.

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

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.

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

The 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.

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:

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 contains the generic worker[T, JobType] struct, event loop implementation, and public API for the VarMQ worker pool.
• internal/pool/pool.go provides the lock-free doubly-linked list container with sync.Pool integration for efficient node caching.
• internal/pool/node.go defines the Node[T] wrapper managing per-goroutine job channels, stop signals, and idle-time tracking.
• worker_binder.go implements the fluent binding interface connecting workers to FIFO, priority, persistent, and distributed queues.
• 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, 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 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 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.

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