conc

The conc package provides several tools for structured concurrency in Go:

• conc.WaitGroup: A safer version of sync.WaitGroup that handles panics.
• pool.Pool: A concurrency-limited task runner.
• pool.ResultPool: A concurrent task runner that collects task results.
• pool.ErrorPool: A concurrent task runner for fallible tasks.
• pool.ContextPool: A concurrent task runner that cancels tasks on failure.
• stream.Stream: Processes an ordered stream of tasks in parallel with serial callbacks.
• iter.Map: Concurrently maps a slice.
• iter.ForEach: Concurrently iterates over a slice.
• panics.Catcher: Catches panics in your own goroutines.

Unlike standard goroutines which crash the entire process on panic, conc.WaitGroup catches panics. When wg.Wait() is called, if any spawned goroutine panicked, Wait() will panic itself, decorating the value with a stacktrace from the child goroutine to preserve debugging information.

func main() {
    var wg conc.WaitGroup
    wg.Go(doSomethingThatMightPanic)
    // panics with a nice stacktrace
    wg.Wait()
}

Install conc using the standard Go get command:

go get github.com/sourcegraph/conc

To prevent goroutine leaks, conc encourages scoped concurrency where every goroutine has an owner. Use conc.WaitGroup to spawn goroutines with .Go() and ensure .Wait() is called before the WaitGroup goes out of scope. If you need a goroutine to outlive the caller, pass the *conc.WaitGroup into the spawning function.

func main() {
    var wg conc.WaitGroup
    defer wg.Wait()

    startTheThing(&wg)
}

func startTheThing(wg *conc.WaitGroup) {
    wg.Go(func() { ... })
}

Use conc.WaitGroup as a drop-in, safer replacement for sync.WaitGroup. It simplifies the syntax by allowing you to pass a function directly to .Go(), which handles the boilerplate of adding to the group and signaling completion.

func main() {
    var wg conc.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Go(doSomething)
    }
    wg.Wait()
}

To process elements from a channel (stream) using a fixed number of goroutines, use pool.New() combined with .WithMaxGoroutines(n). This prevents unbounded goroutine creation.

func process(stream chan int) {
    p := pool.New().WithMaxGoroutines(10)
    for elem := range stream {
        elem := elem
        p.Go(func() {
            handle(elem)
        })
    }
    p.Wait()
}

To process elements from an input channel and ensure they are sent to an output channel in the same order they were received, use stream.New(). You can limit concurrency using .WithMaxGoroutines(n).

Each task passed to .Go() must return a stream.Callback (a function) that performs the actual output operation (e.g., sending to a channel).

func mapStream(
    in chan int,
    out chan int,
    f func(int) int,
) {
    s := stream.New().WithMaxGoroutines(10)
    for elem := range in {
        elem := elem
        s.Go(func() stream.Callback {
            res := f(elem)
            return func() { out <- res }
        })
    }
    s.Wait()
}

Use iter.Map to transform a slice into a new slice concurrently. The function passed to iter.Map should accept a pointer to the element (e.g., func(*int) int).

func concMap(
    input []int,
    f func(*int) int,
) []int {
    return iter.Map(input, f)
}

Instead of manually managing workers and channels to process a slice, use iter.ForEach to execute a function on every element of a slice concurrently.

func process(values []int) {
    iter.ForEach(values, handle)
}

All pools are created using pool.New() or pool.NewWithResults[T](), and can be configured with the following methods:

• p.WithMaxGoroutines(): Configures the maximum number of goroutines in the pool.
• p.WithErrors(): Configures the pool to run tasks that return errors.
• p.WithContext(ctx): Configures the pool to run tasks that should be canceled on first error.
• p.WithFirstError(): (For ErrorPool) Configures the pool to only keep the first returned error rather than an aggregated error.
• p.WithCollectErrored(): (For ResultContextPool) Configures result pools to collect results even when the task errored.