1	import Base
2	import ./QueuePolicy.bend as Policy
3	import ./Protocol.bend as Protocol
4	import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
5	
6	# Owns one accepted coordinator request until dispatch or rejection.
7	type 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.
13	type 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.
18	type Admission is Type:
19	  Enqueued{queue: Queue}
20	  Rejected{queue: Queue, pending: Pending, reason: Policy.Rejection}
21	
22	# Carries the state for TakeResult.
23	type TakeResult is Type:
24	  Empty{queue: Queue}
25	  Taken{queue: Queue, pending: Pending}
26	
27	# Carries the state for HeadMetadata.
28	type 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.
34	type PeekResult is Type:
35	  PeekResult{queue: Queue, head: Maybe<&2, HeadMetadata>}
36	
37	# Carries the state for CursorPeek.
38	type CursorPeek is Type:
39	  CursorPeek{queue: Queue, cursor_id: Maybe<&2, U32>,
40	    connection_id: U32}
41	
42	# Carries the state for CancelResult.
43	type CancelResult is Type:
44	  CancelResult{queue: Queue, cancelled: List<&1, Pending>,
45	    count: Nat, bytes: Nat}
46	
47	# Carries the state for CancelDisposition.
48	type CancelDisposition is Type:
49	  Retained{pending: Pending}
50	  Cancelled{pending: Pending}
51	
52	# Carries the state for CancelFold.
53	type CancelFold is Type:
54	  CancelFold{kept: List<&1, Pending>,
55	    cancelled: List<&1, Pending>, count: Nat, bytes: Nat}
56	
57	# Handles empty.
58	def empty() -> Queue:
59	  Queue{Nil{}, Nil{}, 0n, 0n}
60	
61	def u32.clone.pair(+value: U32) -> U32 & U32:
62	  (value, value)
63	
64	def 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.
76	def 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	
82	def 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	
91	def 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	
98	def 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	
108	def 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	
119	def 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	
124	def 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	
138	def 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.
149	def 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	
157	def 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.
176	def 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	
187	def 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	
199	def 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	
207	def 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	
234	def 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	
246	def 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	
258	def 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.
272	def 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	
283	def 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	
294	def 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	
307	def 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	
325	def 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	
339	def 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	
347	def peek.cursor.result(result: PeekResult) -> CursorPeek:
348	  match result:
349	    case PeekResult{queue, _}: peek.cursor.normalized(queue)
350	
351	def peek.cursor(queue: Queue) -> CursorPeek:
352	  peek.cursor.result(peek(queue))
353	
354	def 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.
369	def 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	
374	def cancel.disposition(
375	  pending: Pending,
376	  matches: Bool
377	) -> CancelDisposition:
378	  match matches:
379	    case True{}: Cancelled{pending}
380	    case False{}: Retained{pending}
381	
382	def 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	
398	def 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	
409	def 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	
420	def 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	
428	def 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	
437	def 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	
448	def 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	
462	def 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.
469	def 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	
476	def 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)