← Interview Prep

Go Drills — Concurrency (Advanced)

The harder concurrency drills: a bounded connection pool, single-flight deduplication, and a streaming fan-out stage. Patterns you get asked to build live, on a whiteboard or in an editor.

These sit a level above the introductory channel/worker-pool drills. Each one hinges on an invariant that must hold under arbitrary interleavings — "no more than N in flight", "fn runs exactly once", "no goroutine leaks on cancel". The tests deliberately hammer them with many goroutines, so the only way to trust your answer is to run it under the race detector:

go test -race ./... — the race detector is not optional here. A pool that hands out one too many connections, a single-flight that calls fn twice, or a stream stage that leaks a goroutine on cancel will often pass a naive test and only surface under -race and load. Design for the invariant, then let -race try to break it.

Bounded connection pool (ctx-aware Acquire)

Problem. Build a bounded connection pool — like a pool to a database or a device: at most size live connections. Connections are created lazily via a factory. When the pool is exhausted, Acquire blocks until someone calls Release, or until the context expires.

NewPool(size int, factory func() (any, error)) *Pool
(*Pool) Acquire(ctx) (any, error)  // idle conn | new (if < size) | wait
(*Pool) Release(c any)             // return a connection to the pool
(*Pool) Close()                    // afterwards Acquire returns ErrClosed

Invariants the tests check (run go test -race):

factory is called no more than size times — connections are reused, not recreated.
At any instant no more than size connections are checked out.
Acquire on a full pool respects ctx (timeout / cancel).
A released connection is reused by the next Acquire.

Approach

The clean pattern uses two buffered channels of capacity size, and lets the channels themselves be the accounting — no manual counter to get wrong:

idle chan any — connections ready to hand out. Buffered to size.
tokens chan struct{} — creation permits. Sending into tokens succeeds only size times before it is full, so it caps the total number of connections ever created.

Acquire is a three-tier select, in priority order:

Fast path — non-blocking receive from idle. If a connection is sitting ready, take it.
Grow — non-blocking send into tokens. If it succeeds, we have reserved one of the size creation slots, so call factory(). On error we roll the token back (<-p.tokens) so a failed creation does not permanently consume a slot.
Wait — the pool is full and we may not create more, so block on idle or ctx.Done(), whichever fires first.

Why it is correct. The invariant "at most size live" holds because a connection only comes into existence after a successful tokens <- struct{}{}, and tokens has capacity size; failed creations return their token. "At most size checked out" holds because idle also has capacity size and every live connection is either checked out or sitting in idle — so Release's send into idle can never block. The context is honoured because tier 3 is the only blocking wait, and it selects on ctx.Done() alongside idle.

Solution (Go)

// Дрилл 10 (решение). Bounded connection pool.
package d10pool

import (
	"context"
	"errors"
	"sync"
)

var ErrClosed = errors.New("pool is closed")

type Pool struct {
	idle    chan any      // готовые к выдаче соединения
	tokens  chan struct{} // разрешения на создание (ограничивают общее число)
	factory func() (any, error)

	mu     sync.Mutex
	closed bool
}

func NewPool(size int, factory func() (any, error)) *Pool {
	if size < 1 {
		size = 1
	}
	return &Pool{
		idle:    make(chan any, size),
		tokens:  make(chan struct{}, size),
		factory: factory,
	}
}

func (p *Pool) isClosed() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.closed
}

