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	  Stop{}
17	
18	def empty() -> Counts:
19	  Counts{0, 0, 0, 0, 0, 0, 0n, 0n, 0n}
20	
21	def acquire.connection(full: Bool, buffer_full: Bool, counts: Counts) -> Counts & Bool:
22	  match full buffer_full counts:
23	    case True{} _ Counts{c, a, b, ch, ah, bh, cr, ar, br}:
24	      (Counts{c, a, b, ch, ah, bh, 1n+cr, ar, br}, False{})
25	    case False{} True{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
26	      (Counts{c, a, b, ch, ah, bh, cr, ar, 1n+br}, False{})
27	    case False{} False{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
28	      +nc = (c + 1 : U32)
29	      +nb = (b + 1 : U32)
30	      (Counts{nc, a, nb, U32.max(ch, nc), ah, U32.max(bh, nb), cr, ar, br}, True{})
31	
32	def acquire.active(full: Bool, counts: Counts) -> Counts & Bool:
33	  match full counts:
34	    case True{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
35	      (Counts{c, a, b, ch, ah, bh, cr, 1n+ar, br}, False{})
36	    case False{} Counts{c, a, b, ch, ah, bh, cr, ar, br}:
37	      +na = (a + 1 : U32)
38	      (Counts{c, na, b, ch, U32.max(ah, na), bh, cr, ar, br}, True{})
39	
40	def acquire(kind: U32, caps: Caps, +counts: Counts) -> Counts & Bool:
41	  match kind:
42	    case 0:
43	      Caps{cc, ac, bc} = caps
44	      Counts{c, a, b, ch, ah, bh, cr, ar, br} = counts
45	      acquire.connection(U32.is_ge(c, cc), U32.is_ge(b, bc), counts)
46	    case _:
47	      Caps{cc, ac, bc} = caps
48	      Counts{c, a, b, ch, ah, bh, cr, ar, br} = counts
49	      acquire.active(U32.is_ge(a, ac), counts)
50	
51	def release(kind: U32, counts: Counts) -> Counts:
52	  match kind counts:
53	    case 0 Counts{c, a, b, ch, ah, bh, cr, ar, br}:
54	      Counts{(c - 1 : U32), a, (b - 1 : U32), ch, ah, bh, cr, ar, br}
55	    case _ Counts{c, a, b, ch, ah, bh, cr, ar, br}:
56	      Counts{c, (a - 1 : U32), b, ch, ah, bh, cr, ar, br}
57	
58	def stopped(closing: Bool, counts: Counts) -> Bool:
59	  Counts{c, a, b, ch, ah, bh, cr, ar, br} = counts
60	  closing && U32.is_zero(c) && U32.is_zero(a) && U32.is_zero(b)
61	
62	type Step is Type:
63	  Waiting{}
64	  Received{command: Maybe<&1, Command>}
65	  Acquired{reply: Chan(Bool), result: Counts & Bool}
66	
67	# The private inbox is reusable: IO.join would close it after one receive.
68	@unsafe
69	def loop(+inbox: Chan(Command), +caps: Caps, +counts: Counts, +closing: Bool, step: Step) -> IO(Unit):
70	  match step:
71	    case Waiting{}:
72	      do IO<Unit>:
73	        got : Maybe<&1, Command> <- Chan.recv(Command, inbox)
74	        loop(inbox, caps, counts, closing, Received{got})
75	    case Acquired{reply, result}:
76	      (next, admitted) = result
77	      do IO<Unit>:
78	        sent : Bool <- Chan.send(Bool, reply, admitted)
79	        loop(inbox, caps, next, closing, Waiting{})
80	    case Received{got}:
81	      match got:
82	        case None{}:
83	          IO.pure(Unit, Unit{})
84	        case Some{command}:
85	          match command:
86	            case Acquire{kind, reply}:
87	              loop(inbox, caps, counts, closing, Acquired{reply, acquire(kind, caps, counts)})
88	            case Release{kind, reply}:
89	              +next = release(kind, counts)
90	              do IO<Unit>:
91	                sent : Bool <- Chan.send(Unit, reply, Unit{})
92	                finish(stopped(closing, next), inbox, caps, next, closing)
93	            case Snapshot{reply}:
94	              do IO<Unit>:
95	                sent : Bool <- Chan.send(Counts, reply, counts)
96	                loop(inbox, caps, counts, closing, Waiting{})
97	            case Stop{}:
98	              finish(stopped(True{}, counts), inbox, caps, counts, True{})
99	
100	@unsafe
101	def finish(done: Bool, +inbox: Chan(Command), caps: Caps, counts: Counts, closing: Bool) -> IO(Unit):
102	  match done:
103	    case True{}:
104	      Chan.close(Command, inbox)
105	    case False{}:
106	      loop(inbox, caps, counts, closing, Waiting{})
107	
108	def open(caps: Caps) -> IO(Chan(Command)):
109	  do IO<Chan(Command)>:
110	    +inbox : Chan(Command) <- Chan.new(Command, 0)
111	    IO.spawn(Unit, loop(inbox, caps, empty(), False{}, Waiting{}))
112	    return inbox
113	
114	def take(+inbox: Chan(Command), kind: U32) -> IO(Bool):
115	  do IO<Bool>:
116	    +reply : Chan(Bool) <- Chan.new(Bool, 1)
117	    sent : Bool <- Chan.send(Command, inbox, Acquire{kind, reply})
118	    IO.join(Bool, reply)
119	
120	def give(+inbox: Chan(Command), kind: U32) -> IO(Unit):
121	  do IO<Unit>:
122	    +reply : Chan(Unit) <- Chan.new(Unit, 1)
123	    sent : Bool <- Chan.send(Command, inbox, Release{kind, reply})
124	    IO.join(Unit, reply)
125	
126	def snapshot.sent(sent: Bool, +reply: Chan(Counts)) -> IO(Maybe<&2, Counts>):
127	  match sent:
128	    case False{}:
129	      do IO<Maybe<&2, Counts>>:
130	        Chan.close(Counts, reply)
131	        return None{}
132	    case True{}:
133	      do IO<Maybe<&2, Counts>>:
134	        counts : Counts <- IO.join(Counts, reply)
135	        return Some{counts}
136	
137	# A successful rendezvous send was received: Stop cannot erase a queued request.
138	# After controller closure a failed send returns None without waiting on its reply.
139	def snapshot(+inbox: Chan(Command)) -> IO(Maybe<&2, Counts>):
140	  do IO<Maybe<&2, Counts>>:
141	    +reply : Chan(Counts) <- Chan.new(Counts, 1)
142	    sent : Bool <- Chan.send(Command, inbox, Snapshot{reply})
143	    snapshot.sent(sent, reply)
144	
145	def close(inbox: Chan(Command)) -> IO(Unit):
146	  do IO<Unit>:
147	    sent : Bool <- Chan.send(Command, inbox, Stop{})
148	    return Unit{}