betterthreads

package module
v0.0.0-...-0eaf207 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 17, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

README

betterthreads

A fast, lock-free, elastic goroutine pool for Go with an ergonomic API.

betterthreads is built to beat the popular Go pools (ants, flock) and hand-rolled patterns (channel semaphores, errgroup) on the workloads pools actually exist for — high-contention submission, tiny-task overhead, CPU-bound work, and memory footprint — while staying race-clean and simple to use.

// The whole point, in one line:
betterthreads.Go(func() { doWork() })   // like `go`, but pooled, bounded, and reusable

Install

go get github.com/guno1928/betterthreads

Requires Go 1.21+ (uses min/max builtins and typed atomics). Tested on Go 1.26.

Quick start

Fire-and-forget on the global pool
betterthreads.Go(func() { fmt.Println("hello") })
A dedicated, bounded pool
p := betterthreads.New(betterthreads.WithMaxWorkers(1024))
defer p.Release()

p.Go(func() { handle(req) })
Typed futures (generics)
future := betterthreads.Submit(p, func() (int, error) {
    return 6 * 7, nil
})
answer, err := future.Await() // 42, nil
Parallel map (order-preserving, bounded concurrency)
urls := []string{"a", "b", "c"}
bodies := betterthreads.Map(p, urls, func(u string) string {
    return fetch(u)
})
An errgroup-style group
g := p.Group()
for _, job := range jobs {
    g.Go(func() error { return process(job) })
}
if err := g.Wait(); err != nil {  // returns the first error
    log.Fatal(err)
}

API

Function Description
Go(task func()) Run task on the global default pool (fire-and-forget).
Async[T](fn func() (T, error)) *Future[T] Submit to the default pool, get a typed future.
Default() *Pool The global pool.
New(opts ...Option) *Pool Create a pool.
(*Pool) Go(task func()) Submit; blocks with backpressure only if at capacity.
(*Pool) Submit(task func()) error Like Go, returns ErrPoolReleased if released.
(*Pool) TryGo(task func()) bool Non-blocking submit; returns false if saturated.
(*Pool) Release() Stop the pool, waiting for in-flight tasks.
(*Pool) Running() int / (*Pool) Cap() int Live worker count / max workers (-1 = unbounded).
Submit[T](p *Pool, fn func() (T, error)) *Future[T] Typed future on p.
(*Future[T]) Await() (T, error) / Done() <-chan struct{} Block for / select on the result.
Map[In, Out](p *Pool, items []In, fn func(In) Out) []Out Parallel, order-preserving map.
(*Pool) Group() *Group An errgroup-style group bound to p.
(*Group) Go(fn func() error) / Wait() error Run; wait and get the first error.
Options
Option Default Meaning
WithMinWorkers(n) 2×GOMAXPROCS Warm floor: pre-spawned at creation, never retired below this.
WithMaxWorkers(n) unbounded Hard cap on concurrent workers.
Unlimited() — Grow workers without bound (elastic).
WithGrowBatch(n) 8192 How many workers to spawn per growth step when saturated.
WithShrinkAfter(d) 5s Retire excess workers back down to the warm floor after d of low use.
WithFastSteal() off Trade single-task latency for max dispatch throughput (steal 2 victims, not all).
WithRingSize(n) 1024 Per-shard task-queue capacity (rounded up to a power of two).
WithSpinCount(n) 30 How long idle workers spin before parking (latency vs. CPU).
WithPanicHandler(fn) swallow Called with the recovered value when a task panics.

Panics in tasks never crash the pool; workers always recover.

Warm pool for event-loop dispatch

A pool that stays warm and only ever pulls a ready worker matches or beats go func() — because go pays goroutine creation (newproc) on every call, while pulling from a warm pool is a lock-free ring push a hot worker grabs. The gap is largest with multiple concurrent dispatchers (per-core event loops), where sharded submit removes all contention. Configure a large warm floor that grows in batches under overload and shrinks back when idle — ideal in front of a custom epoll / event loop:

