sync

🔄 go-sync

A small Go library (package sync) with focused concurrency helpers:

• Convenience aliases for common sync primitives and typed atomics
• Hook-driven execution (Wait, Timeout, Worker)
• Group helpers (ErrorGroup, ErrorsGroup, SingleFlightGroup)
• Typed wrappers for sync.Pool, sync.Map, and atomic.Value
• A bytes.Buffer pool specialized for copy-and-reuse workflows

📦 Install

Requires Go 1.26 or newer.

go get github.com/alexfalkowski/go-sync

🧪 Local validation

Local Make targets use shared tooling from the bin submodule. After cloning the repository for development, initialize it before running Make targets:

git submodule sync && git submodule update --init
make dep

Then run the narrow target you need, such as make specs.

Benchmarks

make benchmarks
make benchtime=1s benchmarks
make benchmark

Fuzzes

make fuzzes
make map-fuzz
make value-fuzz
make pool-fuzz
make group-fuzz
make worker-fuzz
make package=. name=FuzzMapStringIntOperations fuzztime=1000x fuzz

🧭 Package layout

The public API is intentionally small:

• Aliases: Once, Mutex, RWMutex, WaitGroup, Int32, Int64, Uint32, Uint64, Uintptr, Bool, Pointer[T]
• Hooks and timeout helpers: Hook, Handler, ErrorHandler, ErrNoOnRunProvided, ErrTimeout, Wait, Timeout, IsTimeoutError
• Worker: ErrWorkerFull, NewWorker, Worker.Schedule, Worker.TrySchedule, Worker.Wait
• Future: Async, Future[T], Future.Await
• Groups: ErrorGroup, ErrorsGroup, NewSingleFlightGroup, SingleFlightGroup, AnySingleFlightGroup, SingleFlightResult, AnySingleFlightResult
• Pools and wrappers: AnyPool, NewPool, Pool[T], NewBufferPool, BufferPool, NewValue, Value[T], AnyValue, NewMap, Map[K, V], AnyMap

Most wrappers preserve the semantics of the standard library type they wrap while making those semantics easier to use from generic code.

🔁 Aliases

The package re-exports a few commonly used concurrency primitives and helper types for convenience:

• Once, Mutex, RWMutex, and WaitGroup alias their counterparts in sync.
• Int32, Int64, Uint32, Uint64, Uintptr, Bool, and Pointer[T] alias typed atomics from sync/atomic.
• ErrorGroup aliases errgroup.Group.
• AnyPool aliases the non-generic sync.Pool; use Pool[T] for a typed pool.
• AnyMap and AnyValue alias the non-generic sync.Map and atomic.Value; use Map[K, V] and Value[T] for typed wrappers.
• AnySingleFlightGroup and AnySingleFlightResult alias singleflight.Group and singleflight.Result; use SingleFlightGroup[T] and SingleFlightResult[T] for typed variants.

These are type aliases rather than wrappers, so their behavior is exactly the same as the underlying type.

🪝 Hooks

Most execution helpers accept a sync.Hook:

• OnRun(context.Context) error is required.
• OnError(context.Context, error) error is optional.
• If OnRun is nil, helpers return sync.ErrNoOnRunProvided.
• Helpers validate OnRun before context cancellation or timeout shortcuts, so a nil OnRun returns sync.ErrNoOnRunProvided even if the context is already canceled or timeout <= 0.
• Hook callbacks must not panic; helpers do not recover panics, pass them to OnError, or return them as errors.

OnError is only called when OnRun returns a non-nil error. If OnError returns a different error, that new error is returned.

How that returned error is observed depends on the helper:

• Wait and Timeout return it only if OnRun finishes before their timeout/cancellation path wins.
• Worker never returns handler errors from Schedule or TrySchedule; use OnError for logging or side effects.

package main

