tunny — Senior Level¶

Table of Contents¶

1. Introduction
2. Prerequisites
3. Reading the Source — Why and How
4. Top-Down Tour of the Code
5. The Pool Struct
6. The workerWrapper Struct
7. The Dispatcher Loop
8. How Process Actually Works
9. How ProcessTimed Adds Deadlines
10. How ProcessCtx Plugs In
11. How BlockUntilReady Slots In
12. How Interrupt Is Wired
13. How Terminate Is Invoked
14. Close — The Two-Phase Shutdown
15. SetSize — Live Resize
16. Payload Channels Explained
17. Comparison With Stdlib Worker-Pool Patterns
18. Comparison With ants Internals
19. Comparison With workerpool Internals
20. Where Allocations Happen
21. Race Conditions Tunny Avoids
22. Race Conditions You Can Introduce
23. Memory Model Considerations
24. Scheduler Interaction
25. Extending tunny in Your Own Code
26. Patterns the Internals Suggest
27. Best Practices Implied by the Source
28. Tricky Points
29. Tricky Questions
30. Cheat Sheet
31. Self-Assessment Checklist
32. Summary
33. Further Reading
34. Related Topics
35. Diagrams

Introduction¶

Focus: "What does tunny actually do under the hood, and when do those internals matter to me?"

You have used Process, ProcessCtx, and the Worker interface in production. Now you want to understand exactly what happens between the moment you call pool.Process(x) and the moment you receive a result. The senior file is where we open the engine.

Tunny is small enough that you can hold the entire implementation in your head. Roughly:

• A Pool struct with two slices (workers and workerWrappers), a count, and a mutex.
• A workerWrapper type — one per worker — that drives its Worker through a state machine in a goroutine.
• Two channels per workerWrapper: a payload-in channel and a result-out channel.
• A dispatcher loop hidden inside each workerWrapper.run method that reads from the inbox, calls BlockUntilReady, calls Process, sends the result, and listens for interrupt/terminate signals.

That is it. There is no scheduler, no priority queue, no balancing logic — just N goroutines, each independently reading from its own channel. The pool dispatches by writing to whichever channel is currently accepting.

This file walks through the source code in detail, points out the design choices, and connects what you see in the source to behavior you observe at the API.

After reading you should:

• Be able to draw the data structures and the dispatcher loop from memory.
• Be able to predict what tunny does in any unusual scenario — concurrent Close, deeply nested ProcessCtx, factory panic.
• Be able to extend tunny safely (custom Worker wrappers, instrumentation hooks).
• Be able to compare tunny's design with ants, workerpool, and a naive hand-rolled channel pool.

Prerequisites¶

• Comfortable with everything in junior.md and middle.md.
• Familiar with Go's channel semantics in detail — select, close, send-on-closed (panics), receive-from-closed (zero value).
• Familiar with sync.Mutex and sync.RWMutex and the difference.
• Familiar with sync.WaitGroup and its Add/Done/Wait semantics.
• Have read the source of sync/atomic at least once.
• Know what a select on a nil channel does (blocks forever).
• Know the Go memory model basics — happens-before relations between channel send/receive and Unlock/Lock.

If any of these are shaky, take a detour. The senior file refers to all of them repeatedly.

Reading the Source¶

Open github.com/Jeffail/tunny/tunny.go and tunny/workerwrapper.go in your editor side by side. The whole library is on the order of 400 lines of Go. You can read it end to end in under an hour.

We will reference its structure throughout. The exact line numbers shift across versions, but the names of types and methods are stable.

Get the source locally:

go get github.com/Jeffail/tunny
go env GOMODCACHE
# then navigate into the version's directory

Or just read it on GitHub. The point is to have it open while reading this file.

Top-Down Tour¶

The library exposes:

• The Pool type (in tunny.go).
• Constructor functions: New, NewFunc, NewCallback.
• The Worker interface.
• Process methods: Process, ProcessTimed, ProcessCtx.
• Pool management methods: Close, GetSize, SetSize, QueueLength.
• Sentinel errors: ErrPoolNotRunning, ErrJobTimedOut.

Internally, it has:

• The workerWrapper type (one per worker).
• A few helper closures for NewFunc/NewCallback adapters.

That is the entire surface and the entire internal surface. There are no hidden subsystems.

The Pool Struct¶

A simplified view of the Pool struct as it appears in tunny:

type Pool struct {
    queuedJobs int64

    ctor      func() Worker
    workers   []*workerWrapper
    reqChan   chan workRequest

    workerMut sync.Mutex
}

Field walkthrough:

• queuedJobs int64 — incremented when a caller starts to wait, decremented when the wait ends. Read by QueueLength. Accessed via sync/atomic.
• ctor — the factory passed to New. Used by SetSize to grow the pool.
• workers — a slice of *workerWrapper, one per slot.
• reqChan — the unbuffered channel that callers send to and workers receive from. The heart of the dispatcher.
• workerMut — guards mutations of workers. Process reads workers indirectly (via reqChan), so the mutex is mostly for SetSize and Close.

Critical observation: there is no per-pool state machine. The pool does not track "running vs stopped" with an enum. Its only state is in the channels — open or closed, with values pending or not. This makes the implementation small and the behaviour, while subtle, derivable from channel semantics.

A subtle consequence: there is no IsRunning() predicate. You cannot ask the pool whether it has been closed. You can only attempt a Process and observe a panic, or attempt a ProcessTimed and observe ErrPoolNotRunning.

The workerWrapper Struct¶

Each worker is wrapped by a workerWrapper. Simplified:

type workerWrapper struct {
    worker        Worker
    interruptChan chan struct{}
    reqChan       chan workRequest
    closeChan     chan struct{}
    closedChan    chan struct{}
}

And:

type workRequest struct {
    jobChan       chan<- interface{}
    retChan       <-chan interface{}
    interruptFunc func()
}

Lots of channels. Each one has a specific job:

• worker — the user's Worker implementation.
• interruptChan — closed by the pool to signal "stop running, you are being killed".
• reqChan — the shared pool-level channel. Multiple workerWrappers receive from this channel. Whichever is ready first gets the request.
• closeChan — used to ask the wrapper to terminate (different from interrupting an in-flight job).
• closedChan — closed by the wrapper when it has fully terminated. The pool waits on this.

The workRequest is the structure that flows from the caller to the worker. It bundles three things:

• jobChan — the channel on which the caller has written the payload.
• retChan — the channel on which the caller is waiting for a result.
• interruptFunc — a callback the worker invokes when interrupted.

This three-channel structure is what makes ProcessCtx and ProcessTimed possible without modifying the worker dispatch.

The Dispatcher Loop¶

The heart of tunny is workerWrapper.run. Its essence:

func (w *workerWrapper) run() {
    defer func() {
        close(w.closedChan)
    }()
    defer w.worker.Terminate()

    for {
        w.worker.BlockUntilReady()

        select {
        case <-w.closeChan:
            return
        case req := <-w.reqChan:
            payload := <-req.jobChan
            result := w.worker.Process(payload)
            req.retChan <- result
        case <-w.interruptChan:
            // interrupted before reading a payload — ignore
        }
    }
}

This is the pseudocode shape. The real version has slightly more careful handling of edge cases, but this captures the structure.

Walk through it:

1. Terminate is deferred. Whenever the loop exits, Terminate runs. Followed by closing closedChan to signal "I have exited."
2. The outer for loop runs forever until something breaks it.
3. Each iteration begins with BlockUntilReady. The worker can block here until it is willing to accept work.
4. After BlockUntilReady returns, a select chooses between three events:
5. The pool wants the worker to terminate (closeChan is closed). The loop exits.
6. A caller has a request (reqChan has a value). The worker reads the payload, processes it, sends the result.
7. The pool sent an interrupt (interruptChan got a value) but the worker was not in Process. This event is silently consumed and the loop continues.

The third case is interesting: interruptChan is the pool's way of poking the worker. If the worker is currently waiting in BlockUntilReady or in the outer select, the interrupt is consumed but does nothing — the worker has no work to interrupt. If the worker is mid-Process when the interrupt arrives, the pool's code path goes a different way: it calls worker.Interrupt() directly, and the worker's Process may observe it. The interruptChan in this select is really for the case where an interrupt arrives just after a worker finished processing but before it gets to a new iteration — a small race window that this branch swallows safely.

How Process Actually Works¶

When you call pool.Process(payload), here is what happens at the line-of-source level:

func (p *Pool) Process(payload interface{}) interface{} {
    atomic.AddInt64(&p.queuedJobs, 1)

    req, open := <-p.reqChan
    if !open {
        panic(ErrPoolNotRunning)
    }

    req.jobChan <- payload
    payload = <-req.retChan

    atomic.AddInt64(&p.queuedJobs, -1)
    return payload
}

Step by step:

