pargo

README

Pargo

A lightweight, generic, context-aware concurrency library for processing collections in Go.

Pargo helps you process collections concurrently without writing goroutines, channels, or sync.WaitGroup.

Designed for modern Go, Pargo leverages generics, context propagation, and configurable worker pools to make parallel collection processing simple, predictable, and production-ready.

---

Why Pargo?

Go provides excellent concurrency primitives, but processing collections in parallel often requires repetitive boilerplate.

Instead of writing this:

jobs := make(chan User)
results := make(chan Profile)

var wg sync.WaitGroup

for i := 0; i < workers; i++ {
	go worker(...)
}

You simply write:

profiles, err := pargo.Map(
	ctx,
	users,
	fetchProfile,
)

Cleaner.

Safer.

More maintainable.

---

Features

• Generic Map
• Generic Filter
• Generic Reduce
• Generic FlatMap
• Context-aware operations
• Configurable worker pools
• Ordered result collection
• Parallel reduction
• Functional options
• Panic recovery
• Error propagation
• Zero external runtime dependencies
• High-performance concurrent execution

---

Installation

go get github.com/mohammedimrankasab/pargo

---

Quick Start

Map

numbers := []int{1, 2, 3, 4}

result, err := pargo.Map(
	context.Background(),
	numbers,
	func(ctx context.Context, idx, value int) (int, error) {
		return value * value, nil
	},
)

fmt.Println(result)

Output

[1 4 9 16]

---

Filter

numbers := []int{1,2,3,4,5,6}

result, err := pargo.Filter(
	context.Background(),
	numbers,
	func(ctx context.Context, idx, value int) (bool, error) {
		return value%2 == 0, nil
	},
)

fmt.Println(result)

Output

[2 4 6]

---

Reduce

sum, err := pargo.Reduce(
	context.Background(),
	[]int{1,2,3,4,5},
	0,
	func(ctx context.Context, acc, value int) (int, error) {
		return acc + value, nil
	},
)

fmt.Println(sum)

Output

15

---

FlatMap

words := []string{
	"go",
	"rocks",
}

chars, err := pargo.FlatMap(
	context.Background(),
	words,
	func(ctx context.Context, idx int, word string) ([]string, error) {
		return strings.Split(word, ""), nil
	},
)

fmt.Println(chars)

---

Worker Configuration

The default worker count is runtime.GOMAXPROCS(0).

Override it when needed.

result, err := pargo.Map(
	ctx,
	items,
	mapper,
	pargo.WithWorkers(8),
)

---

Context Support

Every operation respects context cancellation.

ctx, cancel := context.WithTimeout(
	context.Background(),
	time.Second,
)
defer cancel()

result, err := pargo.Map(ctx, items, mapper)

---

Error Handling

If any worker returns an error:

• Remaining work is cancelled
• The first error is returned
• No goroutines are leaked

---

Reduce Semantics

Reduce performs chunk-based parallel reduction.

For deterministic results, the reducer should be associative, for example:

• Sum
• Product
• Minimum
• Maximum
• Logical AND / OR

Reducers that depend on evaluation grouping or have side effects may produce different results than a purely sequential reduction.

---

Benchmarks

See:

• benchmark.md

Current benchmark environment:

• Apple M5
• macOS
• arm64
• Go 1.26

---

Examples

examples/basic
examples/filter
examples/cancellation
examples/ordered
examples/workers
examples/reduce

---

Testing

Run the complete validation suite:

go test ./...
go test -race ./...
go test ./... -cover
go test -bench=. -benchmem

---

Design Principles

Pargo is built around a few core ideas:

• Idiomatic Go
• Generics first
• Context everywhere
• Functional options
• Ordered results
• Small API surface
• Zero external dependencies
• High test coverage
• Predictable behavior

---

Roadmap

Version	Status	Features
v0.1	✅	Parallel Map
v0.2	✅	Parallel Filter
v0.3	✅	Parallel Reduce
v0.4	✅	Parallel FlatMap
v0.5	Planned	Batch Processing
v0.6	Planned	Retry Support
v0.7	Planned	Streaming Collections
v1.0	Planned	Stable Public API

---

Project Status

🚧 Pre-1.0

Pargo is under active development.

