bend-kit-concurrency Releases (2): 0.2.0.0 0x67bc295335339eaf9387ddb5aa34190a indexed https://jend.leeeet.dev/files.txt?path=bend-kit-concurrency@0.2.0.0 0.1.0.0 0x9c5a51d7f05acfc6ef075fa6810ffa30 unindexed https://jend.leeeet.dev/files.txt?path=bend-kit-concurrency@0.1.0.0 Definitions (30): Concurrency.count.cons (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=count.cons) # The length of xs, beside xs. def count.cons(-A: Type, h: A, r: Nat & List) -> Nat & List: Concurrency.count (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=count) def count(-A: Type, xs: List) -> Nat & List: Concurrency.split.cons (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=split.cons) # The first n elements of xs, and the rest. def split.cons(-A: Type, h: A, r: List & List) -> List & List: Concurrency.split (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=split) def split(-A: Type, xs: List, n: Nat) -> List & List: Concurrency.par_map.list.go (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_map.list.go) # lr holds two halves of nl and nr elements. Depth d runs 2^(d+1) parallel parts. def par_map.list.go( ~A: Type, ~B: Type, ~f: A -> B, d: Nat, +nl: Nat, +nr: Nat, lr: List & List ) -> List: Concurrency.par_map.list.top (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_map.list.top) def par_map.list.top(~A: Type, ~B: Type, ~f: A -> B, d: Nat, nx: Nat & List) -> List: Concurrency.par_map (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_map) # List.map(f, xs), on at most `workers` parallel parts. def par_map(~A: Type, ~B: Type, ~f: A -> B, workers: U32, xs: List) -> List: Concurrency.fold (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=fold) def fold(~A: Data, ~f: A -> A -> A, xs: List, acc: A) -> A: Concurrency.par_reduce.list.go (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_reduce.list.go) def par_reduce.list.go( ~A: Data, ~f: A -> A -> A, d: Nat, +nl: Nat, +nr: Nat, +z: A, lr: List & List ) -> A: Concurrency.par_reduce.list.top (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_reduce.list.top) def par_reduce.list.top(~A: Data, ~f: A -> A -> A, d: Nat, +z: A, nx: Nat & List) -> A: Concurrency.par_reduce (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_reduce) # The left fold of xs from z, on at most `workers` parallel parts. f must be # associative and z its identity. def par_reduce(~A: Data, ~f: A -> A -> A, workers: U32, +z: A, xs: List) -> A: Concurrency.par_map.array.go (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_map.array.go) # An Array is a balanced tree, so each ANode above depth d is a parallel call. def par_map.array.go(~A: Type, ~B: Type, ~f: A -> B, a: Array, d: Nat) -> Array: Concurrency.par_map.array (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_map.array) # f on every slot of a, on at most `workers` parallel parts. Slots keep their index. def par_map.array(~A: Type, ~B: Type, ~f: A -> B, workers: U32, a: Array) -> Array: Concurrency.par_reduce.array.go (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_reduce.array.go) def par_reduce.array.go(~A: Type, ~f: A -> A -> A, a: Array, d: Nat) -> A: Concurrency.par_reduce.array (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=par_reduce.array) # The slots of a combined in index order with an associative f, on at most # `workers` parallel parts. An Array is never empty, so no identity is needed. def par_reduce.array(~A: Type, ~f: A -> A -> A, workers: U32, a: Array) -> A: Concurrency.Worker (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=Worker) # A pool worker between channel steps. It owns its state s alone. type Worker<-S: Type, -A: Type, -B: Type> is Type: WIdle{s: S} WGot{s: S, got: Maybe<&1, A & Chan(B)>} WDone{sb: S & B, reply: Chan(B)} Concurrency.pool.worker (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=pool.worker) # Takes jobs until the queue is closed and empty, then answers its state. # Each job takes three steps, so 3 * (jobs + 1) fuel is enough. def pool.worker( ~S: Type, ~A: Type, ~B: Type, ~work: S -> A -> IO(S & B), fuel: Nat, +q: Chan(A & Chan(B)), w: Worker ) -> IO(S): Concurrency.pool.put (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=pool.put) # Queues each job with its own reply channel, and answers the replies in order. def pool.put(~A: Type, ~B: Type, xs: List, +q: Chan(A & Chan(B))) -> IO(List): Concurrency.pool.start (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=pool.start) def pool.start( ~S: Type, ~A: Type, ~B: Type, ~work: S -> A -> IO(S & B), states: List, +fuel: Nat, +q: Chan(A & Chan(B)) ) -> IO(List): Concurrency.pool.join (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=pool.join) def pool.join(-A: Type, cs: List) -> IO(List): Concurrency.pool.run (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=pool.run) def pool.run( ~S: Type, ~A: Type, ~B: Type, ~work: S -> A -> IO(S & B), states: List, nx: Nat & List ) -> IO(List & List): Concurrency.pool (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=pool) # Runs work(s, x) for every x in xs, on one worker per state in states. A # worker owns its state and threads it through its jobs, so an affine handle # such as a connection pool is never shared. Answers the final states in # worker order and the results in the order of xs. def pool( ~S: Type, ~A: Type, ~B: Type, ~work: S -> A -> IO(S & B), states: List, xs: List ) -> IO(List & List): Concurrency.Relay (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=Relay) # A select relay between channel steps. type Relay<-A: Type> is Type: RWait{} RGot{got: Maybe<&1, A>} RSent{ok: Result<&1, &1, U32 & A, Unit>} Concurrency.select.relay (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=select.relay) # Forwards each value of src to out as (i, value), then reports on done. # It stops when src closes, or when out is closed under it. def select.relay( -A: Type, fuel: Nat, +i: U32, +src: Chan(A), +out: Chan(U32 & A), +done: Chan(Unit), r: Relay ) -> IO(Result<&1, &1, Unit, Unit>): Concurrency.select.start (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=select.start) # ponytail: relay fuel is the largest Nat literal, and a value takes three # steps, so a relay stops after about 1.4 billion values. Drop the fuel if # Base gains an unbounded loop. def select.start( -A: Type, chans: List, +i: U32, +out: Chan(U32 & A), +done: Chan(Unit) ) -> IO(Nat): Concurrency.select.close (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=select.close) # Closes out once all n relays have reported. def select.close(-A: Type, n: Nat, +out: Chan(U32 & A), +done: Chan(Unit)) -> IO(Unit): Concurrency.select.go (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=select.go) def select.go(-A: Type, chans: List, +out: Chan(U32 & A), +done: Chan(Unit)) -> IO(Chan(U32 & A)): Concurrency.select (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=select) # One channel that carries (i, value) for each value received on chans[i], in # the order they arrive. Chan.recv on it waits for whichever source is ready # first, and answers None once every source is closed. The selector owns the # receiving side of its sources. Close it to stop early: each relay then drops # the one value it holds, if any, and stops. def select(-A: Type, chans: List) -> IO(Chan(U32 & A)): Concurrency.timeout.won (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=timeout.won) def timeout.won(-A: Type, +c: Chan(Maybe<&1, A>), got: Maybe<&1, Maybe<&1, A>>) -> IO(Maybe<&1, A>): Concurrency.timeout (https://jend.leeeet.dev/definition.txt?path=bend-kit-concurrency@0.2.0.0/concurrency.bend&name=timeout) # Some{result} if act finishes within ms milliseconds, else None at the # deadline. act is not cancelled: it runs on, its late result is dropped, and # the program waits for it (and for the timer) before it exits. def timeout(-A: Type, ms: U32, act: IO(A)) -> IO(Maybe<&1, A>):