1. queuedJobs is incremented. QueueLength reports this counter.
2. The caller receives a workRequest from the pool's reqChan. This unblocks when a worker has written to reqChan saying "I am ready." If the pool is closed, reqChan is closed and the receive returns (zero, false), panicking.
3. The caller writes the payload to the worker's jobChan.
4. The caller waits on the worker's retChan for the result.
5. queuedJobs is decremented.
6. The result is returned.

Note the unusual flow: the caller does not write the payload to reqChan. The caller reads from reqChan. The worker is the one that wrote to reqChan. This inversion is what makes tunny's API synchronous from the caller's view and elegant from the worker's view.

Visually:

Worker (loop):
    BlockUntilReady()
    create workRequest with my jobChan/retChan
    write workRequest to pool.reqChan ─┐
                                       │
                                       ▼
Caller:
    read workRequest from pool.reqChan
    write payload to req.jobChan ─┐
                                  ▼
Worker:
    read payload from req.jobChan
    result := Process(payload)
    write result to req.retChan ─┐
                                 ▼
Caller:
    read result from req.retChan
    return

Three channel operations end-to-end. The worker's jobChan and retChan are created per cycle (or per worker, depending on version) — both buffered channels of size 1 typically.

How ProcessTimed Adds Deadlines¶

ProcessTimed is similar to Process but with a time.Timer watching the whole thing:

func (p *Pool) ProcessTimed(payload interface{}, timeout time.Duration) (interface{}, error) {
    atomic.AddInt64(&p.queuedJobs, 1)
    defer atomic.AddInt64(&p.queuedJobs, -1)

    timer := time.NewTimer(timeout)
    defer timer.Stop()

    var req workRequest
    var open bool
    select {
    case req, open = <-p.reqChan:
        if !open {
            return nil, ErrPoolNotRunning
        }
    case <-timer.C:
        return nil, ErrJobTimedOut
    }

    select {
    case req.jobChan <- payload:
    case <-timer.C:
        req.interruptFunc()
        return nil, ErrJobTimedOut
    }

    select {
    case result := <-req.retChan:
        return result, nil
    case <-timer.C:
        req.interruptFunc()
        return nil, ErrJobTimedOut
    }
}

Three select blocks, one per phase:

1. Queue phase. Either we get a worker or the timer fires. If the timer fires, no worker is consumed.
2. Send phase. Either we hand the payload off or the timer fires. If the timer fires, we invoke interruptFunc to let the worker know.
3. Receive phase. Either we get a result or the timer fires. Same interrupt path.

The interruptFunc is a closure created by the workerWrapper. When called, it calls worker.Interrupt() and signals via interruptChan. The worker may or may not honour the signal.

Important: the timer is a single timer for all three phases. Time spent in phase 1 reduces what is available for phases 2 and 3. So ProcessTimed(x, 1*time.Second) means "give me success or error within 1 second total."

How ProcessCtx Plugs In¶

ProcessCtx is essentially ProcessTimed with a context replacing the timer:

func (p *Pool) ProcessCtx(ctx context.Context, payload interface{}) (interface{}, error) {
    atomic.AddInt64(&p.queuedJobs, 1)
    defer atomic.AddInt64(&p.queuedJobs, -1)

    var req workRequest
    var open bool
    select {
    case req, open = <-p.reqChan:
        if !open {
            return nil, ErrPoolNotRunning
        }
    case <-ctx.Done():
        return nil, ctx.Err()
    }

    select {
    case req.jobChan <- payload:
    case <-ctx.Done():
        req.interruptFunc()
        return nil, ctx.Err()
    }

    select {
    case result := <-req.retChan:
        return result, nil
    case <-ctx.Done():
        req.interruptFunc()
        return nil, ctx.Err()
    }
}

Same structure, different cancellation source. The difference matters:

• ProcessTimed uses a timer that fires once at a fixed time.
• ProcessCtx uses ctx.Done(), which may fire for any reason — deadline, manual cancel, parent cancel.

This is why ProcessCtx is more general. In production code use it; in tests, sometimes ProcessTimed is simpler.

How BlockUntilReady Slots In¶

BlockUntilReady is called inside the worker's loop, before the worker offers itself to the dispatcher. Until BlockUntilReady returns, the worker is not writing to reqChan, so callers cannot reach it.

This is the design's elegant point: the worker is the agent that decides when it is ready. The dispatcher is dumb — it just lets workers compete on reqChan. If a worker is not ready, it does not write. If multiple workers are ready, whichever one's send to reqChan is selected first wins.

There is no "is this worker ready?" query. Readiness is an in-band state expressed by being in or out of reqChan send.

Concretely:

• 4 workers, all in BlockUntilReady (suspended on a rate limiter).
• Caller arrives, does <-reqChan. Blocks. Nobody is sending.
• One worker's BlockUntilReady returns. Worker writes to reqChan. The send succeeds. Caller unblocks.
• The other three workers remain in BlockUntilReady.

This is why a long BlockUntilReady reduces effective pool size dynamically. If you have a slow rate limiter, your pool is "smaller than its nominal size" for periods of time.

How Interrupt Is Wired¶

Two places where Interrupt may be invoked:

1. From ProcessTimed / ProcessCtx when the deadline fires mid-Process. This is req.interruptFunc() in the source above.
2. From Close/SetSize when a worker is being terminated mid-Process. This is internal to the pool.

req.interruptFunc is a closure that does:

func() {
    select {
    case w.interruptChan <- struct{}{}:
    default:
    }
    w.worker.Interrupt()
}

Two things:

• Try to push onto interruptChan so the worker's outer select sees it (in case the worker has just returned from Process but not yet read a new request).
• Call worker.Interrupt() directly. This is the synchronous notification to the user's code.

Note Interrupt is called from the caller's goroutine, not the worker's. The worker is still in Process. The user's code must therefore be safe for concurrent access between Process (worker goroutine) and Interrupt (caller goroutine). Hence the mutexes in middle-level patterns.

A subtle issue: if the worker has already returned from Process and the Interrupt call lands a moment later, the user's Interrupt runs against a stale state (a cancel that has already been invoked, or a channel already closed). Idempotent Interrupt implementations handle this gracefully.

How Terminate Is Invoked¶

Terminate is invoked exactly once per worker — when the worker's goroutine is about to exit. The workerWrapper.run method has:

defer w.worker.Terminate()

This defer runs when the loop returns, which happens when:

• Close was called on the pool. The pool closed each worker's closeChan, which broke the loop.
• SetSize was called to shrink the pool. Excess workers had their closeChan closed.

In both cases, Terminate is the last thing the worker does, after any in-flight Process has returned.

You can rely on Terminate running:

• Exactly once.
• After all Process calls on this worker have finished.
• In the worker's goroutine.

You cannot rely on Terminate running:

• If the program exits without Close (e.g. panic).
• If the worker's goroutine is killed by some external means (not possible in normal Go).

Close — The Two-Phase Shutdown¶

Pool.Close is more nuanced than it looks:

func (p *Pool) Close() {
    p.workerMut.Lock()
    defer p.workerMut.Unlock()

    for _, w := range p.workers {
        w.stop()
    }
    for _, w := range p.workers {
        w.join()
    }
    close(p.reqChan)
    p.workers = nil
}

w.stop() and w.join() correspond to "ask to stop" and "wait until stopped" on the workerWrapper.

The two-phase pattern (stop then join) means all workers are asked to stop simultaneously, then we wait for each to exit. If we did it serially (stop, join, stop, join…), shutdown would be O(n * Process_duration). With the two-phase pattern it is O(max Process_duration).

Then reqChan is closed. After this point, any caller still in <-reqChan receives (zero, false) and panics with ErrPoolNotRunning.

The workers = nil line is mostly defensive — it releases memory.

Close is not idempotent in the canonical sense — calling it twice panics because reqChan is already closed. To make it idempotent, wrap it yourself with a sync.Once.

SetSize — Live Resize¶

SetSize changes the number of workers at runtime:

func (p *Pool) SetSize(n int) {
    p.workerMut.Lock()
    defer p.workerMut.Unlock()

    lWorkers := len(p.workers)
    if lWorkers == n {
        return
    }
    if lWorkers < n {
        // Spawn new workers
        for i := lWorkers; i < n; i++ {
            p.workers = append(p.workers, newWorkerWrapper(p.reqChan, p.ctor()))
        }
        return
    }
    // Shrink — ask the extras to stop
    for _, w := range p.workers[n:] {
        w.stop()
    }
    for _, w := range p.workers[n:] {
        w.join()
    }
    p.workers = p.workers[:n]
}

Growing:

• Construct fresh workerWrapper instances. Each one starts its goroutine. The new workers immediately participate in dispatch via the shared reqChan.

Shrinking:

• Stop the excess workers. Wait for them to finish. Trim the slice.

Implication: SetSize can be slow if shrinking workers are mid-Process. The mutex is held throughout, so concurrent calls to Close or another SetSize are serialised.

You can use SetSize to implement crude auto-scaling, but be careful about flapping. Most production tunny users size once at startup and never resize.

Payload Channels Explained¶

Each workerWrapper has two channels involved in the round-trip:

