Documentation
¶
Overview ¶
Package pipeline runs one shard's supply cycle: it looks at what is due, asks the planner what it may serve, fetches that, and pushes it to the candidate cache the workers drain.
One cycle has five phases, and they name everything here:
sleep -> tallying -> planning -> fetching -> pushing
There is no loop and no goroutine in this package. Cycle paces itself - it sleeps at the FRONT, for whatever remains of the interval since the last cycle began - so a caller drives it in a tight loop and holds no cadence policy of its own:
for {
select {
case <-ctx.Done():
return
default:
}
observe(p.Cycle(ctx))
}
Any delay the caller adds between calls is therefore absorbed rather than added to the period.
Cycle NEVER returns an error, only a Result carrying one: every failure it can hit is already dealt with here, and the next cycle retries. A caller reads the Result for its log line and carries on.
SetInterval and SetMinGap are live and safe to call from anywhere; Cycle itself is not safe for concurrent use - one Pipeline per shard, driven by one goroutine.
Index ¶
- Constants
- type Pipeline
- func (p *Pipeline) Cycle(ctx context.Context) Result
- func (p *Pipeline) Interval() time.Duration
- func (p *Pipeline) MinGap() time.Duration
- func (p *Pipeline) SetInterval(d time.Duration)
- func (p *Pipeline) SetMinGap(d time.Duration)
- func (p *Pipeline) Shard() int
- func (p *Pipeline) WorkingFor() time.Duration
- type Result
- type Source
Constants ¶
const ( // DefaultInterval is the starting cycle period. A caller that derives its own replaces it through // SetInterval; this only keeps a freshly-built pipeline from scanning flat out. DefaultInterval = 50 * time.Millisecond // DefaultMinGap is the starting quiet time between cycles. DefaultMinGap = 20 * time.Millisecond )
const NoBand = math.MaxInt
NoBand is the band reported when nothing is due - on this shard (Result.Band) or anywhere in the fleet (Result.GlobalBand).
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline runs one shard's supply cycle.
Cycle is NOT safe for concurrent use - one Pipeline per shard, driven by one goroutine, is the whole intended shape, and the cadence timestamps are unsynchronized on that basis. SetInterval is the one exception and may be called from anywhere.
func New ¶
func New(shard int, source Source, planner *planner.Planner, cache *candidates.Cache) (*Pipeline, error)
New returns a Pipeline for one shard, paced at DefaultInterval and DefaultMinGap until told otherwise. It is where a wiring mistake is caught, which is why Cycle itself has no error return to spend on one.
func (*Pipeline) Cycle ¶
Cycle runs one full cycle - sleeping first for whatever the cadence still owes - and reports what happened. It never returns an error; see the package doc.
func (*Pipeline) SetInterval ¶
SetInterval updates the cycle period - start of tallying to start of tallying - live. The next cycle picks it up: the value is read once per cycle rather than captured, so a caller that re-derives it takes effect without a restart. Zero paces nothing, which is what driving cycles back to back wants.
func (*Pipeline) SetMinGap ¶
SetMinGap updates the minimum quiet time between the END of one cycle and the START of the next. It is the fuse for the case the interval alone cannot cover: a cycle that outruns its interval would otherwise leave no gap at all and scan back to back, which is the duty cycle the interval exists to prevent. Zero disables it.
func (*Pipeline) WorkingFor ¶
WorkingFor is how long the current cycle has been inside its queries, or zero when none is - excluding the pace it sleeps first.
A DURATION rather than a bool, and that is the load-bearing part. It exists for a caller publishing this shard's liveness on its own clock, where a completed cycle is the ordinary evidence but one scan can outrun any sane publishing cadence on a deep backlog (phase one is O(backlog) on every dialect without the run-condition early-stop). A bool cannot serve that: a cycle whose scan fails INSTANTLY is also briefly inside its queries - building the error, recording the phase, logging it - and a caller sampling often enough will keep catching that. Measured at ~1.2% of samples with a failing scan, which is easily enough to keep a broken shard looking alive indefinitely. Only a cycle that has run longer than the caller's own expectations is evidence, and a duration lets the caller decide what that means.
Safe to call from any goroutine.
type Result ¶
type Result struct {
// Per-phase durations. Total spans tallying through pushing and EXCLUDES Slept, so it is the cycle's
// cost rather than its period; the period is Slept+Total.
Slept time.Duration
Tallying time.Duration
Planning time.Duration
Fetching time.Duration
Pushing time.Duration
Total time.Duration
// Band is this shard's own minimum due band, GlobalBand the best band any shard holds. Band >
// GlobalBand means this shard was outranked and served nothing - the ordinary strict-priority case,
// not a fault. Either is NoBand when nothing is due.
Band int
GlobalBand int
// Selected is how many candidates were pushed, Discarded how many un-popped ones that replaced.
// Discarded rising toward Selected means the cycle is turning faster than the workers drain.
Selected int
Discarded int
// Err is set when the cycle ended early, wrapped with the phase that failed. The cycle has already
// dealt with it; this is for the log line.
Err error
}
Result is one cycle's outcome, for logging and metrics. Nothing here is a control signal: a caller reads it and carries on.
type Source ¶
type Source interface {
// ScanBand reports this shard's minimum due priority band and one tally per fairness key at that
// band, or math.MaxInt and no tallies when nothing is due here. It returns O(distinct keys) rows,
// never O(backlog); each tally's Count is expected to be capped at the planning capacity.
ScanBand(ctx context.Context, shard int) (band int, tallies []planner.Tally, err error)
// FetchSteps returns up to perKey step ids for each of keys at the given band, OLDEST FIRST within
// each key. The ordering is the Source's responsibility - the cycle consumes the lists in order and
// does not re-sort them. A key may come back short or missing.
FetchSteps(ctx context.Context, shard, band int, keys []string, perKey int) (map[string][]int, error)
}
Source is the database side of a cycle - the only two queries a cycle makes, both read-only.
Source Files