import (
    "context"
    "errors"
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    runErr := errors.New("boom")
    hook := sync.Hook{
        OnRun: func(context.Context) error {
            return runErr
        },
        OnError: func(_ context.Context, err error) error {
            return fmt.Errorf("wrapped: %w", err)
        },
    }

    err := hook.Error(context.Background(), hook.OnRun(context.Background()))
    fmt.Println(errors.Is(err, runErr))
}

⏱️ Wait vs Timeout

Wait and Timeout both run Hook.OnRun, but they differ:

• Wait: best-effort wait up to timeout; returns nil on timeout/cancel and does not cancel OnRun.
• Timeout: derives a timeout context for OnRun; returns the context cause when the context ends first (sync.ErrTimeout, context.Canceled, or a parent-provided cause).
• If the input context is already done, Wait returns nil immediately and Timeout returns the input context's cause immediately (neither invokes OnRun).
• If timeout <= 0, Wait returns nil immediately, while Timeout returns sync.ErrTimeout immediately.

[!IMPORTANT] Neither Wait nor Timeout forcibly stops OnRun. If the handler ignores context cancellation, it can continue running after the helper returns.

Use sync.IsTimeoutError(err) to check if an error matches sync.ErrTimeout or context.DeadlineExceeded.

⏳ Wait example (best effort)

package main

import (
    "context"
    "errors"
    "fmt"
    "time"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    err := sync.Wait(context.Background(), 10*time.Millisecond, sync.Hook{
        OnRun: func(context.Context) error {
            time.Sleep(time.Second)
            return errors.New("finished too late")
        },
    })

    // true: Wait timed out first.
    fmt.Println(err == nil)
}

🚦 Timeout example (propagated cancellation)

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    err := sync.Timeout(context.Background(), 10*time.Millisecond, sync.Hook{
        OnRun: func(ctx context.Context) error {
            <-ctx.Done()
            return context.Cause(ctx)
        },
    })

    fmt.Println(sync.IsTimeoutError(err))
}

👷 Worker

Worker schedules asynchronous handlers with bounded concurrency.

• Zero value is not ready; use NewWorker(count).
• NewWorker(count) returns a ready-to-use pointer to a worker with at most count in-flight handlers.
• Do not copy a Worker after first use; pass and store *Worker values.
• Schedule is context-only: it blocks until a slot is acquired or ctx is done, and does not derive or bound any deadline itself.
• TrySchedule attempts to acquire a slot immediately and returns sync.ErrWorkerFull if capacity is unavailable.
• The context passed to Schedule is also the context passed to OnRun; to bound only the wait for a slot, pass a ctx with a deadline to Schedule. To give the handler its own run budget starting when it actually begins, wrap ctx with context.WithTimeout (or similar) inside OnRun.
• Schedule and TrySchedule return only scheduling errors.
• Schedule reports context.Cause(ctx) or ErrNoOnRunProvided; TrySchedule reports the input context cause, ErrWorkerFull, or ErrNoOnRunProvided.
• Once a handler has been scheduled, scheduling returns nil even if that handler later observes ctx.Done().
• Wait(ctx) waits for all successfully scheduled handlers to complete, returning nil, or returns context.Cause(ctx) if ctx is done first; it does not cancel running handlers.
• If the input context is already canceled, Schedule and TrySchedule return the input context's cause immediately and do not schedule OnRun.

[!NOTE] Worker scheduling methods report scheduling errors only. Handler errors are routed through Hook.OnError and are not returned by Schedule or TrySchedule.

If count == 0, Schedule always blocks until ctx is done and TrySchedule returns sync.ErrWorkerFull immediately.

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    worker := sync.NewWorker(4)
    defer func() {
        if err := worker.Wait(context.Background()); err != nil {
            log.Printf("wait failed: %v", err)
        }
    }()

    for i := 0; i < 3; i++ {
        job := i
        err := worker.Schedule(context.Background(), sync.Hook{
            OnRun: func(context.Context) error {
                fmt.Println("job", job)
                return nil
            },
            OnError: func(_ context.Context, err error) error {
                log.Printf("job failed: %v", err)
                return err
            },
        })
        if err != nil {
            log.Printf("schedule failed: %v", err)
        }
    }
}