• jobChan — the caller writes the payload here.
• retChan — the worker writes the result here.

These two channels are per workerWrapper, but reused across calls. Or, depending on the version, created fresh per call. The semantics are the same: one-shot from the user's perspective. Each call has one payload, one result.

Why two channels and not one (with a struct containing both directions)?

• Symmetry. Going forward and coming back are different events; separate channels make them independently observable.
• Cancellation. If the deadline fires between phase 2 and phase 3, the caller can bail out of retChan without affecting jobChan's state.

The buffering: jobChan is unbuffered or size-1 depending on version; retChan is similar. The choice has no observable effect for the user because the protocol is strictly sequential.

Comparison With Stdlib Worker-Pool Patterns¶

There is no "stdlib worker pool", but there are canonical patterns. The closest:

Pattern A — for range over a channel of work¶

jobs := make(chan job)
for i := 0; i < n; i++ {
    go func() {
        for j := range jobs {
            process(j)
        }
    }()
}
// produce jobs
close(jobs)

Differences from tunny:

• No result return. You would add a result channel.
• No cancellation per job.
• No BlockUntilReady hook.
• No per-worker Terminate.
• Submission is non-blocking once jobs is buffered.

This pattern is fine for many cases but is essentially the bare bones of what tunny offers.

Pattern B — semaphore + goroutines¶

sem := make(chan struct{}, n)
for _, j := range jobs {
    j := j
    sem <- struct{}{}
    go func() {
        defer func() { <-sem }()
        process(j)
    }()
}

Differences from tunny:

• One goroutine per job, not one per worker slot.
• No reuse of goroutines.
• No per-worker state.

Performance is similar for short jobs. For long-lived workers with state, tunny's worker reuse wins.

Pattern C — errgroup.WithContext + manual semaphore¶

g, ctx := errgroup.WithContext(ctx)
sem := semaphore.NewWeighted(int64(n))
for _, j := range jobs {
    j := j
    if err := sem.Acquire(ctx, 1); err != nil { break }
    g.Go(func() error {
        defer sem.Release(1)
        return process(ctx, j)
    })
}
err := g.Wait()

This is "rich man's tunny". You get context propagation, error aggregation, bounded concurrency. You do not get per-worker state or BlockUntilReady.

Use this when:

• Each job's resource needs are different (some need a DB, some need an HTTP client).
• You want error aggregation rather than per-call results.

Use tunny when:

• All jobs are the same shape.
• Per-worker state is meaningful (buffers, codecs, models).

Comparison With ants Internals¶

ants (panjf2000/ants) takes a very different approach. Key differences:

1. Dynamic pool. ants can grow up to a configured maximum, and shrink when workers are idle. tunny is fixed-size unless you call SetSize.
2. Closure-on-submit. ants takes a closure when you Submit(f). tunny takes a payload to a pre-defined Process.
3. Pre-spawned vs lazy. ants can pre-spawn or grow on demand. tunny pre-spawns all workers.
4. Object pool for workers. ants uses a sync.Pool-like recycling for worker goroutines to handle the dynamic case. tunny does not — workers are static.

Internals snapshot of ants:

• A wrapping pool struct with a goWorkerCache *sync.Pool and workers workerArray.
• Submit(task) either picks a free worker from workers or constructs a new one (up to capacity).
• Workers are returned to the cache when idle.
• Idle workers can be reaped after a configurable interval.

Differences in spirit:

• ants is closer in shape to a Java ThreadPoolExecutor with core/max sizing and idle reaping.
• tunny is closer to a "fixed bank of long-lived agents" — more like a worker pool of Erlang processes.

Neither is "better". They serve different workloads. ants for spiky, heterogeneous, mostly-IO tasks. tunny for steady-state CPU-bound with stateful workers.

Comparison With workerpool Internals¶

gammazero/workerpool is even simpler:

pool := workerpool.New(n)
pool.Submit(func() { ... })
pool.StopWait()

Submission is non-blocking unless the pool's internal queue is also bounded (it is, but large). The queue is chan func() of some default size.

Internals:

• A WorkerPool struct with a task channel and a max workers count.
• Workers are spawned lazily as tasks queue up, up to the max.
• Idle workers are reaped.

Differences from tunny:

• workerpool is "submit a closure" style — like ants but simpler.
• workerpool has no per-worker state.
• workerpool's submission can block if its queue fills (rare, but possible).

If you want stdlib-feel goroutine reuse with a func() API, workerpool is the cleanest choice. If you want stateful workers, tunny.

Where Allocations Happen¶

Profiling tunny code, the typical allocation sites are:

1. Per call: Process(payload any). The any boxing may allocate if payload is not interface-friendly (struct types, basic types in some Go versions).
2. Per call: result interface boxing. Same reason.
3. Per call: timer for ProcessTimed. time.NewTimer allocates a Timer struct.
4. Per pool construction: workers slice. Once.
5. Per worker construction: the workerWrapper. Once per worker.

Things that do not allocate per call (in normal use):

• The workRequest. Re-used or stack-allocated.
• The worker's jobChan and retChan. Created once.

If you want a zero-allocation pool, you must:

• Use a payload type that does not box (e.g. small fixed-size).
• Avoid ProcessTimed (or pool the timer yourself).

For most workloads, the per-call allocations are negligible compared to the work itself. But for sub-microsecond workloads, they may dominate.

Race Conditions Tunny Avoids¶

Tunny is designed to be race-free. The race detector should be silent under normal use. Specifically:

• Process is safe to call from any number of goroutines.
• The dispatcher uses channels (atomic by nature) rather than shared variables.
• queuedJobs is touched only via sync/atomic.
• workers slice is mutated only under workerMut.

You can verify with:

go test -race ./...

The race detector hooks into channel ops and atomic ops, so any unintentional shared-memory access would be caught.

Race Conditions You Can Introduce¶

While tunny is race-free, you can introduce races in the worker function. The most common:

Race 1 — worker function reads/writes a package-level variable¶

var counter int

pool := tunny.NewFunc(8, func(p any) any {
    counter++ // RACE
    return counter
})

Eight workers, eight goroutines, all hitting counter without synchronization.

Fix: use atomic.Int64.

Race 2 — Interrupt reads state Process writes¶

Already covered in middle.md. Always synchronize between methods.

Race 3 — sharing a *bytes.Buffer across workers¶

shared := &bytes.Buffer{}
pool := tunny.New(8, func() tunny.Worker {
    return &w{buf: shared} // all workers share `shared`
})

Eight workers all writing to one buffer. Race. Fix: per-worker buffer.

Race 4 — closure capturing a loop variable¶

for _, x := range items {
    go func() {
        pool.Process(x) // captures loop variable
    }()
}

Classic Go bug. Fix: x := x before the goroutine.

The race detector catches all of these.

Memory Model Considerations¶

A channel send happens-before the matching receive. So:

• The caller's writes to the payload happen-before the worker's reads from the payload (because send-then-receive on jobChan).
• The worker's writes to the result happen-before the caller's reads (because send-then-receive on retChan).

You do not need extra synchronization on the payload or result. Tunny's channels provide the memory ordering for free.

This is one of the unsung values of channel-based concurrency: visibility is automatic. Compare to a hand-rolled "set a shared variable then signal a condition variable" pattern, where you must be careful about memory orderings.

Scheduler Interaction¶

Tunny's worker goroutines are ordinary goroutines. The Go scheduler treats them like any other. A few interactions worth understanding:

Sysmon and preemption¶

A worker in a tight loop without function calls used to be uninterruptable. Since Go 1.14, async preemption catches these cases. So a worker that does for { x++ } will not freeze the scheduler. But the worker will not exit — async preemption just lets other goroutines run.

runtime.LockOSThread¶

If a worker calls runtime.LockOSThread, it pins itself to an OS thread for the duration. The pinning is per-goroutine, not per-worker. If you Lock in one Process and forget to Unlock before returning, the goroutine remains pinned, which can subtly affect future Process calls.

Always defer runtime.UnlockOSThread() if you must lock.

GOMAXPROCS¶

The worker goroutines are subject to GOMAXPROCS. A pool of 8 workers on GOMAXPROCS=2 will have only 2 running at any time. The pool size and the parallelism are not the same.

For CPU-bound work, choose pool size ≤ GOMAXPROCS. Anything more is gratuitous goroutines.

Cgo¶

A cgo call temporarily increases the thread count. If many workers do cgo calls simultaneously, the OS thread count grows. Tunny doesn't help or hinder this; it is just a side effect of cgo + many goroutines.

Extending tunny in Your Own Code¶

Tunny is small. Most extensions are easier than you think.

Extension 1 — instrumentation wrapper¶

type instrumentedWorker struct {
    inner   tunny.Worker
    metrics MetricsClient
}

