sama
Tiny, fast, generic worker pools for Go.
sama exposes 4 functions with the same spirit and different guarantees:
Kun, fire-and-forget: process items from a channel with a worker pool, no output.Chan, unordered: process items from a channel with a worker pool, returns results as they are ready.San, ordered: same asChan, but preserves input order in the output stream.Tsu, It iterates from 0 to n (not included), calling "do" for each value and returning the results in order.
All functions stop naturally when the input channel is closed and all in-flight work finishes.
Go >= 1.25 (no external dependency, only uses generics and the new Go function from sync.WaitGroup).
// It does not produce any output.
func Kun[T any](ch chan T, do func(v T), concurrency ...int)
// Chan consumes ch with a pool and returns an output channel that yields do(v) results.
// Results are emitted as soon as they're ready (order is NOT guaranteed).
func Chan[Ti, To any](ch <-chan Ti, do func(v Ti) To, concurrency ...int) chan To
// San is like Chan but guarantees that the output preserves the input order.
// The i-th value read from ch produces the i-th value on the returned channel.
func San[Ti, To any](ch <-chan Ti, do func(v Ti) To, concurrency ...int) chan To
// Tsu iterates from 0 to n (not included), calling "do" for each value and returning the results in order.
func Tsu[To any](n int, do func(v int) To, concurrency ...int) chan To
concurrencyis optional. If omitted,samauses an arbitrary default (2xruntime.NumCPU()).Close the input channel to finish; the output channel (for
Chan/San) will close automatically when all work is done.
Installation
Patterns & tips
Sample use-case
in = make(chan string)
// San preserves input order in the output.
out = sama.San(in, func(s string) string {
// Simulate variable latency to show ordering guarantee.
if s == "bravo" {
time.Sleep(50 * time.Millisecond)
}
return strings.ToUpper(s)
})
)
go func() {
for _, v := range []string{"alpha", "bravo", "charlie", "delta"} {
in <- v
}
close(in)
}()
// Order matches the input exactly.
for res := range out {
fmt.Println(res)
}
// Output:
// ALPHA
// BRAVO
// CHARLIE
// DELTA
Backpressure & buffering
The input channel acts as backpressure. If producers outpace consumers, either:
- Increase
concurrency, or - Use a buffered
chanfor the input.
For Chan/San, the returned output channel is buffered enough for good throughput.
Still, if a downstream consumer is slow, overall speed will match the slowest stage.
Cancellation & shutdown
Since the API does not accept context.Context, cancellation is cooperative:
- Close the input channel to signal completion.
- Ensure your
dofunction returns promptly (check your own contexts inside do if it performs I/O).
Errors
Use sum types or tuples to propagate errors:
Val string
Err error
}
out := sama.Chan(in, func(s string) Out {
v, err := doWork(s)
return Out{Val: v, Err: err}
}, 8)
When to choose which:
Kun: side effects only (DB writes, HTTP calls where you handle errors internally).Chan: maximum throughput, order doesn't matter (idempotent / commutative workloads).San: streaming and order matters (like re-sequencing responses for a client).Tsu: running in parallel a function iterating from 0 to n not included, returning the result in order.
Guarantees
- No leaks: all goroutines exit once the input channel is closed and all work completes.
- Ordering:
San,Tsustable (input index order).Channone (as-completed).
- Throughput:
Chantends to be the fastest.San,Tsuadd a tiny reorder buffer proportional to "out-of-order window".