1	import 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	
21	type Action is Data:
22	  Requeue{ms: U32}
23	  AwaitChange{}
24	  Retry{msg: String}
25	
26	type 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.
32	type Timer is Data:
33	  Timer{key: String, gen: U32, ms: U32}
34	
35	def 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	
42	def Mp.has.fin(r: Map<&2, U32> & Bool) -> Bool:
43	  match r:
44	    case (m, b):
45	      b
46	
47	def Mp.has(+m: Map<&2, U32>, key: String) -> Bool:
48	  Mp.has.fin(Map.has(&2, U32, m, key))
49	
50	def Mp.get.fin(r: Map<&2, U32> & U32) -> U32:
51	  match r:
52	    case (m, v):
53	      v
54	
55	def Mp.get(+m: Map<&2, U32>, key: String) -> U32:
56	  Mp.get.fin(Map.get(U32, 0, m, key))
57	
58	def Mp.on(m: Map<&2, U32>, key: String) -> Map<&2, U32>:
59	  Map.set(&2, U32, m, key, 1)
60	
61	def Mp.off(m: Map<&2, U32>, key: String) -> Map<&2, U32>:
62	  Map.del(&2, U32, m, key)
63	
64	def 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	
67	def 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.
83	def 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	
89	def 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	
96	def 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	
104	def 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.
120	def 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	
125	def Queue.pop_all.q(r: Queue & Maybe<&2, String>) -> Queue:
126	  match r:
127	    case (q, m):
128	      q
129	
130	def 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).
138	def 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	
145	def 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.
156	def Queue.backoff(+fails: U32) -> U32:
157	  Queue.backoff.go(fails, U32.is_ge(fails, 18))
158	
159	def 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	
166	def 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.
174	def 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	
180	def 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	
189	def 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	
197	def 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	
207	def 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.
215	def Queue.done(+q: Queue, +key: String, +a: Action) -> Queue & List<&2, Timer>:
216	  Queue.done.go(q, key, a, Queue.running(q, key))
217	
218	def 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.
226	def 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	
231	def 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	
236	def 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	
244	def 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.
254	def Queue.forget(+q: Queue, +key: String) -> Queue:
255	  Queue.forget.go(q, key, Bool.or(Queue.running(q, key), Queue.waiting(q, key)))
256	
257	def 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))