func (w *instrumentedWorker) Process(p any) any {
    timer := w.metrics.StartTimer("process")
    defer timer.Stop()
    return w.inner.Process(p)
}
func (w *instrumentedWorker) BlockUntilReady() {
    w.metrics.StartTimer("block").Stop() // ... not great but illustrative
    w.inner.BlockUntilReady()
}
func (w *instrumentedWorker) Interrupt()  { w.inner.Interrupt() }
func (w *instrumentedWorker) Terminate()  { w.inner.Terminate() }

pool := tunny.New(n, func() tunny.Worker {
    return &instrumentedWorker{inner: newRealWorker(), metrics: m}
})

A decorator that adds metrics without touching the inner type. Composes cleanly.

Extension 2 — typed generic wrapper¶

We saw Pool[In, Out] in middle.md. That is the recommended core extension.

Extension 3 — priority pool¶

Build two pools, route by priority. We saw this in middle.md.

Extension 4 — custom dispatch policy¶

If you want anything other than "first available worker", you must build a layer in front of tunny. Tunny's dispatch is channel-based and offers no hooks.

E.g. "least-loaded worker first":

type LeastLoadedPool struct {
    pools []*tunny.Pool
    loads []*atomic.Int64
}

func (p *LeastLoadedPool) Process(payload any) any {
    minIdx := 0
    minLoad := p.loads[0].Load()
    for i := 1; i < len(p.pools); i++ {
        if l := p.loads[i].Load(); l < minLoad {
            minIdx = i
            minLoad = l
        }
    }
    p.loads[minIdx].Add(1)
    defer p.loads[minIdx].Add(-1)
    return p.pools[minIdx].Process(payload)
}

Many tiny pools, one per worker, you dispatch yourself. Loses some of tunny's elegance but lets you implement custom policies.

Extension 5 — replay journal¶

Persist payloads to disk before submission, so you can replay if the process crashes. This is a real production pattern for tasks that take minutes to hours.

func (p *DurablePool) Process(payload any) any {
    p.journal.Append(payload)
    defer p.journal.Mark(payload, "done")
    return p.inner.Process(payload)
}

The journal is outside tunny — tunny does not know about durability. Build it yourself.

Patterns the Internals Suggest¶

A few patterns become obvious once you understand the internals.

Pattern: leverage BlockUntilReady for fair queues¶

If different workers represent different priorities, give the high-priority workers a quicker BlockUntilReady so they win the race to reqChan. Crude but effective.

Pattern: many small pools instead of one big¶

Because dispatch is "first-ready-wins", having distinct pools per resource class avoids head-of-line blocking. A slow worker in pool A does not affect callers in pool B.

Pattern: instrument at the workerWrapper level, not the inner Worker¶

If you wrap Worker, you intercept exactly what tunny calls. You see real Process durations, real BlockUntilReady waits, real Interrupt events. The inner type is unaware.

Pattern: design Process to be short and pure¶

Long Process blocks the worker. Pure Process makes testing easy. Side-effectful Process complicates observability. Where possible, push side effects (logging, metrics) to the wrapper layer.

Best Practices Implied by the Source¶

Some practices fall out naturally once you understand the internals:

1. Always defer Close. You leak goroutines otherwise — confirmed by the source: workers run until closeChan is closed.
2. Make Interrupt idempotent. The source may call it after Process returns; idempotency is required for correctness.
3. Do not capture the *tunny.Pool inside the worker. The worker should be self-contained. Capturing the pool risks recursive calls and deadlock.
4. Keep BlockUntilReady predictable. Long, irregular BlockUntilReady distorts your effective pool size.
5. Treat workerWrapper.run as the authoritative state machine. Any extension that claims to control worker lifecycle must respect this loop's order.

Tricky Points¶

Tricky 1 — Close does not interrupt in-flight Process¶

Close waits for in-flight Process to finish. It does not interrupt them. If your worker is mid-Process doing a 10-minute compute, Close will block for 10 minutes.

If you need fast shutdown, send interrupts to all workers before calling Close:

for _, w := range p.workers {
    w.worker.Interrupt() // but the source doesn't expose this directly
}

You cannot do this through tunny's API — workers is not exposed. Workaround: store a slice of your own workers and interrupt them yourself before Close.

Tricky 2 — ProcessTimed and BlockUntilReady interaction¶

A worker that is in BlockUntilReady is not registered to receive a payload. If a ProcessTimed caller's timer fires while no worker is ready, the caller exits cleanly with ErrJobTimedOut. No worker is interrupted because none was assigned. Your Interrupt method is not called. This is correct behaviour, but it can be confusing: a metric for "Interrupts" will show fewer than "timeouts" because many timeouts never reach a worker.

Tricky 3 — interruptFunc racing with Process returning¶

If Process is about to return at the same moment interruptFunc is called, you have a race window. The caller will get the timeout error; the worker's return value is discarded.

What about the worker's state? If Process set w.cancel = nil before interruptFunc reads it, Interrupt is a no-op. If Process is reading from w.cancel while interruptFunc writes, that is a race — protect with mutex.

In production, treat Interrupt as fire-and-forget. Do not rely on it doing anything observable.

Tricky 4 — SetSize does not affect in-flight calls¶

If you have 8 workers, 4 are processing, you call SetSize(2) to shrink. Tunny stops 6 workers — but workers 1 through 4 are currently busy. Tunny waits for them to finish current Process, then terminates them. So SetSize(2) is asynchronous in this sense. From the caller perspective the size is 2 immediately; from the dispatch perspective the effective concurrency catches up later.

Tricky 5 — closed reqChan panic¶

If you call Process after Close, you panic. The exact panic message depends on Go version but it is one of:

panic: send on closed channel
panic: receive from closed channel (... actually returns false ...)

Real tunny: receive from closed reqChan returns (zero, false). The library detects this and panics with ErrPoolNotRunning. So the panic message is ErrPoolNotRunning, not the raw channel-closed message. Verify with:

defer func() {
    fmt.Println(recover())
}()
pool.Close()
pool.Process(nil)

Output approximately: tunny pool is not running.

Tricky Questions¶

Q1. Two callers call Process simultaneously. The pool has 1 worker. What is the order of dispatch?

A1. Whichever caller's <-reqChan is selected first by the runtime. The order is not arrival order — Go does not promise FIFO on channel receives. It is approximately fair under typical workloads, but not strictly so.

Q2. A ProcessCtx deadline fires. Interrupt is called. The worker's Process ignores the interrupt and runs for another minute. What is the state of the pool during that minute?

A2. The worker is unavailable. The pool's effective size has dropped by 1. After the minute, the worker returns, the result is discarded, the worker re-enters BlockUntilReady and becomes available again.

Q3. Close runs concurrently with Process on the same pool. Possible outcomes?

A3. Race. Possible outcomes:

• Process completes before Close reaches it — fine.
• Process is mid-channel-op when Close closes channels — panic.
• Process reads from reqChan after Close — returns (zero, false) → panic with ErrPoolNotRunning.

Always coordinate Close with the lifecycle of Process-calling goroutines.

Q4. Why does Process not return an error?

A4. Design choice — Process is the simplest API. Errors are encoded in the result. ProcessTimed/ProcessCtx return errors because timeouts/cancellations are distinct from "the work failed".

Q5. Could tunny implement Process with error return as (any, error) always? What would change?

A5. It could, but the simple case becomes more verbose. The library chose to keep Process simple. The cost is the awkward distinction between Process (no error) and ProcessTimed/ProcessCtx (error).

Q6. What is the lifecycle of req.jobChan — is it reused across calls?

A6. In current tunny, jobChan and retChan are owned by the workerWrapper and are channels created with size 1, persisting across calls. Each cycle of the worker loop sends a fresh workRequest containing these channels (closed over from the workerWrapper) to reqChan. The caller reads, sends payload, reads result.

Q7. What happens if the worker panics in Process — does the worker goroutine die?

A7. A panic in Process propagates up to workerWrapper.run. If the panic is unrecovered, the whole program crashes (panics in goroutines kill the process). Tunny does not insert a recover, so always handle panics inside your worker.

If you want defensive behaviour, write a wrapper:

type panicSafeWorker struct {
    inner tunny.Worker
}

func (w *panicSafeWorker) Process(p any) (out any) {
    defer func() {
        if r := recover(); r != nil {
            out = fmt.Errorf("panic in worker: %v", r)
        }
    }()
    return w.inner.Process(p)
}
// other methods delegate to inner

This catches the panic before tunny would have crashed.

Q8. Is the worker's Process re-entrant? Can it call pool.Process on the same pool?

A8. Re-entrant in the sense that Go allows it — but you can deadlock. If all other workers are also busy and call back into the pool, no one is available to serve. Forbidden in practice.

Q9. Why does tunny use buffered vs unbuffered channels?

A9. Mostly unbuffered. The dispatch is rendezvous-style. The exception is the small buffers on jobChan/retChan so that the worker can send-then-receive in the natural order without blocking on the channel itself. The channel buffer is small because the protocol is strictly sequential.

Q10. Could you build tunny on top of sync.Cond instead of channels?

A10. In principle, yes. The result would be uglier — sync.Cond requires explicit signal/broadcast and explicit mutex. Channels give you select for free, which is what ProcessTimed and ProcessCtx need.