⏳ Future

Async starts a typed operation immediately and returns a Future[T] for its eventual result.

• Async runs the operation in a new goroutine.
• The work context passed to Async controls the operation.
• Async invokes the operation even when its work context is already canceled; the operation is responsible for observing ctx.Done().
• Future.Await controls only how long the caller waits; canceling its context does not cancel the operation.
• The value and error are cached, so Await can be called repeatedly by one or more callers after completion.
• If await cancellation is selected, Await checks completion once more; a result published by that check wins, otherwise the await context's cause is returned.
• Operations must not panic; Future does not recover panics.

package main

import (
    "context"
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    future := sync.Async(context.Background(), func(context.Context) (int, error) {
        return 42, nil
    })

    value, err := future.Await(context.Background())
    fmt.Println(value, err == nil)
}

👥 Group

🧩 ErrorGroup / ErrorsGroup / WaitGroup

sync.ErrorGroup is a type alias for errgroup.Group. sync.ErrorsGroup waits for every scheduled function and returns all non-nil errors joined with errors.Join. sync.WaitGroup is a type alias for sync.WaitGroup.

package main

import (
    "errors"
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    var g sync.ErrorGroup

    g.Go(func() error { return nil })
    g.Go(func() error { return errors.New("boom") })

    fmt.Println(g.Wait() != nil)
}

Use ErrorsGroup when callers need every error rather than only the first one:

ErrorsGroup retains recorded errors for its lifetime. Use a fresh ErrorsGroup for each independent batch of work. Functions passed to ErrorsGroup.Go must not panic; panics are not joined into the error returned by Wait. Call SetLimit(n) to bound how many functions run concurrently; a negative n, and the zero value, mean unbounded. SetLimit must not be called while any function started by Go is still running. Use TryGo for non-blocking, best-effort submission: it starts a function only if a concurrency slot is currently free, returning false without starting it otherwise, while still joining every error into a later Wait. Start the first function before calling Wait for an empty group, and wait for a batch to finish before starting the next independent batch. Do not copy an ErrorsGroup after first use.

package main

import (
    "errors"
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    var g sync.ErrorsGroup

    first := errors.New("first")
    second := errors.New("second")

    g.Go(func() error { return first })
    g.Go(func() error { return second })

    err := g.Wait()
    fmt.Println(errors.Is(err, first), errors.Is(err, second))
}

✈️ SingleFlightGroup

SingleFlightGroup[T] deduplicates concurrent work by key.

• Zero value is ready for use; NewSingleFlightGroup[T]() is optional and returns a ready-to-use pointer.
• Do(key, fn) returns (value, err, shared).
• DoChan(key, fn) returns a receive-only channel of SingleFlightResult[T].
• shared == true means the result was given to multiple callers.
• On fn error, Do returns zero T plus the error, and DoChan sends a result with zero T plus the error.
• If T is an interface type and fn returns a nil interface value, Do and DoChan expose it as zero T.
• DoChan follows singleflight.Group.DoChan: the returned channel receives one buffered result and is not closed.
• Completed results are not cached; Forget only affects a call that is still in flight.
• Do not copy a SingleFlightGroup[T] after first use.

package main

import (
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    var g sync.SingleFlightGroup[int]

    v, err, shared := g.Do("key", func() (int, error) {
        return 42, nil
    })
    fmt.Println(v, err == nil, shared)
}

🏊 Pool

🧺 Generic Pool

Pool[T] is a typed wrapper around sync.Pool.

• Stores *T values.
• Zero value is ready for use.
• NewPool[T]() returns a ready-to-use pointer with the default new(T) constructor.
• Set New func() *T when values need custom initialization.
• When New is nil, Get allocates new(T) when the pool is empty.
• Follows normal sync.Pool semantics (runtime may drop entries anytime).
• Does not reset values automatically on Put; callers are responsible for reuse hygiene.
• Put(nil) is a no-op.
• Do not copy a Pool[T] after first use.

