~james/core

core/spindle/mill/fair.go -rw-r--r-- 3.1 KiB
452f17b7 — james chore(import): snapshot james.dad.town/core at f70c71c7696a 16 hours ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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
}