Cheat Sheet¶

Types:
  Pool                 - the public type
  workerWrapper        - internal, one per worker
  workRequest          - internal, one per call

Channels per workerWrapper:
  closeChan            - "stop now"
  closedChan           - "I have stopped"
  interruptChan        - "stop your current Process"
  reqChan              - "I am ready" (write here for dispatch)

Per-call channels (via workRequest):
  jobChan              - payload from caller to worker
  retChan              - result from worker to caller

Worker loop:
  loop {
    BlockUntilReady()
    select { closeChan -> exit; reqChan <- request -> serve; interruptChan -> ignore }
    on serve:
      jobChan -> payload
      Process(payload)
      retChan <- result
  }

Process flow:
  <-reqChan (get worker)
  jobChan <- payload
  retChan -> result

ProcessTimed/ProcessCtx flow:
  three select blocks, one per phase, each with deadline branch
  on timeout: interruptFunc() called

Close:
  close each workerWrapper.closeChan
  wait for each workerWrapper.closedChan
  close pool.reqChan

Self-Assessment Checklist¶

• Can draw the Pool and workerWrapper data structures.
• Can recite the dispatcher loop pseudocode.
• Can explain why dispatch is via worker-writes-to-channel, not pool-writes-to-channel.
• Can predict what ProcessTimed does in each of three phases.
• Can name three races you can introduce despite tunny being race-free.
• Can describe the lifecycle of a worker from ctor to Terminate.
• Can compare tunny's design with ants and workerpool.
• Can write a wrapper Worker that adds metrics around Process and BlockUntilReady.

Summary¶

• Tunny is a small library. Its core is workerWrapper.run, a goroutine loop with one select block.
• Dispatch is inverted: workers write to reqChan saying "I am ready"; callers read from it to discover an available worker.
• Per-call payload and result flow through small per-call channels.
• ProcessTimed and ProcessCtx wrap each phase in a select with a deadline branch.
• Interrupt is sent from the caller's goroutine to the worker; the worker must synchronize internally if it reads shared state.
• Close is a two-phase shutdown: stop all, then join all.
• Allocations are mostly per-call interface boxing and time.Timer (for ProcessTimed).
• Tunny does not catch worker panics — wrap your worker if you cannot trust it.
• The library composes well with rate.Limiter, context.Context, errgroup, sync.Pool, and slog.

Further Reading¶

• The tunny source. Read it twice: once after junior level, once after senior level. Different things stand out.
• The Go memory model document.
• "Channels Reading" — explanations of select semantics in detail.
• The source of panjf2000/ants and gammazero/workerpool for contrast.
• Articles on "worker pool patterns in Go" — there are many; tunny's is one of several valid shapes.

Related Topics¶

• Junior tunny and middle tunny — the API levels this file builds on.
• Professional tunny (professional.md) — production deployment patterns informed by what you know now.
• Channels deep dive (../../02-channels/) — the foundation.
• sync.WaitGroup (../../03-sync-package/) — used inside Close.
• GMP scheduler (../../01-goroutines/) — what runs the workers.

Diagrams¶

Diagram 1 — Pool data structure¶

+---------------------------------------+
|              Pool                      |
|  queuedJobs    int64                   |
|  ctor          func() Worker            |
|  workers       []*workerWrapper        |
|  reqChan       chan workRequest        |
|  workerMut     sync.Mutex              |
+---------------------------------------+
              │
              ├─> []*workerWrapper of length n
              │       │
              │       ├─ workerWrapper #0 (goroutine)
              │       ├─ workerWrapper #1 (goroutine)
              │       └─ ...
              │
              └─> reqChan: shared by all workerWrappers
                  workers WRITE to it ("I am ready")
                  callers READ from it ("give me a worker")

Diagram 2 — workerWrapper.run loop¶

┌─────────────────────────────────┐
│         workerWrapper.run        │
└───────────┬─────────────────────┘
            │
            ▼
   ┌─────────────────────┐
   │ BlockUntilReady()    │
   └─────────┬───────────┘
             │
             ▼
   ┌──────────────────────────────┐
   │ select {                      │
   │   case <-closeChan: exit      │
   │   case reqChan <- req:        │
   │       <-jobChan -> payload    │
   │       Process(payload)        │
   │       retChan <- result       │
   │   case <-interruptChan: ignore│
   │ }                              │
   └─────────┬───────────┬─────────┘
             │           │
             │       (other)
             ▼
       (back to top)

Diagram 3 — Process call timing¶

time ─────────────────────────────────────▶

Caller A:        |--wait--|--send--|--wait--|--ret--|
Worker W:        |--ready|--get--|---Process---|--push--|
                          ^      ^             ^      ^
                       reqChan jobChan      Process retChan

Caller B (later):                              |--wait--|...

Diagram 4 — ProcessTimed phases¶

phase 1: wait for a worker (queue)
   start ───────────────── timer ────────────── deadline
   waiting on reqChan...
   either: get worker, proceed to phase 2
       or: deadline fires, return ErrJobTimedOut

phase 2: send the payload (handoff)
   waiting on req.jobChan <- payload
   either: send succeeds, proceed to phase 3
       or: deadline fires, call interruptFunc, return ErrJobTimedOut

phase 3: wait for the result (work)
   waiting on <-req.retChan
   either: result arrives, return (result, nil)
       or: deadline fires, call interruptFunc, return ErrJobTimedOut

Diagram 5 — Comparison of dispatch models¶

tunny model (worker-pull):

       reqChan: <─ W1 (writes "I am ready")
                  <─ W2 (writes "I am ready")
                  <─ W3 (writes "I am ready")
       caller:  ── reads, picks one


closure-on-submit model (e.g. ants):

       taskChan: <── caller submits closure
                  <── caller submits closure
       worker:   ── reads, executes


tunny inverts the usual flow. Callers don't push tasks
into a shared queue; workers offer themselves. Same channel,
different semantics.

These five diagrams cover every internal aspect of tunny worth understanding. Keep them in mind when reading the source.

Extended Internal Investigation — Stepping Through With a Debugger¶

To really cement understanding, run the following minimal program under dlv:

package main

import (
    "fmt"

    "github.com/Jeffail/tunny"
)

func main() {
    pool := tunny.NewFunc(2, func(p any) any {
        return p.(int) * 2
    })
    defer pool.Close()
    fmt.Println(pool.Process(7))
}

Set breakpoints in tunny.go at:

• The top of New (constructor).
• The top of Pool.Process.
• The top of workerWrapper.run.
• The top of Pool.Close.

Step through. Observe:

1. New runs ctor twice. Two workerWrapper instances exist.
2. Each workerWrapper is started with go w.run(). Two goroutines now exist (plus main).
3. Each run reaches the select. Both are blocking on reqChan's send.
4. Process(7) enters. The caller reads from reqChan. One of the two waiting workers wins and is now "in-flight".
5. The caller writes 7 to req.jobChan. The worker reads.
6. The worker calls Process (which doubles 7 to 14). The result is written to retChan.
7. The caller reads 14 from retChan and returns.
8. The winning worker's run loops back to BlockUntilReady and the select.
9. defer pool.Close() fires. Both workers' closeChan is closed. Both run loops exit. Both Terminate runs.

Doing this once with a debugger crystallises the model.

Final Note on the Senior Level¶

You now understand tunny at the level of the maintainer. You can:

• Read the source and predict every behavior.
• Spot edge cases in production logs by translating symptoms to internal mechanisms.
• Extend tunny without forking it — via wrappers and adapter types.
• Compare tunny to alternatives with depth, not surface impressions.

The professional file builds on this with production deployment patterns — observability, graceful shutdown, integration with HTTP frameworks, case studies. The technical knowledge is established. The professional level is about operating tunny under real production conditions: capacity planning, monitoring, incident response.

Take a break, write a small experiment that touches the internals, then read on.

Deep Dive — Channel Semantics Used by Tunny¶

To understand tunny's source, you must understand Go channel semantics at a level beyond "send and receive". Some of these are subtle and used in clever ways.

Send on a closed channel¶

A send on a closed channel panics. Tunny relies on this to detect double-close:

close(p.reqChan)
close(p.reqChan) // panic: close of closed channel

It also affects callers: a Process call running concurrently with Close may panic with "send on closed channel" if the timing is wrong. Tunny detects this in the receive direction (closed channel returns (zero, false)) and reports ErrPoolNotRunning.

Receive from a closed channel¶

Receive from a closed channel returns the zero value of the channel's element type and false for the two-value receive form. Tunny's Process uses this:

req, open := <-p.reqChan
if !open {
    panic(ErrPoolNotRunning)
}

The open flag is true if the channel is still open and an actual value was received, false if the channel was closed and the zero value was returned.

Select with multiple ready channels¶

select chooses pseudo-randomly among multiple ready channels. Tunny relies on this for fairness: when multiple workers are ready and a caller arrives, no worker is privileged.

