Documentation
¶
Overview ¶
Package workers holds the demand side of dispatch: a CREW of goroutines that pop step candidates from the cache and hand each to one callback. The supply side - selecting which candidates are there to be popped - is elsewhere.
The crew knows nothing about steps, flows, or databases. It owns three things and no more: how many goroutines exist, when one more is worth spawning, and how they shut down without racing each other. What "processing a candidate" means is the caller's, passed in as a func.
crew, err := workers.New(cache, gate, process) crew.SetMax(ceiling) crew.Start(ctx, resident) ... cache.Close() // releases the goroutines parked waiting for work gate.Close() // releases the goroutines waiting on a permit crew.Drain()
It is GROW-ON-DEMAND, and the whole rule is one question asked at one place: a worker that has just taken a candidate adds a peer if NOBODY is left idle. So the crew holds a standing reserve of one, and that reserve is what makes a single edge trigger sufficient - see considerGrowth.
It SHRINKS the same way - locally, with no coordinator: a worker that has held a candidate for too little of its own recent wall clock retires on a coin flip, never below Min. See considerRetirement.
The gate belongs to the caller: the crew holds *a* gate without knowing what it gates, takes a permit before removing work from the cache, and hands it back when the handler says so. It is deliberately NOT consulted by the growth rule; blocking on it is what the rule reads instead, since a worker waiting for a permit is counted idle.
The cache and the gate, not ctx, are the stop signals: close them, then Drain. Every Set* is safe to call from anywhere at any time; Start and Drain are the caller's lifecycle and must not overlap.
Index ¶
- type Crew
- func (c *Crew) Drain()
- func (c *Crew) Idle() int
- func (c *Crew) Max() int
- func (c *Crew) Min() int
- func (c *Crew) Resident() int
- func (c *Crew) SetGateWaitObserver(f func(shard int, waited time.Duration))
- func (c *Crew) SetLogger(l *slog.Logger)
- func (c *Crew) SetMax(n int)
- func (c *Crew) Start(ctx context.Context, resident int)
- type Gate
- type ProcessFunc
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Crew ¶
type Crew struct {
// contains filtered or unexported fields
}
Crew is a grow-on-demand set of goroutines draining one candidate cache through one gate.
Every Set* is safe to call from anywhere at any time. Start and Drain are the caller's lifecycle and must not overlap.
func New ¶
func New(cache *candidates.Cache, gate Gate, process ProcessFunc) (*Crew, error)
New returns a crew over a cache, the gate that bounds its concurrency, and the callback that handles what it pops. Nothing is spawned until Start.
func (*Crew) Drain ¶
func (c *Crew) Drain()
Drain closes the crew to new goroutines and waits for every existing one to finish. The caller must close the cache AND the gate FIRST, or this blocks forever on goroutines parked in one of them.
Closing before waiting is the load-bearing half: a worker can try to spawn a peer at any instant, and an Add racing a Wait panics. Only workers spawn, so the flag alone covers it - there is no separate grower goroutine to stop first.
func (*Crew) Resident ¶
Resident is how many goroutines exist. It rises on demand and falls as workers retire, never below Min.
func (*Crew) SetGateWaitObserver ¶
SetLogger sets the logger. Nil restores the discarding default. SetGateWaitObserver registers a callback invoked with how long each successful Acquire blocked. It is optional and unset by default (the crew emits no metrics - instrument names are a public surface, so they belong to the owner), and it is called on EVERY acquire including the uncontended ones, so an observer that counts sees acquires and one that sums sees the mean.
It exists because the gate's own free-permit count is instantaneous: a reservation saturated for a whole window without ever being sampled empty is indistinguishable from an idle one. The wait is the durable half. The callback runs on the worker's own goroutine, before it takes any work, so it must not block.
func (*Crew) SetMax ¶
SetMax caps how far the crew may grow. Live: it is read on every spawn decision rather than captured, so a caller that re-derives it (from a connection budget, a fleet count, anything) takes effect at once.
It retires nothing itself - lowering it stops growth rather than forcing a shrink. A crew above a lowered Max comes down only as its own workers measure themselves surplus, and only as far as Min.
func (*Crew) Start ¶
Start spawns the resident set and returns immediately; the goroutines run until the cache closes.
resident is separate from Max because they answer different questions. The resident set is what the caller wants running unconditionally - sized from whatever throughput it expects to sustain - while Max is the ceiling growth may reach, which is typically far larger and deliberately never spawned up front. It is ALSO Min, the floor retirement may not cross - which is what makes a caller passing resident == Max (a pinned crew) opt out of both growth and retirement without having to say so.
The stop signals are the cache and the gate, not ctx, and that is not an oversight. A goroutine with no candidate to run is blocked in one of them, which nothing but a close will release; and ctx here is the one handed to process, which the caller usually wants live until after every in-flight handler has committed its work. So the caller closes both and then calls Drain.
type Gate ¶
type Gate interface {
Acquire(ctx context.Context, shard int) (gated context.Context, release func(), ok bool)
}
Gate bounds how many goroutines may be working against whatever resource actually binds. The crew treats it opaquely: take a permit before removing work from the cache, hand it back when the handler says so.
Acquire blocks, and reports !ok only when the gate CLOSES - one of the two stop signals. The crew only ever ENTERS: whatever the gate does for work on its way out is the handler's business, not the crew's.
It returns a context, which the crew passes to the handler in place of the one it was given. That is how a gate that carries per-unit-of-work state - an admission time, a band, whatever identifies this work to the resource - gets it to the handler without the crew knowing any of it exists. A gate with nothing to add returns the context unchanged.
One method, and no capacity query, is the whole interface on purpose. The crew never asks whether the gate has room; it lets a worker BLOCK and reads that instead, because a worker waiting for a permit holds no candidate and so counts idle - which stops growth at its own next check. Saturation therefore needs no reporting, and a gate that meters more than one thing needs no say in how it is summarised.
type ProcessFunc ¶
ProcessFunc handles one popped candidate. It is called on a crew goroutine, and the crew cares about nothing it does except that it eventually returns - an error is logged, a panic is caught, and either way the goroutine goes back for the next candidate.
release hands the gate's permit back. The HANDLER owns when, because only it knows the point in its own work at which the bounded resource stops being held - typically just before a long call that holds nothing. It is safe to call more than once and safe never to call: the crew wraps it in sync.OnceFunc and calls it unconditionally on the way out, so no early return can leak a permit.
Source Files