1	import Base
2	import ./RequestQueue.bend as RequestQueue
3	import ./Protocol.bend as Protocol
4	
5	# Carries the state for DispatchReservation.
6	type DispatchReservation is Type:
7	  DispatchReservation{connection_id: U32, request_bytes: Nat,
8	    payload_limit: U32, peak_bytes: Nat}
9	
10	# Carries the state for FrameReservation.
11	type FrameReservation is Type:
12	  FrameReservation{connection_id: U32, reserved_bytes: Nat}
13	
14	# Carries the state for InFlight.
15	type InFlight is Type:
16	  ReservedRequest{reservation: DispatchReservation,
17	    request: Protocol.Request}
18	  DispatchedRequest{reservation: DispatchReservation}
19	
20	# Carries the state for DispatchStart.
21	type DispatchStart is Type:
22	  DispatchStarted{flight: InFlight, request: Protocol.Request}
23	  DispatchAlreadyStarted{flight: InFlight}
24	
25	# Carries the state for DispatchCompletion.
26	type DispatchCompletion is Type:
27	  DispatchCompleted{reservation: DispatchReservation,
28	    response: Protocol.Response}
29	  DispatchCompletionRejected{flight: InFlight,
30	    response: Protocol.Response}
31	
32	# Carries the state for TxDecision.
33	type TxDecision is Data:
34	  Reserve{}
35	  WaitForCapacity{}
36	  Reject{}
37	
38	# Carries the state for TxReservation.
39	type TxReservation is Data:
40	  TxReserved{used_bytes: Nat}
41	  TxWaiting{used_bytes: Nat}
42	  TxRejected{used_bytes: Nat}
43	
44	# Carries the state for RxInterest.
45	type RxInterest is Data:
46	  ReadEnabled{}
47	  ReadPaused{}
48	  ClosePeer{}
49	
50	# Carries the state for RxReservationDecision.
51	type RxReservationDecision is Data:
52	  ReserveFrame{reserved_bytes: Nat}
53	  WaitForFrameCredit{used_bytes: Nat}
54	  RejectFrame{used_bytes: Nat}
55	
56	
57	# Carries the state for BoundDecision.
58	type BoundDecision is Data:
59	  Bounded{payload_bytes: U32, peak_bytes: Nat}
60	  InvalidRequest{}
61	  TooLarge{}
62	
63	# Carries the state for QueueDecision.
64	type QueueDecision is Data:
65	  KeepHead{}
66	  DispatchHead{}
67	  RejectHead{}
68	  ExpireHead{}
69	
70	# Carries the state for CursorAdmissionDecision.
71	type CursorAdmissionDecision is Data:
72	  PrepareCursorNext{}
73	  WaitForEarlierCursorNext{}
74	
75	# Carries the state for HeadPolicy.
76	type HeadPolicy is Data:
77	  HeadPolicy{fits_total: Bool, fits_available: Bool, expired: Bool}
78	
79	# Carries the state for QueueReservation.
80	type QueueReservation is Type:
81	  QueueKeep{queue: RequestQueue.Queue}
82	  QueueDispatch{queue: RequestQueue.Queue}
83	  QueueReject{queue: RequestQueue.Queue}
84	  QueueExpire{queue: RequestQueue.Queue}
85	
86	def tx.flight.start(flight: InFlight) -> DispatchStart:
87	  match flight:
88	    case ReservedRequest{reservation, request}:
89	      DispatchStarted{DispatchedRequest{reservation}, request}
90	    case flight: DispatchAlreadyStarted{flight}
91	
92	def tx.flight.complete(
93	  flight: InFlight,
94	  response: Protocol.Response
95	) -> DispatchCompletion:
96	  match flight:
97	    case DispatchedRequest{reservation}:
98	      DispatchCompleted{reservation, response}
99	    case flight: DispatchCompletionRejected{flight, response}
100	
101	def tx.flight.reserved(
102	  connection_id: U32,
103	  request_bytes: Nat,
104	  bound: BoundDecision,
105	  request: Protocol.Request
106	) -> InFlight:
107	  match bound:
108	    case Bounded{payload_limit, peak_bytes}:
109	      ReservedRequest{DispatchReservation{connection_id, request_bytes,
110	        payload_limit, peak_bytes}, request}
111	    case _:
112	      ReservedRequest{DispatchReservation{connection_id, request_bytes,
113	        0, 28n}, request}
114	
115	def tx.frame.transfer(reservation: DispatchReservation) -> FrameReservation:
116	  match reservation:
117	    case DispatchReservation{connection_id, _, _, peak_bytes}:
118	      FrameReservation{connection_id, peak_bytes}
119	
120	def tx.frame.control(connection_id: U32) -> FrameReservation:
121	  FrameReservation{connection_id, 0n}
122	
123	def tx.frame.measure(reservation: FrameReservation)
124	  -> FrameReservation & Nat:
125	  match reservation:
126	    case FrameReservation{connection_id, +reserved_bytes}:
127	      (FrameReservation{connection_id, reserved_bytes}, reserved_bytes)
128	
129	def slots.control.bytes(connections: Nat) -> Nat:
130	  Nat.mul(connections, 28n)
131	
132	def slots.header.bytes(connections: Nat) -> Nat:
133	  Nat.mul(connections, 24n)
134	
135	def tx.available(available: Bool) -> TxDecision:
136	  match available:
137	    case True{}: Reserve{}
138	    case False{}: WaitForCapacity{}
139	
140	def tx.decide(fits_total: Bool, fits_available: Bool) -> TxDecision:
141	  match fits_total:
142	    case False{}: Reject{}
143	    case True{}: tx.available(fits_available)
144	
145	def tx.request(+used: Nat, +required: Nat, +capacity: Nat) -> TxDecision:
146	  tx.decide(Nat.is_le(required, capacity),
147	    Nat.is_le(Nat.add(used, required), capacity))
148	
149	def tx.reserve.result(
150	  +used: Nat,
151	  +required: Nat,
152	  decision: TxDecision
153	) -> TxReservation:
154	  match decision:
155	    case Reserve{}: TxReserved{Nat.add(used, required)}
156	    case WaitForCapacity{}: TxWaiting{used}
157	    case Reject{}: TxRejected{used}
158	
159	def tx.reserve(+used: Nat, +required: Nat, +capacity: Nat) -> TxReservation:
160	  tx.reserve.result(used, required, tx.request(used, required, capacity))
161	
162	def tx.get.peak(value_bytes: Nat) -> Nat:
163	  Nat.add(Nat.mul(2n, value_bytes), 28n)
164	
165	def tx.capacity(response_bytes: U32) -> Nat:
166	  Nat.add(Nat.mul(2n, U32.to_nat(response_bytes)), 28n)
167	
168	def tx.get.bound.peak(
169	  +value_bytes: Nat,
170	  peak_bytes: Nat,
171	  fits_capacity: Bool
172	) -> BoundDecision:
173	  match fits_capacity:
174	    case True{}: Bounded{U32.from_nat(value_bytes), peak_bytes}
175	    case False{}: TooLarge{}
176	
177	def tx.get.bound.decided(
178	  fits_payload: Bool,
179	  +value_bytes: Nat,
180	  +capacity: Nat
181	) -> BoundDecision:
182	  match fits_payload:
183	    case False{}: TooLarge{}
184	    case True{}:
185	      tx.get.bound.peak(value_bytes,
186	        tx.get.peak(value_bytes),
187	        Nat.is_le(tx.get.peak(value_bytes), capacity))
188	
189	def tx.get.bound(
190	  +value_bytes: U32,
191	  max_payload_bytes: U32,
192	  capacity: Nat
193	) -> BoundDecision:
194	  tx.get.bound.decided(U32.is_le(value_bytes, max_payload_bytes),
195	    U32.to_nat(value_bytes), capacity)
196	
197	def tx.response.bound(+payload_bytes: U32) -> BoundDecision:
198	  Bounded{payload_bytes, tx.get.peak(U32.to_nat(payload_bytes))}
199	
200	def tx.payload.bound.decided(
201	  +payload_bytes: Nat,
202	  fits: Bool
203	) -> BoundDecision:
204	  match fits:
205	    case False{}: TooLarge{}
206	    case True{}:
207	      Bounded{U32.from_nat(payload_bytes), tx.get.peak(payload_bytes)}
208	
209	def tx.payload.bound(
210	  +payload_bytes: Nat,
211	  max_payload_bytes: U32
212	) -> BoundDecision:
213	  tx.payload.bound.decided(payload_bytes,
214	    Nat.is_le(payload_bytes, U32.to_nat(max_payload_bytes)))
215	
216	def tx.nat.duplicate.step(pair: Nat & Nat) -> Nat & Nat:
217	  match pair:
218	    case (first, second): (Nat.add(1n, first), Nat.add(1n, second))
219	
220	def tx.nat.duplicate(value: Nat) -> Nat & Nat:
221	  match value:
222	    case 0n: (0n, 0n)
223	    case 1n+rest: tx.nat.duplicate.step(tx.nat.duplicate(rest))
224	
225	def tx.u32.duplicate.from.nat(copies: Nat & Nat) -> U32 & U32:
226	  match copies:
227	    case (first, second): (U32.from_nat(first), U32.from_nat(second))
228	
229	def tx.u32.duplicate(value: U32) -> U32 & U32:
230	  tx.u32.duplicate.from.nat(tx.nat.duplicate(U32.to_nat(value)))
231	
232	def tx.cursor.bound.payload(copies: Nat & Nat) -> BoundDecision:
233	  match copies:
234	    case (payload_bytes, peak_bytes):
235	      Bounded{U32.from_nat(payload_bytes), tx.get.peak(peak_bytes)}
236	
237	def tx.cursor.bound.measured(
238	  +required_bytes: Nat,
239	  fits: Bool
240	) -> BoundDecision:
241	  match fits:
242	    case False{}: TooLarge{}
243	    case True{}:
244	      tx.cursor.bound.payload(tx.nat.duplicate(required_bytes))
245	
246	def tx.cursor.bound.required(
247	  copies: Nat & Nat,
248	  response_copies: U32 & U32
249	) -> BoundDecision:
250	  match copies:
251	    case (check_bytes, required_bytes):
252	      match response_copies:
253	        case (response_for_check, _):
254	          tx.cursor.bound.measured(required_bytes,
255	            Nat.is_le(check_bytes, U32.to_nat(response_for_check)))
256	
257	def tx.cursor.bound(+retained_bytes: Nat, response_bytes: U32) -> BoundDecision:
258	  tx.cursor.bound.required(tx.nat.duplicate(Nat.add(retained_bytes, 9n)),
259	    tx.u32.duplicate(response_bytes))
260	
261	def tx.queue.eligible(decision: TxDecision) -> QueueDecision:
262	  match decision:
263	    case Reserve{}: DispatchHead{}
264	    case WaitForCapacity{}: KeepHead{}
265	    case Reject{}: RejectHead{}
266	
267	def tx.cursor.admission(seen: Bool) -> CursorAdmissionDecision:
268	  match seen:
269	    case True{}: WaitForEarlierCursorNext{}
270	    case False{}: PrepareCursorNext{}
271	
272	def tx.queue.decide(policy: HeadPolicy) -> QueueDecision:
273	  match policy:
274	    case HeadPolicy{fits_total, fits_available, expired}:
275	      match expired:
276	        case True{}: ExpireHead{}
277	        case False{}:
278	          tx.queue.eligible(tx.decide(fits_total, fits_available))
279	
280	def tx.queue.reserved(
281	  queue: RequestQueue.Queue,
282	  decision: QueueDecision
283	) -> QueueReservation:
284	  match decision:
285	    case KeepHead{}: QueueKeep{queue}
286	    case DispatchHead{}: QueueDispatch{queue}
287	    case RejectHead{}: QueueReject{queue}
288	    case ExpireHead{}: QueueExpire{queue}
289	
290	def tx.queue.reserve(
291	  queue: RequestQueue.Queue,
292	  policy: HeadPolicy
293	) -> QueueReservation:
294	  tx.queue.reserved(queue, tx.queue.decide(policy))
295	
296	def tx.queue.decide.budget(
297	  +used: Nat,
298	  +required: Nat,
299	  +capacity: Nat,
300	  expired: Bool
301	) -> QueueDecision:
302	  match expired:
303	    case True{}: ExpireHead{}
304	    case False{}: tx.queue.eligible(tx.request(used, required, capacity))
305	
306	def tx.queue.head.policy(
307	  head: Maybe<&2, RequestQueue.HeadMetadata>,
308	  +used: Nat,
309	  +capacity: Nat,
310	  now: Nat
311	) -> QueueDecision:
312	  match head:
313	    case None{}:
314	      tx.queue.decide.budget(used, 28n, capacity, False{})
315	    case Some{RequestQueue.HeadMetadata{_, _, deadline_ms, _, _,
316	        response_peak}}:
317	      tx.queue.decide.budget(used, response_peak, capacity,
318	        Nat.is_le(deadline_ms, now))
319	
320	def tx.queue.reserve.peeked(
321	  result: RequestQueue.PeekResult,
322	  +used: Nat,
323	  +capacity: Nat,
324	  now: Nat
325	) -> QueueReservation:
326	  match result:
327	    case RequestQueue.PeekResult{queue, head}:
328	      tx.queue.reserved(queue, tx.queue.head.policy(head, used, capacity, now))
329	
330	def rx.live(owns_credit: Bool) -> RxInterest:
331	  match owns_credit:
332	    case True{}: ReadEnabled{}
333	    case False{}: ReadPaused{}
334	
335	def rx.interest(terminal: Bool, owns_credit: Bool) -> RxInterest:
336	  match terminal:
337	    case True{}: ClosePeer{}
338	    case False{}: rx.live(owns_credit)
339	
340	def rx.frame.peak(payload_bytes: Nat) -> Nat:
341	  Nat.add(48n, Nat.mul(3n, payload_bytes))
342	
343	def rx.frame.reservation(
344	  +used_bytes: Nat,
345	  +required_bytes: Nat,
346	  +_capacity_bytes: Nat,
347	  fits_total: Bool,
348	  fits_available: Bool
349	) -> RxReservationDecision:
350	  match fits_total:
351	    case False{}: RejectFrame{used_bytes}
352	    case True{}:
353	      match fits_available:
354	        case True{}: ReserveFrame{Nat.add(used_bytes, required_bytes)}
355	        case False{}: WaitForFrameCredit{used_bytes}
356	
357	def rx.frame.reserve(
358	  +used_bytes: Nat,
359	  +required_bytes: Nat,
360	  +capacity_bytes: Nat
361	) -> RxReservationDecision:
362	  rx.frame.reservation(used_bytes, required_bytes, capacity_bytes,
363	    Nat.is_le(required_bytes, capacity_bytes),
364	    Nat.is_le(Nat.add(used_bytes, required_bytes), capacity_bytes))