Until v1.0.0 the public API may evolve based on community feedback.

Bug reports, feature requests, and pull requests are always welcome.

---

Contributing

Please read CONTRIBUTING.md before opening a pull request.

---

License

Released under the MIT License.

Documentation

Overview

Package pargo provides generic, context-aware primitives for processing collections concurrently.

Pargo offers simple, production-ready APIs for common collection operations, including Map, Filter, and Reduce. All operations support context cancellation, configurable worker pools, and deterministic ordering where applicable.

Example:

result, err := pargo.Map(ctx, items, mapper)

By default, Pargo uses runtime.GOMAXPROCS(0) worker goroutines.

Index

• func Filter[T any](ctx context.Context, items []T, fn FilterFunc[T], opts ...Option) ([]T, error)
• func FlatMap[T any, R any](ctx context.Context, items []T, fn FlatMapFunc[T, R], opts ...Option) ([]R, error)
• func Map[T any, R any](ctx context.Context, items []T, fn MapFunc[T, R], opts ...Option) ([]R, error)
• func Reduce[T any](ctx context.Context, items []T, initial T, fn ReduceFunc[T], opts ...Option) (T, error)
• type FilterFunc
• type FlatMapFunc
• type MapFunc
• type Option
  • func WithWorkers(n int) Option
• type ReduceFunc

Functions

func Filter

func Filter[T any](
	ctx context.Context,
	items []T,
	fn FilterFunc[T],
	opts ...Option,
) ([]T, error)

Filter evaluates fn for each element of items concurrently and returns a new slice containing only the elements for which fn returns true.

The relative order of retained elements matches their order in the input slice.

If any invocation of fn returns an error, processing stops, the remaining work is canceled, and the error is returned.

The number of workers can be configured using WithWorkers.

func FlatMap

func FlatMap[T any, R any](
	ctx context.Context,
	items []T,
	fn FlatMapFunc[T, R],
	opts ...Option,
) ([]R, error)

FlatMap applies fn concurrently to every element and flattens the results into a single slice while preserving the input order.

If any invocation of fn returns an error, processing stops, the remaining work is canceled, and the error is returned.

The number of workers can be configured using WithWorkers.

func Map

func Map[T any, R any](
	ctx context.Context,
	items []T,
	fn MapFunc[T, R],
	opts ...Option,
) ([]R, error)

Map applies fn to each element of items concurrently and returns a new slice containing the transformed values.

The output slice preserves the order of the input slice regardless of the order in which workers complete.

If any invocation of fn returns an error, processing stops, the remaining work is canceled, and the error is returned.

The number of workers can be configured using WithWorkers. By default, Map uses runtime.GOMAXPROCS(0) workers.

func Reduce

func Reduce[T any](
	ctx context.Context,
	items []T,
	initial T,
	fn ReduceFunc[T],
	opts ...Option,
) (T, error)

Reduce combines the elements of items into a single value using fn.

Reduction is performed concurrently by first reducing independent chunks and then combining the partial results.

For deterministic results, fn should be associative (for example, addition, multiplication, minimum, or maximum). Reducers whose results depend on evaluation grouping or side effects may produce different results than a purely sequential reduction.

If fn returns an error, processing stops and the error is returned.

Types

type FilterFunc

type FilterFunc[T any] func(
	ctx context.Context,
	idx int,
	value T,
) (bool, error)

FilterFunc determines whether an element should be retained.

Returning true keeps the element in the output slice. Returning false discards it.

type FlatMapFunc

type FlatMapFunc[T any, R any] func(context.Context, int, T) ([]R, error)

type MapFunc

type MapFunc[T any, R any] func(context.Context, int, T) (R, error)

MapFunc transforms a value of type T into a value of type R.

The index parameter corresponds to the element's position in the input slice.

type Option

type Option func(*config)

Option configures the behavior of a Pargo operation.

func WithWorkers

func WithWorkers(n int) Option

WithWorkers specifies the maximum number of worker goroutines used by an operation.

Values less than one are ignored, causing the default worker count to be used instead.

type ReduceFunc

type ReduceFunc[T any] func(
	ctx context.Context,
	accumulator T,
	value T,
) (T, error)

ReduceFunc combines an accumulated value with the next input value and returns the updated accumulator.