func (p *Pool) Acquire(ctx context.Context) (any, error) {
	if p.isClosed() {
		return nil, ErrClosed
	}

	// 1) быстрый путь: готовое соединение
	select {
	case c := <-p.idle:
		return c, nil
	default:
	}

	// 2) есть ли право создать новое (пул ещё не заполнен)?
	select {
	case p.tokens <- struct{}{}:
		c, err := p.factory()
		if err != nil {
			<-p.tokens // откатить резерв при ошибке
			return nil, err
		}
		return c, nil
	default:
	}

	// 3) пул полон: ждём освобождения ИЛИ отмены контекста
	select {
	case c := <-p.idle:
		return c, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func (p *Pool) Release(c any) {
	// idle буферизован на size, а живых соединений не больше size,
	// поэтому запись никогда не блокирует.
	p.idle <- c
}

func (p *Pool) Close() {
	p.mu.Lock()
	p.closed = true
	p.mu.Unlock()
}

Pitfalls

A manual counter under a mutex is the tempting-but-fragile alternative: you have to increment/decrement it in exactly the right places and it is easy to leak a slot on the factory-error path. Letting tokens and idle be the accounting removes a whole class of bugs.
Forgetting to roll back the token on factory error — without <-p.tokens, a transient creation failure permanently shrinks pool capacity by one.
Calling factory while holding a lock. Creation can be slow (dialing a socket); doing it under a mutex serialises all acquirers. Here creation happens after a lock-free channel send, so concurrent growth is fine.
Blocking on Release. It only works because the "at most size live" invariant guarantees idle has room. Break that invariant (e.g. hand out more than size) and Release can deadlock.
Tier ordering matters. If you skip the fast path and always try to grow first, a pool that already has idle connections will still burn creation tokens — reuse must be preferred.
Close semantics. This minimal version just flips a flag so future Acquire returns ErrClosed; it does not drain idle or unblock waiters already parked in tier 3. In production you would close a done channel and select on it in the wait, and actually close the pooled connections.

Single-flight: deduplicating concurrent identical requests

Problem. A Group deduplicates concurrent calls that share the same key: while one call to fn is in progress, other callers with the same key do not launch fn — they wait and receive the same result. This is the classic defence against a thundering herd (e.g. a cache miss should send exactly one request to the database, not thousands).

(*Group[K,V]) Do(key K, fn func() (V, error)) (V, error, bool)
                                                        ^ shared:
  true if the result was shared with other callers (there were duplicates).

After fn finishes the key is removed, so the next Do(key) calls fn again.

Invariants (tests + go test -race):

For N simultaneous Do with one key, fn is called exactly once.
Everyone receives the same (value, err).
Different keys run independently.

Approach

State is a map[K]*call[V] guarded by a sync.Mutex, where each in-flight call holds a sync.WaitGroup plus the eventual val/err. The WaitGroup is a one-shot broadcast latch — the leader does Add(1) up front and Done() when the result is ready; followers do Wait().

The critical-section discipline is what makes it correct and concurrent:

Under the lock, and only the map bookkeeping: check whether a call for this key already exists. If so, this is a follower — bump dups, unlock, Wait(), and return the leader's result with shared=true.
Otherwise this is the leader: create the call, wg.Add(1), publish it in the map, then unlock.
Outside the lock, the leader runs fn(), stores the result, and calls wg.Done() to release all followers.
Re-lock briefly to delete the key (so the next call re-runs fn) and read dups to compute shared.

Why it is correct. "Exactly once" follows from the map + mutex: the map lookup and the insert happen atomically in one critical section, so only the first goroutine to reach it finds the key absent and becomes the leader; every later goroutine finds the key present and becomes a follower. The WaitGroup gives a happens-before edge — the leader's writes to val/err precede Done(), which precedes every follower's Wait() return — so followers read the result without a data race and without their own lock.

The load-bearing detail: fn must run OUTSIDE the mutex. If you called fn while holding g.mu, you would serialise every key through a single lock — a slow fn for key A would block unrelated key B, destroying the "different keys run independently" invariant and turning your deduplicator into a global bottleneck. The mutex protects only the map; the work runs unlocked.

Solution (Go)

// Дрилл 11 (решение). Single-flight.
package d11flight

import "sync"

type call[V any] struct {
	wg   sync.WaitGroup
	val  V
	err  error
	dups int
}

type Group[K comparable, V any] struct {
	mu sync.Mutex
	m  map[K]*call[V]
}

func (g *Group[K, V]) Do(key K, fn func() (V, error)) (V, error, bool) {
	g.mu.Lock()
	if g.m == nil {
		g.m = make(map[K]*call[V])
	}
	if c, ok := g.m[key]; ok {
		// вызов уже в работе -- ждём его результат
		c.dups++
		g.mu.Unlock()
		c.wg.Wait()
		return c.val, c.err, true
	}

	c := new(call[V])
	c.wg.Add(1)
	g.m[key] = c
	g.mu.Unlock()

	// fn вне мьютекса, иначе сериализуем все ключи
	c.val, c.err = fn()
	c.wg.Done()

	g.mu.Lock()
	delete(g.m, key)
	shared := c.dups > 0
	g.mu.Unlock()

	return c.val, c.err, shared
}

Pitfalls

Running fn under the lock — the headline mistake. It compiles, passes a single-key test, and quietly serialises all keys. Always drop the lock before doing work.
Deleting the key too early or too late. Delete after Done() so late-arriving followers still latch onto the in-flight call; if you delete before the work finishes, a new caller starts a second fn.
Follower reading val/err without the WaitGroup edge. The Wait()/Done() pair is what makes those reads race-free — reaching into c.val before Wait() returns is a data race the detector will flag.
A panicking fn leaves Done() uncalled, so every follower blocks forever. Real singleflight recovers the panic (or uses defer) and re-raises it to all callers; this drill omits that for brevity.
Duplicate suppression can propagate one failure to many. If the single fn returns an error, every shared caller gets that same error — usually desirable, but worth calling out when the caller expected independent retries.
Counting shared from the leader side. dups is incremented under the lock by followers, and read under the lock by the leader after Done(), so the count is stable — don't try to read it lock-free.

Streaming bounded fan-out with backpressure and cancellation

Problem. Build StreamMap[T, R](ctx, in, workers, fn) -> <-chan Result[R]: process a stream from the input channel in concurrently, subject to four constraints:

At most workers elements are in processing simultaneously (the in-flight cap).
Results are emitted to the output channel as they become ready (order is not guaranteed).
Backpressure works: if the receiver reads out slowly, the stage must not drain in faster than it emits results (don't buffer the world in memory).
On ctx cancellation the stage stops quickly with no goroutine leak.
out is closed when in is drained (closed) and all workers have stopped.
workers < 1 is treated as 1.

Unlike the finite-slice fan-out drill (a known-length slice mapped to a result map), this is a stream — a channel of unknown/unbounded length turned into a stream of results, with ctx and backpressure. It is one stage of a pipeline.

Approach

Spin up exactly workers goroutines, all reading from the same in. Because there are exactly workers readers and each holds at most one item while it computes fn, the count of readers is the in-flight cap — no semaphore needed. Every worker loops on a two-way select:

<-ctx.Done() → return immediately (cancellation).
t, ok := <-in → if !ok, in is closed and the stream is over, so return; otherwise compute fn(ctx, t).

Emitting the result is itself a select on out <- res or <-ctx.Done(), so a slow or vanished consumer cannot pin a worker forever.

Backpressure via an unbuffered out. Because out is unbuffered, a send blocks until the consumer actually receives. A blocked worker is not looping back to read in, so a slow consumer transitively throttles input consumption — memory stays bounded by workers, not by the size of the stream.

Closing out exactly once. The workers are the senders, so none of them may close out (a send on a closed channel panics; multiple closers panic). Instead a single closer goroutine does wg.Wait() then close(out) — it runs only after every worker has returned (either in drained or ctx cancelled), which is the precise moment the invariant "in empty AND all workers stopped" is satisfied.

Why there is no leak. Every blocking point — the receive from in and the send to out — is guarded by ctx.Done(). On cancel, each worker unblocks and returns, wg falls to zero, the closer closes out, and the whole stage tears down without a parked goroutine.

Solution (Go)

// Дрилл 12 (решение). Потоковый bounded fan-out с backpressure и отменой.
package d12stream

import (
	"context"
	"sync"
)

type Result[R any] struct {
	Val R
	Err error
}

func StreamMap[T any, R any](
	ctx context.Context,
	in <-chan T,
	workers int,
	fn func(context.Context, T) (R, error),
) <-chan Result[R] {
	if workers < 1 {
		workers = 1
	}
	out := make(chan Result[R]) // небуферизованный -> backpressure от получателя

	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case t, ok := <-in:
					if !ok {
						return // in закрыт -> поток кончился
					}
					v, err := fn(ctx, t)
					select {
					case out <- Result[R]{Val: v, Err: err}:
					case <-ctx.Done():
						return
					}
				}
			}
		}()
	}

	// out закрывает один отправитель — после того, как все воркеры встали
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

Pitfalls

Buffering out "for performance" silently breaks backpressure: the stage will race ahead draining in and pile results up in the buffer. Keep out unbuffered (or size the buffer deliberately, knowing it relaxes the guarantee).
Sending to out without a ctx.Done() arm. If the consumer stops reading after cancel, a bare out <- res blocks forever and leaks the worker. Both the receive from in and the send to out need the cancel arm.
Closing out from a worker. With multiple senders this panics (double close, or send-on-closed). Exactly one goroutine, after wg.Wait(), may close it.
Forgetting wg.Add(1) before the goroutine starts (or calling it inside the goroutine) — the closer can then race to Wait() before the counter is set and close out prematurely.
A per-item semaphore instead of fixed workers. It works, but the fixed-worker design gets the in-flight cap for free — "exactly workers readers of in" already bounds concurrency, so no extra token channel is needed.
Assuming output order matches input order. Results are emitted as they finish; if the caller needs ordering, tag items with an index and reassemble downstream.
Passing ctx into fn but ignoring it there. The stage cancels promptly only if fn itself honours ctx — a long, uncancellable fn will delay teardown until it returns.

Run every drill with go test -race ./.... Related: the introductory concurrency drills cover worker pools, channels, and the finite-slice fan-out these problems build on.