def pool.worker

bend-kit-concurrency@0.2.0.0 · concurrency.bend:150

import Base

150# Takes jobs until the queue is closed and empty, then answers its state.
151# Each job takes three steps, so 3 * (jobs + 1) fuel is enough.
152def pool.worker(
153  ~S: Type, ~A: Type, ~B: Type, ~work: S -> A -> IO(S & B), fuel: Nat,
154  +q: Chan(A & Chan(B)), w: Worker<S, A, B>
155) -> IO(S):
156  match fuel:
157    case 0n:
158      IO.die(S, 1, "Conc.pool: a worker ran out of fuel")
159    case 1n+p:
160      match w:
161        case WIdle{s}:
162          do IO<S>:
163            got : Maybe<&1, A & Chan(B)> <- Chan.recv(A & Chan(B), q)
164            pool.worker(~S, ~A, ~B, ~work, p, q, WGot{s, got})
165        case WGot{s, got}:
166          match got:
167            case None{}:
168              IO.pure(S, s)
169            case Some{(a, reply)}:
170              do IO<S>:
171                sb : S & B <- work(s, a)
172                pool.worker(~S, ~A, ~B, ~work, p, q, WDone{sb, reply})
173        case WDone{sb, reply}:
174          (s, b) = sb
175          do IO<S>:
176            sent : Result<&1, &1, B, Unit> <- Chan.send(B, reply, b)
177            pool.worker(~S, ~A, ~B, ~work, p, q, WIdle{s})