pool := betterthreads.New(
    betterthreads.WithMinWorkers(80_000),        // 80k always-warm workers
    betterthreads.WithGrowBatch(128),            // add 128 at a time when all are busy
    betterthreads.WithMaxWorkers(200_000),       // hard ceiling (or Unlimited())
    betterthreads.WithShrinkAfter(30*time.Second), // return to 80k after 30s of low use
    betterthreads.WithFastSteal(),               // max dispatch throughput
)
// hot path: pull a worker and hand it the request, fire-and-forget
pool.Go(func() { handle(conn) })

Measured dispatch cost — pulling from a pre-warmed 20k-worker pool vs. go func() (Ryzen 7 5700X, Go 1.26.2):

Dispatch pattern Raw go func() ants flock betterthreads
Single producer (one loop) 181 ns 494 ns 171 ns 178 ns (≈ raw / flock)
Parallel producers (sharded loops) 289 ns 341 ns — 72 ns (4× faster than raw); ~50 ns with WithFastSteal()

So for the fire-and-forget dispatch a per-core epoll setup does, pulling from the warm pool is ~4× faster than go func() — because your event loop pushes work and moves on; it never blocks per task (that's the one pattern where raw wins — see the caveat below).

Design

The engine is a per-P sharded, lock-free, elastic work-stealing pool:

  • Contention-free submit. The submitting goroutine is mapped to a shard by its current processor (runtime.procPin via //go:linkname), so concurrent producers hit different shards and rarely contend. This is why betterthreads scales on multi-producer submission where a single-queue pool serializes.
  • Lock-free queues. Each shard is a bounded Vyukov MPMC ring buffer (sync/atomic, cache-line padded, zero-alloc enqueue — the task closure is stored directly).
  • Work stealing. A worker drains its home shard, then steals from other shards (fast per-call RNG from turbo). WithFastSteal() probes only two random victims instead of all shards — higher dispatch throughput at the cost of single-task tail latency.
  • Adaptive spinning + semaphore parking. After finishing work a worker spins briefly (PAUSE → Gosched) so streaming tasks are caught in nanoseconds without a park/wake round-trip; the number of concurrent spinners is bounded (GOMAXPROCS/2) so they never starve producers. One worker stays "hot" during warm periods for low latency. Idle workers park on the runtime semaphore (sync.runtime_Semacquire via //go:linkname) — the same primitive the scheduler uses, and dramatically cheaper than a channel for waking thousands of workers at once (this is what makes IO-heavy and high-worker-count workloads beat raw).
  • Warm floor + elastic growth + shrink. WithMinWorkers(n) pre-spawns a warm floor that is never retired below. A background grower samples total queue backlog on a ~1ms tick and spawns workers in batches (WithGrowBatch) only when backlog is sustained across ticks — which distinguishes IO-bound tasks (workers block, backlog persists) from tiny tasks (workers keep up, backlog drains within a tick), so it never over-spawns for fast workloads. After WithShrinkAfter of low use, excess workers retire back down to the warm floor.

Everything on the hot path is sync/atomic — no mutexes. Every design choice was kept only if a benchmark proved it faster (a 15-experiment parameter sweep); see below.

Benchmarks

Hardware: AMD Ryzen 7 5700X (8C/16T), Go 1.26.2, windows/amd64. Contenders: raw goroutines (go + WaitGroup), channel semaphore, errgroup, ants v2.12.1, flock v0.1.0, and betterthreads (default config). Sixteen benchmark files cover four task profiles (no-op, CPU-light, CPU-heavy, IO-bound 10ms sleep), plus contention, memory, scaling, latency, overload, and warm-pool dispatch. Reproduce with:

go test -bench=. -benchmem ./benchmarks/

Throughput — million tasks/sec (higher is better). Winner in bold. Default config.

Benchmark Raw goroutines ChannelSem errgroup ants flock betterthreads
No-op latency (200k, wait all) 5.33 3.66 3.17 3.21 4.32 5.50
No-op throughput (1 producer) 5.17 3.57 3.14 3.43 5.59 6.20
IO-bound latency (1 producer) 1.80 1.44 1.37 1.46 0.83 1.86
CPU-light 5.17 3.26 3.08 2.53 7.12 6.66
CPU-heavy 0.498 0.491 0.480 0.454 0.440 0.495
Bursty (50×2k bursts) 1.87 1.47 1.26 1.29 1.67 1.66
Task-count scale n=1M 4.96 3.24 2.86 3.18 4.75 7.50
Memory bench (100k no-op) 5.37 3.76 3.32 3.32 4.66 5.66

Per-op cost — ns/op (lower is better).

Benchmark Raw goroutines ChannelSem errgroup ants flock betterthreads
IO-bound throughput (parallel) 569 511 503 461 1106 230
Parallel submit (16 producers) 274 344 355 303 91.6 87.1
High concurrency 299 367 331 318 97.6 80.8
Submit latency (single task) 786 960 1100 875 1,591,436 2,163

Memory / footprint.

Metric Raw goroutines ants flock betterthreads
Goroutines @50k tasks in flight (cap 1000) 19,555 1,041 1,038 1,040
allocs/op (100k no-op) 100,001 100,099 200,001 100,001

Scoreboard (default config): betterthreads beats both pool competitors (ants, flock) on nearly every benchmark and beats raw goroutines on most — including IO-latency (1.86M, now ahead of raw thanks to the semaphore park), IO-throughput (2× ants), high-concurrency, memory, and every task-count size. flock edges it on CPU-light and raw on bursty/mixed by small margins in the default config; WithFastSteal() flips those (CPU-light 6.7 → 11.8M/s, parallel-submit 87 → 47 ns, high-conc 81 → 50 ns) at the cost of single-task latency. flock also suffers pathological blowups betterthreads never does (593 MB / 74M allocs at cap=100; 1.6 ms single-task latency). The lone loss everywhere is single-task ping-pong latency — see below.

The one honest caveat

There is exactly one axis where betterthreads (and every pool, including ants) cannot beat raw goroutines: single-task ping-pong latency — submitting one task and then blocking until it finishes, repeatedly. That measures goroutine create-and-handoff, which go func() does optimally via runnext (the new goroutine runs on the same core the creator is about to yield). A pool's worker lives on a different core, so the wakeup back to the blocked caller is cross-core. No unsafe or assembly changes this — it's a runtime privilege.

This pattern does not occur in fire-and-forget dispatch (an event loop pushes work and moves on; it never blocks per task) — which is why betterthreads beats raw on warm-pool dispatch (above). Use it for high-throughput, many-producer, CPU-bound, memory-sensitive, or bounded-concurrency workloads. Use bare go only when you truly want one goroutine per task and never wait on it — and even then, a warm pool is faster to dispatch from.

Correctness

The lock-free engine is verified under the race detector (go test -race): 32-way concurrent producers, panic recovery, graceful release draining in-flight work, ordered Map, futures, groups, warm-floor pre-spawn, and grow-then-shrink-to-min behavior.

License

MIT.

Documentation

Index

Examples

Constants

This section is empty.

Variables

View Source 
var ErrPoolReleased = errors.New("betterthreads: pool released")

Functions

func Go 

func Go(task func())
Example 
package main

import (
	"fmt"
	"sync"

	"github.com/guno1928/betterthreads"
)

func main() {
	var wg sync.WaitGroup
	wg.Add(1)
	betterthreads.Go(func() {
		fmt.Println("ran on the default pool")
		wg.Done()
	})
	wg.Wait()
}

Output:
ran on the default pool

func Map 

func Map[In, Out any](p *Pool, items []In, fn func(In) Out) []Out
Example 
package main

import (
	"fmt"

	"github.com/guno1928/betterthreads"
)

func main() {
	p := betterthreads.New()
	defer p.Release()

	nums := []int{1, 2, 3, 4, 5}
	squares := betterthreads.Map(p, nums, func(n int) int { return n * n })
	fmt.Println(squares)
}

Output:
[1 4 9 16 25]

Types

type Future 

type Future[T any] struct {
	// contains filtered or unexported fields
}

func Async 

func Async[T any](fn func() (T, error)) *Future[T]

func Submit 

func Submit[T any](p *Pool, fn func() (T, error)) *Future[T]
Example 
package main

import (
	"fmt"

	"github.com/guno1928/betterthreads"
)

func main() {
	p := betterthreads.New()
	defer p.Release()

	future := betterthreads.Submit(p, func() (int, error) {
		return 6 * 7, nil
	})

	answer, err := future.Await()
	fmt.Println(answer, err)
}

Output:
42 <nil>

func (*Future[T]) Await 

func (f *Future[T]) Await() (T, error)

func (*Future[T]) Done 

func (f *Future[T]) Done() <-chan struct{}

type Group 

type Group struct {
	// contains filtered or unexported fields
}
Example (FirstError) 
package main

import (
	"errors"
	"fmt"

	"github.com/guno1928/betterthreads"
)

func main() {
	p := betterthreads.New()
	defer p.Release()

	g := p.Group()
	g.Go(func() error { return nil })
	g.Go(func() error { return errors.New("task failed") })
	fmt.Println(g.Wait())
}

Output:
task failed

func (*Group) Go 

func (g *Group) Go(fn func() error)

func (*Group) Wait 

func (g *Group) Wait() error

type Option 

type Option func(*config)

func Unlimited 

func Unlimited() Option

func WithFastSteal 

func WithFastSteal() Option

func WithGrowBatch 

func WithGrowBatch(n int) Option

func WithMaxWorkers 

func WithMaxWorkers(n int) Option

func WithMinWorkers 

func WithMinWorkers(n int) Option
Example 
package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/guno1928/betterthreads"
)

func main() {
	pool := betterthreads.New(
		betterthreads.WithMinWorkers(64),
		betterthreads.WithGrowBatch(128),
		betterthreads.WithShrinkAfter(30*time.Second),
		betterthreads.WithFastSteal(),
	)
	defer pool.Release()

	var wg sync.WaitGroup
	wg.Add(1)
	pool.Go(func() {
		fmt.Println("handled on a warm worker")
		wg.Done()
	})
	wg.Wait()
}

Output:
handled on a warm worker

func WithPanicHandler 

func WithPanicHandler(h func(any)) Option

func WithRingSize 

func WithRingSize(n int) Option

func WithShrinkAfter 

func WithShrinkAfter(d time.Duration) Option

func WithSpinCount 

func WithSpinCount(n int) Option

type Pool 

type Pool struct {
	// contains filtered or unexported fields
}
Example 
package main

import (
	"fmt"
	"sync"

	"github.com/guno1928/betterthreads"
)

func main() {
	p := betterthreads.New(betterthreads.WithMaxWorkers(8))
	defer p.Release()

	var wg sync.WaitGroup
	wg.Add(1)
	p.Go(func() {
		fmt.Println("hello from a pooled worker")
		wg.Done()
	})
	wg.Wait()
}

Output:
hello from a pooled worker

func Default 

func Default() *Pool

func New 

func New(opts ...Option) *Pool

func (*Pool) Cap 

func (p *Pool) Cap() int

func (*Pool) Go 

func (p *Pool) Go(task func())

func (*Pool) Group 

func (p *Pool) Group() *Group
Example 
package main

import (
	"fmt"
	"sync"

	"github.com/guno1928/betterthreads"
)

func main() {
	p := betterthreads.New()
	defer p.Release()

	g := p.Group()
	var mu sync.Mutex
	sum := 0
	for i := 1; i <= 100; i++ {
		g.Go(func() error {
			mu.Lock()
			sum += i
			mu.Unlock()
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		fmt.Println("error:", err)
	}
	fmt.Println("sum:", sum)
}

Output:
sum: 5050

func (*Pool) Release 

func (p *Pool) Release()

func (*Pool) Running 

func (p *Pool) Running() int

func (*Pool) Submit 

func (p *Pool) Submit(task func()) error

func (*Pool) TryGo 

func (p *Pool) TryGo(task func()) bool

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.