src/coordinator/RequestQueue.bend
mylsm-lsm-store@0.6.0.0 · 481 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/coordinator/QueuePolicy.bend as Policy
3import mylsm-lsm-store@0.6.0.0/src/coordinator/Protocol.bend as Protocol
4import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5
6# Owns one accepted coordinator request until dispatch or rejection.
7type Pending is Type:
8 Pending{connection_id: U32, accepted_sequence: Nat,
9 deadline_ms: Nat, encoded_bytes: Nat, response_peak: Nat,
10 request: Protocol.Request}
11
12# Carries the state for Queue.
13type Queue is Type:
14 Queue{front: List<&1, Pending>, back: List<&1, Pending>,
15 count: Nat, bytes: Nat}
16
17# Carries the state for Admission.
18type Admission is Type:
19 Enqueued{queue: Queue}
20 Rejected{queue: Queue, pending: Pending, reason: Policy.Rejection}
21
22# Carries the state for TakeResult.
23type TakeResult is Type:
24 Empty{queue: Queue}
25 Taken{queue: Queue, pending: Pending}
26
27# Carries the state for HeadMetadata.
28type HeadMetadata is Data:
29 HeadMetadata{connection_id: U32, accepted_sequence: Nat,
30 deadline_ms: Nat, opcode: U32, request_bytes: Nat,
31 response_peak: Nat}
32
33# Carries the state for PeekResult.
34type PeekResult is Type:
35 PeekResult{queue: Queue, head: Maybe<&2, HeadMetadata>}
36
37# Carries the state for CursorPeek.
38type CursorPeek is Type:
39 CursorPeek{queue: Queue, cursor_id: Maybe<&2, U32>,
40 connection_id: U32}
41
42# Carries the state for CancelResult.
43type CancelResult is Type:
44 CancelResult{queue: Queue, cancelled: List<&1, Pending>,
45 count: Nat, bytes: Nat}
46
47# Carries the state for CancelDisposition.
48type CancelDisposition is Type:
49 Retained{pending: Pending}
50 Cancelled{pending: Pending}
51
52# Carries the state for CancelFold.
53type CancelFold is Type:
54 CancelFold{kept: List<&1, Pending>,
55 cancelled: List<&1, Pending>, count: Nat, bytes: Nat}
56
57# Handles empty.
58def empty() -> Queue:
59 Queue{Nil{}, Nil{}, 0n, 0n}
60
61def u32.clone.pair(+value: U32) -> U32 & U32:
62 (value, value)
63
64def active.counted(
65 front: List<&1, Pending>,
66 back: List<&1, Pending>,
67 bytes: Nat,
68 counts: Nat & Nat
69) -> Queue & Bool:
70 match counts:
71 case (checked_count, retained_count):
72 (Queue{front, back, retained_count, bytes},
73 Nat.is_lt(0n, checked_count))
74
75# Handles active.
76def active(queue: Queue) -> Queue & Bool:
77 match queue:
78 case Queue{front, back, +count, bytes}:
79 active.counted(front, back, bytes,
80 Policy.nat.clone.pair(count))
81
82def reverse.into(
83 remaining: List<&1, Pending>,
84 output: List<&1, Pending>
85) -> List<&1, Pending>:
86 match remaining:
87 case Nil{}: output
88 case Con{item, tail}:
89 reverse.into(tail, Con{item, output})
90
91def pending.size(pending: Pending) -> Pending & Nat:
92 match pending:
93 case Pending{connection_id, accepted_sequence, deadline_ms,
94 +encoded_bytes, response_peak, request}:
95 (Pending{connection_id, accepted_sequence, deadline_ms,
96 encoded_bytes, response_peak, request}, encoded_bytes)
97
98def enqueue.accepted(
99 queue: Queue,
100 pending: Pending,
101 count: Nat,
102 bytes: Nat
103) -> Admission:
104 match queue:
105 case Queue{front, back, _, _}:
106 Enqueued{Queue{front, Con{pending, back}, count, bytes}}
107
108def enqueue.result(
109 queue: Queue,
110 pending: Pending,
111 decision: Policy.Admission
112) -> Admission:
113 match decision:
114 case Policy.Enqueued{Policy.Occupancy{count, bytes}}:
115 enqueue.accepted(queue, pending, count, bytes)
116 case Policy.Rejected{_, reason}:
117 Rejected{queue, pending, reason}
118
119def queue.metrics(queue: Queue) -> Queue & Nat & Nat:
120 match queue:
121 case Queue{front, back, +count, +bytes}:
122 (Queue{front, back, count, bytes}, count, bytes)
123
124def enqueue.with.metrics(
125 pending: Pending,
126 request_bytes: Nat,
127 limits: Nat & Nat,
128 metrics: Queue & Nat & Nat
129) -> Admission:
130 match limits:
131 case (max_count, max_bytes):
132 match metrics:
133 case (retained_queue, (count, bytes)):
134 enqueue.result(retained_queue, pending,
135 Policy.admit.state(Policy.Occupancy{count, bytes},
136 request_bytes, max_count, max_bytes))
137
138def enqueue.measured(
139 queue: Queue,
140 measured: Pending & Nat,
141 limits: Nat & Nat
142) -> Admission:
143 match measured:
144 case (pending, request_bytes):
145 enqueue.with.metrics(pending, request_bytes, limits,
146 queue.metrics(queue))
147
148# Handles enqueue.
149def enqueue(
150 queue: Queue,
151 pending: Pending,
152 max_count: Nat,
153 max_bytes: Nat
154) -> Admission:
155 enqueue.measured(queue, pending.size(pending), (max_count, max_bytes))
156
157def take.front(
158 front: List<&1, Pending>,
159 back: List<&1, Pending>,
160 count: Nat,
161 bytes: Nat
162) -> TakeResult:
163 match front:
164 case Nil{}: Empty{Queue{Nil{}, back, count, bytes}}
165 case Con{item, tail}:
166 match item:
167 case Pending{connection_id, accepted_sequence, deadline_ms,
168 +encoded_bytes, response_peak, request}:
169 Taken{
170 Queue{tail, back, Nat.sub(count, 1n),
171 Nat.sub(bytes, encoded_bytes)},
172 Pending{connection_id, accepted_sequence, deadline_ms,
173 encoded_bytes, response_peak, request}}
174
175# Handles take.
176def take(queue: Queue) -> TakeResult:
177 match queue:
178 case Queue{front, back, count, bytes}:
179 match front:
180 case Nil{}:
181 match back:
182 case Nil{}: Empty{Queue{Nil{}, Nil{}, count, bytes}}
183 case Con{_, _}:
184 take.front(reverse.into(back, Nil{}), Nil{}, count, bytes)
185 case Con{_, _}: take.front(front, back, count, bytes)
186
187def request.opcode.copies(
188 version: U32,
189 generation: U32,
190 request_id: U32,
191 payload: Bytes.Bytes,
192 copies: U32 & U32 & U32
193) -> Protocol.Request & U32:
194 match copies:
195 case (metadata_opcode, (request_opcode, _)):
196 (Protocol.Request{version, generation, request_id, request_opcode,
197 payload}, metadata_opcode)
198
199def request.opcode.copy(request: Protocol.Request)
200 -> Protocol.Request & U32:
201 match request:
202 case Protocol.Request{version, generation, request_id, +operation,
203 payload}:
204 request.opcode.copies(version, generation, request_id, payload,
205 Protocol.u32.duplicate.triple(operation))
206
207def pending.head.metadata.values(
208 ids: U32 & U32,
209 sequences: Nat & Nat,
210 deadlines: Nat & Nat,
211 byte_counts: Nat & Nat,
212 peak_counts: Nat & Nat,
213 request_pair: Protocol.Request & U32
214) -> Pending & HeadMetadata:
215 match ids:
216 case (pending_id, metadata_id):
217 match sequences:
218 case (pending_sequence, metadata_sequence):
219 match deadlines:
220 case (pending_deadline, metadata_deadline):
221 match byte_counts:
222 case (pending_bytes, metadata_bytes):
223 match peak_counts:
224 case (pending_peak, metadata_peak):
225 match request_pair:
226 case (request, opcode):
227 (Pending{pending_id, pending_sequence,
228 pending_deadline, pending_bytes, pending_peak,
229 request},
230 HeadMetadata{metadata_id, metadata_sequence,
231 metadata_deadline, opcode, metadata_bytes,
232 metadata_peak})
233
234def pending.head.metadata(pending: Pending)
235 -> Pending & HeadMetadata:
236 match pending:
237 case Pending{connection_id, accepted_sequence, deadline_ms,
238 encoded_bytes, response_peak, request}:
239 pending.head.metadata.values(u32.clone.pair(connection_id),
240 Policy.nat.clone.pair(accepted_sequence),
241 Policy.nat.clone.pair(deadline_ms),
242 Policy.nat.clone.pair(encoded_bytes),
243 Policy.nat.clone.pair(response_peak),
244 request.opcode.copy(request))
245
246def peek.front.head(
247 tail: List<&1, Pending>,
248 back: List<&1, Pending>,
249 count: Nat,
250 bytes: Nat,
251 pair: Pending & HeadMetadata
252) -> PeekResult:
253 match pair:
254 case (retained, metadata):
255 PeekResult{Queue{Con{retained, tail}, back, count, bytes},
256 Some{metadata}}
257
258def peek.front(
259 front: List<&1, Pending>,
260 back: List<&1, Pending>,
261 count: Nat,
262 bytes: Nat
263) -> PeekResult:
264 match front:
265 case Nil{}:
266 PeekResult{Queue{front, back, count, bytes}, None{}}
267 case Con{pending, tail}:
268 peek.front.head(tail, back, count, bytes,
269 pending.head.metadata(pending))
270
271# Handles peek.
272def peek(queue: Queue) -> PeekResult:
273 match queue:
274 case Queue{front, back, count, bytes}:
275 match front:
276 case Nil{}:
277 match back:
278 case Nil{}: PeekResult{Queue{Nil{}, Nil{}, count, bytes}, None{}}
279 case Con{_, _}:
280 peek.front(reverse.into(back, Nil{}), Nil{}, count, bytes)
281 case Con{_, _}: peek.front(front, back, count, bytes)
282
283def request.cursor.identifier.read(
284 version: U32,
285 generation: U32,
286 request_id: U32,
287 pair: Bytes.Bytes & Maybe<&2, U32>
288) -> Protocol.Request & Maybe<&2, U32>:
289 match pair:
290 case (payload, cursor_id):
291 (Protocol.Request{version, generation, request_id, 13, payload},
292 cursor_id)
293
294def request.cursor.identifier(request: Protocol.Request)
295 -> Protocol.Request & Maybe<&2, U32>:
296 match request:
297 case Protocol.Request{version, generation, request_id, operation,
298 payload}:
299 match operation:
300 case 13:
301 request.cursor.identifier.read(version, generation, request_id,
302 Bytes.get.u32be(payload, 0))
303 case opcode:
304 (Protocol.Request{version, generation, request_id, opcode,
305 payload}, None{})
306
307def peek.cursor.front.request(
308 pair: Protocol.Request & Maybe<&2, U32>,
309 +connection_id: U32,
310 accepted_sequence: Nat,
311 deadline_ms: Nat,
312 encoded_bytes: Nat,
313 response_peak: Nat,
314 tail: List<&1, Pending>,
315 back: List<&1, Pending>,
316 count: Nat,
317 bytes: Nat
318) -> CursorPeek:
319 match pair:
320 case (retained_request, cursor_id):
321 CursorPeek{Queue{Con{Pending{connection_id, accepted_sequence,
322 deadline_ms, encoded_bytes, response_peak, retained_request},
323 tail}, back, count, bytes}, cursor_id, connection_id}
324
325def peek.cursor.front(
326 pending: Pending,
327 tail: List<&1, Pending>,
328 back: List<&1, Pending>,
329 count: Nat,
330 bytes: Nat
331) -> CursorPeek:
332 match pending:
333 case Pending{connection_id, accepted_sequence, deadline_ms,
334 encoded_bytes, response_peak, request}:
335 peek.cursor.front.request(request.cursor.identifier(request),
336 connection_id, accepted_sequence, deadline_ms, encoded_bytes,
337 response_peak, tail, back, count, bytes)
338
339def peek.cursor.normalized(queue: Queue) -> CursorPeek:
340 match queue:
341 case Queue{front, back, count, bytes}:
342 match front:
343 case Nil{}: CursorPeek{Queue{front, back, count, bytes}, None{}, 0}
344 case Con{pending, tail}:
345 peek.cursor.front(pending, tail, back, count, bytes)
346
347def peek.cursor.result(result: PeekResult) -> CursorPeek:
348 match result:
349 case PeekResult{queue, _}: peek.cursor.normalized(queue)
350
351def peek.cursor(queue: Queue) -> CursorPeek:
352 peek.cursor.result(peek(queue))
353
354def head.peak(queue: Queue, peak: Nat) -> Queue:
355 match queue:
356 case Queue{front, back, count, bytes}:
357 match front:
358 case Nil{}: Queue{Nil{}, back, count, bytes}
359 case Con{pending, tail}:
360 match pending:
361 case Pending{connection_id, accepted_sequence, deadline_ms,
362 encoded_bytes, _, request}:
363 Queue{Con{Pending{connection_id, accepted_sequence,
364 deadline_ms, encoded_bytes, peak, request}, tail}, back,
365 count, bytes}
366
367
368# Handles append.
369def append(left: List<&1, Pending>, right: List<&1, Pending>) -> List<&1, Pending>:
370 match left:
371 case Nil{}: right
372 case Con{item, tail}: Con{item, append(tail, right)}
373
374def cancel.disposition(
375 pending: Pending,
376 matches: Bool
377) -> CancelDisposition:
378 match matches:
379 case True{}: Cancelled{pending}
380 case False{}: Retained{pending}
381
382def pending.cancel.disposition.id(
383 copies: U32 & U32,
384 accepted_sequence: Nat,
385 deadline_ms: Nat,
386 encoded_bytes: Nat,
387 response_peak: Nat,
388 request: Protocol.Request,
389 +target: U32
390) -> CancelDisposition:
391 match copies:
392 case (connection_id, comparison_id):
393 cancel.disposition(
394 Pending{connection_id, accepted_sequence, deadline_ms,
395 encoded_bytes, response_peak, request},
396 U32.is_eq(target, comparison_id))
397
398def pending.cancel.disposition(
399 +target: U32,
400 pending: Pending
401) -> CancelDisposition:
402 match pending:
403 case Pending{connection_id, accepted_sequence, deadline_ms,
404 encoded_bytes, response_peak, request}:
405 pending.cancel.disposition.id(u32.clone.pair(connection_id),
406 accepted_sequence, deadline_ms, encoded_bytes, response_peak,
407 request, target)
408
409def cancel.fold.retained(
410 state: CancelFold,
411 measured: Pending & Nat
412) -> CancelFold:
413 match state:
414 case CancelFold{kept, cancelled, count, bytes}:
415 match measured:
416 case (pending, size):
417 CancelFold{Con{pending, kept}, cancelled,
418 Nat.add(count, 1n), Nat.add(bytes, size)}
419
420def cancel.fold.cancelled(
421 state: CancelFold,
422 pending: Pending
423) -> CancelFold:
424 match state:
425 case CancelFold{kept, cancelled, count, bytes}:
426 CancelFold{kept, Con{pending, cancelled}, count, bytes}
427
428def cancel.fold.add(
429 state: CancelFold,
430 disposition: CancelDisposition
431) -> CancelFold:
432 match disposition:
433 case Retained{pending}:
434 cancel.fold.retained(state, pending.size(pending))
435 case Cancelled{pending}: cancel.fold.cancelled(state, pending)
436
437def cancel.fold.loop(
438 entries: List<&1, Pending>,
439 +target: U32,
440 state: CancelFold
441) -> CancelFold:
442 match entries:
443 case Nil{}: state
444 case Con{pending, rest}:
445 cancel.fold.loop(rest, target,
446 cancel.fold.add(state, pending.cancel.disposition(target, pending)))
447
448def cancel.result.values(
449 kept: List<&1, Pending>,
450 cancelled: List<&1, Pending>,
451 counts: Nat & Nat,
452 byte_counts: Nat & Nat
453) -> CancelResult:
454 match counts:
455 case (queue_count, result_count):
456 match byte_counts:
457 case (queue_bytes, result_bytes):
458 CancelResult{
459 Queue{reverse.into(kept, Nil{}), Nil{}, queue_count, queue_bytes},
460 reverse.into(cancelled, Nil{}), result_count, result_bytes}
461
462def cancel.result(state: CancelFold) -> CancelResult:
463 match state:
464 case CancelFold{kept, cancelled, count, bytes}:
465 cancel.result.values(kept, cancelled,
466 Policy.nat.clone.pair(count), Policy.nat.clone.pair(bytes))
467
468# Handles cancel connection.
469def cancel_connection(queue: Queue, target: U32) -> CancelResult:
470 match queue:
471 case Queue{front, back, _, _}:
472 cancel.result(cancel.fold.loop(
473 append(front, reverse.into(back, Nil{})), target,
474 CancelFold{Nil{}, Nil{}, 0n, 0n}))
475
476def cancelled.discard(entries: List<&1, Pending>) -> Unit:
477 match entries:
478 case Nil{}: Unit{}
479 case Con{pending, rest}:
480 match pending:
481 case Pending{_, _, _, _, _, _}: cancelled.discard(rest)