pond

pond is a high-performance Go library for managing concurrency via the Worker Pool pattern. It allows you to run many tasks while limiting the number of active goroutines to prevent resource exhaustion (e.g., limiting HTTP requests, database connections, or API rate limits).

Key Features:

• Automatic Scaling: Worker goroutines are created when needed and removed when idle (scale to zero).
• Concurrency Control: Limit the maximum number of concurrent tasks.
• Flexible Task Submission: Supports fire-and-forget, waiting for completion, and submitting groups of tasks.
• Type Safety (v2): Type-safe APIs for tasks returning results or errors.
• Robustness: Panics are captured and returned as errors.
• Advanced Management: Supports subpools, dynamic resizing, and configurable parent contexts for cancellation.

Use pool.NewGroup() to create a TaskGroup. You can submit multiple tasks to the group using group.Submit(...). Calling group.Wait() will block until all tasks in the group are complete. If tasks return errors (via SubmitErr), group.Wait() will return the first error encountered.

// Create a pool with limited concurrency
pool := pond.NewPool(10)

// Create a task group
group := pool.NewGroup()

// Submit a group of tasks
for i := 0; i < 20; i++ {
	i := i
	group.Submit(func() {
		fmt.Printf("Running group task #%d\n", i)
	})
}

// Wait for all tasks in the group to complete
err := group.Wait()

Use pool.NewGroupContext(ctx) to create a task group linked to a context.Context. If the context is cancelled or times out, the group's Wait() method will return an error. To ensure 'in-flight' (currently running) tasks are also stopped, tasks should reference group.Context() within their execution logic.

// Create a pool with limited concurrency
pool := pond.NewPool(10)

// Create a context with a 5s timeout
timeout, _ := context.WithTimeout(context.Background(), 5*time.Second)

// Create a task group with a context
group := pool.NewGroupContext(timeout)

// Submit a group of tasks
for i := 0; i < 20; i++ {
	i := i
	group.Submit(func() {
		fmt.Printf("Running group task #%d\n", i)
	})
}

// Wait for all tasks in the group to complete or the timeout to occur, whichever comes first
err := group.Wait()

Install the pond library using go get to manage concurrent tasks with a worker pool pattern. This version (v2) includes features like bounded/unbounded queues, type-safe result/error tasks, and panic recovery.

go get -u github.com/alitto/pond/v2

By default, a pool uses context.Background(). You can provide a custom context using the pond.WithContext(ctx) option during pool creation. When this context is cancelled, the pool stops accepting new work and drains the queue (queued tasks are cancelled and not executed) to shut down cleanly.

// Create a custom context that can be cancelled
customCtx, cancel := context.WithCancel(context.Background())

// This creates a pool that is stopped when customCtx is cancelled
pool := pond.NewPool(10, pond.WithContext(customCtx))

Both TaskGroup and ResultTaskGroup provide a Context() method. This context is tied to the group's lifecycle:

1. It is cancelled if the parent context passed during creation is cancelled.
2. It is cancelled if any task within the group returns an error.
3. It is cancelled if Stop() is called on the group.

When a task is running during cancellation, the group allows the task to complete before returning from Wait(). However, tasks should ideally monitor group.Context().Err() to stop their own work early.

Use pond.NewPool(n) to create a pool with a fixed maximum number of concurrent workers. You can submit tasks using pool.Submit(func()). To ensure all tasks finish before the program exits, use pool.StopAndWait().

package main

import (
	"fmt"

	"github.com/alitto/pond/v2"
)

func main() {

	// Create a pool with limited concurrency
	pool := pond.NewPool(100)

	// Submit 1000 tasks
	for i := 0; i < 1000; i++ {
		i := i
		pool.Submit(func() {
			fmt.Printf("Running task #%d\n", i)
		})
	}

	// Stop the pool and wait for all submitted tasks to complete
	pool.StopAndWait()
}

If you do not want to manage a pool instance, you can use the global default pool via pond.Submit(...) or pond.SubmitErr(...). The default pool is unbounded and scales automatically based on the number of tasks submitted.

// Submit a task to the default pool and wait for it to complete
err := pond.SubmitErr(func() error {
	fmt.Println("Running task in default pool")
	return nil
}).Wait()

if err != nil {
	fmt.Printf("Failed to run task: %v", err)
} else {
	fmt.Println("Task completed successfully")
}

A subpool is a pool created from an existing pool that uses a fraction of the parent pool's maximum worker capacity. Use pool.NewSubpool(n) to create one.

// Create a pool with limited concurrency
pool := pond.NewPool(10)

// Create a subpool with a fraction of the parent pool's maximum number of workers
subpool := pool.NewSubpool(5)

// Submit a task to the subpool
subpool.Submit(func() {
	fmt.Println("Running task in subpool")
})

// Stop the subpool and wait for all submitted tasks to complete
subpool.StopAndWait()

Use pond.NewResultPool[T](n) combined with pool.SubmitErr(func() (T, error)) to submit tasks that return both a value and an error. Use .Wait() to retrieve both.

// Create a concurrency limited pool that accepts tasks that return a string
pool := pond.NewResultPool[string](10)

// Submit a task that returns a string value or an error
task := pool.SubmitErr(func() (string, error) {
	return "Hello, World!", nil
})

// Wait for the task to complete and get the result
result, err := task.Wait()
// result = "Hello, World!" and err = nil