package main

import (
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    type item struct {
        ID int
    }

    pool := sync.Pool[item]{
        New: func() *item {
            return &item{ID: 10}
        },
    }
    it := pool.Get()
    pool.Put(it)

    fmt.Println("ok")
}

🧽 BufferPool

BufferPool is a convenience wrapper over Pool[bytes.Buffer].

• Zero value is not ready; use NewBufferPool() which returns a ready-to-use pointer.
• Get returns *bytes.Buffer.
• Get returns an empty buffer.
• Put resets the buffer (nil-safe no-op).
• Copy returns a cloned []byte (non-aliasing, nil-safe).

package main

import (
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    bp := sync.NewBufferPool()
    buf := bp.Get()
    defer bp.Put(buf)

    buf.WriteString("hello")
    out := bp.Copy(buf)
    fmt.Println(string(out))
}

⚛️ Value

Value[T] is a typed wrapper around atomic.Value.

• Zero value is ready (NewValue is optional and returns a ready-to-use pointer).
• Do not copy a Value[T] after first use.
• Load and Swap return zero T if unset.
• Same underlying constraints as atomic.Value apply.
• If T is an interface type, storing or swapping a nil interface value panics just like atomic.Value.Store(nil) or atomic.Value.Swap(nil).
• When T is an interface or any, stored and swapped values must still be consistent with atomic.Value's concrete-type rules.
• CompareAndSwap follows atomic.Value.CompareAndSwap; non-comparable dynamic values in old can panic, and a nil interface value for new panics.
• On an unset Value, CompareAndSwap can initialize only when old is a nil interface value; comparing against zero T does not generally initialize.

package main

import (
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    var v sync.Value[int]
    fmt.Println(v.Load())

    v.Store(1)
    fmt.Println(v.Swap(2))
    fmt.Println(v.CompareAndSwap(2, 3))
}

🗺️ Map

Map[K, V] is a typed wrapper around sync.Map.

• Zero value is ready (NewMap is optional and returns a ready-to-use pointer).
• Do not copy a Map[K, V] after first use.
• Load, LoadOrStore, LoadAndDelete, and Swap return zero V when needed; use boolean flags to distinguish missing keys.
• If K is an interface type and a nil interface key is stored, Range exposes it as zero K (for example, nil for interface K).
• If V is an interface type and a nil interface value is stored, value-returning methods expose it as zero V (for example, nil for interface V).
• Range follows sync.Map.Range semantics and does not provide a consistent snapshot during concurrent mutation.
• Clear removes all entries.
• CompareAndSwap / CompareAndDelete follow sync.Map comparability rules; non-comparable dynamic old values can panic.

package main

import (
    "fmt"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    m := sync.NewMap[string, int]()

    m.Store("one", 1)
    v, ok := m.Load("one")
    fmt.Println(v, ok)

    prev, loaded := m.LoadOrStore("one", 99)
    fmt.Println(prev, loaded)

    m.Range(func(k string, v int) bool {
        fmt.Println(k, v)
        return true
    })
}

Nil interface edge-case example:

package main

import (
    "fmt"
    "io"

    "github.com/alexfalkowski/go-sync"
)

func main() {
    var m sync.Map[string, io.Reader]
    var r io.Reader

    m.Store("reader", r)
    m.Range(func(_ string, value io.Reader) bool {
        fmt.Println(value == nil)
        return true
    })
}

🔎 Background / References

This library draws inspiration from:

• https://go.dev/blog/pipelines
• https://go.dev/talks/2012/concurrency.slide
• https://gobyexample.com/timeouts
• https://go.dev/wiki/Timeouts
• https://github.com/lotusirous/go-concurrency-patterns

For executable, CI-verified usage examples, see example_test.go. Those examples back the rendered package documentation on pkg.go.dev.