src/coordinator/QueuePolicy.bend

mylsm-lsm-store@0.6.0.1 · 147 lines · 16 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)