src/queue.bend
bend-kube-runtime@0.1.1.0 · 262 lines · 0 definitions
1import Base
2
3# Queue
4# =====
5#
6# The controller's work queue, with client-go's workqueue semantics and
7# kube-runtime's scheduler folded in:
8#
9# - A key waits in the queue at most once (dedup).
10# - A key is processed by at most one worker at a time: adding a key that
11# is running marks it dirty, and it is queued again when its run ends.
12# - At most max keys run at once.
13# - A run ends with an Action: requeue after a delay, wait for a change, or
14# retry after an error with per-key exponential backoff (client-go's
15# ItemExponentialFailureRateLimiter: 5ms doubling, at most 1000s); a
16# success forgets the key's failures.
17# - Delayed requeues are timers the driver sleeps on; each carries the
18# key's generation, and an add (or a newer timer) bumps it, so a stale
19# timer wakes to nothing.
20
21type Action is Data:
22 Requeue{ms: U32}
23 AwaitChange{}
24 Retry{msg: String}
25
26type Queue is Data:
27 Queue{front: List<&2, String>, back: List<&2, String>, inq: Map<&2, U32>,
28 running: Map<&2, U32>, dirty: Map<&2, U32>, gens: Map<&2, U32>, fails: Map<&2, U32>,
29 nrun: U32, max: U32}
30
31# A timer for the driver to start: wake key after ms with this generation.
32type Timer is Data:
33 Timer{key: String, gen: U32, ms: U32}
34
35def Queue.new(+max: U32) -> Queue:
36 Queue{Nil{}, Nil{}, Map.new(&2, U32), Map.new(&2, U32), Map.new(&2, U32), Map.new(&2, U32),
37 Map.new(&2, U32), 0, max}
38
39# Maps
40# ----
41
42def Mp.has.fin(r: Map<&2, U32> & Bool) -> Bool:
43 match r:
44 case (m, b):
45 b
46
47def Mp.has(+m: Map<&2, U32>, key: String) -> Bool:
48 Mp.has.fin(Map.has(&2, U32, m, key))
49
50def Mp.get.fin(r: Map<&2, U32> & U32) -> U32:
51 match r:
52 case (m, v):
53 v
54
55def Mp.get(+m: Map<&2, U32>, key: String) -> U32:
56 Mp.get.fin(Map.get(U32, 0, m, key))
57
58def Mp.on(m: Map<&2, U32>, key: String) -> Map<&2, U32>:
59 Map.set(&2, U32, m, key, 1)
60
61def Mp.off(m: Map<&2, U32>, key: String) -> Map<&2, U32>:
62 Map.del(&2, U32, m, key)
63
64def Mp.bump(+m: Map<&2, U32>, +key: String) -> Map<&2, U32>:
65 Map.set(&2, U32, m, key, (Mp.get(m, key) + 1 : U32))
66
67def Queue.add.go(f: List<&2, String>, b: List<&2, String>, inq: Map<&2, U32>,
68 run: Map<&2, U32>, d: Map<&2, U32>, g: Map<&2, U32>, fl: Map<&2, U32>, n: U32, mx: U32,
69 +key: String, running: Bool, waiting: Bool) -> Queue:
70 match running waiting:
71 case True{} _:
72 Queue{f, b, inq, run, Mp.on(d, key), g, fl, n, mx}
73 case False{} True{}:
74 Queue{f, b, inq, run, d, g, fl, n, mx}
75 case False{} False{}:
76 Queue{f, key <> b, Mp.on(inq, key), run, d, g, fl, n, mx}
77
78# Adding
79# ------
80
81# Queues key unless it waits already; a running key is marked dirty
82# instead. Either way its pending timer, if any, goes stale.
83def Queue.add(q: Queue, +key: String) -> Queue:
84 match q:
85 case Queue{f, b, +inq, +run, d, +g, fl, n, mx}:
86 Queue.add.go(f, b, inq, run, d, Mp.bump(g, key), fl, n, mx, key, Mp.has(run, key),
87 Mp.has(inq, key))
88
89def Queue.add_all(ks: List<&2, String>, q: Queue) -> Queue:
90 match ks:
91 case Nil{}:
92 q
93 case Con{k, t}:
94 Queue.add_all(t, Queue.add(q, k))
95
96def Queue.pop.flip(f: List<&2, String>, i: Map<&2, U32>, r: Map<&2, U32>, d: Map<&2, U32>,
97 g: Map<&2, U32>, fl: Map<&2, U32>, +n: U32, mx: U32) -> Queue & Maybe<&2, String>:
98 match f:
99 case Con{+k, t}:
100 (Queue{t, Nil{}, Mp.off(i, k), Mp.on(r, k), d, g, fl, (n + 1 : U32), mx}, Some{k})
101 case Nil{}:
102 (Queue{Nil{}, Nil{}, i, r, d, g, fl, n, mx}, None{})
103
104def Queue.pop.go(f: List<&2, String>, b: List<&2, String>, i: Map<&2, U32>, r: Map<&2, U32>,
105 d: Map<&2, U32>, g: Map<&2, U32>, fl: Map<&2, U32>, +n: U32, mx: U32, free: Bool) ->
106 Queue & Maybe<&2, String>:
107 match f free:
108 case Con{+k, t} True{}:
109 (Queue{t, b, Mp.off(i, k), Mp.on(r, k), d, g, fl, (n + 1 : U32), mx}, Some{k})
110 case Nil{} True{}:
111 Queue.pop.flip(List.reverse(&2, String, b), i, r, d, g, fl, n, mx)
112 case xs False{}:
113 (Queue{xs, b, i, r, d, g, fl, n, mx}, None{})
114
115# Taking
116# ------
117
118# The next waiting key, if a worker slot is free: it leaves the queue and
119# starts running.
120def Queue.pop(q: Queue) -> Queue & Maybe<&2, String>:
121 match q:
122 case Queue{f, b, i, r, d, g, fl, +n, +mx}:
123 Queue.pop.go(f, b, i, r, d, g, fl, n, mx, U32.is_lt(n, mx))
124
125def Queue.pop_all.q(r: Queue & Maybe<&2, String>) -> Queue:
126 match r:
127 case (q, m):
128 q
129
130def Queue.pop_all.acc(r: Queue & Maybe<&2, String>, acc: List<&2, String>) -> List<&2, String>:
131 match r:
132 case (q, Some{k}):
133 k <> acc
134 case (q, None{}):
135 acc
136
137# Every key that may start now, in queue order (fuel bounds the count).
138def Queue.pop_all(fuel: Nat, +q: Queue, acc: List<&2, String>) -> Queue & List<&2, String>:
139 match fuel:
140 case 0n:
141 (q, List.reverse(&2, String, acc))
142 case 1n+p:
143 Queue.pop_all(p, Queue.pop_all.q(Queue.pop(q)), Queue.pop_all.acc(Queue.pop(q), acc))
144
145def Queue.backoff.go(+fails: U32, capped: Bool) -> U32:
146 match capped:
147 case True{}:
148 1000000
149 case False{}:
150 U32.min((5 << U32.to_nat(fails) : U32), 1000000)
151
152# Finishing
153# ---------
154
155# The delay after fails failures: 5ms doubling, at most 1000s.
156def Queue.backoff(+fails: U32) -> U32:
157 Queue.backoff.go(fails, U32.is_ge(fails, 18))
158
159def Queue.done.fails(+fl: Map<&2, U32>, +key: String, a: Action) -> Map<&2, U32>:
160 match a:
161 case Retry{m}:
162 Mp.bump(fl, key)
163 case _:
164 Mp.off(fl, key)
165
166def Queue.done.dirty(q: Queue, +key: String, dirty: Bool) -> Queue:
167 match dirty:
168 case True{}:
169 Queue.add(q, key)
170 case False{}:
171 q
172
173# Starts a timer for key, staling any earlier one.
174def Queue.arm(q: Queue, +key: String, ms: U32) -> Queue & List<&2, Timer>:
175 match q:
176 case Queue{f, b, i, r, d, +g, fl, n, mx}:
177 (Queue{f, b, i, r, d, Mp.bump(g, key), fl, n, mx},
178 [Timer{key, (Mp.get(g, key) + 1 : U32), ms}])
179
180def Queue.done.timer(q: Queue, +key: String, a: Action, +fails: U32) -> Queue & List<&2, Timer>:
181 match a:
182 case Requeue{ms}:
183 Queue.arm(q, key, ms)
184 case AwaitChange{}:
185 (q, Nil{})
186 case Retry{m}:
187 Queue.arm(q, key, Queue.backoff(fails))
188
189def Queue.done.run(q: Queue, +key: String, +a: Action) -> Queue & List<&2, Timer>:
190 match q:
191 case Queue{f, b, i, r, +d, g, +fl, +n, mx}:
192 Queue.done.timer(
193 Queue.done.dirty(Queue{f, b, i, Mp.off(r, key), Mp.off(d, key), g,
194 Queue.done.fails(fl, key, a), (n - 1 : U32), mx}, key, Mp.has(d, key)),
195 key, a, Mp.get(fl, key))
196
197def Queue.done.go(q: Queue, +key: String, +a: Action, running: Bool) -> Queue & List<&2, Timer>:
198 match running:
199 case True{}:
200 Queue.done.run(q, key, a)
201 case False{}:
202 (q, Nil{})
203
204# Observers
205# ---------
206
207def Queue.running(q: Queue, key: String) -> Bool:
208 match q:
209 case Queue{f, b, i, r, d, g, fl, n, mx}:
210 Mp.has(r, key)
211
212# A run of key ended with an action. The key stops running; a dirty key is
213# queued again at once; the action may ask for a timer. A key that is not
214# running changes nothing.
215def Queue.done(+q: Queue, +key: String, +a: Action) -> Queue & List<&2, Timer>:
216 Queue.done.go(q, key, a, Queue.running(q, key))
217
218def Queue.wake.go(q: Queue, +key: String, current: Bool) -> Queue:
219 match current:
220 case True{}:
221 Queue.add(q, key)
222 case False{}:
223 q
224
225# A timer woke: the key is queued if the timer is still current.
226def Queue.wake(q: Queue, +key: String, +gen: U32) -> Queue:
227 match q:
228 case Queue{f, b, i, r, d, +g, fl, n, mx}:
229 Queue.wake.go(Queue{f, b, i, r, d, g, fl, n, mx}, key, U32.is_eq(Mp.get(g, key), gen))
230
231def Queue.waiting(q: Queue, key: String) -> Bool:
232 match q:
233 case Queue{f, b, i, r, d, g, fl, n, mx}:
234 Mp.has(i, key)
235
236def Queue.nrun(q: Queue) -> U32:
237 match q:
238 case Queue{f, b, i, r, d, g, fl, n, mx}:
239 n
240
241# Forgetting
242# ----------
243
244def Queue.forget.go(q: Queue, +key: String, busy: Bool) -> Queue:
245 match q busy:
246 case x True{}:
247 x
248 case Queue{f, b, i, r, d, g, fl, n, mx} False{}:
249 Queue{f, b, i, r, d, Mp.off(g, key), Mp.off(fl, key), n, mx}
250
251# Drops what the queue remembers of a key whose object is gone (its timer
252# generation and failures, as client-go's Forget does), so memory follows
253# the live objects. A key still waiting or running keeps its state.
254def Queue.forget(+q: Queue, +key: String) -> Queue:
255 Queue.forget.go(q, key, Bool.or(Queue.running(q, key), Queue.waiting(q, key)))
256
257def Queue.forget_all(ks: List<&2, String>, q: Queue) -> Queue:
258 match ks:
259 case Nil{}:
260 q
261 case Con{k, t}:
262 Queue.forget_all(t, Queue.forget(q, k))