Documentation
¶
Overview ¶
Package candidates holds the per-replica bounded set of step candidates produced by the engine's refillers. It is a hint cache, not a work queue: entries confer no ownership, so a stale or duplicated candidate is harmless. The engine claims the underlying step via CAS before running it.
Index ¶
- type Cache
- func (c *Cache) Capacity() int
- func (c *Cache) Close()
- func (c *Cache) Init(workers int)
- func (c *Cache) Len() int
- func (c *Cache) Offer(j Job, priority int) (admitted bool)
- func (c *Cache) PeekShard() (shard int, ok bool)
- func (c *Cache) Pop() (j Job, ok bool, needRefill bool)
- func (c *Cache) Refill(shard int, batch []Job, floor int) (discarded int)
- func (c *Cache) Resize(workers int)
- func (c *Cache) TryPopFrom(shard int) (j Job, ok bool, needRefill bool)
- func (c *Cache) WaitForWork() (shard int, ok bool)
- type Job
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Cache ¶
type Cache struct {
// contains filtered or unexported fields
}
Cache is a bounded per-replica set of step candidates - hints, not ownership - partitioned by shard. Each shard's refiller wholesale-replaces its own partition; workers Pop from whichever partition advertises the best (lowest) floor. One mutex and one condition variable span all partitions, so a worker blocked on an empty cache wakes when ANY partition fills - N separate cond vars would let a worker sleep through another shard's refill.
func (*Cache) Capacity ¶
Capacity returns the cache's global bound (twice the worker count). It bounds the SUM of the partitions, and it is the capacity each refiller draws its global plan at.
func (*Cache) Close ¶
func (c *Cache) Close()
Close permanently unblocks all waiters. Offer/Refill become no-ops after close.
func (*Cache) Offer ¶
Offer admits a candidate onto its shard's partition (routing is j.Shard) without consulting the plan. Callers are step-origination sites, most often the successor of a step that just completed, so what it answers is whether this replica can run the step now or must wait for a cycle to select it.
Three rules, and each is load-bearing:
- An EMPTY partition admits REGARDLESS of the global bound. That bound is a sum over all partitions, so gating this on it lets one busy shard silence an idle one's doorbell - and a caller reads a decline as "this partition needs nothing", so the step then waits for a scan nobody asked for.
- A WORSE band than the partition's is DECLINED. Admitting it would run band-900 work while band-100 work sits cached right here, a strict-priority inversion rather than the soft staleness the design accepts. Note the direction: a BETTER band is harmless and simply appends.
- Otherwise admit while a slot has been freed (totalLen < size), at the TAIL so nothing the plan chose is reordered. A better band is deliberately not head-inserted; preempting was built, measured as nothing, and removed.
Admissions are counted (partition.offered) so Refill's discard count stays the refiller's own waste.
func (*Cache) PeekShard ¶
PeekShard reports which shard Pop would drain right now, without blocking and without removing anything. ok is false when every partition is empty or the cache is closed.
It is the non-blocking twin of WaitForWork, and exists for the demand side's growth decision: "is there work waiting, and on which shard" is one question, and answering it with two calls would let the two halves disagree. A HINT either way - the answer can be stale before the caller acts on it, which there costs one goroutine that parks harmlessly.
func (*Cache) Pop ¶
Pop removes and returns the front candidate of the best partition, blocking until one is available or the cache closes. Partition selection is lowest floor; ties break by depth (deepest first - round-robin would hand every shard an equal share of workers regardless of backlog), then by lower shard index for determinism.
needRefill signals that the POPPED partition has drained to its low-water mark, so the caller should nudge that shard's refiller (the popped job's Shard names it) and draining overlaps refill.
func (*Cache) Refill ¶
Refill replaces one shard's partition with batch at the given priority floor and wakes up to ONE WAITER PER ADMITTED CANDIDATE - never all of them, since a wake beyond the batch cannot be given anything to do. The floor is stamped onto every batch job, which is what keeps the partition's floor exact.
The replacement is WHOLESALE, and the empty case must not be short-circuited. An empty batch is the scan's statement that nothing is due on that shard, so every candidate still cached there is a dead hint a worker would pop and burn a claim-CAS round trip on. An early return on len(batch)==0 - what this once did - keeps the whole dead batch, and nothing breaks loudly: the cache self-corrects within a cycle, so it reads as wasted work rather than a wedge, which is why the bug was invisible.
The caller must NOT route a FAILED scan here. An error means "unknown", not "nothing is due", and replacing a healthy partition with nothing on a transient blip idles its workers in Pop.
discarded counts the candidates thrown away un-popped, MINUS those the doorbell admitted - it is the refiller's oversupply signal, and charging it for work it never selected would read worst exactly when the doorbell is working best.
func (*Cache) Resize ¶
Resize re-bounds the cache to a new worker count, live, and applies to ALL partitions: when the sum overflows the new bound, each partition's tail is trimmed proportionally to its depth (largest-remainder rounding, lower shard first on ties, so the trim is deterministic). A trimmed candidate stays `pending` in the database and is simply re-selected - exactly what already happens when an Offer's head-insert pushes one past the bound - so this adds no new state, only a smaller bound.
The engine calls this when the observed replica count changes and each shard's connection budget is re-divided. The cache MUST follow that split: it is sized from what the replica can actually CLAIM, and one still holding a cache sized for the WHOLE fleet's budget is handed far more candidates than it can claim - stale hints whose claim CAS loses to a peer, and wasted round-trips, precisely when the fleet is busiest.
func (*Cache) TryPopFrom ¶
TryPopFrom removes and returns the front candidate of ONE shard's partition without blocking. ok is false when that partition is empty (lost the race, or it drained) or the cache is closed.
THE TWO CASES ARE DELIBERATELY NOT DISTINGUISHED HERE, because the caller must treat them identically: retry the park. It is WaitForWork that reports a close, and a worker that returned on an empty partition instead of looping would erode the crew under exactly the contention that caused the race - on a signal that says nothing about whether that worker was surplus.
needRefill signals that this partition has drained to its low-water mark, exactly as Pop's does.
func (*Cache) WaitForWork ¶
WaitForWork blocks until some partition holds a candidate, and reports which shard Pop would drain. ok is false ONLY when the cache CLOSES, so a worker parked here distinguishes "drained" from "nothing yet" without a second signal - the same contract Pop has always had.
It exists so a worker can park holding NOTHING. A worker that took a permit and then blocked waiting for work would hoard admission capacity it is not using, so the order must be park, then acquire, then take work non-blockingly - which is what splits the old Pop into this plus TryPopFrom.
The shard is a HINT and nothing more. By the time the caller acts on it another worker may have taken the peeked entry and a better-banded arrival may have landed, which is why the pop that follows must tolerate finding the partition empty.
type Job ¶
Job holds a step ID, its shard index, and its priority band for the worker pool. Priority is assigned by the cache itself (Refill stamps its floor onto every batch job; Offer stamps the offered priority onto the single job it admits), so a partition's floor is always derivable from its head.
Source Files