1	# Private bounded transport accounting. Replies contain copied data, never resources.
2	import Base
3	import ../resources/resources.bend as Res
4	
5	type 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	
10	type Caps is Data:
11	  Caps{connections: U32, active: U32, buffered: U32}
12	
13	type 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.
22	type Lease is Type:
23	  Lease{pool: Res.Pool, held: Res.Grant}
24	
25	type 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	
29	type 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.
33	type Booked is Type:
34	  Booked{ledger: Ledger, admitted: Bool}
35	  Broken{}
36	
37	type Taken is Type:
38	  Admitted{lease: Lease}
39	  Full{lease: Lease}
40	  Lost{}
41	
42	def lease.open(+owner: Nat, capacity: U32) -> Lease:
43	  Lease{Res.new(owner, U32.to_nat(capacity)), Res.Grant{owner, 0n}}
44	
45	def 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	
49	def 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	
56	def 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	
63	def lease.take(lease: Lease) -> Taken:
64	  Lease{pool, held} = lease
65	  lease.took(Res.reserve(pool, 1n), held)
66	
67	def 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	
74	def 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	
81	def lease.give(lease: Lease) -> Maybe<&1, Lease>:
82	  Lease{pool, held} = lease
83	  lease.split(Res.split(held, 1n), pool)
84	
85	def count(+pool: Res.Pool) -> U32:
86	  U32.from_nat(Res.reserved(pool))
87	
88	def 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	
93	def final(ledger: Ledger) -> Counts:
94	  Pair.fst(Counts, Ledger, counts(ledger))
95	
96	# Raises each high-water mark to its current count.
97	def 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	
103	def 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.
113	def 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	
120	def 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.
130	def 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	
139	def 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	
148	def 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	
157	def 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	
164	def 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	
171	def 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	
178	def 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	
187	def 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	
191	def idle(closing: Bool, counted: Counts & Ledger) -> Bool & Ledger:
192	  (counts, ledger) = counted
193	  (stopped(closing, counts), ledger)
194	
195	type 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
205	def 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 : Bool <- 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 : Bool <- 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 : Bool <- 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 : Bool <- Chan.send(Bool, reply, running)
256	                loop(inbox, False{}, closing, Waiting{ledger})
257	            case Running{reply}:
258	              do IO<Counts>:
259	                sent : Bool <- 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	
264	def 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.
271	def 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	
277	def boolean.sent(sent: Bool, +reply: Chan(Bool)) -> IO(Bool):
278	  match sent:
279	    case False{}:
280	      do IO<Bool>:
281	        Chan.close(Bool, reply)
282	        return False{}
283	    case True{}:
284	      IO.join(Bool, reply)
285	
286	# False also covers copied controls used after actual controller closure.
287	def quiesce(+inbox: Chan(Command)) -> IO(Bool):
288	  do IO<Bool>:
289	    +reply : Chan(Bool) <- Chan.new(Bool, 1)
290	    sent : Bool <- Chan.send(Command, inbox, Quiesce{reply})
291	    boolean.sent(sent, reply)
292	
293	def running(+inbox: Chan(Command)) -> IO(Bool):
294	  do IO<Bool>:
295	    +reply : Chan(Bool) <- Chan.new(Bool, 1)
296	    sent : Bool <- Chan.send(Command, inbox, Running{reply})
297	    boolean.sent(sent, reply)
298	
299	def take(+inbox: Chan(Command), kind: U32) -> IO(Bool):
300	  do IO<Bool>:
301	    +reply : Chan(Bool) <- Chan.new(Bool, 1)
302	    sent : Bool <- 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.
308	def give.sent(sent: Bool, +reply: Chan(Unit)) -> IO(Unit):
309	  match sent:
310	    case False{}:
311	      do IO<Unit>:
312	        Chan.close(Unit, reply)
313	        IO.die(Unit, 1, "Admission.give: controller closed while lease retained")
314	    case True{}:
315	      IO.join(Unit, reply)
316	
317	def give(+inbox: Chan(Command), kind: U32) -> IO(Unit):
318	  do IO<Unit>:
319	    +reply : Chan(Unit) <- Chan.new(Unit, 1)
320	    sent : Bool <- Chan.send(Command, inbox, Release{kind, reply})
321	    give.sent(sent, reply)
322	
323	def snapshot.sent(sent: Bool, +reply: Chan(Counts)) -> IO(Maybe<&2, Counts>):
324	  match sent:
325	    case False{}:
326	      do IO<Maybe<&2, Counts>>:
327	        Chan.close(Counts, reply)
328	        return None{}
329	    case True{}:
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.
336	def snapshot(+inbox: Chan(Command)) -> IO(Maybe<&2, Counts>):
337	  do IO<Maybe<&2, Counts>>:
338	    +reply : Chan(Counts) <- Chan.new(Counts, 1)
339	    sent : Bool <- 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.
344	def close(inbox: Chan(Command)) -> IO(Unit):
345	  do IO<Unit>:
346	    sent : Bool <- Chan.send(Command, inbox, Stop{})
347	    return Unit{}