src/coordinator/TransportBudgetPolicy.bend
mylsm-lsm-store@0.6.0.1 · 364 lines · 56 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/coordinator/RequestQueue.bend as RequestQueue
3import mylsm-lsm-store@0.6.0.1/src/coordinator/Protocol.bend as Protocol
4
5
6type DispatchReservation is Type:
7 DispatchReservation{connection_id: U32, request_bytes: Nat,
8 payload_limit: U32, peak_bytes: Nat}
9
10
11type FrameReservation is Type:
12 FrameReservation{connection_id: U32, reserved_bytes: Nat}
13
14
15type InFlight is Type:
16 ReservedRequest{reservation: DispatchReservation,
17 request: Protocol.Request}
18 DispatchedRequest{reservation: DispatchReservation}
19
20
21type DispatchStart is Type:
22 DispatchStarted{flight: InFlight, request: Protocol.Request}
23 DispatchAlreadyStarted{flight: InFlight}
24
25
26type DispatchCompletion is Type:
27 DispatchCompleted{reservation: DispatchReservation,
28 response: Protocol.Response}
29 DispatchCompletionRejected{flight: InFlight,
30 response: Protocol.Response}
31
32
33type TxDecision is Data:
34 Reserve{}
35 WaitForCapacity{}
36 Reject{}
37
38
39type TxReservation is Data:
40 TxReserved{used_bytes: Nat}
41 TxWaiting{used_bytes: Nat}
42 TxRejected{used_bytes: Nat}
43
44
45type RxInterest is Data:
46 ReadEnabled{}
47 ReadPaused{}
48 ClosePeer{}
49
50
51type RxReservationDecision is Data:
52 ReserveFrame{reserved_bytes: Nat}
53 WaitForFrameCredit{used_bytes: Nat}
54 RejectFrame{used_bytes: Nat}
55
56
57
58type BoundDecision is Data:
59 Bounded{payload_bytes: U32, peak_bytes: Nat}
60 InvalidRequest{}
61 TooLarge{}
62
63
64type QueueDecision is Data:
65 KeepHead{}
66 DispatchHead{}
67 RejectHead{}
68 ExpireHead{}
69
70
71type CursorAdmissionDecision is Data:
72 PrepareCursorNext{}
73 WaitForEarlierCursorNext{}
74
75
76type HeadPolicy is Data:
77 HeadPolicy{fits_total: Bool, fits_available: Bool, expired: Bool}
78
79
80type QueueReservation is Type:
81 QueueKeep{queue: RequestQueue.Queue}
82 QueueDispatch{queue: RequestQueue.Queue}
83 QueueReject{queue: RequestQueue.Queue}
84 QueueExpire{queue: RequestQueue.Queue}
85
86def 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
92def 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
101def 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
115def tx.frame.transfer(reservation: DispatchReservation) -> FrameReservation:
116 match reservation:
117 case DispatchReservation{connection_id, _, _, peak_bytes}:
118 FrameReservation{connection_id, peak_bytes}
119
120def tx.frame.control(connection_id: U32) -> FrameReservation:
121 FrameReservation{connection_id, 0n}
122
123def 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
129def slots.control.bytes(connections: Nat) -> Nat:
130 Nat.mul(connections, 28n)
131
132def slots.header.bytes(connections: Nat) -> Nat:
133 Nat.mul(connections, 24n)
134
135def tx.available(available: Bool) -> TxDecision:
136 match available:
137 case True{}: Reserve{}
138 case False{}: WaitForCapacity{}
139
140def 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
145def 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
149def 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
159def tx.reserve(+used: Nat, +required: Nat, +capacity: Nat) -> TxReservation:
160 tx.reserve.result(used, required, tx.request(used, required, capacity))
161
162def tx.get.peak(value_bytes: Nat) -> Nat:
163 Nat.add(Nat.mul(2n, value_bytes), 28n)
164
165def tx.capacity(response_bytes: U32) -> Nat:
166 Nat.add(Nat.mul(2n, U32.to_nat(response_bytes)), 28n)
167
168def 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
177def 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
189def 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
197def tx.response.bound(+payload_bytes: U32) -> BoundDecision:
198 Bounded{payload_bytes, tx.get.peak(U32.to_nat(payload_bytes))}
199
200def 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
209def 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
216def 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
220def 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
225def 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
229def tx.u32.duplicate(value: U32) -> U32 & U32:
230 tx.u32.duplicate.from.nat(tx.nat.duplicate(U32.to_nat(value)))
231
232def 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
237def 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
246def 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
257def 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
261def tx.queue.eligible(decision: TxDecision) -> QueueDecision:
262 match decision:
263 case Reserve{}: DispatchHead{}
264 case WaitForCapacity{}: KeepHead{}
265 case Reject{}: RejectHead{}
266
267def tx.cursor.admission(seen: Bool) -> CursorAdmissionDecision:
268 match seen:
269 case True{}: WaitForEarlierCursorNext{}
270 case False{}: PrepareCursorNext{}
271
272def 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
280def 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
290def tx.queue.reserve(
291 queue: RequestQueue.Queue,
292 policy: HeadPolicy
293) -> QueueReservation:
294 tx.queue.reserved(queue, tx.queue.decide(policy))
295
296def 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
306def 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
320def 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
330def rx.live(owns_credit: Bool) -> RxInterest:
331 match owns_credit:
332 case True{}: ReadEnabled{}
333 case False{}: ReadPaused{}
334
335def rx.interest(terminal: Bool, owns_credit: Bool) -> RxInterest:
336 match terminal:
337 case True{}: ClosePeer{}
338 case False{}: rx.live(owns_credit)
339
340def rx.frame.peak(payload_bytes: Nat) -> Nat:
341 Nat.add(48n, Nat.mul(3n, payload_bytes))
342
343def 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
357def 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))