syncx

Typed concurrency patterns for Go 1.22+.

syncx provides reusable, context-aware primitives for worker pools, stream pipelines, pub/sub, throttling, circuit breaking, debouncing, and more. Every blocking operation accepts a context.Context, so waits can always be bounded by cancellation or a deadline. All background goroutines terminate when their input closes or their context is cancelled — no leaks.

Install

go get github.com/Ismahsantiago/syncx

Pattern guide

Pick a primitive by the problem it solves:

Primitive	Package	Solves
Future	syncx	Start slow work now (RPC, query), use the result later; awaitable by many goroutines.
Retry	syncx	Absorb transient failures with linear backoff instead of duplicating retry loops.
Semaphore	syncx	Bound how many goroutines touch a scarce resource at once.
Barrier	syncx	Align goroutines at a checkpoint between phases; reusable per round.
OrDone	syncx	Make for v := range ch cancellable without nested selects.
Map / Filter / Reduce / Collect	stream	Transform and aggregate channel streams without per-stage goroutine boilerplate.
FanIn	stream	Merge N producers into one consumer loop.
FanOut	stream	Parallelize one stream across N consumers (round-robin, each value to one consumer).
Broadcast / Tee	stream	Give every consumer the full stream (each value to all consumers).
Pipeline	stream	Declare a multi-stage flow in one call; one context shuts down all stages.
Pool	workerpool	Process a job stream with N workers; results carry input, output, and error.
Batcher	workerpool	Group items by size or timeout to amortize per-call overhead (bulk inserts, batched APIs).
PubSub	pubsub	In-process events: publishers target topic names, subscribers attach/detach at runtime.
Throttle	control	Cap operation rate with a token bucket (stay under upstream rate limits).
CircuitBreaker	control	Fail fast when a dependency keeps failing; let it recover instead of piling on.
Debounce	control	Collapse bursts of events into one reaction with the latest value.

Each primitive has one or more runnable Example... in its package documentation — typical usage plus edge cases like error handling, cancellation, and recovery — see the GoDoc, or the example_*_test.go files. 34 examples in total across the five packages.

Quick example

ctx := context.Background()
jobs := make(chan int)

go func() {
    defer close(jobs)
    for i := 0; i < 5; i++ {
        jobs <- i
    }
}()

pool := workerpool.New(3, func(ctx context.Context, n int) (int, error) {
    return n * 2, nil
})

for r := range pool.Start(ctx, jobs) {
    fmt.Println(r.Input, r.Value, r.Err)
}

Benchmarks

bench/ compares select syncx primitives against the most widely used equivalent from the Go ecosystem — x/sync/semaphore, x/sync/errgroup, sony/gobreaker, cenkalti/backoff — with reference numbers and a methodology note. It is its own Go module (replaced to this one), so running it does not pull those dependencies into syncx itself:

cd bench
go test -run=^$ -bench=. -benchmem ./...

See bench/README.md for results and how to read them.

Quality

• Unit tests for every package, run with -race
• Runnable Example... tests for every primitive, including error paths, cancellation, and recovery — not just the happy path
• Race-safe APIs designed for concurrent use
• Zero runtime dependencies

License

MIT