def owner

bend-kit-camber@0.15.0.1 · camber.bend:260

import Base

260# Only idle close ends this loop. Busy close never waits on a stalled handler.
261# @unsafe is solely the lifecycle receive loop, not a termination/proof claim.
262@unsafe def owner(~R: Type, ~P: Data, ~I: Data, ~E: Data, ~O: Data,
263  ~handler: R -> P -> I -> IO(R & Result<&2, &2, E, O>), ~destroy: R -> IO(Unit),
264  +messages: Chan(Message<R, P, I, E, O>), +room: U32, step: OwnerStep<R, P, I, E, O>
265) -> IO(Unit):
266  match step:
267    case Waiting{resources, ledger}:
268      do IO<Unit>:
269        message : Message<R, P, I, E, O> <- receive(Message<R, P, I, E, O>, messages)
270        owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Received{message, ledger, resources})
271    case Counted{reply, counted, resources}:
272      (counts, kept) = counted
273      do IO<Unit>:
274        sent : Result<&1, &1, Counts, Unit> <- Chan.send(Counts, reply, counts)
275        owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, kept})
276    case Closing{reply, counted, resources}:
277      (counts, kept) = counted
278      Counts{+capacity, in_use, +completed, +rejected} = counts
279      match in_use:
280        case 0n:
281          do IO<Unit>:
282            dispose(~R, ~destroy, resources)
283            Chan.close(Message<R, P, I, E, O>, messages)
284            sent : Result<&1, &1, CloseResult, Unit> <- Chan.send(CloseResult, reply, Closed{})
285            drain(~R, ~P, ~I, ~E, ~O, ~destroy, U32.to_nat(room), messages)
286        case 1n+p:
287          do IO<Unit>:
288            sent : Result<&1, &1, CloseResult, Unit> <- Chan.send(CloseResult, reply, Busy{Counts{capacity, 1n+p, completed, rejected}})
289            owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resources, kept})
290    case Received{message, ledger, resources}:
291      match message:
292        case Returned{resource}:
293          do IO<Unit>:
294            given : Ledger <- ledger.must(ledger.give(ledger), "camber: a resource returned without a matching submit")
295            owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{resource <> resources, given})
296        case Inspect{reply}:
297          owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Counted{reply, ledger.counts(ledger), resources})
298        case Shutdown{reply}:
299          owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Closing{reply, ledger.counts(ledger), resources})
300        case Submit{principal, input, reply}:
301          match resources:
302            case Nil{}:
303              do IO<Unit>:
304                sent : Result<&1, &1, Outcome<E, O>, Unit> <- Chan.send(Outcome<E, O>, reply, Exhausted{})
305                owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{Nil{}, ledger.refuse(ledger)})
306            case Con{resource, rest}:
307              do IO<Unit>:
308                lent : Ledger <- ledger.must(ledger.lend(ledger), "camber: a held resource found no free pool unit")
309                IO.spawn(Unit, work(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, resource, principal, input, messages, reply))
310                owner(~R, ~P, ~I, ~E, ~O, ~handler, ~destroy, messages, room, Waiting{rest, lent})