camber.bend
bend-kit-camber@0.1.0.0 · 293 lines · 0 definitions
1# Bounded affine dependency bundles with typed concurrent operations and explicit close.
2import Base
3
4type Counts is Data:
5 Counts{capacity: Nat, in_use: Nat, completed: Nat, rejected: Nat}
6
7type Outcome<-E: Data, -O: Data> is Data:
8 Success{value: O}
9 Expected{error: E}
10 Exhausted{}
11 Stopped{}
12
13type CloseResult is Data:
14 Closed{}
15 Busy{counts: Counts}
16
17type Message<-R: Type, -P: Data, -I: Data, -E: Data, -O: Data> is Type:
18 Submit{principal: P, input: I, reply: Chan(Outcome<E, O>)}
19 Returned{resource: R}
20 Inspect{reply: Chan(Counts)}
21 Shutdown{reply: Chan(CloseResult)}
22
23# Copy only this context, never the resources. The inbox bounds queued commands;
24# callers must count blocked submissions inside their transport admission limit.
25type Context<-R: Type, -P: Data, -I: Data, -E: Data, -O: Data> is Data:
26 Context{messages: Chan(Message<R, P, I, E, O>)}
27
28def outcome.value(-E: Data, -O: Data, result: Result<&2, &2, E, O>) -> Outcome<E, O>:
29 match result:
30 case Fail{error}:
31 Expected{error}
32 case Done{value}:
33 Success{value}
34
35# The resources return through the private owner protocol BEFORE a disposable
36# caller reply. A closed caller reply cannot erase a handle or strand capacity.
37def work.return(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data,
38 messages: Chan(Message<R, P, I, E, O>), reply: Chan(Outcome<E, O>),
39 result: R & Result<&2, &2, E, O>
40) -> IO(Unit):
41 (returned, outcome) = result
42 do IO<Unit>:
43 accepted : Bool <- Chan.send(Message<R, P, I, E, O>, messages, Returned{returned})
44 sent : Bool <- Chan.send(Outcome<E, O>, reply, outcome.value(E, O, outcome))
45 return Unit{}
46
47def work(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
48 ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>),
49 resource: R, principal: P, input: I,
50 messages: Chan(Message<R, P, I, E, O>), reply: Chan(Outcome<E, O>)
51) -> IO(Unit):
52 do IO<Unit>:
53 result : R & Result<&2, &2, E, O> <- handler(resource, principal, input)
54 work.return(R, P, I, E, O, messages, reply, result)
55
56def dispose(~R: Type, ~destroy: R -> IO(Unit), resources: List<R>) -> IO(Unit):
57 match resources:
58 case Nil{}:
59 IO.pure(Unit, Unit{})
60 case Con{resource, rest}:
61 do IO<Unit>:
62 destroy(resource)
63 dispose(~R, ~destroy, rest)
64
65# Closing Base channels retains buffered messages. Drain them and answer stopped
66# callers, rather than stranding accepted commands behind the close command.
67def drain.message(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
68 ~destroy: R -> IO(Unit), message: Message<R, P, I, E, O>
69) -> IO(Unit):
70 match message:
71 case Submit{principal, input, reply}:
72 do IO<Unit>:
73 sent : Bool <- Chan.send(Outcome<E, O>, reply, Stopped{})
74 return Unit{}
75 case Inspect{reply}:
76 Chan.close(Counts, reply)
77 case Shutdown{reply}:
78 do IO<Unit>:
79 sent : Bool <- Chan.send(CloseResult, reply, Closed{})
80 return Unit{}
81 case Returned{resource}:
82 destroy(resource)
83
84def drain.step(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
85 ~destroy: R -> IO(Unit), pending: Maybe<&1, Message<R, P, I, E, O>>,
86 next: Unit -> IO(Unit)
87) -> IO(Unit):
88 match pending:
89 case None{}:
90 IO.pure(Unit, Unit{})
91 case Some{message}:
92 do IO<Unit>:
93 drain.message(~R, ~P, ~I, ~E, ~O, ~destroy, message)
94 next(Unit{})
95
96# The closed inbox contains at most room buffered messages.
97def drain(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
98 ~destroy: R -> IO(Unit), fuel: Nat, +messages: Chan(Message<R, P, I, E, O>)
99) -> IO(Unit):
100 match fuel:
101 case 0n:
102 IO.pure(Unit, Unit{})
103 case 1n+p:
104 do IO<Unit>:
105 pending : Maybe<&1, Message<R, P, I, E, O>> <- Chan.recv(Message<R, P, I, E, O>, messages)
106 drain.step(~R, ~P, ~I, ~E, ~O, ~destroy, pending, u => drain(~R, ~P, ~I, ~E, ~O, ~destroy, p, messages))
107
108type OwnerStep<-R: Type, -P: Data, -I: Data, -E: Data, -O: Data> is Type:
109 Waiting{resources: List<R>, counts: Counts}
110 Received{message: Message<R, P, I, E, O>, counts: Counts, resources: List<R>}
111
112# IO.join closes a one-shot channel; the owner inbox must remain reusable.
113def receive.value(-A: Type, value: Maybe<&1, A>) -> IO(A):
114 match value:
115 case None{}:
116 IO.die(A, 1, "camber internal channel closed unexpectedly")
117 case Some{message}:
118 IO.pure(A, message)
119
120def receive(-A: Type, channel: Chan(A)) -> IO(A):
121 do IO<A>:
122 value : Maybe<&1, A> <- Chan.recv(A, channel)
123 receive.value(A, value)
124
125# Only idle close ends this loop. Busy close never waits on a stalled handler.
126# @unsafe is solely the lifecycle receive loop, not a termination/proof claim.
127@unsafe def owner(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
128 ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>), ~destroy: R -> IO(Unit),
129 +messages: Chan(Message<R, P, I, E, O>), +room: U32, step: OwnerStep<R, P, I, E, O>
130) -> IO(Unit):
131 match step:
132 case Waiting{resources, counts}:
133 do IO<Unit>:
134 message : Message<R, P, I, E, O> <- receive(Message<R, P, I, E, O>, messages)
135 owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Received{message, counts, resources})
136 case Received{message, counts, resources}:
137 match message:
138 case Returned{resource}:
139 Counts{capacity, in_use, completed, rejected} = counts
140 owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resource <> resources, Counts{capacity, Nat.sub(in_use, 1n), 1n+completed, rejected}})
141 case Inspect{reply}:
142 Counts{+capacity, +in_use, +completed, +rejected} = counts
143 do IO<Unit>:
144 sent : Bool <- Chan.send(Counts, reply, Counts{capacity, in_use, completed, rejected})
145 owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, Counts{capacity, in_use, completed, rejected}})
146 case Shutdown{reply}:
147 Counts{+capacity, in_use, +completed, +rejected} = counts
148 match in_use:
149 case 0n:
150 do IO<Unit>:
151 dispose(~R, ~destroy, resources)
152 Chan.close(Message<R, P, I, E, O>, messages)
153 sent : Bool <- Chan.send(CloseResult, reply, Closed{})
154 drain(~R, ~P, ~I, ~E, ~O, ~destroy, U32.to_nat(room), messages)
155 case 1n+ +p:
156 do IO<Unit>:
157 sent : Bool <- Chan.send(CloseResult, reply, Busy{Counts{capacity, 1n+p, completed, rejected}})
158 owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, Counts{capacity, 1n+p, completed, rejected}})
159 case Submit{principal, input, reply}:
160 Counts{capacity, in_use, completed, rejected} = counts
161 match resources:
162 case Nil{}:
163 do IO<Unit>:
164 sent : Bool <- Chan.send(Outcome<E, O>, reply, Exhausted{})
165 owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{Nil{}, Counts{capacity, in_use, completed, 1n+rejected}})
166 case Con{resource, rest}:
167 do IO<Unit>:
168 IO.spawn(Unit, work(~R, ~P, ~I, ~E, ~O, ~handler, resource, principal, input, messages, reply))
169 owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{rest, Counts{capacity, 1n+in_use, completed, rejected}})
170
171def build.step(~R: Type, ~E: Data, ~destroy: R -> IO(Unit),
172 result: Result<&2, &1, E, R>, resources: List<R>,
173 next: List<R> -> IO(Result<&2, &1, E, List<R>>)
174) -> IO(Result<&2, &1, E, List<R>>):
175 match result:
176 case Fail{error}:
177 do IO<Result<&2, &1, E, List<R>>>:
178 dispose(~R, ~destroy, resources)
179 return Fail{error}
180 case Done{resource}:
181 next(resource <> resources)
182
183def build(~K: Data, ~R: Type, ~E: Data, ~create: K -> Nat -> IO(Result<&2, &1, E, R>), ~destroy: R -> IO(Unit),
184 remaining: Nat, +config: K, resources: List<R>
185) -> IO(Result<&2, &1, E, List<R>>):
186 match remaining:
187 case 0n:
188 IO.pure(Result<&2, &1, E, List<R>>, Done{resources})
189 case 1n+ +p:
190 do IO<Result<&2, &1, E, List<R>>>:
191 result : Result<&2, &1, E, R> <- create(config, p)
192 build.step(~R, ~E, ~destroy, result, resources, next => build(~K, ~R, ~E, ~create, ~destroy, p, config, next))
193
194def start.ready(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
195 ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>), ~destroy: R -> IO(Unit),
196 capacity: Nat, +room: U32, result: Result<&2, &1, E, List<R>>
197) -> IO(Result<&2, &2, E, Context<R, P, I, E, O>>):
198 match result:
199 case Fail{error}:
200 IO.pure(Result<&2, &2, E, Context<R, P, I, E, O>>, Fail{error})
201 case Done{resources}:
202 do IO<Result<&2, &2, E, Context<R, P, I, E, O>>>:
203 channel : Chan(Message<R, P, I, E, O>) <- Chan.new(Message<R, P, I, E, O>, room)
204 +messages : Chan(Message<R, P, I, E, O>) = channel
205 IO.spawn(Unit, owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, Counts{capacity, 0n, 0n, 0n}}))
206 return Done{Context{messages}}
207
208# The initializer closes partial handles within a failed bundle; start closes
209# all prior whole bundles. No workers exist until the complete set opens.
210def start(~K: Data, ~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
211 ~create: K -> Nat -> IO(Result<&2, &1, E, R>),
212 ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>), ~destroy: R -> IO(Unit),
213 config: K, +capacity: Nat, room: U32
214) -> IO(Result<&2, &2, E, Context<R, P, I, E, O>>):
215 do IO<Result<&2, &2, E, Context<R, P, I, E, O>>>:
216 result : Result<&2, &1, E, List<R>> <- build(~K, ~R, ~E, ~create, ~destroy, capacity, config, Nil{})
217 start.ready(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, capacity, room, result)
218
219def submit(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data,
220 context: Context<R, P, I, E, O>, principal: P, input: I, reply: Chan(Outcome<E, O>)
221) -> IO(Bool):
222 Context{messages} = context
223 Chan.send(Message<R, P, I, E, O>, messages, Submit{principal, input, reply})
224
225def call.accepted(-E: Data, -O: Data, accepted: Bool, +reply: Chan(Outcome<E, O>)) -> IO(Outcome<E, O>):
226 match accepted:
227 case False{}:
228 do IO<Outcome<E, O>>:
229 Chan.close(Outcome<E, O>, reply)
230 return Stopped{}
231 case True{}:
232 do IO<Outcome<E, O>>:
233 result : Outcome<E, O> <- IO.join(Outcome<E, O>, reply)
234 Chan.close(Outcome<E, O>, reply)
235 return result
236
237def call(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data,
238 context: Context<R, P, I, E, O>, principal: P, input: I
239) -> IO(Outcome<E, O>):
240 do IO<Outcome<E, O>>:
241 channel : Chan(Outcome<E, O>) <- Chan.new(Outcome<E, O>, 1)
242 +reply : Chan(Outcome<E, O>) = channel
243 accepted : Bool <- submit(R, P, I, E, O, context, principal, input, reply)
244 call.accepted(E, O, accepted, reply)
245
246def inspect.value(value: Maybe<&1, Counts>) -> Maybe<&2, Counts>:
247 match value:
248 case None{}:
249 None{}
250 case Some{counts}:
251 Some{counts}
252
253def inspect.sent(sent: Bool, +reply: Chan(Counts)) -> IO(Maybe<&2, Counts>):
254 match sent:
255 case False{}:
256 do IO<Maybe<&2, Counts>>:
257 Chan.close(Counts, reply)
258 return None{}
259 case True{}:
260 do IO<Maybe<&2, Counts>>:
261 value : Maybe<&1, Counts> <- Chan.recv(Counts, reply)
262 Chan.close(Counts, reply)
263 return inspect.value(value)
264
265def inspect(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data, context: Context<R, P, I, E, O>) -> IO(Maybe<&2, Counts>):
266 Context{messages} = context
267 do IO<Maybe<&2, Counts>>:
268 channel : Chan(Counts) <- Chan.new(Counts, 1)
269 +reply : Chan(Counts) = channel
270 sent : Bool <- Chan.send(Message<R, P, I, E, O>, messages, Inspect{reply})
271 inspect.sent(sent, reply)
272
273def close.sent(sent: Bool, +reply: Chan(CloseResult)) -> IO(CloseResult):
274 match sent:
275 case False{}:
276 do IO<CloseResult>:
277 Chan.close(CloseResult, reply)
278 return Closed{}
279 case True{}:
280 do IO<CloseResult>:
281 value : CloseResult <- IO.join(CloseResult, reply)
282 Chan.close(CloseResult, reply)
283 return value
284
285# Stop external admissions before close. Busy preserves live context/capacity;
286# retry after admitted operations finish. Stalled operations remain observable.
287def close(-R: Type, -P: Data, -I: Data, -E: Data, -O: Data, context: Context<R, P, I, E, O>) -> IO(CloseResult):
288 Context{messages} = context
289 do IO<CloseResult>:
290 channel : Chan(CloseResult) <- Chan.new(CloseResult, 1)
291 +reply : Chan(CloseResult) = channel
292 sent : Bool <- Chan.send(Message<R, P, I, E, O>, messages, Shutdown{reply})
293 close.sent(sent, reply)