concurrency-semaphore-pool

Answer

The implementation uses a buffered channel as a semaphore to bound concurrency, and a pending-map reorder buffer to emit results in submission order despite out-of-order completion. Context cancellation propagates to all in-flight workers, and the first job error cancels the remaining work.

Core design

             ┌──────────────────┐
   jobs[] ──►│  semaphore (chan)│◄── at most N tokens
             └──────┬───────────┘
                    │ launch goroutines
                    ▼
             ┌──────────────┐
             │  jobFn(ctx)  │  (each acquires + releases a token)
             └──────┬───────┘
                    │ result{idx, val, err}
                    ▼
             ┌──────────────┐
             │ pending map  │  reorder by index
             └──────┬───────┘
                    ▼
             results[] in order

Code listing — pool.go

package semaphorepool

import (
    "context"
    "sync"
)

// SemaphorePool is a generic worker pool that runs at most maxWorkers
// concurrent goroutines.
type SemaphorePool[T, R any] struct {
    maxWorkers int
    jobFn      func(context.Context, T) (R, error)
}

// New creates a pool that runs at most n goroutines concurrently.
func New[T, R any](n int, fn func(context.Context, T) (R, error)) *SemaphorePool[T, R] {
    return &SemaphorePool[T, R]{maxWorkers: n, jobFn: fn}
}

type result[T any] struct {
    idx int
    val T
    err error
}

// Process runs all jobs with bounded concurrency and returns results
// in the same order as the input slice.
func (p *SemaphorePool[T, R]) Process(ctx context.Context, jobs []T) ([]R, error) {
    if len(jobs) == 0 {
        return nil, nil
    }

    n := len(jobs)
    results := make([]R, n)
    sem := make(chan struct{}, p.maxWorkers)
    resCh := make(chan result[R], n)

    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    var wg sync.WaitGroup
    var firstErr error
    var setErrOnce sync.Once

    // Submit all jobs, acquiring semaphore tokens
    for i, job := range jobs {
        if ctx.Err() != nil {
            break
        }
        select {
        case sem <- struct{}{}:
        case <-ctx.Done():
            break
        }

        wg.Add(1)
        go func(idx int, job T) {
            defer func() {
                <-sem
                wg.Done()
            }()
            if ctx.Err() != nil {
                resCh <- result[R]{idx: idx, err: ctx.Err()}
                return
            }
            val, err := p.jobFn(ctx, job)
            if err != nil {
                setErrOnce.Do(func() {
                    firstErr = err
                    cancel()
                })
            }
            resCh <- result[R]{idx: idx, val: val, err: err}
        }(i, job)
    }

    go func() {
        wg.Wait()
        close(resCh)
    }()

    // Reorder results using a pending map keyed by index
    pending := make(map[int]result[R])
    nextIdx := 0
    for r := range resCh {
        pending[r.idx] = r
        for {
            pr, ok := pending[nextIdx]
            if !ok {
                break
            }
            results[pr.idx] = pr.val
            if pr.err != nil && firstErr == nil {
                firstErr = pr.err
            }
            delete(pending, nextIdx)
            nextIdx++
        }
    }

    if firstErr == nil {
        firstErr = ctx.Err()
    }
    return results, firstErr
}

Key design decisions