src/coordinator/RequestQueue.bend

mylsm-lsm-store@0.6.0.1 · 481 lines · 51 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as Policy
3import mylsm-lsm-store@0.6.0.1/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)