How to Wait for All Jobs to Finish in VarMQ: Complete Guide

Use worker.WaitUntilFinished() to block until every queued job is dequeued and processed, or call worker.WaitAndStop() to finish all work and shut down the worker cleanly.

VarMQ is a Go job queue library that processes jobs concurrently through pool-based workers. When you need to ensure all enqueued tasks complete before proceeding—such as during graceful shutdowns or test assertions—the library provides built-in synchronization primitives that block until the worker becomes idle.

Core Blocking Methods in VarMQ

The Worker type in worker.go exposes three primary APIs for waiting on job completion. Each serves a distinct use case depending on whether you want to continue processing, pause temporarily, or stop entirely.

WaitUntilFinished

The WaitUntilFinished() method blocks the calling goroutine while the worker remains busy. It returns only when the internal condition indicates no pending jobs and zero in-flight processing.

In worker.go, this method implements a condition variable pattern:

// Simplified logic from worker.go#L209-L227
func (w *Worker) WaitUntilFinished() error {
    w.mu.Lock()
    defer w.mu.Unlock()
    
    for w.condition() {
        w.waiters.Wait() // Blocks on sync.Cond
    }
    return nil
}

The underlying condition closure evaluates to true while the worker is running and (queues.Len() > 0 or curProcessing > 0), or while the worker is paused/stopped but curProcessing > 0. When releaseWaiters broadcasts—triggered when curProcessing drops to zero—the blocked caller wakes and returns.

PauseAndWait

When you need to stop accepting new jobs but allow current work to finish, use PauseAndWait(). This atomic operation first transitions the worker to paused status, then invokes WaitUntilFinished().

According to the source implementation, pausing prevents the worker from pulling new jobs from the queue set, while the wait ensures all curProcessing goroutines complete. After returning, you can safely call Resume() to continue processing or perform maintenance without data loss.

WaitAndStop

For graceful shutdown scenarios, WaitAndStop() combines waiting with termination. This method waits for all in-flight jobs to finish using the same condition logic as WaitUntilFinished(), then stops the worker cleanly by terminating internal goroutines.

How the Waiting Mechanism Works Internally

The synchronization relies on a sync.Cond (w.waiters) coupled with atomic counters. Understanding this flow helps diagnose blocking behavior in production:

1. Condition Evaluation: The condition closure in worker.go checks w.status, queues.Len(), and the atomic curProcessing counter. It returns true while work remains active.

2. Efficient Blocking: Rather than polling, WaitUntilFinished() locks the worker mutex and calls w.waiters.Wait(), yielding the CPU until broadcast.

3. Release Trigger: The private releaseWaiters() method (located around worker.go#L183-L194) broadcasts on the condition variable whenever curProcessing reaches zero and either the worker is paused or the queue set is empty.

4. Completion Guarantee: Upon return from WaitUntilFinished(), the mutex guarantees that curProcessing == 0 and queues.Len() == 0 (or the worker is stopped), ensuring all jobs have finished.

Practical Code Examples

Wait for All Jobs After Enqueuing

After starting a worker and adding payloads, block until the queue drains:

package main

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

func main() {
    // Initialize queue and worker
    queue := varmq.NewQueue()
    worker := varmq.NewWorker(queue)
    
    if err := worker.Start(); err != nil {
        log.Fatalf("start failed: %v", err)
    }
    
    // Enqueue work
    for i := 0; i < 100; i++ {
        queue.Add(fmt.Sprintf("task-%d", i))
    }
    
    // Block until queue empty and all jobs processed
    if err := worker.WaitUntilFinished(); err != nil {
        log.Fatalf("wait failed: %v", err)
    }
    
    fmt.Println("All 100 jobs completed")
}

This pattern appears frequently in worker_test.go (lines 87-105), where tests verify queue drainage after batch inserts.

Pause Processing and Wait

Use this pattern when you need to perform atomic operations or maintenance without shutting down:

// Prevent worker from taking new jobs
if err := worker.Pause(); err != nil {
    log.Fatal(err)
}

// Wait for current batch to finish
if err := worker.WaitUntilFinished(); err != nil {
    log.Fatal(err)
}

// At this point: no new jobs started, all active jobs done
performMaintenance()

// Resume normal operation
if err := worker.Resume(); err != nil {
    log.Fatal(err)
}

Alternatively, use the convenience method:

if err := worker.PauseAndWait(); err != nil {
    log.Fatal(err)
}

Graceful Shutdown with WaitAndStop

For application shutdown hooks, ensure all jobs complete before exiting:

func shutdownWorker(worker *varmq.Worker) {
    log.Println("Initiating graceful shutdown...")
    
    if err := worker.WaitAndStop(); err != nil {
        log.Printf("Error during shutdown: %v", err)
        return
    }
    
    log.Println("Worker stopped after finishing all jobs")
}

WaitAndStop() handles the synchronization internally, making it safer than manually calling Stop() which might interrupt running jobs.

Wait for a Single Specific Job

Each Job struct embeds its own sync.WaitGroup. To block on an individual job rather than the entire worker:

// AddJob returns *Job immediately
job := queue.AddJob(myPayload)

// Process other logic...

// Block only until this specific job completes
job.Wait()

// Implementation reference: job.go#L179-L183

This is useful when you need results from a specific task while allowing the worker to continue processing unrelated jobs.

Summary

• WaitUntilFinished() blocks until the queue empties and all in-flight jobs complete, using an efficient sync.Cond mechanism defined in worker.go.
• PauseAndWait() atomically pauses the worker and waits for current jobs, preventing new work from starting during the wait.
• WaitAndStop() provides graceful shutdown by waiting for completion then stopping the worker cleanly.
• Individual job waiting is available via job.Wait() for per-task synchronization.
• The waiting logic relies on atomic curProcessing counters and condition variables rather than busy-waiting, ensuring efficient CPU usage.

Frequently Asked Questions

What is the difference between WaitUntilFinished and PauseAndWait?

WaitUntilFinished() blocks until current work finishes while the worker remains in its current state—if running, it will continue pulling new jobs from the queue. PauseAndWait() first transitions the worker to paused status (preventing new job acquisition) and then waits. Use PauseAndWait() when you need to guarantee no new jobs start during the wait period, such as during maintenance windows or when preparing for a controlled shutdown.

Is WaitUntilFinished safe to call from multiple goroutines?

Yes, WaitUntilFinished() is safe for concurrent use. The implementation locks the worker's mutex (w.mu) before checking the condition and waiting on the sync.Cond. Multiple goroutines can safely call this method; all will block until the completion condition is met, at which point the condition variable broadcasts to wake all waiters simultaneously.

How do I wait for a specific job instead of all jobs?

Use the Wait() method on the individual Job pointer returned by queue.AddJob(). As implemented in job.go (lines 179-183), each job maintains its own sync.WaitGroup that completes when the job handler finishes. This allows fine-grained synchronization without blocking the entire worker or affecting other concurrent jobs.

What happens if I call WaitAndStop on an already stopped worker?

WaitAndStop() first checks the worker's status. If the worker is already stopped, the method returns immediately without error. The implementation ensures idempotent behavior: if curProcessing is already zero and the worker status is stopped, the condition evaluates to false instantly, and the method proceeds to the stop logic, which safely handles the already-stopped state.

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