It looks like that https://pkg.go.dev/golang.org/x/sync/errgroup has most of the features described here. Have you considered improving https://pkg.go.dev/golang.org/x/sync/errgroup instead of adding a new one?

Being honest, I didn't know that package existed. My searches related to the worker pool pattern never once surfaced sync/errgroup as an option.

While I admit there is definitely significant feature overlap upon first look, I'm not quite convinced these aren't two distinct use-cases.

By my (possibly naïve) understanding of worker pools, a pool simply (and continually) dispatches tasks to a set number of concurrent workers regardless of task outcome, and remains available to run future tasks until the pool is stopped. I believe that sync/errgroup (and I've only briefly studied this package so very possible I'm wrong here), on the other hand, will terminate upon encountering an error by any task, and only remains usable so long as no task has errored.

Assuming my observations are true, I can see distinct value provided by the approach of sync/errgroup and that of this proposal.

The Run method makes this API prone to goroutine leaks and synchronization bugs: it is too easy to accidentally leak the Run goroutine, especially given that it doesn't come with a mechanism to way for that goroutine to finish. (For more detail, see my GopherCon '18 talk, particularly starting around slide 75.)

That leaves New, Push, and Wait, which are analogous to errgroup.Group's SetLimit (#27837), Go, and Wait respectively, but you are correct that errgroup specifically focuses on error aggregation (to facilitate fork/join-style concurrency) whereas the API proposed here intentionally does not (to facilitate reuse).

That makes it more similar to cmd/go/internal/par.Queue, which provides only NewQueue(maxActive int), Add(func()), and Idle() <-chan struct{}. I think that is the more appropriate API here — it has the same concurrency-limiting properties, but without the leak-prone Run method and with a somewhat more flexible way to select on completion.

The Run method makes this API prone to goroutine leaks and synchronization bugs: it is too easy to accidentally leak the Run goroutine, especially given that it doesn't come with a mechanism to way for that goroutine to finish. (For more detail, see my GopherCon '18 talk, particularly starting around slide 75.)

This actually had crossed my mind, but I decided to punt pending some feedback. One thought was to add a func (p WorkerPool) WaitStop() to wait for the pool itself to stop gracefully. Also, this design doesn't require that Run(context.Context) be run as a goroutine, it can absolutely be called normally and will block until the context is cancelled. Tasks can also be added via Push() before the call to Run(context.Context) or from another goroutine, but I'm not sure that alleviates your concerns.

After learning about the existence sync/errgroup, I've been reading through it's history and came across your talk. Will be listening to it tonight. In the mean time, very interested in hearing other ideas on how to combat this.

That leaves New, Push, and Wait, which are analogous to errgroup.Group's SetLimit (#27837), Go, and Wait respectively, but you are correct that errgroup specifically focuses on error aggregation (to facilitate fork/join-style concurrency) whereas the API proposed here intentionally does not (to facilitate reuse).

That makes it more similar to cmd/go/internal/par.Queue, which provides only NewQueue(maxActive int), Add(func()), and Idle() <-chan struct{}. I think that is the more appropriate API here — it has the same concurrency-limiting properties, but without the leak-prone Run method and with a somewhat more flexible way to select on completion.

I'll need to do some studying of par.Queue to competently comment, but my understanding is that it is an internal package, thus not available for use. Are you suggesting making that API commonly available or using it as an influence to modify this proposal's defined API?

On first look, the ergonomics of par.Queue seem a bit more "clever" than I was going for. The WaitGroup-style Wait() feels very clear and natural.

Thanks for the feedback!

I've finally had some time to revisit this and have incorporated feedback from @bcmills (thank you, btw):

sync/workerpool

package workerpool

type Task interface {
	Invoke(ctx context.Context) error
}

type WorkerPool struct {}

func New(n int64) WorkerPool
func (p *WorkerPool) Push(t Task, opts ...TaskOption)
func (p *WorkerPool) Run(ctx context.Context) <-chan struct{}

Example

package main

import (
    "context"
    "golang.org/x/sync/workerpool"
    "time"
)

func main() {

    pool := workerpool.New(2)
    ctx, cancel := context.WithTimeout(context.Background(), 5 * time.Second)
    defer cancel()
    
    done := wp.Run(ctx) // runs until context is cancelled

    // pool.Push(Task 1)
    // pool.Push(Task 2)

    // block until the workerpool is stopped and done channel is closed
    <-done

}

I have also updated my reference implementation:

https://pkg.go.dev/github.com/tniswong/workerpool/v2@v2.0.0
https://github.com/tniswong/workerpool/tree/master/v2