commit 70a33087d2bb443dd107b3a4f2aa21bb1695b6cc
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Thu, 21 Aug 2025 09:37:08 -0300
initial
Signed-off-by: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Diffstat:
16 files changed, 1506 insertions(+), 0 deletions(-)
diff --git a/README.md b/README.md
@@ -0,0 +1,4 @@
+# jackhammer
+
+This package provides experimental utilities that extend Go's standard library.
+
diff --git a/channel/channel.go b/channel/channel.go
@@ -0,0 +1,145 @@
+package channel
+
+import (
+ "context"
+ "reflect"
+ "slices"
+ "time"
+)
+
+// Drain the provided channel.
+func Drain[T any](ch <-chan T) {
+ for range ch {
+ }
+}
+
+// TryReceive selects on a channel returning true if there was a value.
+func TryReceive[T any](ch <-chan T) (t T, ok bool) {
+ select {
+ case t = <-ch:
+ ok = true
+ default:
+ }
+ return t, ok
+}
+
+// TrySend attempts a sends on a channel returning true if success.
+func TrySend[T any](ch chan<- T, t T) bool {
+ select {
+ case ch <- t:
+ return true
+ default:
+ return false
+ }
+}
+
+// ReceiveIn selects on the provided channel up to duration `d` and returns the value.
+// If the duration elapses, the function returns the zero value of `T` and false.
+func ReceiveIn[T any](ch <-chan T, d time.Duration) (T, bool) {
+ timer := time.NewTimer(d)
+ select {
+ case t := <-ch:
+ return t, true
+ case <-timer.C:
+ return *new(T), false
+ }
+}
+
+// Collect all values from a channel into a slice.
+func Collect[T any](in <-chan T) []T {
+ out := []T{}
+ for t := range in {
+ out = append(out, t)
+ }
+ return out
+}
+
+// CollectAllCtx buffers every value received value into a slice until all input channels are closed
+// or the context is cancelled.
+func CollectAllCtx[T any](ctx context.Context, chs ...chan T) []T {
+ out := []T{}
+ for v := range FanInCtx(ctx, chs...) {
+ out = append(out, v)
+ }
+ return out
+}
+
+// CollectAll buffers every value received value into a slice until all input channels are closed.
+func CollectAll[T any](chs ...chan T) (out []T) {
+ return CollectAllCtx(context.Background(), chs...)
+}
+
+// BufferedSend uses slice allocation to buffer items providing a dynamically
+// buffered channel send. This send will never block.
+//
+// If the channel is open, we send on it.
+// If the channel is full, we select off the slice, append to it and send it
+// back onto the channel.
+//
+// The loop will drain the channel until it becomes available to send. After
+// the function returns, the value v is guaranteed to be sent on the channel,
+// however it is not guaranteed that this is the only value sent.
+//
+// The input channel MUST have capacity of 1.
+func BufferedSend[S ~[]E, E any](in chan S, v E) {
+ if cap(in) != 1 {
+ panic("BufferedSend is only valid for single buffered channels")
+ }
+
+ var items S = []E{v}
+
+ for {
+ select {
+ case buf := <-in:
+ items = append(buf, items...)
+ case in <- items:
+ return
+ }
+ }
+}
+
+// FanIn returns a channel that emits every value received from the input channel set.
+// The output channel is closed when all input channels are closed.
+func FanIn[T any](chs ...chan T) chan T {
+ return FanInCtx(context.Background(), chs...)
+}
+
+// FanInCtx returns a channel that emits every value received from the input channel set.
+// The output channel is closed when all input channels are closed, or when the context
+// is cancelled.
+func FanInCtx[T any](ctx context.Context, chs ...chan T) chan T {
+ out := make(chan T)
+
+ go func() {
+ defer close(out)
+
+ cases := make([]reflect.SelectCase, 0, len(chs)+1)
+
+ cases = append(cases, reflect.SelectCase{
+ Dir: reflect.SelectRecv,
+ Chan: reflect.ValueOf(ctx.Done()),
+ })
+
+ for _, ch := range chs {
+ cases = append(cases, reflect.SelectCase{
+ Dir: reflect.SelectRecv,
+ Chan: reflect.ValueOf(ch),
+ })
+ }
+
+ for len(cases) > 1 {
+ ii, recv, ok := reflect.Select(cases)
+ if ok {
+ out <- recv.Interface().(T)
+ } else {
+ cases = slices.Delete(cases, ii, ii+1)
+ }
+ if ii == 0 {
+ // This is always the ctx.Done channel.
+ return
+ }
+ }
+ }()
+
+ return out
+}
diff --git a/channel/channel_test.go b/channel/channel_test.go
@@ -0,0 +1,150 @@
+package channel
+
+import (
+ "context"
+ "slices"
+ "testing"
+)
+
+func TestCollectAll(t *testing.T) {
+ t.Run("normal", func(t *testing.T) {
+ ctx := context.Background()
+
+ channels := []chan int{}
+
+ for ii := 0; ii < 10; ii += 1 {
+ channels = append(channels, make(chan int))
+ }
+
+ var got []int
+ var want []int
+
+ done := make(chan any)
+
+ go func() {
+ got = CollectAllCtx(ctx, channels...)
+ close(done)
+ }()
+
+ for ii := 0; ii < 100; ii += 1 {
+ want = append(want, ii)
+ channels[ii%len(channels)] <- ii
+ }
+
+ for _, ch := range channels {
+ close(ch)
+ }
+
+ <-done
+
+ if !slices.Equal(got, want) {
+ t.Fatalf("got=%v, want=%v", got, want)
+ }
+ })
+
+ t.Run("context cancel", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+
+ channels := []chan int{}
+
+ for ii := 0; ii < 10; ii += 1 {
+ channels = append(channels, make(chan int))
+ }
+
+ var got []int
+ var want []int
+
+ done := make(chan any)
+
+ go func() {
+ got = CollectAllCtx(ctx, channels...)
+ close(done)
+ }()
+
+ for ii := 0; ii < 100; ii += 1 {
+ want = append(want, ii)
+ channels[ii%len(channels)] <- ii
+ }
+
+ cancel()
+
+ <-done
+
+ if !slices.Equal(got, want) {
+ t.Fatalf("got=%v, want=%v", got, want)
+ }
+ })
+
+}
+
+func TestFanIn(t *testing.T) {
+
+ t.Run("normal", func(t *testing.T) {
+ channels := make([]chan int, 0, 10)
+
+ for ii := 0; ii < 10; ii += 1 {
+ channels = append(channels, make(chan int))
+ }
+
+ go func() {
+ for ii := 0; ii < 100; ii += 1 {
+ channels[ii%len(channels)] <- ii
+ }
+ for _, ch := range channels {
+ close(ch)
+ }
+ }()
+
+ var got []int
+ var want []int
+
+ for ii := 0; ii < 100; ii += 1 {
+ want = append(want, ii)
+ }
+
+ for n := range FanIn(channels...) {
+ got = append(got, n)
+ }
+
+ slices.Sort(got)
+
+ if !slices.Equal(got, want) {
+ t.Fatalf("got=%v, want=%v", got, want)
+ }
+ })
+
+ t.Run("context cancel", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+
+ channels := make([]chan int, 0, 10)
+
+ for ii := 0; ii < 10; ii += 1 {
+ channels = append(channels, make(chan int))
+ }
+
+ go func() {
+ for ii := 0; ii < 100; ii += 1 {
+ channels[ii%len(channels)] <- ii
+ }
+ cancel()
+ }()
+
+ var got []int
+ var want []int
+
+ for ii := 0; ii < 100; ii += 1 {
+ want = append(want, ii)
+ }
+
+ for n := range FanInCtx(ctx, channels...) {
+ got = append(got, n)
+ }
+
+ slices.Sort(got)
+
+ if !slices.Equal(got, want) {
+ t.Fatalf("got=%v, want=%v", got, want)
+ }
+ })
+
+}
diff --git a/errorsx/errorsx.go b/errorsx/errorsx.go
@@ -0,0 +1,29 @@
+package errorsx
+
+import (
+ "strings"
+)
+
+type PolyError struct {
+ Errs []error
+}
+
+func (err PolyError) Unwrap() []error {
+ return err.Errs
+}
+
+func (err PolyError) Error() string {
+ var b strings.Builder
+ b.WriteString("multiple errors:\n")
+ for ii, e := range err.Errs {
+ if e == nil {
+ continue
+ }
+ b.WriteString(" - ")
+ b.WriteString(e.Error())
+ if ii != len(err.Errs)-1 {
+ b.WriteByte('\n')
+ }
+ }
+ return b.String()
+}
diff --git a/go.mod b/go.mod
@@ -0,0 +1,5 @@
+module git.sr.ht/~jackmordaunt/jackhammer
+
+go 1.23.6
+
+require golang.org/x/sync v0.12.0
diff --git a/go.sum b/go.sum
@@ -0,0 +1,2 @@
+golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
+golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
diff --git a/iterx/iterx.go b/iterx/iterx.go
@@ -0,0 +1,159 @@
+package iterx
+
+import (
+ "iter"
+ "runtime"
+ "sync"
+
+ "git.sr.ht/~jackmordaunt/jackhammer/channel"
+ "git.sr.ht/~jackmordaunt/jackhammer/errorsx"
+)
+
+// Fanout processes a sequence concurrently.
+func Fanout[V any](seq iter.Seq[V], fn func(value V)) {
+ wg := sync.WaitGroup{}
+ defer wg.Wait()
+
+ for v := range seq {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ fn(v)
+ }()
+ }
+}
+
+// Fanout2 processes a sequence concurrently.
+func Fanout2[K any, V any](seq iter.Seq2[K, V], fn func(key K, value V)) {
+ type pair[K any, V any] struct {
+ key K
+ value V
+ }
+
+ queue := make(chan pair[K, V])
+ wg := sync.WaitGroup{}
+
+ for ii := 0; ii < runtime.NumCPU(); ii += 1 {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for p := range queue {
+ fn(p.key, p.value)
+ }
+ }()
+ }
+
+ for k, v := range seq {
+ queue <- pair[K, V]{key: k, value: v}
+ }
+
+ close(queue)
+ wg.Wait()
+}
+
+// FanoutErr2 processes a sequence cuncurrently, returning an aggregate error.
+func FanoutErr2[K any, V any](seq iter.Seq2[K, V], fn func(key K, value V) error) error {
+ errs := []error{}
+ failures := make(chan error)
+ done := make(chan any)
+
+ Fanout2(seq, func(key K, value V) {
+ if err := fn(key, value); err != nil {
+ failures <- err
+ }
+ })
+
+ go func() {
+ errs = channel.Collect(failures)
+ close(done)
+ }()
+
+ close(failures)
+ <-done
+
+ if len(errs) > 0 {
+ return errorsx.PolyError{Errs: errs}
+ }
+
+ return nil
+}
+
+// For2 processes a sequence with the given function.
+func For2[K any, V any](seq iter.Seq2[K, V], fn func(key K, value V)) {
+ for k, v := range seq {
+ fn(k, v)
+ }
+}
+
+// ForErr2 processes a sequence with the given function.
+func ForErr2[K any, V any](seq iter.Seq2[K, V], fn func(key K, value V) error) error {
+ var errs []error
+
+ for k, v := range seq {
+ if err := fn(k, v); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ if len(errs) > 0 {
+ return errorsx.PolyError{Errs: errs}
+ }
+
+ return nil
+}
+
+// Map transforms an iterator by applying a function to each element and returning the result.
+func Map[S iter.Seq[In], In any, Out any](s S, fn func(In) Out) iter.Seq[Out] {
+ return func(yield func(Out) bool) {
+ for v := range s {
+ if !yield(fn(v)) {
+ break
+ }
+ }
+ }
+}
+
+// MapIndex is like [Map] but provides the iteration index to the transform function.
+func MapIndex[S iter.Seq[In], In any, Out any](s S, fn func(int, In) Out) iter.Seq[Out] {
+ return func(yield func(Out) bool) {
+ ii := -1
+ for v := range s {
+ ii += 1
+ if !yield(fn(ii, v)) {
+ break
+ }
+ }
+ }
+}
+
+// Map2 transforms an iterator by applying a function to each element and returning the result.
+func Map2[S iter.Seq2[In1, In2], In1 any, In2 any, Out any](s S, fn func(In1, In2) Out) iter.Seq[Out] {
+ return func(yield func(Out) bool) {
+ for v1, v2 := range s {
+ if !yield(fn(v1, v2)) {
+ break
+ }
+ }
+ }
+}
+
+// ToMap collects a sequence into a map, using [keyFor] to extract the key from each element.
+func ToMap[S iter.Seq2[int, E], E any, K comparable](s S, keyFor func(int, E) K) map[K]E {
+ out := map[K]E{}
+ for ii, v := range s {
+ out[keyFor(ii, v)] = v
+ }
+ return out
+}
+
+// WithIndex adapts a single-value iterator to a two-value iterator that yeilds the index.
+func WithIndex[T any](s iter.Seq[T]) iter.Seq2[int, T] {
+ return iter.Seq2[int, T](func(yield func(int, T) bool) {
+ i := 0
+ for v := range s {
+ if !yield(i, v) {
+ break
+ }
+ i++
+ }
+ })
+}
diff --git a/iterx/iterx_test.go b/iterx/iterx_test.go
@@ -0,0 +1,30 @@
+package iterx
+
+import (
+ "slices"
+ "strconv"
+ "testing"
+)
+
+func TestMap(t *testing.T) {
+ values := []string{}
+ for ii := 0; ii < 100; ii += 1 {
+ values = append(values, strconv.Itoa(ii))
+ }
+
+ iterator := Map(slices.Values(values), func(s string) int {
+ n, err := strconv.Atoi(s)
+ if err != nil {
+ panic(err)
+ }
+ return n
+ })
+
+ for ii, n := range slices.Collect(iterator) {
+ want := values[ii]
+ got := strconv.Itoa(n)
+ if want != got {
+ t.Fatalf("got=%s, want=%s", got, want)
+ }
+ }
+}
diff --git a/para/gather.go b/para/gather.go
@@ -0,0 +1,16 @@
+package para
+
+import (
+ "golang.org/x/sync/errgroup"
+)
+
+// Gather processes a batch of heterogenous operations.
+func Gather(fns ...func() error) error {
+ work := errgroup.Group{}
+
+ for _, fn := range fns {
+ work.Go(fn)
+ }
+
+ return work.Wait()
+}
diff --git a/para/para.go b/para/para.go
@@ -0,0 +1,72 @@
+package para
+
+import (
+ "context"
+ "fmt"
+ "iter"
+ "sync"
+
+ "golang.org/x/sync/errgroup"
+)
+
+// Fanout processes a sequence concurrently.
+func Fanout[V any](seq iter.Seq[V], fn func(value V)) {
+ wg := sync.WaitGroup{}
+ defer wg.Wait()
+
+ for v := range seq {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ fn(v)
+ }()
+ }
+}
+
+// Batch processes the sequence concurrently. Any error cancels the
+// entire batch. No results are generated. This is helpful for a fanout of
+// side-effects.
+func Batch[T any](ctx context.Context, seq iter.Seq[T], work func(ctx context.Context, ii int, v T) error) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ group, ctx := errgroup.WithContext(ctx)
+
+ ii := 0
+ for v := range seq {
+ func(ii int) {
+ group.Go(func() error {
+ return work(ctx, ii, v)
+ })
+ }(ii)
+ ii += 1
+ }
+
+ return group.Wait()
+}
+
+// MapErr processes the sequence concurrently, transforming [In] to [Out].
+// Any error cancels the entire batch. This is helpful for a fanout of
+// fallible transformations.
+func MapErr[In any, Out any](ctx context.Context, seq []In, work func(ctx context.Context, ii int, v In) (Out, error)) ([]Out, error) {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ group, ctx := errgroup.WithContext(ctx)
+
+ res := make([]Out, len(seq))
+
+ for ii, v := range seq {
+ group.Go(func() error {
+ out, err := work(ctx, ii, v)
+ res[ii] = out
+ return err
+ })
+ }
+
+ if err := group.Wait(); err != nil {
+ return nil, fmt.Errorf("map: %w", err)
+ }
+
+ return res, nil
+}
diff --git a/para/para_test.go b/para/para_test.go
@@ -0,0 +1,212 @@
+package para
+
+import (
+ "context"
+ "fmt"
+ "slices"
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "testing"
+)
+
+func TestFanout(t *testing.T) {
+ t.Run("happy", func(t *testing.T) {
+ n := 10
+
+ input := []int{}
+ for ii := range n {
+ input = append(input, ii)
+ }
+
+ var calls atomic.Int32
+
+ Fanout(slices.Values(input), func(v int) {
+ calls.Add(1)
+ })
+
+ if got := calls.Load(); got != int32(n) {
+ t.Fatalf("calls mismatch: want=%v, got=%v", n, got)
+ }
+ })
+}
+
+func TestBatch(t *testing.T) {
+ t.Run("happy", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ input := []string{}
+ want := []int64{}
+
+ for ii := range 100 {
+ input = append(input, strconv.Itoa(ii))
+ want = append(want, int64(ii))
+ }
+
+ got := make([]int64, len(input))
+
+ err := Batch(ctx, slices.Values(input), func(ctx context.Context, ii int, v string) error {
+ n, err := strconv.ParseInt(v, 10, 64)
+ got[ii] = n
+ return err
+ })
+
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ if !slices.Equal(got, want) {
+ t.Fatalf("want=%v, got=%v", want, got)
+ }
+ })
+
+ t.Run("sad", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ input := []string{}
+
+ n := 5
+
+ for ii := range n {
+ input = append(input, strconv.Itoa(ii))
+ }
+
+ trigger := make(chan any, 1)
+ ready := sync.WaitGroup{}
+ ready.Add(n)
+
+ go func() {
+ ready.Wait()
+ // trigger one closure to return an error
+ trigger <- nil
+ }()
+
+ var nextID atomic.Int32
+ var canceled atomic.Int32
+ var errored atomic.Int32
+
+ err := Batch(ctx, slices.Values(input), func(ctx context.Context, ii int, v string) error {
+ id := nextID.Add(1)
+
+ ready.Done()
+
+ select {
+ case <-ctx.Done():
+ // most should get canceled due to the one that failed
+ // should be n-1
+ canceled.Add(1)
+ t.Logf("%d: canceled", id)
+ return ctx.Err()
+ case <-trigger:
+ // should be one
+ errored.Add(1)
+ t.Logf("%d: errored", id)
+ return fmt.Errorf("an error")
+ }
+ })
+
+ if err == nil {
+ t.Fatalf("expected error")
+ }
+
+ if got := canceled.Load(); got != int32(n-1) {
+ t.Errorf("expected %d cancels, got %v", int32(n-1), got)
+ }
+
+ if got := errored.Load(); got != 1 {
+ t.Errorf("expected 1 error, got %v", got)
+ }
+ })
+}
+
+func TestMapErr(t *testing.T) {
+ t.Run("happy", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ input := []string{}
+ want := []int64{}
+
+ for ii := range 100 {
+ input = append(input, strconv.Itoa(ii))
+ want = append(want, int64(ii))
+ }
+
+ got, err := MapErr(ctx, input, func(ctx context.Context, ii int, v string) (int64, error) {
+ return strconv.ParseInt(v, 10, 64)
+ })
+
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ if !slices.Equal(got, want) {
+ t.Fatalf("want=%v, got=%v", want, got)
+ }
+ })
+
+ t.Run("sad", func(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ input := []string{}
+
+ n := 5
+
+ for ii := range n {
+ input = append(input, strconv.Itoa(ii))
+ }
+
+ trigger := make(chan any, 1)
+ ready := sync.WaitGroup{}
+ ready.Add(n)
+
+ go func() {
+ ready.Wait()
+ // trigger one closure to return an error
+ trigger <- nil
+ }()
+
+ var nextID atomic.Int32
+ var canceled atomic.Int32
+ var errored atomic.Int32
+
+ got, err := MapErr(ctx, input, func(ctx context.Context, ii int, v string) (int64, error) {
+ id := nextID.Add(1)
+
+ ready.Done()
+
+ select {
+ case <-ctx.Done():
+ // most should get canceled due to the one that failed
+ // should be n-1
+ canceled.Add(1)
+ t.Logf("%d: canceled", id)
+ return 0, ctx.Err()
+ case <-trigger:
+ // should be one
+ errored.Add(1)
+ t.Logf("%d: errored", id)
+ return 0, fmt.Errorf("an error")
+ }
+ })
+
+ if err == nil {
+ t.Fatalf("expected error")
+ }
+
+ if len(got) != 0 {
+ t.Fatalf("expected result to be empty, got %v", got)
+ }
+
+ if got := canceled.Load(); got != int32(n-1) {
+ t.Errorf("expected %d cancels, got %v", int32(n-1), got)
+ }
+
+ if got := errored.Load(); got != 1 {
+ t.Errorf("expected 1 error, got %v", got)
+ }
+ })
+}
diff --git a/slicesx/slicesx.go b/slicesx/slicesx.go
@@ -0,0 +1,325 @@
+// Package slicesx provides slice utilities that compliment Go standard
+// slices package.
+package slicesx
+
+import (
+ "errors"
+ "fmt"
+ "maps"
+ "slices"
+ "sync"
+
+ "git.sr.ht/~jackmordaunt/jackhammer/errorsx"
+ "git.sr.ht/~jackmordaunt/jackhammer/iterx"
+)
+
+// Fanout across all elements of a slice in parallel.
+// The function receives the index and element.
+func Fanout[S ~[]E, E any](seq S, fn func(int, E)) {
+ iterx.Fanout2(slices.All(seq), fn)
+}
+
+// FanoutErr is like [Fanout] but returns an aggregated error.
+func FanoutErr[S ~[]E, E any](seq S, fn func(int, E) error) error {
+ return iterx.FanoutErr2(slices.All(seq), fn)
+}
+
+// Map transforms a slice by applying a function to each element and returning the result.
+func Map[S ~[]E, E any, Out any](s S, fn func(E) Out) []Out {
+ out := make([]Out, len(s))
+ for ii := range s {
+ out[ii] = fn(s[ii])
+ }
+ return out
+}
+
+// MapIndex is like [Map] but provides the iteration index to the transform function.
+func MapIndex[S ~[]E, E any, Out any](s S, fn func(int, E) Out) []Out {
+ out := make([]Out, len(s))
+ for ii := range s {
+ out[ii] = fn(ii, s[ii])
+ }
+ return out
+}
+
+// MapSafe is like [Map] but recovers from panics and returns an error instead.
+func MapSafe[S ~[]E, E any, Out any](s S, fn func(E) Out) (_ []Out, err error) {
+ defer func() {
+ if v := recover(); v != nil {
+ if e, ok := v.(error); ok {
+ err = e
+ } else {
+ err = fmt.Errorf("%v", v)
+ }
+ }
+ }()
+ return Map(s, fn), err
+}
+
+// MapIndexSafe is like [MapIndex] but recovers from panics and returns an error instead.
+func MapIndexSafe[S ~[]E, E any, Out any](s S, fn func(int, E) Out) (_ []Out, err error) {
+ defer func() {
+ if v := recover(); v != nil {
+ if e, ok := v.(error); ok {
+ err = e
+ } else {
+ err = fmt.Errorf("%v", v)
+ }
+ }
+ }()
+ return MapIndex(s, fn), err
+}
+
+// ParaMap transforms a slice by applying a function to each element and returning the result.
+// Transforms are executed concurrently.
+func ParaMap[S ~[]E, E any, Out any](s S, fn func(E) Out) []Out {
+ out := make([]Out, len(s))
+ wg := sync.WaitGroup{}
+ for ii := range s {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ out[ii] = fn(s[ii])
+ }()
+ }
+ wg.Wait()
+ return out
+}
+
+// MapErr transforms a slice by applying a fallible function to each element and
+// returning the result.
+//
+// If the function returns an error the map operation completes, returning an
+// aggregated error for each failed transform. The output list will be short
+// by the number of failures. Length of the output is not guaranteed to match that
+// of the input.
+func MapErr[S ~[]E, E any, Out any](s S, fn func(E) (Out, error)) (out []Out, err error) {
+ out = make([]Out, 0, len(s))
+ for ii := range s {
+ v, e := fn(s[ii])
+ if e != nil {
+ err = errors.Join(err, e)
+ } else {
+ out = append(out, v)
+ }
+ }
+ return out, err
+}
+
+// ParaMapErr is [MapErr], but all instances of fn run concurrently.
+func ParaMapErr[S ~[]E, E any, Out any](s S, fn func(E) (Out, error)) (out []Out, err error) {
+ out = make([]Out, len(s))
+ errs := make([]error, len(s))
+ wg := sync.WaitGroup{}
+
+ for ii := range s {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ v, e := fn(s[ii])
+ if e != nil {
+ errs[ii] = e
+ } else {
+ out[ii] = v
+ }
+ }()
+ }
+
+ wg.Wait()
+
+ for ii, err := range errs {
+ if err != nil {
+ out = slices.Delete(out, ii, ii+1)
+ }
+ }
+
+ if len(errs) > 0 {
+ return out, errorsx.PolyError{Errs: errs}
+ }
+
+ return out, nil
+}
+
+// For processes a sequence with the given function.
+func For[S ~[]E, E any](seq S, fn func(int, E)) {
+ iterx.For2(slices.All(seq), fn)
+}
+
+// ForErr processes a sequence with the given function.
+func ForErr[S ~[]E, E any](seq S, fn func(int, E) error) error {
+ return iterx.ForErr2(slices.All(seq), fn)
+}
+
+// Generate returns a slice of [n] items produced by [fn].
+func Generate[E any](n int, fn func(int) E) []E {
+ out := make([]E, n)
+ for ii := range n {
+ out[ii] = fn(ii)
+ }
+ return out
+}
+
+// Repeat returns a slice of [n] items of value [v].
+func Repeat[E any](n int, v E) []E {
+ out := make([]E, n)
+ for ii := range n {
+ out[ii] = v
+ }
+ return out
+}
+
+// PopFront returns the first element from the slice.
+func PopFront[S ~[]E, E any](s *S) E {
+ defer func() { (*s) = (*s)[1:] }()
+ return (*s)[0]
+}
+
+// PopFrontSafe returns the first element from the slice, or false.
+func PopFrontSafe[S ~[]E, E any](s *S) (e E, ok bool) {
+ if s == nil {
+ return e, false
+ }
+ if len(*s) == 0 {
+ return e, false
+ }
+ return PopFront(s), true
+}
+
+// PopBack returns the last element from the slice.
+func PopBack[S ~[]E, E any](s *S) E {
+ defer func() { (*s) = (*s)[:len(*s)-1] }()
+ return (*s)[len(*s)-1]
+}
+
+// PopBackSafe returns the last element from the slice, or false.
+func PopBackSafe[S ~[]E, E any](s *S) (e E, ok bool) {
+ if s == nil {
+ return e, false
+ }
+ if len(*s) == 0 {
+ return e, false
+ }
+ return PopBack(s), true
+}
+
+// Filter out elements from a slice when the predicate returns false.
+func Filter[S ~[]E, E any](s S, fn func(E) bool) S {
+ out := make(S, 0, len(s))
+ for _, v := range s {
+ if fn(v) {
+ out = append(out, v)
+ }
+ }
+ return slices.Clip(out)
+}
+
+// FilterMap maps [In] to [Out], ignoring entries that return false.
+func FilterMap[S ~[]In, In any, Out any](s S, fn func(In) (Out, bool)) []Out {
+ out := make([]Out, 0, len(s))
+ for _, v := range s {
+ if v, ok := fn(v); ok {
+ out = append(out, v)
+ }
+ }
+ return slices.Clip(out)
+}
+
+// TakeRange selects all items between [start] and [end] and removes them from the slice.
+// Does not bounds check.
+func TakeRange[S ~[]E, E any](s *S, start, end int) S {
+ taken := (*s)[start:end]
+ *s = slices.Delete(*s, start, end)
+ return taken
+}
+
+// TakeRangeSafe selects all items between [start] and [end] and removes them from the slice.
+// Checks the bounds.
+func TakeRangeSafe[S ~[]E, E any](s *S, start, end int) (S, bool) {
+ if start < 0 || start >= end || end < 0 || end > len(*s)-1 || start > len(*s)-1 {
+ return nil, false
+ }
+ taken := (*s)[start:end]
+ *s = slices.Delete(*s, start, end)
+ return taken, true
+}
+
+// TakeAllFunc selects all items for which [fn] returns true and removes them from the slice.
+func TakeAllFunc[S ~[]E, E any](s *S, fn func(E) bool) (S, bool) {
+ indexes := []int{}
+ for ii, v := range *s {
+ if fn(v) {
+ indexes = append(indexes, ii)
+ }
+ }
+ if len(indexes) == 0 {
+ return nil, false
+ }
+ out := make(S, 0, len(indexes))
+ for _, index := range indexes {
+ out = append(out, (*s)[index])
+ }
+ *s = slices.DeleteFunc(*s, fn)
+ return out, true
+}
+
+// Take selects the item and removes it, without bounds checks.
+func Take[S ~[]E, E any](s *S, index int) E {
+ taken := (*s)[index]
+ *s = slices.Delete(*s, index, index+1)
+ return taken
+}
+
+// TakeSafe selects the item and removes it, returning true if found.
+func TakeSafe[S ~[]E, E any](s *S, index int) (E, bool) {
+ if index < 0 || index > len(*s)-1 {
+ return *new(E), false
+ }
+ taken := (*s)[index]
+ *s = slices.Delete(*s, index, index+1)
+ return taken, true
+}
+
+// TakeFunc selects the first item that satisfies [fn] and removes it.
+func TakeFunc[S ~[]E, E any](s *S, fn func(E) bool) (E, bool) {
+ ii := slices.IndexFunc(*s, fn)
+ if ii < 0 {
+ return *new(E), false
+ }
+ taken := (*s)[ii]
+ *s = slices.Delete(*s, ii, ii+1)
+ return taken, true
+}
+
+// Normalize de-duplicates a slices of comparable elements.
+// Doesn't rely on sorting.
+func Normalize[S ~[]E, E comparable](s S) S {
+ s = slices.Clone(s)
+ seen := map[E]struct{}{}
+ for _, e := range s {
+ seen[e] = struct{}{}
+ }
+ return slices.Collect(maps.Keys(seen))
+}
+
+// Flatten reduces a 2D slices to a 1D slice.
+func Flatten[E any](s [][]E) []E {
+ count := 0
+ for _, ss := range s {
+ count += len(ss)
+ }
+
+ out := make([]E, 0, count)
+
+ for _, ss := range s {
+ for _, e := range ss {
+ out = append(out, e)
+ }
+ }
+
+ return out
+}
+
+// ToMap converts a slice to a map, using [keyFor] to extract the key from each element.
+func ToMap[S ~[]E, E any, K comparable](s S, keyFor func(int, E) K) map[K]E {
+ return iterx.ToMap(slices.All(s), keyFor)
+}
diff --git a/slicesx/slicesx_test.go b/slicesx/slicesx_test.go
@@ -0,0 +1,218 @@
+package slicesx
+
+import (
+ "slices"
+ "strconv"
+ "testing"
+)
+
+func TestMap(t *testing.T) {
+ list := Generate(3, func(ii int) int {
+ return ii
+ })
+
+ t.Run("MapSafe", func(t *testing.T) {
+ if _, err := MapSafe(list, func(s int) int {
+ panic("test panic")
+ }); err == nil {
+ t.Fatalf("expected error, not nil")
+ }
+ })
+
+ t.Run("MapIndexSafe", func(t *testing.T) {
+ if _, err := MapIndexSafe(list, func(ii int, s int) int {
+ panic("test panic")
+ }); err == nil {
+ t.Fatalf("expected error, not nil")
+ }
+ })
+
+ t.Run("MapSafe", func(t *testing.T) {
+ if got, err := MapSafe(list, func(s int) string {
+ return strconv.Itoa(s)
+ }); err != nil {
+ t.Fatalf("unexpected error, not nil")
+ } else {
+ want := []string{"0", "1", "2"}
+ if !slices.Equal(got, want) {
+ t.Fatalf("result mismatch: got=%v, want=%v", got, want)
+ }
+ }
+ })
+
+ t.Run("MapIndexSafe", func(t *testing.T) {
+ if got, err := MapIndexSafe(list, func(ii int, s int) string {
+ return strconv.Itoa(s + ii)
+ }); err != nil {
+ t.Fatalf("unexpected error, not nil")
+ } else {
+ want := []string{"0", "2", "4"}
+ if !slices.Equal(got, want) {
+ t.Fatalf("result mismatch: got=%v, want=%v", got, want)
+ }
+ }
+ })
+
+ t.Run("Map", func(t *testing.T) {
+ got := Map(list, func(s int) string {
+ return strconv.Itoa(s)
+ })
+ want := []string{"0", "1", "2"}
+ if !slices.Equal(got, want) {
+ t.Fatalf("result mismatch: got=%v, want=%v", got, want)
+ }
+ })
+
+ t.Run("MapIndex", func(t *testing.T) {
+ got := MapIndex(list, func(ii int, s int) string {
+ return strconv.Itoa(s + ii)
+ })
+ want := []string{"0", "2", "4"}
+ if !slices.Equal(got, want) {
+ t.Fatalf("result mismatch: got=%v, want=%v", got, want)
+ }
+ })
+}
+
+func TestPop(t *testing.T) {
+ t.Run("PopFront", func(t *testing.T) {
+ original := Generate(5, func(ii int) int {
+ return ii
+ })
+
+ list := Generate(5, func(ii int) int {
+ return ii
+ })
+
+ for _, want := range original {
+ length := len(list)
+
+ head, _ := PopFrontSafe(&list)
+
+ if len(list) == length {
+ t.Fatalf("failed to modify slice")
+ }
+
+ if head != want {
+ t.Fatalf("popped wrong value: got=%d, want=%d", head, want)
+ }
+ }
+
+ v, ok := PopFrontSafe(&list)
+ if v != 0 || ok == true {
+ t.Fatalf("unexpected value from popping empty list: %v", v)
+ }
+ })
+
+ t.Run("PopBack", func(t *testing.T) {
+ original := Generate(5, func(ii int) int {
+ return ii
+ })
+
+ slices.Reverse(original)
+
+ list := Generate(5, func(ii int) int {
+ return ii
+ })
+
+ for _, want := range original {
+ length := len(list)
+
+ tail, _ := PopBackSafe(&list)
+
+ if len(list) == length {
+ t.Fatalf("failed to modify slice")
+ }
+
+ if tail != want {
+ t.Fatalf("popped wrong value: got=%d, want=%d", tail, want)
+ }
+ }
+
+ v, ok := PopBackSafe(&list)
+ if v != 0 || ok == true {
+ t.Fatalf("unexpected value from popping empty list: %v", v)
+ }
+ })
+}
+
+func TestFilterMap(t *testing.T) {
+ input := []string{
+ "0",
+ "nan",
+ "1",
+ "nan",
+ "2",
+ "nan",
+ "3",
+ "nan",
+ "4",
+ "nan",
+ "5",
+ "nan",
+ "6",
+ "nan",
+ "7",
+ "nan",
+ "8",
+ "nan",
+ "9",
+ }
+
+ got := FilterMap(input, func(v string) (int, bool) {
+ n, err := strconv.Atoi(v)
+ return n, err == nil
+ })
+
+ want := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+
+ if !slices.Equal(got, want) {
+ t.Fatalf("got=%v, want=%v", got, want)
+ }
+}
+
+func TestTakeAllFunc(t *testing.T) {
+ input := Generate(10, func(n int) int { return n })
+
+ evens, ok := TakeAllFunc(&input, func(n int) bool {
+ return n%2 == 0
+ })
+
+ if !ok {
+ t.Fatalf("expected ok, got not ok")
+ }
+
+ wantEvens := []int{0, 2, 4, 6, 8}
+
+ if !slices.Equal(evens, wantEvens) {
+ t.Fatalf("got=%v, want=%v", evens, wantEvens)
+ }
+
+ wantInput := []int{1, 3, 5, 7, 9}
+
+ if !slices.Equal(input, wantInput) {
+ t.Fatalf("got=%v, want=%v", input, wantInput)
+ }
+}
+
+func TestTakeFunc(t *testing.T) {
+ input := Generate(10, func(n int) int { return n })
+
+ got, ok := TakeFunc(&input, func(n int) bool {
+ return n == 5
+ })
+
+ if !ok {
+ t.Fatalf("expected ok, got not ok")
+ }
+
+ want := 5
+
+ if want != got {
+ t.Fatalf("got=%v, want=%v", got, want)
+ }
+
+ if slices.Contains(input, 5) {
+ t.Fatalf("failed to remove %d from list: %v", want, input)
+ }
+}
diff --git a/syncx/mutex.go b/syncx/mutex.go
@@ -0,0 +1,47 @@
+package syncx
+
+import "sync"
+
+// Mutex protects a value with a mutex.
+type Mutex[T any] struct {
+ lock sync.Mutex
+ value T
+}
+
+func NewMutex[T any](v T) *Mutex[T] {
+ return &Mutex[T]{value: v}
+}
+
+func (m *Mutex[T]) With(fn func(t T)) {
+ m.lock.Lock()
+ defer m.lock.Unlock()
+ fn(m.value)
+}
+
+func (m *Mutex[T]) Update(fn func(t T) T) {
+ m.lock.Lock()
+ defer m.lock.Unlock()
+ m.value = fn(m.value)
+}
+
+func (m *Mutex[T]) Set(value T) {
+ m.lock.Lock()
+ defer m.lock.Unlock()
+ m.value = value
+}
+
+func (m *Mutex[T]) Take() T {
+ m.lock.Lock()
+ defer m.lock.Unlock()
+ v := m.value
+ m.value = *new(T)
+ return v
+}
+
+func (m *Mutex[T]) Swap(t T) T {
+ m.lock.Lock()
+ defer m.lock.Unlock()
+ v := m.value
+ m.value = t
+ return v
+}
diff --git a/syncx/pool.go b/syncx/pool.go
@@ -0,0 +1,33 @@
+package syncx
+
+import "sync"
+
+// Pool wraps a [sync.Pool] providing a type-safe API.
+// [validate] is used to optionally ignore [Put] operations.
+type Pool[T any] struct {
+ sync.Pool
+ allocate func() T
+ validate func(t T) bool
+}
+
+func NewPool[T any](allocate func() T, validate func(T) bool) *Pool[T] {
+ return &Pool[T]{
+ allocate: allocate,
+ validate: validate,
+ Pool: sync.Pool{
+ New: func() any {
+ return allocate()
+ },
+ },
+ }
+}
+
+func (p *Pool[T]) Get() T {
+ return p.Pool.Get().(T)
+}
+
+func (p *Pool[T]) Put(t T) {
+ if p.validate(t) {
+ p.Pool.Put(t)
+ }
+}
diff --git a/syncx/slice.go b/syncx/slice.go
@@ -0,0 +1,59 @@
+package syncx
+
+import (
+ "iter"
+ "slices"
+)
+
+// Slice is a mutex protected slice.
+type Slice[T any] struct {
+ Value Mutex[[]T]
+}
+
+func (s *Slice[T]) Append(v T) {
+ s.Value.Update(func(list []T) []T {
+ return append(list, v)
+ })
+}
+
+func (s *Slice[T]) Prepend(v T) {
+ s.Value.Update(func(list []T) []T {
+ return append([]T{v}, list...)
+ })
+}
+
+func (s *Slice[T]) Delete(start, end int) {
+ s.Value.Update(func(list []T) []T {
+ return slices.Delete(list, start, end)
+ })
+}
+
+func (s *Slice[T]) Contains(fn func(T) bool) bool {
+ return s.Index(fn) > -1
+}
+
+func (s *Slice[T]) Index(fn func(T) bool) int {
+ _, ii := s.Find(fn)
+ return ii
+}
+
+func (s *Slice[T]) Find(fn func(T) bool) (T, int) {
+ for ii, v := range s.Iter() {
+ if fn(v) {
+ return v, ii
+ }
+ }
+ return *new(T), -1
+}
+
+func (s *Slice[T]) Iter() iter.Seq2[int, T] {
+ return func(yield func(int, T) bool) {
+ s.Value.With(func(list []T) {
+ for ii, v := range list {
+ if !yield(ii, v) {
+ break
+ }
+ }
+ })
+ }
+}
