admission.bend
bend-kit-http@0.28.0.0 · 148 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 Stop{}
17
18def empty() -> Counts:
19 Counts{0, 0, 0, 0, 0, 0, 0n, 0n, 0n}
20
21def 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
32def 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
40def 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
51def 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
58def 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
62type 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
69def 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
101def 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
108def 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
114def 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
120def 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
126def 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.
139def 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
145def close(inbox: Chan(Command)) -> IO(Unit):
146 do IO<Unit>:
147 sent : Bool <- Chan.send(Command, inbox, Stop{})
148 return Unit{}