src/coordinator/QueuePolicy.bend
mylsm-lsm-store@0.6.0.0 · 147 lines · 0 definitions
1import Base
2
3# Separates an oversized single request from temporary queue saturation.
4type Decision is Data:
5 Accepted{count: Nat, bytes: Nat}
6 QueueFull{}
7 RequestTooLarge{}
8
9# Stores the current queued request count and encoded byte total.
10type State is Data:
11 Occupancy{count: Nat, bytes: Nat}
12
13# Distinguishes temporary occupancy from a request exceeding all capacity.
14type Rejection is Data:
15 Saturated{}
16 Oversized{}
17
18# Returns changed occupancy on acceptance and exact prior state on rejection.
19type Admission is Data:
20 Enqueued{state: State}
21 Rejected{state: State, reason: Rejection}
22
23def nat.clone.pair(+value: Nat) -> Nat & Nat:
24 (value, value)
25
26def admit.total.pair(checked: Nat, output: Nat, max_queue_bytes: Nat) -> Bool & Nat:
27 (Nat.is_le(checked, max_queue_bytes), output)
28
29def admit.decision(count: Nat, within_request_limit: Bool, within_count: Bool, pair: Bool & Nat) -> Decision:
30 match within_request_limit:
31 case False{}: RequestTooLarge{}
32 case True{}:
33 match within_count:
34 case False{}: QueueFull{}
35 case True{}:
36 match pair:
37 case (False{}, _): QueueFull{}
38 case (True{}, bytes): Accepted{Nat.add(count, 1n), bytes}
39
40def admit.inputs(
41 count: Nat,
42 queued_pair: Nat & Nat,
43 request_pair: Nat & Nat,
44 max_queue_bytes: Nat,
45 within_request_limit: Bool,
46 within_count: Bool
47) -> Decision:
48 match queued_pair:
49 case (queued_check, queued_output):
50 match request_pair:
51 case (request_check, request_output):
52 admit.decision(count, within_request_limit, within_count,
53 admit.total.pair(Nat.add(queued_check, request_check),
54 Nat.add(queued_output, request_output), max_queue_bytes))
55
56def admit.counted.request.check(
57 count: Nat,
58 checked_count: Nat,
59 queued_bytes: Nat,
60 request_output: Nat,
61 max_count: Nat,
62 checked_limit: Nat,
63 output_limit: Nat,
64 request_check_pair: Nat & Nat
65) -> Decision:
66 match request_check_pair:
67 case (request_check, request_size_check):
68 admit.inputs(count, nat.clone.pair(queued_bytes), (request_check, request_output), output_limit,
69 Nat.is_le(request_size_check, checked_limit), Nat.is_lt(checked_count, max_count))
70
71def admit.counted.requested(
72 count: Nat,
73 checked_count: Nat,
74 queued_bytes: Nat,
75 request_pair: Nat & Nat,
76 max_count: Nat,
77 limits: Nat & Nat
78) -> Decision:
79 match request_pair:
80 case (request_check, request_output):
81 match limits:
82 case (checked_limit, output_limit):
83 admit.counted.request.check(count, checked_count, queued_bytes, request_output,
84 max_count, checked_limit, output_limit, nat.clone.pair(request_check))
85
86def admit.counted.limits(
87 pair: Nat & Nat,
88 queued_bytes: Nat,
89 request_bytes: Nat,
90 max_count: Nat,
91 limits: Nat & Nat
92) -> Decision:
93 match pair:
94 case (count, checked_count):
95 admit.counted.requested(count, checked_count, queued_bytes,
96 nat.clone.pair(request_bytes), max_count, limits)
97
98def admit.counted(
99 pair: Nat & Nat,
100 queued_bytes: Nat,
101 request_bytes: Nat,
102 max_count: Nat,
103 max_queue_bytes: Nat
104) -> Decision:
105 admit.counted.limits(pair, queued_bytes, request_bytes, max_count, nat.clone.pair(max_queue_bytes))
106
107# Computes bounded queue admission from the request size and current occupancy.
108def admit(
109 count: Nat,
110 queued_bytes: Nat,
111 request_bytes: Nat,
112 max_count: Nat,
113 max_queue_bytes: Nat
114) -> Decision:
115 admit.counted(nat.clone.pair(count), queued_bytes, request_bytes, max_count, max_queue_bytes)
116
117def admit.state.result(original: State, decision: Decision) -> Admission:
118 match decision:
119 case Accepted{count, bytes}: Enqueued{Occupancy{count, bytes}}
120 case QueueFull{}: Rejected{original, Saturated{}}
121 case RequestTooLarge{}: Rejected{original, Oversized{}}
122
123def admit.state.copies(
124 count_copies: Nat & Nat,
125 byte_copies: Nat & Nat,
126 request_bytes: Nat,
127 max_count: Nat,
128 max_queue_bytes: Nat
129) -> Admission:
130 match count_copies:
131 case (checked_count, retained_count):
132 match byte_copies:
133 case (checked_bytes, retained_bytes):
134 admit.state.result(Occupancy{retained_count, retained_bytes},
135 admit(checked_count, checked_bytes, request_bytes,
136 max_count, max_queue_bytes))
137
138def admit.state(
139 original: State,
140 request_bytes: Nat,
141 max_count: Nat,
142 max_queue_bytes: Nat
143) -> Admission:
144 match original:
145 case Occupancy{count, bytes}:
146 admit.state.copies(nat.clone.pair(count), nat.clone.pair(bytes),
147 request_bytes, max_count, max_queue_bytes)