// Package work bounds concurrent expensive operations.
//
// Nothing in quarry limited concurrency: every zit subprocess, every
// buffered archive, ran in the request goroutine with the arrival rate
// as the only bound. That is fine until it is not — the file page
// saturated at 32 requests per second and then went BACKWARDS, serving
// 24 while latency quadrupled, because each request forked a process
// competing with the others.
//
// A Lane is the smallest thing that fixes it: a concurrency limit plus a
// queue cap, and a refusal past the cap. Refusing matters as much as
// limiting. A request accepted into a queue it cannot be served from in
// time consumes the capacity that would have served someone else, so
// past the cap a Lane sheds rather than accepts — the caller gets a 503
// it can retry, not a timeout it cannot distinguish from a hang.
//
// Lanes are separate rather than one global pool because the operations
// have wildly different costs: a 256 MiB archive and a 3 ms name lookup
// in one pool means the archive decides how many name lookups happen.
package work
import (
"context"
"errors"
"sort"
"sync"
"sync/atomic"
"time"
"craftdesign.group/quarry/internal/fault"
)
// ErrBusy reports that a lane is at its queue cap. Callers map it to a
// retryable refusal; ops.ErrUnavailable is the ops-layer spelling.
//
// It stays a bare sentinel because errors.Is compares against it. What
// reaches the journal is busy(), which wraps it with the numbers that
// make the refusal actionable — the sentinel alone said "work: lane is
// saturated" 833 times, naming no lane, no depth and no repository,
// while the lane raising it knew all three.
var ErrBusy = errors.New("work: lane is saturated")
// busy is ErrBusy with this lane's load attached. Building it costs an
// allocation on a path that is already shedding, which is the cheapest
// moment in the system to spend one.
func (l *Lane) busy() error {
st := l.Stats()
return fault.Wrap(ErrBusy, l.name+" lane is full").
Cost("work is being refused rather than queued; callers are told to retry").
Do("raise this lane's limit or queue, or add a node — a shed count that keeps climbing is capacity, not a bug").
With("lane", st.Name, "running", st.InFlight, "limit", st.Limit,
"waiting", st.Waiting, "queue", st.Queue, "shed", st.Shed)
}
// Lane admits at most `limit` operations at once and queues at most
// `queue` more before refusing.
type Lane struct {
name string
slots chan struct{}
waiting atomic.Int64
maxWait int64
// Counters for the admin surface. The reason this package exists is
// to make load legible, so the numbers are part of the contract, not
// debug instrumentation.
admitted atomic.Int64
shed atomic.Int64
inFlight atomic.Int64
// What the lane is holding, for the panel that has to explain a
// refusal. Counters say how much; these say what.
itemMu sync.Mutex
items map[int64]*item
seq atomic.Int64
}
// NewLane builds a lane. limit is how many run concurrently; queue is
// how many may WAIT beyond that before the lane sheds. A queue of zero
// means "run limit at a time, refuse the rest immediately".
func NewLane(name string, limit, queue int) *Lane {
if limit < 1 {
limit = 1
}
if queue < 0 {
queue = 0
}
l := &Lane{name: name, slots: make(chan struct{}, limit), maxWait: int64(queue)}
return l
}
// Do runs fn if the lane can admit it, blocking while a slot frees.
// It returns ErrBusy without running fn when the queue is full, and the
// context's error if the caller gives up while waiting.
func (l *Lane) Do(ctx context.Context, fn func() error) error {
return l.DoNamed(ctx, "", "", fn)
}
// DoNamed is Do with the work named, so the lane can say what it is
// holding rather than only how much.
//
// verb is what the unit does ("ingest", "archive"); subject is what it
// does it to, normally a repository. Both are for a reader: a lane at
// its limit is a number, and "three ingests, the oldest of them pkgsrc
// for eleven minutes" is a diagnosis.
//
// The bookkeeping takes a mutex on a path that already blocks on a
// channel, which is affordable because these lanes admit a handful of
// units at a time and each runs for seconds — the store-subprocess gate
// with its per-request traffic is a different mechanism for that
// reason. An unnamed unit still runs; it simply has nothing to show.
func (l *Lane) DoNamed(ctx context.Context, verb, subject string, fn func() error) error {
return l.DoNamedAside(ctx, verb, subject, func(func(func() error) error) error { return fn() })
}
// DoNamedAside is DoNamed for a unit that may have to wait on something
// other than this lane partway through — an ingest round queued for a
// rebuild slot. fn is handed aside: aside(wait) gives the unit's slot
// and its queue position back for the length of wait, so ordinary units
// run meanwhile, and takes a slot again before returning — without the
// queue cap, since the unit was already admitted. If wait fails, or ctx
// ends while the unit waits to get back in, aside returns that error
// without the slot, and fn must return without doing more lane work.
// aside is called only from fn's own goroutine.
func (l *Lane) DoNamedAside(ctx context.Context, verb, subject string, fn func(aside func(wait func() error) error) error) error {
if l == nil {
return fn(func(wait func() error) error { return wait() })
}
// Reserve a waiting position first, so the check and the wait cannot
// race: without this, N goroutines could each observe a short queue
// and then all wait.
if l.waiting.Add(1) > l.maxWait+int64(cap(l.slots)) {
l.waiting.Add(-1)
l.shed.Add(1)
return l.busy()
}
// Tracked from the moment it QUEUES, not from admission: a unit
// waiting behind three others is the thing an operator is looking
// for, and one only listed once it runs would make a full queue
// look like an empty one.
id := l.track(verb, subject)
defer l.untrack(id)
counted, held := true, false
defer func() {
if held {
l.inFlight.Add(-1)
<-l.slots
}
if counted {
l.waiting.Add(-1)
}
}()
enter := func() error {
select {
case l.slots <- struct{}{}:
case <-ctx.Done():
return ctx.Err()
}
held = true
l.inFlight.Add(1)
l.mark(id, true)
return nil
}
if err := enter(); err != nil {
return err
}
l.admitted.Add(1)
aside := func(wait func() error) error {
held = false
l.inFlight.Add(-1)
<-l.slots
counted = false
l.waiting.Add(-1)
l.mark(id, false)
if err := wait(); err != nil {
return err
}
counted = true
l.waiting.Add(1)
return enter()
}
return fn(aside)
}
// Item is one unit the lane is holding: what it is, what it is for, how
// long it has been there, and whether it is running or still queued.
type Item struct {
Verb string `json:"verb"`
Subject string `json:"subject,omitempty"`
Age time.Duration `json:"-"`
Running bool `json:"running"`
}
type item struct {
verb, subject string
since time.Time
running bool
}
func (l *Lane) track(verb, subject string) int64 {
id := l.seq.Add(1)
l.itemMu.Lock()
defer l.itemMu.Unlock()
if l.items == nil {
l.items = map[int64]*item{}
}
l.items[id] = &item{verb: verb, subject: subject, since: time.Now()}
return id
}
func (l *Lane) mark(id int64, running bool) {
l.itemMu.Lock()
defer l.itemMu.Unlock()
if it, ok := l.items[id]; ok {
it.running = running
}
}
func (l *Lane) untrack(id int64) {
l.itemMu.Lock()
defer l.itemMu.Unlock()
delete(l.items, id)
}
// Items is what the lane is holding, oldest first — running before
// queued at the same age, because the running ones are what the queued
// ones are waiting for.
func (l *Lane) Items() []Item {
if l == nil {
return nil
}
l.itemMu.Lock()
out := make([]Item, 0, len(l.items))
now := time.Now()
for _, it := range l.items {
out = append(out, Item{Verb: it.verb, Subject: it.subject,
Age: now.Sub(it.since), Running: it.running})
}
l.itemMu.Unlock()
sort.Slice(out, func(i, j int) bool {
if out[i].Running != out[j].Running {
return out[i].Running
}
if out[i].Age != out[j].Age {
return out[i].Age > out[j].Age
}
return out[i].Subject < out[j].Subject
})
return out
}
// Stats is a lane's load, for the fleet admin surface.
type Stats struct {
Name string `json:"name"`
Limit int `json:"limit"`
Queue int `json:"queue"`
InFlight int64 `json:"inFlight"`
Waiting int64 `json:"waiting"`
Admitted int64 `json:"admitted"`
// Shed counts refusals. A nonzero and growing value is the signal
// that this node is out of capacity — the number that turns "it
// feels slow" into "add a node".
Shed int64 `json:"shed"`
}
func (l *Lane) Stats() Stats {
if l == nil {
return Stats{}
}
waiting := l.waiting.Load() - l.inFlight.Load()
if waiting < 0 {
waiting = 0
}
return Stats{
Name: l.name, Limit: cap(l.slots), Queue: int(l.maxWait),
InFlight: l.inFlight.Load(), Waiting: waiting,
Admitted: l.admitted.Load(), Shed: l.shed.Load(),
}
}