If you need FIFO, you must implement it yourself — select does not give you that.

Select with nil channels¶

A nil channel in a select case is never selected. This is used in some libraries to dynamically disable a branch:

var ch chan int = nil
select {
case <-ch: // never fires
case <-other: // always candidate
}

Tunny does not use this trick, but it is good to know when reading other channel-heavy code.

Buffered vs unbuffered for the dispatcher¶

Tunny's reqChan is unbuffered. This means a worker's send blocks until a caller receives. The rendezvous semantics ensure that:

• A worker is not "lost" — there is exactly one caller paired with each worker.
• A caller is not "lost" — there is exactly one worker paired with each request.

If reqChan were buffered, a worker could write to it and immediately loop back to BlockUntilReady. Then if no caller arrived, the worker would write again. The buffer would grow unboundedly. So unbuffered is correct.

Per-call channel ownership¶

The jobChan and retChan are owned by the worker. They are created during worker construction and reused across calls. The worker's send to reqChan includes pointers to its jobChan and retChan, so the caller knows where to send the payload and where to listen for the result.

This is a subtle but important detail: the worker is the "active" party. The pool does not own the per-call channels; the workers do.

Deep Dive — Why Tunny Doesn't Use sync.Mutex Much¶

Read the tunny source and you will see sync.Mutex used sparingly. Mostly only workerMut in Pool for mutating workers. Why?

Because tunny is built around channels, and channels are the synchronization primitive. A send-and-receive pair on a channel provides a happens-before relation in the Go memory model — equivalent to releasing and acquiring a mutex, but cheaper in some cases and more expressive (you can select).

When you read the tunny source, treat each channel send/receive as a synchronization point. The "lock" is the channel; the "guarded data" is whatever was written before the send.

Compare to a hand-rolled pool using mutexes — you would have a sync.Mutex around a slice of available workers, with Lock / Unlock calls. More verbose, less expressive, similar performance.

Deep Dive — How the Three Constructor Functions Relate¶

tunny.NewFunc, tunny.NewCallback, and tunny.New are not separate code paths — they are layered.

tunny.NewFunc is implemented as:

func NewFunc(n int, f func(interface{}) interface{}) *Pool {
    return New(n, func() Worker {
        return &closureWorker{processor: f}
    })
}

type closureWorker struct {
    processor func(interface{}) interface{}
}

func (w *closureWorker) Process(payload interface{}) interface{} {
    return w.processor(payload)
}
func (w *closureWorker) BlockUntilReady() {}
func (w *closureWorker) Interrupt()       {}
func (w *closureWorker) Terminate()       {}

Same for NewCallback:

func NewCallback(n int) *Pool {
    return New(n, func() Worker {
        return &callbackWorker{}
    })
}

type callbackWorker struct{}

func (w *callbackWorker) Process(payload interface{}) interface{} {
    f, ok := payload.(func())
    if !ok {
        return nil
    }
    f()
    return nil
}
func (w *callbackWorker) BlockUntilReady() {}
func (w *callbackWorker) Interrupt()       {}
func (w *callbackWorker) Terminate()       {}

So:

• NewFunc = New + an adapter that wraps a closure.
• NewCallback = New + an adapter that treats the payload as a closure.
• New is the only "real" constructor.

This confirms our advice: when you need anything more than the bare adapter, use New. The other two are conveniences.

Deep Dive — Worker Goroutine Stacks¶

When tunny starts n worker goroutines, each one gets its own stack. Goroutine stacks in Go start at ~2 KB and grow as needed. A worker that just loops on the dispatcher uses a tiny stack — single-digit KB.

But: a worker that calls a deep recursive function inside Process can grow its stack to whatever size it needs. The stack does not shrink between calls (in current Go). So if your worker once called a function that recursed 1000 frames, the worker holds that stack memory forever.

Implication: if your Process has wildly varying stack depth, your workers may end up with much larger stacks than typical. To estimate worst-case memory, multiply n_workers * max_stack_depth.

For most workloads this is negligible. For workers that occasionally do deep recursion (image processing libraries with recursive algorithms, JSON decoders with very nested objects), it can matter.

Deep Dive — time.Timer Semantics in ProcessTimed¶

ProcessTimed uses time.NewTimer to enforce the deadline. The timer fires once at the specified time. The implementation:

timer := time.NewTimer(timeout)
defer timer.Stop()

timer.Stop() is important: it cancels the timer and ensures the timer's goroutine is reclaimed. If you do not Stop a timer, it may still fire and consume resources (though Go's runtime handles this gracefully).

The <-timer.C channel is the fire signal. The timer's goroutine writes to timer.C when the duration elapses.

A subtle point: if the timer has already fired and timer.C has a value pending, Stop returns false (it could not stop because it was already fired). In that case you may want to drain timer.C:

if !timer.Stop() {
    <-timer.C
}

Tunny does not drain. That is fine because the timer's value is small and gets GC'd; the channel is per-call.

Deep Dive — Context Cancellation Propagation¶

context.Context's Done() channel is closed when the context is cancelled. Receiving from a closed channel returns immediately with the zero value. ctx.Err() returns either context.Canceled or context.DeadlineExceeded.

In ProcessCtx, the select branches on <-ctx.Done():

select {
case ...:
case <-ctx.Done():
    req.interruptFunc()
    return nil, ctx.Err()
}

req.interruptFunc() is the closure tunny built. It does the work of sending the interrupt signal to the worker.

ctx.Err() is returned so the caller can distinguish:

• context.DeadlineExceeded — the context's deadline elapsed.
• context.Canceled — the context was explicitly cancelled.

If you set a deadline shorter than your time.Now() (i.e. already expired), ctx.Done() is closed immediately. ProcessCtx would return error without ever submitting work. This is correct behaviour.

Deep Dive — Worker Replacement Semantics¶

What if you build a new pool with the same ctor after closing an old one? Are the workers reused?

No. Tunny does not pool the Worker instances across pools. Each tunny.New call constructs fresh workers. The closeChan and closedChan are also fresh per worker.

This is "clean state" at the cost of repeated construction. If your factory is expensive, do not create and destroy pools — keep them around.

If you need to reset state on existing workers (e.g. flush caches), you must expose a method on your Worker type and call it manually. Tunny does not have a Reset hook on the interface.

Deep Dive — How QueueLength Compares to Other Metrics¶

Pool.QueueLength returns atomic.LoadInt64(&p.queuedJobs). This is the number of callers currently in Process (or ProcessTimed/ProcessCtx), regardless of phase.

This is not the same as:

• The number of pending payloads in a queue (there is no queue per se).
• The number of running Process invocations (some callers may still be in queue).

What QueueLength actually measures: how many concurrent callers are currently using the pool. If your pool size is n and QueueLength is m:

• If m <= n: every caller has a worker.
• If m > n: n callers have workers, m - n callers are queued.

So QueueLength - PoolSize is the actual queue depth, when positive. Below zero means there is slack.

For monitoring, you often want both:

inflight := min(QueueLength, PoolSize)
queued := max(0, QueueLength - PoolSize)

Export both as separate metrics.

Deep Dive — What runtime.NumGoroutine Shows¶

runtime.NumGoroutine() reports the total goroutine count. A tunny pool of size n contributes n goroutines, plus the main / caller goroutines.

You can use this to detect leaks:

g0 := runtime.NumGoroutine()
pool := tunny.NewFunc(8, work)
g1 := runtime.NumGoroutine()
// g1 - g0 should be 8
pool.Close()
time.Sleep(10 * time.Millisecond) // allow goroutines to exit
g2 := runtime.NumGoroutine()
// g2 should be ~g0

If g2 is much larger than g0, you have leaked goroutines from somewhere (not necessarily tunny).

Deep Dive — Trace Output During a Tunny Run¶

If you compile with runtime/trace and run a small tunny program, you can see exactly when each goroutine is created and scheduled. The output looks roughly like:

goroutine 1 (main): created at t=0
goroutine 2 (workerWrapper.run #0): created at t=0.0001
goroutine 3 (workerWrapper.run #1): created at t=0.0001
...
goroutine 10 (caller): Process(x) at t=0.001
goroutine 2: wakes up, reads payload, runs Process at t=0.001
goroutine 10: waits at t=0.001
goroutine 2: finishes Process, sends result at t=0.002
goroutine 10: wakes up, returns at t=0.002

This trace confirms the model. Run it once on your own machine. Tools like go tool trace give you an interactive viewer.

Deep Dive — Goroutine Profile Stacks¶

If you take a goroutine profile of a process running tunny:

curl -s http://localhost:6060/debug/pprof/goroutine?debug=1

You will see stacks like:

goroutine 42 [chan receive]:
github.com/Jeffail/tunny.(*workerWrapper).run(0xc000020000)
    /path/to/tunny/workerwrapper.go:25 +0x80
created by github.com/Jeffail/tunny.New
    /path/to/tunny/tunny.go:75 +0xc0

Many such stacks indicate many idle workers. That is normal — workers sit on a channel receive while idle.

If you see different stacks, e.g.:

goroutine 42 [select]:
github.com/Jeffail/tunny.(*Pool).ProcessTimed.func1()
    ...

…you have callers in flight in ProcessTimed.

Profile your service in production. The shapes of stacks tell you what tunny is doing.

Deep Dive — Edge: A Worker That Returns Early¶

If your worker's Process returns very quickly — say, hits an error and returns in 1 microsecond — what happens?

The worker:

1. Returns the result.
2. Loops back to BlockUntilReady.
3. If BlockUntilReady is a no-op, immediately writes to reqChan again.

So the worker becomes available again in microseconds. Tunny does not slow this down or introduce minimum cycles. The pool's throughput for trivial work is limited by channel operation cost (which is fast — single-digit nanoseconds in Go).

This is why tunny does not benchmark well for nanosecond workloads: the channel overhead dominates. Use tunny for microsecond-or-larger workloads.

Deep Dive — Why Tunny Doesn't Use Lock-Free Data Structures¶

Channels are essentially lock-free for the common case. The runtime implementation of channels uses atomic operations and a small spinlock; for uncontended cases, it is very fast.

Could tunny use a custom lock-free queue? In principle yes. In practice, channels are fast enough and the resulting code is much simpler. The maintainer chose simplicity.

If you find tunny too slow for your workload, the answer is unlikely to be "swap channels for a lock-free queue." It is more likely "use a different design entirely — submit-and-forget, or a custom dispatcher."

Deep Dive — The interface{} Tax¶

Every Process call boxes the payload into interface{} (any). Boxing means:

• If the payload is a pointer, no allocation — the interface contains the pointer directly.
• If the payload is a small value (int, bool), Go may inline it into the interface header — no allocation.
• If the payload is a larger value (struct), Go allocates space on the heap and stores a pointer in the interface.

Similarly for the return value.

For most workloads, this is invisible. For very hot paths, allocate-bound performance, this can matter. Workarounds:

1. Pre-allocate payload structs and pass pointers. A *Job does not allocate; a Job does.
2. Use a payload type that fits in a register. An int or uintptr does not allocate.
3. Batch many tasks per Process call. Amortize the box cost.

If you need zero-allocation per task, tunny is the wrong tool — you want a custom hand-rolled pool. But that is a niche concern.

Deep Dive — Compaction of the Workers Slice¶

When SetSize shrinks the pool, the workers slice is sliced:

p.workers = p.workers[:n]

This does not free the underlying array. The capacity is unchanged. If you frequently grow and shrink, the slice retains the largest allocation.

For long-lived pools this is fine. For pools that go through many resizes, you may want:

p.workers = append([]*workerWrapper(nil), p.workers[:n]...)

…which makes a fresh underlying array. But this is rarely needed in practice.

Deep Dive — defer Ordering in workerWrapper.run¶

The run function has:

defer func() { close(w.closedChan) }()
defer w.worker.Terminate()

Defers run in reverse order. So Terminate runs first, then closedChan is closed. The closedChan close is the signal "this worker has fully exited including Terminate." The pool's Close waits on closedChan, so after Close returns, you know Terminate has run on every worker.

This ordering is important. If you reverse it:

defer w.worker.Terminate()             // runs last
defer func() { close(w.closedChan) }() // runs first

…then closedChan is closed before Terminate runs. The pool's Close would return before Terminate had finished. Resource leaks could ensue if Close-returns triggers something that needs Terminate to have completed.

Always think about defer order in lifecycle code.

Deep Dive — Async vs Sync Panic Behavior¶

A panic in a goroutine kills the whole process. Tunny's workerWrapper.run does not have a recover. So a panic in:

• BlockUntilReady — process crashes.
• Process — process crashes.
• Interrupt — process crashes (running in caller's goroutine).
• Terminate — process crashes.

If you cannot trust your worker, wrap it:

func (w *safeWorker) Process(p any) (out any) {
    defer func() {
        if r := recover(); r != nil {
            out = fmt.Errorf("panic in worker: %v", r)
        }
    }()
    return w.inner.Process(p)
}

Similarly for the other three methods if needed. Most workers do not need recover in BlockUntilReady or Terminate, but you might.

A philosophical question: should the library do the recover for you? The maintainer's choice is no — explicit is better than implicit. You decide whether to silence panics.

Deep Dive — Goroutine Identity¶

Each worker is a single goroutine. You cannot identify it from outside (Go does not expose goroutine IDs reliably). If you need to log "which worker handled this request", you must give workers explicit IDs:

type identifiedWorker struct {
    id int
    // ...
}

var nextID atomic.Int64
pool := tunny.New(n, func() tunny.Worker {
    return &identifiedWorker{id: int(nextID.Add(1))}
})

Then log w.id inside Process. This is useful for tracing which worker is slow, which one has stale state, etc.

Deep Dive — for select{} vs for { select{} }¶

In workerWrapper.run, the loop is:

for {
    w.worker.BlockUntilReady()
    select {
        // ...
    }
}

Not:

for {
    select {
        // ...
    }
}

The difference: BlockUntilReady is called every iteration before the select. If the iteration exits with no work (interrupt branch taken), BlockUntilReady is called again on the next iteration. So BlockUntilReady may be called many times between actual Process calls.

This means: your BlockUntilReady should be cheap if no work has actually flowed. If you do a heavy operation in BlockUntilReady (e.g. log "ready"), you will see many such logs even if no Process is happening — because of the interrupt-noop branch.

Comparison Study — Tunny vs Java's ThreadPoolExecutor¶

The closest familiar analogy outside Go is Java's ThreadPoolExecutor.

Differences:

• Tunny is much smaller. It does not aspire to ThreadPoolExecutor's flexibility.
• Tunny's "synchronous Process" forces backpressure on callers. Java's "submit returns Future" allows callers to fire and forget, but then they must manage the futures themselves.
• Java's Future supports cancellation that interrupts the running thread (via Thread.interrupt()). Tunny's Interrupt is similar in spirit.

If you are coming from Java, the closest mental shift is "Process is submit(callable).get() in one call". Less flexible, but the simplicity is its own value.

Comparison Study — Tunny vs Tokio's JoinSet¶

In Rust async, the closest cousin is something like JoinSet:

let mut set = JoinSet::new();
for x in inputs {
    set.spawn(async move { work(x).await });
}
while let Some(r) = set.join_next().await {
    // ...
}

Differences:

• Rust async is implicit concurrency on a single executor; Go's goroutines are explicit. The pool concept is more pronounced in Go.
• Tokio's JoinSet is fire-and-forget plus aggregation. Tunny's Process is synchronous per call.
• Tokio's runtime manages worker threads. Tunny does not — it manages goroutines, leaving threads to the runtime.

If you are coming from Rust, the closest mental model is "tunny is a typed channel with N consumers that you can address synchronously."

Engineering Trade-offs in Tunny's Design¶

Looking at the source as a whole, the tunny maintainer made these choices:

1. Synchronous API over asynchronous. Returns from Process carry the result directly. Cost: no fire-and-forget. Benefit: simple call sites.
2. interface{} over generics. Pre-generics-era code. Cost: type assertions everywhere. Benefit: works on any Go version.
3. Fixed-size pool over dynamic. Cost: no auto-scaling. Benefit: predictable resource use, simpler code.
4. Worker interface over closure. Cost: more boilerplate for trivial cases. Benefit: per-worker state, lifecycle hooks.
5. Channels over locks. Cost: harder to debug. Benefit: select-based cancellation is easy.

Each choice is defensible. Some lean against modern Go idioms (generics, especially), but the API is stable and the code is small.

If tunny were rewritten from scratch today, I would expect:

• Generic types: Pool[In, Out] natively.
• ProcessCtx as the primary Process.
• A configurable Sizing policy (fixed, dynamic with bounds).

But the existing API is good enough that a rewrite would be net negative — the cost of breaking everyone's code would exceed the benefit of marginally cleaner internals.

Hands-On Exercise — Read and Annotate the Source¶

Find tunny's source on your machine:

go env GOMODCACHE
find $(go env GOMODCACHE) -name 'tunny.go' -path '*/Jeffail/tunny*'

Open it. Find these methods:

• New
• NewFunc
• NewCallback
• Process
• ProcessTimed
• ProcessCtx
• Close
• SetSize
• QueueLength

For each, write one sentence explaining what it does in terms of the channels. If you can do this for all of them, you understand tunny.

Find workerwrapper.go and identify:

• The workerWrapper struct fields and their roles.
• The run method's loop.
• The interrupt helper.

If you can recite the run loop from memory, you have absorbed everything in this file.

Hands-On Exercise — Fork Tunny and Add a BeforeProcess Hook¶

This is a meaningful but small extension. Add a hook to the Worker interface (or as an optional interface):

type Hookable interface {
    BeforeProcess(payload any)
    AfterProcess(payload any, result any)
}

Inside workerWrapper.run, before and after calling Process, check if the worker implements Hookable and call the hook.

This exercise:

• Forces you to navigate the source.
• Shows you how to extend tunny without breaking back-compat (optional interface).
• Demonstrates that the library is small and modifiable.

You will not actually merge this upstream — it is just for understanding. After doing it, revert. You will know tunny better than before.

Hands-On Exercise — Implement Tunny From Scratch¶

The ultimate test: implement tunny yourself, from memory, in one sitting. Goal: ~200 lines. You need:

• A Pool struct.
• A workerWrapper per worker, with goroutine.
• Process, Close, and the Worker interface.

Skip the conveniences (NewFunc, NewCallback, ProcessTimed, ProcessCtx). Just the core.

If you can do this, you understand tunny at the level of the maintainer.

Final Architecture Reflection¶

Tunny is a tiny library that fully embraces "channels are for communication, not just synchronization." Its design teaches a few lessons that apply broadly:

1. Workers can offer themselves rather than being offered work. The inverted dispatch is small, elegant, and naturally fair.
2. Channels carry not just data but ownership. A workRequest containing pointers to jobChan and retChan hands the worker's mailbox to the caller temporarily.
3. A small interface with lifecycle hooks beats a big closure. The four-method Worker interface is just enough.
4. Errors should live at the API boundary, not the worker boundary. Process returns any; only the timeout/context variants return errors.
5. Tiny library is feature. Tunny does not aspire to be a framework; it is a library you can read on a screen.

These principles, learned through reading tunny, will improve your Go concurrency code for years.

Onward¶

You have now seen tunny from API to source. The professional file applies this knowledge to production operations: deployment, observability, graceful shutdown, capacity planning, incident response. Less code, more ops mindset.

After professional, the specification file is a quick reference; the interview, tasks, find-bug, and optimize files are sharpening exercises. Pick what is useful for your situation.

Whatever your path, you are no longer using tunny — you are operating it. That is the senior level achievement.

Appendix — Reference Snippets¶

Snippet 1 — minimal Pool reimplementation skeleton¶

package mypool

import (
    "sync"
)

type Pool struct {
    workers []*workerWrapper
    reqChan chan workRequest
    mu      sync.Mutex
}

type Worker interface {
    Process(any) any
    BlockUntilReady()
    Interrupt()
    Terminate()
}

type workRequest struct {
    payload   any
    result    chan any
    interrupt func()
}

type workerWrapper struct {
    w        Worker
    reqChan  chan workRequest
    closeCh  chan struct{}
    doneCh   chan struct{}
}

func New(n int, ctor func() Worker) *Pool {
    p := &Pool{reqChan: make(chan workRequest)}
    for i := 0; i < n; i++ {
        ww := &workerWrapper{
            w:        ctor(),
            reqChan:  p.reqChan,
            closeCh:  make(chan struct{}),
            doneCh:   make(chan struct{}),
        }
        p.workers = append(p.workers, ww)
        go ww.run()
    }
    return p
}

func (w *workerWrapper) run() {
    defer close(w.doneCh)
    defer w.w.Terminate()
    for {
        w.w.BlockUntilReady()
        req := workRequest{
            result:    make(chan any, 1),
            interrupt: w.w.Interrupt,
        }
        select {
        case <-w.closeCh:
            return
        case <-w.handoff(req):
        }
        // process
        out := w.w.Process(req.payload)
        req.result <- out
    }
}

func (w *workerWrapper) handoff(req workRequest) chan struct{} {
    ready := make(chan struct{}, 1)
    go func() {
        select {
        case <-w.closeCh:
        case w.reqChan <- req:
            ready <- struct{}{}
        }
    }()
    return ready
}

func (p *Pool) Process(payload any) any {
    req := <-p.reqChan
    // ... too sketchy for production, but conveys shape
    req.payload = payload // bug: req is value type, this is local
    return <-req.result
}

func (p *Pool) Close() {
    p.mu.Lock()
    defer p.mu.Unlock()
    for _, w := range p.workers {
        close(w.closeCh)
    }
    for _, w := range p.workers {
        <-w.doneCh
    }
    close(p.reqChan)
}

This is illustrative, not production code. The real tunny is more careful — particularly around the payload mutation. But this captures the shape.

Snippet 2 — observability wrapper¶

type observedWorker struct {
    inner   tunny.Worker
    name    string
}

func (w *observedWorker) Process(p any) (out any) {
    timer := prometheus.NewTimer(processDuration.WithLabelValues(w.name))
    defer timer.ObserveDuration()
    defer func() {
        if r := recover(); r != nil {
            panicCounter.WithLabelValues(w.name).Inc()
            out = errPanic
        }
    }()
    return w.inner.Process(p)
}

func (w *observedWorker) BlockUntilReady() {
    timer := prometheus.NewTimer(blockDuration.WithLabelValues(w.name))
    defer timer.ObserveDuration()
    w.inner.BlockUntilReady()
}

func (w *observedWorker) Interrupt() {
    interruptCounter.WithLabelValues(w.name).Inc()
    w.inner.Interrupt()
}

func (w *observedWorker) Terminate() {
    terminateCounter.WithLabelValues(w.name).Inc()
    w.inner.Terminate()
}

This is the wrapper you would use in production to get metrics without modifying the inner Worker.

Snippet 3 — health probe¶

func (s *Service) HealthCheck() error {
    if s.pool.QueueLength() > 1000 {
        return fmt.Errorf("queue too long: %d", s.pool.QueueLength())
    }
    if s.pool.GetSize() == 0 {
        return errors.New("pool has zero workers")
    }
    return nil
}

Integrate into your service's /health endpoint. Trip degradation when the pool is overloaded.

Snippet 4 — graceful shutdown¶

func (s *Service) Shutdown(ctx context.Context) error {
    // Stop accepting new requests at the HTTP layer first.
    s.server.Shutdown(ctx)

    // Drain in-flight pool work.
    done := make(chan struct{})
    go func() {
        for s.pool.QueueLength() > 0 {
            select {
            case <-ctx.Done():
                return
            case <-time.After(100 * time.Millisecond):
            }
        }
        close(done)
    }()
    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-done:
    }

    // Now safe to close the pool.
    s.pool.Close()
    return nil
}

Three phases: stop accepting, drain in-flight, close. Each is bounded by the context.

Snippet 5 — testing aid¶

type testWorker struct {
    called atomic.Int64
    fn     func(any) any
}

func (w *testWorker) Process(p any) any {
    w.called.Add(1)
    return w.fn(p)
}

func (w *testWorker) BlockUntilReady() {}
func (w *testWorker) Interrupt()       {}
func (w *testWorker) Terminate()       {}

func TestPoolDispatches(t *testing.T) {
    w := &testWorker{fn: func(p any) any { return p }}
    pool := tunny.New(2, func() tunny.Worker { return w })
    t.Cleanup(pool.Close)

    for i := 0; i < 10; i++ {
        if got := pool.Process(i); got != i {
            t.Errorf("Process(%d) = %v", i, got)
        }
    }
    if w.called.Load() != 10 {
        t.Errorf("called %d times, want 10", w.called.Load())
    }
}

Workers can carry test instrumentation directly. Use this pattern to verify dispatch behaviour in your tests.

Closing the Senior Chapter¶

The senior level is about seeing through the abstraction. You know what tunny does, you know how it does it, you know why those design choices were made, and you can extend or replace tunny as your workload demands.

The professional level is about operating the abstraction. Same code; different concerns. Production deployment, monitoring, incident response, capacity planning.

Recommended next step: read professional.md when you have at least one tunny-based service in production (or about to deploy). The material there is concrete and addresses real ops problems.

Extended Internals — Walking Through the workerWrapper Annotated¶

Here is an annotated version of workerWrapper.run. Numbers point to interesting lines.

func (w *workerWrapper) run() {
    defer close(w.closedChan)              // (1)
    defer w.worker.Terminate()             // (2)

    for {                                   // (3)
        w.worker.BlockUntilReady()         // (4)

        select {                            // (5)
        case <-w.closeChan:                // (6)
            return
        case w.reqChan <- workRequest{      // (7)
            jobChan:       w.jobChan,
            retChan:       w.retChan,
            interruptFunc: w.interrupt,
        }:
            payload := <-w.jobChan         // (8)
            result := w.worker.Process(payload) // (9)
            w.retChan <- result            // (10)
        case <-w.interruptChan:            // (11)
        }
    }
}

(1) closedChan is closed when this goroutine exits — including after a panic. Combined with the defer order, the pool's Close can wait on closedChan to know that all of Terminate has run.

(2) Terminate is the first defer registered, so it runs LAST. Reverse the order of defers and you change visible behavior — Close would return before Terminate finished, opening race windows.

(3) The infinite loop. The only way out is return, which happens only on closeChan close.

(4) BlockUntilReady is called every iteration. This means it is called many times if many cycles complete. Make it fast for the no-work case.

(5) The select is the heart of dispatch. Three branches.

(6) Close branch. If closeChan is closed, this case is always ready. The function returns, triggering deferreds.