Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120package 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 seattype 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 onfunc (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 tryfunc (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 seatfunc (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 dashboardfunc (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}