def pool.run

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

import Base

215def pool.run(
216  ~S: Type, ~A: Type, ~B: Type, ~work: S -> A -> IO(S & B), states: List<S>,
217  nx: Nat & List<A>
218) -> IO(List<S> & List<B>):
219  (+n, xs) = nx
220  do IO<List<S> & List<B>>:
221    queue : Chan(A & Chan(B)) <- Chan.new(A & Chan(B), U32.from_nat(n))
222    +q : Chan(A & Chan(B)) = queue
223    replies : List<Chan(B)> <- pool.put(~A, ~B, xs, q)
224    Chan.close(A & Chan(B), q)
225    workers : List<Chan(S)> <- pool.start(~S, ~A, ~B, ~work, states, Nat.mul(3n, 1n+n), q)
226    outs : List<B> <- pool.join(B, replies)
227    done : List<S> <- pool.join(S, workers)
228    return (done, outs)