concurrency

README

Concurrency Helpers

This package provides a set of concurrency helpers for Go projects.

WorkerPool for concurrent processing of slices in Go

This package provides a generic and straightforward concurrency worker pool for Go (requires Go 1.26+ for generics). It allows you to process jobs in parallel using a fixed number of workers (goroutines), returning the results and any associated errors through a single result channel.

Features

• Generic Worker Function: Define any input (T) and output (U) types for your jobs and results.
• Configurable Concurrency: Set the number of workers you want to run in parallel.
• Context-Aware: Supports context cancellation to stop worker processing early.
• Buffered Channels: Results are sent back via a buffered channel, and you can range over this channel until all jobs are complete.
• Rich Result Structure: The result channel returns a WorkerPoolResult struct containing the original Job, its JobIndex, the Result, and any Error.

Usage

To use the WorkerPool, import the following package:

import "git.prolicht.digital/golib/concurrency"

The following example demonstrates how to use the WorkerPool to process a list of integers concurrently:

package main

import (
    "context"
    "fmt"
    "log"

    "git.prolicht.digital/golib/concurrency"
)

func main() {
    // Define some sample data
    jobs := []int{1, 2, 3, 4, 5}

    // Define a worker function that doubles the input
    doubleWorker := func(job int) (int, error) {
        // Here you can perform more complex operations
		    if job == 3 {
            return 0, fmt.Errorf("job %d failed", job) // Return an error for job 3
        }
        return job * 2, nil // Return the result for the other jobs
    }

    // Create a context (here, using a background context).
	  // You can also use a context with a timeout or deadline.
    ctx := context.Background()

    // Start the worker pool with 3 threads
    resultChan := concurrency.WorkerPool[int, int](ctx, 3, jobs, doubleWorker)

    // Receive results from the channel
    for result := range resultChan {
        if result.Error != nil {
            // Handle any error returned by the worker
            log.Printf("Error processing job %d (index %d): %v", result.Job, result.JobIndex, result.Error)
            continue
        }
        // Use the result as needed
        fmt.Printf("Job: %d, Index: %d, Result: %d\n", result.Job, result.JobIndex, result.Result)
    }

    fmt.Println("All jobs processed!")
}

It is very important to complete the range loop over the result channel to ensure that all results are processed. Otherwise the worker pool might leak goroutines.

---

Documentation

Overview

Package concurrency provides convenience functions for working with concurrency in Go.

Index

• func WorkerPool[T, U any](ctx context.Context, workers int, jobData []T, workerFn WorkerFunc[T, U]) chan WorkerPoolResult[T, U]
• type PanicError
  • func (e *PanicError) Error() string
• type WorkerFunc
• type WorkerPoolResult

Functions

func WorkerPool

func WorkerPool[T, U any](
	ctx context.Context,
	workers int,
	jobData []T,
	workerFn WorkerFunc[T, U],
) chan WorkerPoolResult[T, U]

WorkerPool executes the worker function for a fixed amount of threads (workers). Each worker function processes one element of the jobData array. The returned result channel gets closed once all jobs have been processed. The channel must be read until it is closed, otherwise the goroutine will leak.

Types

type PanicError

type PanicError struct {
	Message    string
	Stacktrace string
}

PanicError is a custom error that contains information about a panic.

func (*PanicError) Error

func (e *PanicError) Error() string

type WorkerFunc

type WorkerFunc[T, U any] func(job T) (U, error)

WorkerFunc is a function that processes a job of type T and returns a result of type U. The error return value should be nil if the job was processed successfully.

type WorkerPoolResult

type WorkerPoolResult[T, U any] struct {
	// Job is the original job that was processed.
	// This can be useful for correlating results with their input data.
	Job T
	// JobIndex is the index of the job in the original jobData slice.
	// It can be used to maintain the order of results if needed.
	JobIndex int
	// Result is the output of the worker function for the given job.
	// It will be of type U, as defined by the WorkerFunc.
	Result U
	// Error is the error returned by the worker function for the given job.
	// It will be nil if the job was processed successfully.
	Error error
}

WorkerPoolResult is used to return worker pool results via the worker pool channel.