admission.bend
bend-kit-http@0.34.0.0 · 347 lines · 0 definitions
1# Private bounded transport accounting. Replies contain copied data, never resources.
2import Base
3import bend-kit-resources@0.1.0.0/resources.bend as Res
4
5type Counts is Data:
6 Counts{connections: U32, active: U32, buffered: U32,
7 connection_high: U32, active_high: U32, buffered_high: U32,
8 connection_rejected: Nat, active_rejected: Nat, buffered_rejected: Nat}
9
10type Caps is Data:
11 Caps{connections: U32, active: U32, buffered: U32}
12
13type Command is Data:
14 Acquire{kind: U32, reply: Chan(Bool)}
15 Release{kind: U32, reply: Chan(Unit)}
16 Snapshot{reply: Chan(Counts)}
17 Quiesce{reply: Chan(Bool)}
18 Running{reply: Chan(Bool)}
19 Stop{}
20
21# One transport resource: its pool, and every lease taken from it joined in one grant.
22type Lease is Type:
23 Lease{pool: Res.Pool, held: Res.Grant}
24
25type Marks is Data:
26 Marks{connection_high: U32, active_high: U32, buffered_high: U32,
27 connection_rejected: Nat, active_rejected: Nat, buffered_rejected: Nat}
28
29type Ledger is Type:
30 Ledger{connections: Lease, active: Lease, buffered: Lease, marks: Marks}
31
32# Broken means a lease count is wrong, such as a give with no matching take.
33type Booked is Type:
34 Booked{ledger: Ledger, admitted: Bool}
35 Broken{}
36
37type Taken is Type:
38 Admitted{lease: Lease}
39 Full{lease: Lease}
40 Lost{}
41
42def lease.open(+owner: Nat, capacity: U32) -> Lease:
43 Lease{Res.new(owner, U32.to_nat(capacity)), Res.Grant{owner, 0n}}
44
45def ledger.open(caps: Caps) -> Ledger:
46 Caps{cc, ac, bc} = caps
47 Ledger{lease.open(0n, cc), lease.open(1n, ac), lease.open(2n, bc), Marks{0, 0, 0, 0n, 0n, 0n}}
48
49def lease.joined(combined: Res.Combine, pool: Res.Pool) -> Taken:
50 match combined:
51 case Res.Joined{grant}:
52 Admitted{Lease{pool, grant}}
53 case Res.Apart{first, second}:
54 Lost{}
55
56def lease.took(reserved: Res.Reserve, held: Res.Grant) -> Taken:
57 match reserved:
58 case Res.Took{pool, grant}:
59 lease.joined(Res.combine(held, grant), pool)
60 case Res.Refused{pool}:
61 Full{Lease{pool, held}}
62
63def lease.take(lease: Lease) -> Taken:
64 Lease{pool, held} = lease
65 lease.took(Res.reserve(pool, 1n), held)
66
67def lease.released(released: Res.Release, rest: Res.Grant) -> Maybe<&1, Lease>:
68 match released:
69 case Res.Released{pool}:
70 Some{Lease{pool, rest}}
71 case Res.Kept{pool, grant}:
72 None{}
73
74def lease.split(parts: Res.Split, pool: Res.Pool) -> Maybe<&1, Lease>:
75 match parts:
76 case Res.Parts{one, rest}:
77 lease.released(Res.release(pool, one), rest)
78 case Res.Whole{grant}:
79 None{}
80
81def lease.give(lease: Lease) -> Maybe<&1, Lease>:
82 Lease{pool, held} = lease
83 lease.split(Res.split(held, 1n), pool)
84
85def count(+pool: Res.Pool) -> U32:
86 U32.from_nat(Res.reserved(pool))
87
88def counts(ledger: Ledger) -> Counts & Ledger:
89 Ledger{Lease{+cp, cg}, Lease{+ap, ag}, Lease{+bp, bg}, +marks} = ledger
90 Marks{ch, ah, bh, cr, ar, br} = marks
91 (Counts{count(cp), count(ap), count(bp), ch, ah, bh, cr, ar, br}, Ledger{Lease{cp, cg}, Lease{ap, ag}, Lease{bp, bg}, marks})
92
93def final(ledger: Ledger) -> Counts:
94 Pair.fst(Counts, Ledger, counts(ledger))
95
96# Raises each high-water mark to its current count.
97def marked(ledger: Ledger) -> Ledger:
98 Ledger{Lease{+cp, cg}, Lease{+ap, ag}, Lease{+bp, bg}, marks} = ledger
99 Marks{ch, ah, bh, cr, ar, br} = marks
100 Ledger{Lease{cp, cg}, Lease{ap, ag}, Lease{bp, bg},
101 Marks{U32.max(ch, count(cp)), U32.max(ah, count(ap)), U32.max(bh, count(bp)), cr, ar, br}}
102
103def refused(kind: U32, marks: Marks) -> Marks:
104 match kind marks:
105 case 0 Marks{ch, ah, bh, cr, ar, br}:
106 Marks{ch, ah, bh, 1n+cr, ar, br}
107 case 1 Marks{ch, ah, bh, cr, ar, br}:
108 Marks{ch, ah, bh, cr, 1n+ar, br}
109 case _ Marks{ch, ah, bh, cr, ar, br}:
110 Marks{ch, ah, bh, cr, ar, 1n+br}
111
112# A refused buffer returns the connection lease it already took.
113def acquire.rollback(connections: Maybe<&1, Lease>, active: Lease, buffered: Lease, marks: Marks) -> Booked:
114 match connections:
115 case None{}:
116 Broken{}
117 case Some{lease}:
118 Booked{Ledger{lease, active, buffered, refused(2, marks)}, False{}}
119
120def acquire.buffered(taken: Taken, connections: Lease, active: Lease, marks: Marks) -> Booked:
121 match taken:
122 case Admitted{buffered}:
123 Booked{marked(Ledger{connections, active, buffered, marks}), True{}}
124 case Full{buffered}:
125 acquire.rollback(lease.give(connections), active, buffered, marks)
126 case Lost{}:
127 Broken{}
128
129# A connection holds one connection unit and one buffer unit, taken together.
130def acquire.connection(taken: Taken, active: Lease, buffered: Lease, marks: Marks) -> Booked:
131 match taken:
132 case Admitted{connections}:
133 acquire.buffered(lease.take(buffered), connections, active, marks)
134 case Full{connections}:
135 Booked{Ledger{connections, active, buffered, refused(0, marks)}, False{}}
136 case Lost{}:
137 Broken{}
138
139def acquire.active(taken: Taken, connections: Lease, buffered: Lease, marks: Marks) -> Booked:
140 match taken:
141 case Admitted{active}:
142 Booked{marked(Ledger{connections, active, buffered, marks}), True{}}
143 case Full{active}:
144 Booked{Ledger{connections, active, buffered, refused(1, marks)}, False{}}
145 case Lost{}:
146 Broken{}
147
148def acquire(kind: U32, ledger: Ledger) -> Booked:
149 match kind:
150 case 0:
151 Ledger{connections, active, buffered, marks} = ledger
152 acquire.connection(lease.take(connections), active, buffered, marks)
153 case _:
154 Ledger{connections, active, buffered, marks} = ledger
155 acquire.active(lease.take(active), connections, buffered, marks)
156
157def acquire.running(running: Bool, kind: U32, ledger: Ledger) -> Booked:
158 match running:
159 case False{}:
160 Booked{ledger, False{}}
161 case True{}:
162 acquire(kind, ledger)
163
164def release.connection(connections: Maybe<&1, Lease>, buffered: Maybe<&1, Lease>, active: Lease, marks: Marks) -> Booked:
165 match connections buffered:
166 case Some{c} Some{b}:
167 Booked{Ledger{c, active, b, marks}, True{}}
168 case _ _:
169 Broken{}
170
171def release.active(active: Maybe<&1, Lease>, connections: Lease, buffered: Lease, marks: Marks) -> Booked:
172 match active:
173 case Some{a}:
174 Booked{Ledger{connections, a, buffered, marks}, True{}}
175 case None{}:
176 Broken{}
177
178def release(kind: U32, ledger: Ledger) -> Booked:
179 match kind:
180 case 0:
181 Ledger{connections, active, buffered, marks} = ledger
182 release.connection(lease.give(connections), lease.give(buffered), active, marks)
183 case _:
184 Ledger{connections, active, buffered, marks} = ledger
185 release.active(lease.give(active), connections, buffered, marks)
186
187def stopped(closing: Bool, counts: Counts) -> Bool:
188 Counts{c, a, b, ch, ah, bh, cr, ar, br} = counts
189 closing && U32.is_zero(c) && U32.is_zero(a) && U32.is_zero(b)
190
191def idle(closing: Bool, counted: Counts & Ledger) -> Bool & Ledger:
192 (counts, ledger) = counted
193 (stopped(closing, counts), ledger)
194
195type Step is Type:
196 Waiting{ledger: Ledger}
197 Received{ledger: Ledger, command: Maybe<&1, Command>}
198 Acquired{reply: Chan(Bool), booked: Booked}
199 Released{reply: Chan(Unit), booked: Booked}
200 Counted{reply: Chan(Counts), counted: Counts & Ledger}
201 Finishing{checked: Bool & Ledger}
202
203# The private inbox is reusable: IO.join would close it after one receive.
204@unsafe
205def loop(+inbox: Chan(Command), +running: Bool, +closing: Bool, step: Step) -> IO(Counts):
206 match step:
207 case Waiting{ledger}:
208 do IO<Counts>:
209 got : Maybe<&1, Command> <- Chan.recv(Command, inbox)
210 loop(inbox, running, closing, Received{ledger, got})
211 case Acquired{reply, booked}:
212 match booked:
213 case Broken{}:
214 IO.die(Counts, 1, "Admission: a lease count broke")
215 case Booked{ledger, admitted}:
216 do IO<Counts>:
217 sent : Result<&1, &1, Bool, Unit> <- Chan.send(Bool, reply, admitted)
218 loop(inbox, running, closing, Waiting{ledger})
219 case Released{reply, booked}:
220 match booked:
221 case Broken{}:
222 IO.die(Counts, 1, "Admission.give: no lease of this kind is held")
223 case Booked{ledger, admitted}:
224 do IO<Counts>:
225 sent : Result<&1, &1, Unit, Unit> <- Chan.send(Unit, reply, Unit{})
226 loop(inbox, running, closing, Finishing{idle(closing, counts(ledger))})
227 case Counted{reply, counted}:
228 (snapshot, ledger) = counted
229 do IO<Counts>:
230 sent : Result<&1, &1, Counts, Unit> <- Chan.send(Counts, reply, snapshot)
231 loop(inbox, running, closing, Waiting{ledger})
232 case Finishing{checked}:
233 (done, ledger) = checked
234 match done:
235 case True{}:
236 do IO<Counts>:
237 Chan.close(Command, inbox)
238 return final(ledger)
239 case False{}:
240 loop(inbox, running, closing, Waiting{ledger})
241 case Received{ledger, got}:
242 match got:
243 case None{}:
244 IO.pure(Counts, final(ledger))
245 case Some{command}:
246 match command:
247 case Acquire{kind, reply}:
248 loop(inbox, running, closing, Acquired{reply, acquire.running(running, kind, ledger)})
249 case Release{kind, reply}:
250 loop(inbox, running, closing, Released{reply, release(kind, ledger)})
251 case Snapshot{reply}:
252 loop(inbox, running, closing, Counted{reply, counts(ledger)})
253 case Quiesce{reply}:
254 do IO<Counts>:
255 sent : Result<&1, &1, Bool, Unit> <- Chan.send(Bool, reply, running)
256 loop(inbox, False{}, closing, Waiting{ledger})
257 case Running{reply}:
258 do IO<Counts>:
259 sent : Result<&1, &1, Bool, Unit> <- Chan.send(Bool, reply, running)
260 loop(inbox, running, closing, Waiting{ledger})
261 case Stop{}:
262 loop(inbox, running, True{}, Finishing{idle(True{}, counts(ledger))})
263
264def open(caps: Caps) -> IO(Chan(Command)):
265 do IO<Chan(Command)>:
266 +inbox : Chan(Command) <- Chan.new(Command, 0)
267 IO.spawn(Counts, loop(inbox, True{}, False{}, Waiting{ledger.open(caps)}))
268 return inbox
269
270# The affine server owner joins this actual actor result after closing its listener.
271def open.owner(caps: Caps) -> IO(Chan(Command) & Chan(Counts)):
272 do IO<Chan(Command) & Chan(Counts)>:
273 +inbox : Chan(Command) <- Chan.new(Command, 0)
274 finished : Chan(Counts) <- IO.fork(Counts, loop(inbox, True{}, False{}, Waiting{ledger.open(caps)}))
275 return (inbox, finished)
276
277def boolean.sent(sent: Result<&1, &1, Command, Unit>, +reply: Chan(Bool)) -> IO(Bool):
278 match sent:
279 case Fail{command}:
280 do IO<Bool>:
281 Chan.close(Bool, reply)
282 return False{}
283 case Done{u}:
284 IO.join(Bool, reply)
285
286# False also covers copied controls used after actual controller closure.
287def quiesce(+inbox: Chan(Command)) -> IO(Bool):
288 do IO<Bool>:
289 +reply : Chan(Bool) <- Chan.new(Bool, 1)
290 sent : Result<&1, &1, Command, Unit> <- Chan.send(Command, inbox, Quiesce{reply})
291 boolean.sent(sent, reply)
292
293def running(+inbox: Chan(Command)) -> IO(Bool):
294 do IO<Bool>:
295 +reply : Chan(Bool) <- Chan.new(Bool, 1)
296 sent : Result<&1, &1, Command, Unit> <- Chan.send(Command, inbox, Running{reply})
297 boolean.sent(sent, reply)
298
299def take(+inbox: Chan(Command), kind: U32) -> IO(Bool):
300 do IO<Bool>:
301 +reply : Chan(Bool) <- Chan.new(Bool, 1)
302 sent : Result<&1, &1, Command, Unit> <- Chan.send(Command, inbox, Acquire{kind, reply})
303 boolean.sent(sent, reply)
304
305# Correct owners cannot retire the actor while a lease remains counted. If an
306# internal channel is externally closed, report the violation instead of joining
307# an unanswered reply or pretending a retained operation drained.
308def give.sent(sent: Result<&1, &1, Command, Unit>, +reply: Chan(Unit)) -> IO(Unit):
309 match sent:
310 case Fail{command}:
311 do IO<Unit>:
312 Chan.close(Unit, reply)
313 IO.die(Unit, 1, "Admission.give: controller closed while lease retained")
314 case Done{u}:
315 IO.join(Unit, reply)
316
317def give(+inbox: Chan(Command), kind: U32) -> IO(Unit):
318 do IO<Unit>:
319 +reply : Chan(Unit) <- Chan.new(Unit, 1)
320 sent : Result<&1, &1, Command, Unit> <- Chan.send(Command, inbox, Release{kind, reply})
321 give.sent(sent, reply)
322
323def snapshot.sent(sent: Result<&1, &1, Command, Unit>, +reply: Chan(Counts)) -> IO(Maybe<&2, Counts>):
324 match sent:
325 case Fail{command}:
326 do IO<Maybe<&2, Counts>>:
327 Chan.close(Counts, reply)
328 return None{}
329 case Done{u}:
330 do IO<Maybe<&2, Counts>>:
331 counts : Counts <- IO.join(Counts, reply)
332 return Some{counts}
333
334# A successful rendezvous send was received: Stop cannot erase a queued request.
335# After controller closure a failed send returns None without waiting on its reply.
336def snapshot(+inbox: Chan(Command)) -> IO(Maybe<&2, Counts>):
337 do IO<Maybe<&2, Counts>>:
338 +reply : Chan(Counts) <- Chan.new(Counts, 1)
339 sent : Result<&1, &1, Command, Unit> <- Chan.send(Command, inbox, Snapshot{reply})
340 snapshot.sent(sent, reply)
341
342# Retire accounting after listener closure; quiesce is the separate admission
343# cutover. Finite internal fixtures may finish their already-accepted protocol.
344def close(inbox: Chan(Command)) -> IO(Unit):
345 do IO<Unit>:
346 sent : Result<&1, &1, Command, Unit> <- Chan.send(Command, inbox, Stop{})
347 return Unit{}