1	# Private bounded transport accounting. Replies contain copied data, never resources.
2	import Base
3	
4	type 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	
9	type Caps is Data:
10	  Caps{connections: U32, active: U32, buffered: U32}
11	
12	type 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	
20	def empty() -> Counts:
21	  Counts{0, 0, 0, 0, 0, 0, 0n, 0n, 0n}
22	
23	def 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	
34	def 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	
42	def 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	
53	def 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	
60	def 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	
67	def 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	
71	type 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
79	def 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	
126	def 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.
133	def 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	
139	def 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.
149	def 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	
155	def 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	
161	def 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.
170	def 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	
179	def 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	
185	def 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.
198	def 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.
206	def close(inbox: Chan(Command)) -> IO(Unit):
207	  do IO<Unit>:
208	    sent : Bool <- Chan.send(Command, inbox, Stop{})
209	    return Unit{}