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
}
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.
Looks good to me! I think WaitGroup.Go is probably the easiest thing to read for ultra-simple concurrency, but it doesn’t support limits like you are here. Or panics.
I especially like that I don’t need extra logic to handle edge cases like an empty array or an array smaller than the number of workers. And the context handling is baked in without any select statements.
curious if you’ve tested this at scale with really large slices — since every item still gets its own goroutine, wonder if there’s overhead when pCount is way smaller than len(values)
This will not be a problem, since the synchronized call to semaphore.Acquire inside the for-loop will only start new go routines, once the old ones are finished. So there will only ever by pCount parallel go routines at the same time.
This is the reason why the acquire is in the for-loop and not inside the go routine.
Here are some Benchmarking results. I compared the Semaphore Solution with a WaitGroup based approach using similar Error handling (context, panic,…): Code
The Semaphore uses more allocations and worse performance on small workloads, but provides better performance for heavy workloads. The additional allocations are 1 allocation per array-element, since each element will over the course of execution create an additional go routine on the heap, while the waitGroup Solution reuses the same number of go routines for all elements. On the other hand the Semaphore seems to have the edge, if the number of elements or the load of the computation is really high - maybe this has something to do with heap access, maybe the cost of the heap allocations is negligible, if the computation itself also needs to access the heap a lot.
Overall I think for most real life workloads with complex functions (like big data, file access, network calls,…) the semaphore solutions seems better. For very tight CPU bound computations, you should probably go a level lower anyway (either to the GPU or by slicing your memory for locality and L1/L2 cache hits) - this is a high level function, with the focus on high level ergonomics, ease of use and stability (context, panics, …).
Benchmarks using base64.EncodeToString(BIG STRING) as workload: