package mill
import "tangled.org/core/spindle/observability"
// a workflow waiting in place for a seat. when a seat frees up, waiters bid
// one at a time, starting with the owner that has the fewest workflows on
// the fleet, so one owner's burst can't take every seat
type placeWaiter struct {
owner string
repo string
engine string
seq uint64
admitting bool
// set when a bid in triedGen found no seat. the waiter then sits out and
// stops blocking the ones behind it until something changes
tried bool
triedGen uint64
}
func (w *placeWaiter) triedSince(gen uint64) bool {
return w.tried && w.triedGen == gen
}
func (m *Mill) joinPlacement(owner, repo, engine string) *placeWaiter {
m.mu.Lock()
defer m.mu.Unlock()
m.waiterSeq++
w := &placeWaiter{owner: owner, repo: repo, engine: engine, seq: m.waiterSeq}
m.waiters[w] = struct{}{}
return w
}
func (m *Mill) leavePlacement(w *placeWaiter) {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.waiters, w)
m.wakeWaitersLocked()
}
// reports whether w may bid now, plus the generation and channel to wait on
func (m *Mill) placementTurn(w *placeWaiter) (bool, uint64, <-chan struct{}) {
m.mu.Lock()
defer m.mu.Unlock()
if w.triedSince(m.changeGen) {
return false, m.changeGen, m.changeCh
}
running := m.runningByOwnerLocked()
for v := range m.waiters {
if v == w || v.engine != w.engine || v.admitting || v.triedSince(m.changeGen) {
continue
}
if running[v.owner] < running[w.owner] || (running[v.owner] == running[w.owner] && v.seq < w.seq) {
return false, m.changeGen, m.changeCh
}
}
return true, m.changeGen, m.changeCh
}
// the bid found no seat, so let the next waiter try
func (m *Mill) passTurn(w *placeWaiter, gen uint64) {
m.mu.Lock()
defer m.mu.Unlock()
w.tried = true
w.triedGen = gen
m.wakeWaitersLocked()
}
func (m *Mill) setAdmitting(w *placeWaiter, admitting bool) {
m.mu.Lock()
defer m.mu.Unlock()
w.admitting = admitting
if !admitting {
m.wakeWaitersLocked()
}
}
// admitting waiters count too because they're about to hold a seat
func (m *Mill) runningByOwnerLocked() map[string]int {
running := make(map[string]int)
for _, lease := range m.leases {
running[lease.ownerDID]++
}
for w := range m.waiters {
if w.admitting {
running[w.owner]++
}
}
return running
}
// what each repo has leased or waiting right now, for the dashboard
func (m *Mill) workflowCounts() []observability.MillWorkflows {
type key struct{ state, owner, repo string }
counts := make(map[key]int)
m.mu.Lock()
for _, lease := range m.leases {
counts[key{"running", lease.ownerDID, lease.repoDID}]++
}
for w := range m.waiters {
state := "waiting_seat"
if w.admitting {
state = "waiting_quota"
}
counts[key{state, w.owner, w.repo}]++
}
m.mu.Unlock()
names := make(map[string]string)
out := make([]observability.MillWorkflows, 0, len(counts))
for k, n := range counts {
name, ok := names[k.repo]
if !ok {
name = m.db.RepoName(k.repo)
names[k.repo] = name
}
out = append(out, observability.MillWorkflows{State: k.state, OwnerDID: k.owner, RepoDID: k.repo, Repo: name, Count: n})
}
return out
}