I recently reworked my helper function for parallel execution to use a simple Semaphore to limit concurrency with a clean and simple loop. It effectively behaves like a worker pool, running at most `pCount` concurrent go routines. It is controlled via context and cancelling will immediately return:
func RunParallel[T any](ctx context.Context, values []T, pCount int64, f func(idx int, value T)) (err error) {
pContext, cancel := context.WithCancelCause(ctx)
defer cancel(nil)
sem := semaphore.NewWeighted(pCount)
for i, v := range values {
// loop will only advance once the semaphore has an opening
err = sem.Acquire(pContext, 1)
if err != nil {
return context.Cause(pContext)
}
go func() {
defer sem.Release(1)
defer func() {
if p := recover(); p != nil {
// a panic will cancel the context and stop all executions
cancel(fmt.Errorf("panic in parallel execution %d (%+v): %v", i, v, p))
}
}()
f(i, v)
}()
}
// after the loop this will block until the last routines have finished
err = sem.Acquire(pContext, pCount)
if err != nil {
return context.Cause(pContext)
}
return
}
Usage:
input := []string{"A","B","C",....}
results = make([]string{}, len(input))
err = RunParallel(ctx, input, 4, func(i int, s string){
results[i] = calculateExpensiveMagic(s)
})
I’m open for feedback / code review especially in light of the alternatives (waitGroup, workerPool,…) I think this code is very easy to read and behaves stable.