src/queue.bend

bend-kube-runtime@0.2.0.0 · 262 lines · 37 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))