admission.bend
bend-kit-http@0.30.0.0 · 209 lines · 0 definitions
1# Private bounded transport accounting. Replies contain copied data, never resources.
2import Base
3
4type Counts is Data:
5 Counts{connections: U32, active: U32, buffered: U32,
6 connection_high: U32, active_high: U32, buffered_high: U32,
7 connection_rejected: Nat, active_rejected: Nat, buffered_rejected: Nat}
8
9type Caps is Data:
10 Caps{connections: U32, active: U32, buffered: U32}
11
12type Command is Data:
13 Acquire{kind: U32, reply: Chan(Bool)}
14 Release{kind: U32, reply: Chan(Unit)}
15 Snapshot{reply: Chan(Counts)}
16 Quiesce{reply: Chan(Bool)}
17 Running{reply: Chan(Bool)}
18 Stop{}
19
20def empty() -> Counts:
21 Counts{0, 0, 0, 0, 0, 0, 0n, 0n, 0n}
22
23def acquire.connection(full: Bool, buffer_full: Bool, counts: Counts) -> Counts & Bool:
24 match full buffer_full counts:
25 case True{} _ Counts{c, a, b, ch, ah, bh, cr, ar, br}:
26 (Counts{c, a, b, ch, ah, bh, 1n+cr, ar, br}, False{})
27 case False{} True{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
28 (Counts{c, a, b, ch, ah, bh, cr, ar, 1n+br}, False{})
29 case False{} False{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
30 +nc = (c + 1 : U32)
31 +nb = (b + 1 : U32)
32 (Counts{nc, a, nb, U32.max(ch, nc), ah, U32.max(bh, nb), cr, ar, br}, True{})
33
34def acquire.active(full: Bool, counts: Counts) -> Counts & Bool:
35 match full counts:
36 case True{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
37 (Counts{c, a, b, ch, ah, bh, cr, 1n+ar, br}, False{})
38 case False{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
39 +na = (a + 1 : U32)
40 (Counts{c, na, b, ch, U32.max(ah, na), bh, cr, ar, br}, True{})
41
42def acquire(kind: U32, caps: Caps, +counts: Counts) -> Counts & Bool:
43 match kind:
44 case 0:
45 Caps{cc, ac, bc} = caps
46 Counts{c, a, b, ch, ah, bh, cr, ar, br} = counts
47 acquire.connection(U32.is_ge(c, cc), U32.is_ge(b, bc), counts)
48 case _:
49 Caps{cc, ac, bc} = caps
50 Counts{c, a, b, ch, ah, bh, cr, ar, br} = counts
51 acquire.active(U32.is_ge(a, ac), counts)
52
53def acquire.running(running: Bool, kind: U32, caps: Caps, counts: Counts) -> Counts & Bool:
54 match running:
55 case False{}:
56 (counts, False{})
57 case True{}:
58 acquire(kind, caps, counts)
59
60def release(kind: U32, counts: Counts) -> Counts:
61 match kind counts:
62 case 0 Counts{c, a, b, ch, ah, bh, cr, ar, br}:
63 Counts{(c - 1 : U32), a, (b - 1 : U32), ch, ah, bh, cr, ar, br}
64 case _ Counts{c, a, b, ch, ah, bh, cr, ar, br}:
65 Counts{c, (a - 1 : U32), b, ch, ah, bh, cr, ar, br}
66
67def stopped(closing: Bool, counts: Counts) -> Bool:
68 Counts{c, a, b, ch, ah, bh, cr, ar, br} = counts
69 closing && U32.is_zero(c) && U32.is_zero(a) && U32.is_zero(b)
70
71type Step is Type:
72 Waiting{}
73 Received{command: Maybe<&1, Command>}
74 Acquired{reply: Chan(Bool), result: Counts & Bool}
75 Finishing{done: Bool}
76
77# The private inbox is reusable: IO.join would close it after one receive.
78@unsafe
79def loop(+inbox: Chan(Command), +caps: Caps, +counts: Counts, +running: Bool, +closing: Bool, step: Step) -> IO(Counts):
80 match step:
81 case Waiting{}:
82 do IO<Counts>:
83 got : Maybe<&1, Command> <- Chan.recv(Command, inbox)
84 loop(inbox, caps, counts, running, closing, Received{got})
85 case Acquired{reply, result}:
86 (next, admitted) = result
87 do IO<Counts>:
88 sent : Bool <- Chan.send(Bool, reply, admitted)
89 loop(inbox, caps, next, running, closing, Waiting{})
90 case Finishing{done}:
91 match done:
92 case True{}:
93 do IO<Counts>:
94 Chan.close(Command, inbox)
95 return counts
96 case False{}:
97 loop(inbox, caps, counts, running, closing, Waiting{})
98 case Received{got}:
99 match got:
100 case None{}:
101 IO.pure(Counts, counts)
102 case Some{command}:
103 match command:
104 case Acquire{kind, reply}:
105 loop(inbox, caps, counts, running, closing, Acquired{reply, acquire.running(running, kind, caps, counts)})
106 case Release{kind, reply}:
107 +next = release(kind, counts)
108 do IO<Counts>:
109 sent : Bool <- Chan.send(Unit, reply, Unit{})
110 loop(inbox, caps, next, running, closing, Finishing{stopped(closing, next)})
111 case Snapshot{reply}:
112 do IO<Counts>:
113 sent : Bool <- Chan.send(Counts, reply, counts)
114 loop(inbox, caps, counts, running, closing, Waiting{})
115 case Quiesce{reply}:
116 do IO<Counts>:
117 sent : Bool <- Chan.send(Bool, reply, running)
118 loop(inbox, caps, counts, False{}, closing, Waiting{})
119 case Running{reply}:
120 do IO<Counts>:
121 sent : Bool <- Chan.send(Bool, reply, running)
122 loop(inbox, caps, counts, running, closing, Waiting{})
123 case Stop{}:
124 loop(inbox, caps, counts, running, True{}, Finishing{stopped(True{}, counts)})
125
126def open(caps: Caps) -> IO(Chan(Command)):
127 do IO<Chan(Command)>:
128 +inbox : Chan(Command) <- Chan.new(Command, 0)
129 IO.spawn(Counts, loop(inbox, caps, empty(), True{}, False{}, Waiting{}))
130 return inbox
131
132# The affine server owner joins this actual actor result after closing its listener.
133def open.owner(caps: Caps) -> IO(Chan(Command) & Chan(Counts)):
134 do IO<Chan(Command) & Chan(Counts)>:
135 +inbox : Chan(Command) <- Chan.new(Command, 0)
136 finished : Chan(Counts) <- IO.fork(Counts, loop(inbox, caps, empty(), True{}, False{}, Waiting{}))
137 return (inbox, finished)
138
139def boolean.sent(sent: Bool, +reply: Chan(Bool)) -> IO(Bool):
140 match sent:
141 case False{}:
142 do IO<Bool>:
143 Chan.close(Bool, reply)
144 return False{}
145 case True{}:
146 IO.join(Bool, reply)
147
148# False also covers copied controls used after actual controller closure.
149def quiesce(+inbox: Chan(Command)) -> IO(Bool):
150 do IO<Bool>:
151 +reply : Chan(Bool) <- Chan.new(Bool, 1)
152 sent : Bool <- Chan.send(Command, inbox, Quiesce{reply})
153 boolean.sent(sent, reply)
154
155def running(+inbox: Chan(Command)) -> IO(Bool):
156 do IO<Bool>:
157 +reply : Chan(Bool) <- Chan.new(Bool, 1)
158 sent : Bool <- Chan.send(Command, inbox, Running{reply})
159 boolean.sent(sent, reply)
160
161def take(+inbox: Chan(Command), kind: U32) -> IO(Bool):
162 do IO<Bool>:
163 +reply : Chan(Bool) <- Chan.new(Bool, 1)
164 sent : Bool <- Chan.send(Command, inbox, Acquire{kind, reply})
165 boolean.sent(sent, reply)
166
167# Correct owners cannot retire the actor while a lease remains counted. If an
168# internal channel is externally closed, report the violation instead of joining
169# an unanswered reply or pretending a retained operation drained.
170def give.sent(sent: Bool, +reply: Chan(Unit)) -> IO(Unit):
171 match sent:
172 case False{}:
173 do IO<Unit>:
174 Chan.close(Unit, reply)
175 IO.die(Unit, 1, "Admission.give: controller closed while lease retained")
176 case True{}:
177 IO.join(Unit, reply)
178
179def give(+inbox: Chan(Command), kind: U32) -> IO(Unit):
180 do IO<Unit>:
181 +reply : Chan(Unit) <- Chan.new(Unit, 1)
182 sent : Bool <- Chan.send(Command, inbox, Release{kind, reply})
183 give.sent(sent, reply)
184
185def snapshot.sent(sent: Bool, +reply: Chan(Counts)) -> IO(Maybe<&2, Counts>):
186 match sent:
187 case False{}:
188 do IO<Maybe<&2, Counts>>:
189 Chan.close(Counts, reply)
190 return None{}
191 case True{}:
192 do IO<Maybe<&2, Counts>>:
193 counts : Counts <- IO.join(Counts, reply)
194 return Some{counts}
195
196# A successful rendezvous send was received: Stop cannot erase a queued request.
197# After controller closure a failed send returns None without waiting on its reply.
198def snapshot(+inbox: Chan(Command)) -> IO(Maybe<&2, Counts>):
199 do IO<Maybe<&2, Counts>>:
200 +reply : Chan(Counts) <- Chan.new(Counts, 1)
201 sent : Bool <- Chan.send(Command, inbox, Snapshot{reply})
202 snapshot.sent(sent, reply)
203
204# Retire accounting after listener closure; quiesce is the separate admission
205# cutover. Finite internal fixtures may finish their already-accepted protocol.
206def close(inbox: Chan(Command)) -> IO(Unit):
207 do IO<Unit>:
208 sent : Bool <- Chan.send(Command, inbox, Stop{})
209 return Unit{}