When to Use the conc Package for Concurrent Processing in Go

The conc package is the best choice when you need a bounded, FIFO-ordered concurrent queue that limits active goroutines while preserving job submission order and providing observable state for graceful shutdowns.

The conc package in the aperturerobotics/util repository offers a specialized concurrent processing primitive for Go applications. Unlike generic worker pools, it guarantees FIFO ordering of job starts while strictly bounding the number of simultaneous goroutines through a lightweight queue implementation. If your workload requires deterministic execution order alongside resource throttling, the conc package provides a purpose-built solution optimized for fire-and-forget semantics.

What Makes the conc Package Unique

The conc package distinguishes itself through deterministic start order and built-in observability. According to the source code in conc/queue.go, the ConcurrentQueue type manages jobs using a linked list backend and a broadcast primitive for state synchronization. This architecture ensures that jobs begin execution in the exact order they are submitted, even though completion times may vary based on workload duration.

4 Ideal Use Cases for the conc Package

1. Throttling Concurrent Goroutines

When processing bursts of network requests or disk I/O, unbounded concurrency can exhaust file descriptors or memory. The conc package enforces a hard limit on active goroutines through NewConcurrentQueue(maxConcurrency), preventing resource exhaustion while maintaining a backlog of pending work in the internal linkedlist.LinkedList.

2. Preserving FIFO Submission Order

Many concurrent queues start jobs as workers become available, losing the original submission sequence. In conc/queue.go, the implementation stores jobs in a linkedlist.LinkedList (defined in linkedlist/linkedlist.go), ensuring that the earliest submitted job always starts next, regardless of how long previous jobs take to complete.

3. Observing Queue State for Back-Pressure

Applications requiring back-pressure or graceful degradation need visibility into queued versus running job counts. The WatchState method exposes real-time metrics without polling, leveraging the broadcast.Broadcast primitive from broadcast/broadcast.go to notify observers immediately when state changes occur.

4. Fire-and-Forget Job Semantics

The conc package is optimized for func() closures that run to completion without returning values. Each job is a simple callback enqueued via Enqueue(func()), making the API surface minimal and eliminating the complexity of result channels or per-job context management. Cancellation is handled at the queue level via WaitIdle rather than per-job tokens.

How the conc Queue Works Internally

Understanding the internals clarifies why conc behaves differently from channel-based workers. The queue maintains two critical components:

• Storage layer: A linkedlist.LinkedList (from linkedlist/linkedlist.go) holds pending jobs in insertion order, providing O(1) enqueue and dequeue operations.
• Synchronization: The broadcast.Broadcast type (from broadcast/broadcast.go) acts as a condition variable, waking waiters when jobs complete or the queue becomes idle.

When a job finishes, the queue atomically decrements the running count and immediately starts the next job from the linked list head, ensuring the concurrency limit is always respected without gaps. The conc/queue_test.go file demonstrates how this implementation maintains correct ordering under high concurrency.

Practical Code Examples

Basic Throttled Worker Pool

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/aperturerobotics/util/conc"
)

func main() {
	// Allow at most 3 concurrent jobs.
	q := conc.NewConcurrentQueue(3)

	// Enqueue 10 jobs.
	for i := 0; i < 10; i++ {
		id := i // capture loop variable
		q.Enqueue(func() {
			fmt.Printf("job %d started\n", id)
			time.Sleep(500 * time.Millisecond)
			fmt.Printf("job %d finished\n", id)
		})
	}

	// Block until all jobs complete.
	if err := q.WaitIdle(context.Background(), nil); err != nil {
		panic(err)
	}
	fmt.Println("all jobs done")
}

This example demonstrates the core pattern: bounded concurrency with guaranteed start order.

Monitoring Queue State for Back-Pressure

package main

import (
	"context"
	"log"
	"time"

	"github.com/aperturerobotics/util/conc"
)

func main() {
	q := conc.NewConcurrentQueue(2)

	// Monitor queue depth and active workers.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	
	go func() {
		q.WatchState(ctx, nil, func(queued, running int) (bool, error) {
			log.Printf("queued=%d running=%d", queued, running)
			return true, nil // continue watching
		})
	}()

	// Enqueue burst workload.
	for i := 0; i < 5; i++ {
		id := i
		q.Enqueue(func() {
			time.Sleep(2 * time.Second)
			log.Printf("job %d done", id)
		})
	}

	if err := q.WaitIdle(context.Background(), nil); err != nil {
		panic(err)
	}
	cancel()
}

The WatchState callback receives live updates whenever the queue state changes, enabling reactive back-pressure strategies.

Graceful Shutdown with Timeout

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/aperturerobotics/util/conc"
)

func main() {
	q := conc.NewConcurrentQueue(4)

	// Start long-running jobs.
	for i := 0; i < 3; i++ {
		id := i
		q.Enqueue(func() {
			fmt.Printf("job %d started\n", id)
			time.Sleep(10 * time.Second)
			fmt.Printf("job %d finished\n", id)
		})
	}

	// Allow 2 seconds for completion after initial delay.
	time.Sleep(3 * time.Second)
	
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	
	if err := q.WaitIdle(ctx, nil); err != nil {
		fmt.Println("shutdown timed out:", err)
	} else {
		fmt.Println("clean shutdown completed")
	}
}

WaitIdle respects context cancellation, allowing servers to enforce shutdown deadlines while respecting the concurrency limit.

Summary

• Use conc when you need strict FIFO job ordering combined with bounded concurrency limits.
• Throttling is handled automatically by NewConcurrentQueue(maxConcurrency), preventing resource exhaustion during burst workloads.
• Observability comes standard via WatchState and WaitIdle, supporting graceful shutdown and back-pressure monitoring.
• Implementation relies on linkedlist.LinkedList for storage and broadcast.Broadcast for synchronization, as implemented in conc/queue.go.
• Limitations include fire-and-forget semantics only (no per-job results) and queue-level rather than per-job cancellation.

Frequently Asked Questions

How does conc differ from a standard Go worker pool?

Standard worker pools typically use channels and start jobs in whatever order workers become available, losing submission sequence. The conc package explicitly maintains a FIFO queue using linkedlist.LinkedList, ensuring jobs start in submission order regardless of execution duration, while broadcast.Broadcast provides efficient state notifications.

Can I cancel individual jobs in the conc queue?

No. Cancellation operates at the queue level via the context passed to WaitIdle. Individual jobs are simple func() closures without cancellation tokens. For per-job cancellation, consider higher-level utilities or custom implementations with context propagation.

What data structure does conc use to store pending jobs?

The queue uses a linkedlist.LinkedList from linkedlist/linkedlist.go to store pending jobs. This provides O(1) enqueue and dequeue operations while maintaining strict insertion order, which is critical for the FIFO guarantees.

Is the conc package suitable for CPU-bound workloads?

While functional for CPU-bound tasks, the conc package excels at I/O-bound throttling where preserving submission order matters. For pure CPU parallelism without ordering constraints, standard worker pools or errgroup.Group may offer lower overhead, though they lack the deterministic start order and state observability of conc.

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