1 # ts2vg's visibility graphs of time series, in Bend2, on every CPU thread. 2 # 3 # import bend-visgraph@2.0.0.0/visgraph.bend as VG 4 # import bend-visgraph@2.0.0.0/parallel.bend as Par 5 # 6 # Par.build and Par.degrees are VG.build and VG.degrees on every thread. 7 # Each def here that hands one array to two tasks relies on Base's @unsafe 8 # Array.fork: the tasks that share an array read it, write cells no other 9 # task writes, or add to cells atomically. 10 import Base 11 import ./visgraph.bend as VG 12 13 def fork3(a: Array & Array, b: Array & Array, c: Array & Array) -> VG.Cols & VG.Cols: 14 (a1, a2) = a 15 (b1, b2) = b 16 (c1, c2) = c 17 (VG.Cols{a1, b1, c1}, VG.Cols{a2, b2, c2}) 18 19 def fork_cols(c: VG.Cols) -> VG.Cols & VG.Cols: 20 VG.Cols{ys, lefts, rights} = c 21 fork3(Array.fork(F32, ys), Array.fork(U32, lefts), Array.fork(U32, rights)) 22 23 # The stack runs on blocks in parallel; a merge then walks the blocks in 24 # order. 25 type Blocks is Data: 26 BOne{b: VG.Block} 27 BJoin{a: Blocks, b: Blocks} 28 29 def block_of(l: VG.Local) -> Blocks: 30 VG.Local{b, c} = l 31 BOne{b} 32 33 # It forks nothing, so the runtime runs it once the tree has grown, on any 34 # idle thread; work done before a fork runs while the tree grows, on a few 35 # threads. 36 def block_leaf(+s: U32, +e: U32, c: VG.Cols) -> Blocks: 37 VG.Cols{ys, lefts, rights} = c 38 block_of(VG.block_pass(U32.to_nat((e - s : U32)), s, Array.get(F32, ys, s), lefts, rights)) 39 40 def block_tree(depth: Nat, +s: U32, +e: U32, cc: VG.Cols & VG.Cols) -> Blocks: 41 match depth: 42 case 1n++d: 43 (c1, c2) = cc 44 +mid = (s + U32.shr((e - s : U32)) : U32) 45 x y = block_tree(d, s, mid, fork_cols(c1)) block_tree(d, mid, e, fork_cols(c2)) 46 BJoin{x, y} 47 case 0n: 48 (c1, c2) = cc 49 +mid = (s + U32.shr((e - s : U32)) : U32) 50 x y = block_leaf(s, mid, c1) block_leaf(mid, e, c2) 51 BJoin{x, y} 52 53 def merge(bs: Blocks, m: VG.Merged) -> VG.Merged: 54 match bs: 55 case BOne{b}: 56 VG.merge_block(b, m) 57 case BJoin{a, b}: 58 merge(b, merge(a, m)) 59 60 type Chunks is Data: 61 CNil{} 62 CCon{start: U32, count: U32, next: Chunks} 63 64 def close(full: Bool, +start: U32, +end: U32, cs: Chunks) -> Chunks: 65 match full: 66 case True{}: 67 CCon{start, (end - start : U32), cs} 68 case False{}: 69 cs 70 71 type Cut is Type: 72 Cut{lefts: Array, rights: Array, cs: Chunks, m: U32} 73 74 def cut(count: Nat, +i: U32, lr: Array & U32, rr: Array & U32, +start: U32, +acc: U32, +target: U32, cs: Chunks, +m: U32) -> Cut: 75 match count: 76 case 0n: 77 (lefts, l) = lr 78 (rights, r) = rr 79 +last = U32.is_lt(start, i) 80 Cut{lefts, rights, close(last, start, i, cs), VG.pick_u32(last, (m + 1 : U32), m)} 81 case 1n+c: 82 (lefts, +l) = lr 83 (rights, +r) = rr 84 +w = (acc + l + r + 1 : U32) 85 +full = U32.is_ge(w, target) 86 +i2 = (i + 1 : U32) 87 cut(c, i2, Array.get(U32, lefts, i2), Array.get(U32, rights, i2), VG.pick_u32(full, i2, start), VG.pick_u32(full, 0, w), target, close(full, start, i2, cs), VG.pick_u32(full, (m + 1 : U32), m)) 88 89 def fork_depth() -> Nat: 90 8n 91 92 # At most 512 blocks, and at least 32 points in each unless n is below 64. 93 def block_depth(+n: U32) -> Nat: 94 Nat.min(fork_depth(), U32.log2(U32.shrn(n, 6n))) 95 96 # About two chunks for each of the tree's leaves. 97 def target(+visits: F32) -> U32: 98 U32.max(1, F32.to_u32(F32.div(visits, U32.to_f32(U32.shln(4, fork_depth()))))) 99 100 type Halves is Data: 101 Halves{a: Chunks, na: U32, b: Chunks, nb: U32} 102 103 def halve(k: Nat, cs: Chunks, +na: U32, +total: U32, acc: Chunks) -> Halves: 104 match k cs: 105 case 1n+q CCon{+start, +count, next}: 106 halve(q, next, na, total, CCon{start, count, acc}) 107 case _ _: 108 Halves{acc, na, cs, (total - na : U32)} 109 110 def halves(cs: Chunks, +m: U32) -> Halves: 111 halve(U32.to_nat(U32.shr(m)), cs, U32.shr(m), m, CNil{}) 112 113 def chunks(cs: Chunks, o: VG.Out, +index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32) -> VG.Out: 114 match cs: 115 case CCon{+start, count, next}: 116 VG.Out{cols, xs, es} = o 117 VG.Cols{ys, lefts, rights} = cols 118 chunks(next, VG.points(U32.to_nat(count), start, Array.get(U32, lefts, start), Array.get(U32, rights, start), VG.Walked{Array.get(F32, ys, start), Array.get(F32, xs, start), es}, index, g, dir, k, lo, hi), index, g, dir, k, lo, hi) 119 case CNil{}: 120 o 121 122 def leaf(cs: Chunks, c: VG.Cols, xs: Array, +index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32) -> VG.Edges: 123 VG.edges_of(chunks(cs, VG.Out{c, xs, VG.ENil{}}, index, g, dir, k, lo, hi)) 124 125 def tree(depth: Nat, h: Halves, cc: VG.Cols & VG.Cols, xx: Array & Array, +index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32) -> VG.Edges: 126 match depth h: 127 case 1n++d Halves{a, +na, b, +nb}: 128 (c1, c2) = cc 129 (x1, x2) = xx 130 x y = tree(d, halves(a, na), fork_cols(c1), Array.fork(F32, x1), index, g, dir, k, lo, hi) tree(d, halves(b, nb), fork_cols(c2), Array.fork(F32, x2), index, g, dir, k, lo, hi) 131 VG.EJoin{x, y} 132 case _ Halves{a, na, b, nb}: 133 (c1, c2) = cc 134 (x1, x2) = xx 135 x y = leaf(a, c1, x1, index, g, dir, k, lo, hi) leaf(b, c2, x2, index, g, dir, k, lo, hi) 136 VG.EJoin{x, y} 137 138 def split(+index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32, ys: Array, xs: Array, c: Cut) -> VG.Edges: 139 Cut{lefts, rights, cs, +m} = c 140 tree(fork_depth(), halves(cs, m), fork_cols(VG.Cols{ys, lefts, rights}), Array.fork(F32, xs), index, g, dir, k, lo, hi) 141 142 def cut_all(+index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32, +n: U32, ys: Array, xs: Array, lefts: Array, r: Array & F32) -> VG.Edges: 143 (rights, +visits) = r 144 split(index, g, dir, k, lo, hi, ys, xs, cut(U32.to_nat(n), 0, Array.get(U32, lefts, 0), Array.get(U32, rights, 0), 0, 0, target(visits), CNil{}, 0)) 145 146 def merged(+index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32, +n: U32, ys: Array, xs: Array, m: VG.Merged) -> VG.Edges: 147 VG.Merged{st, lefts, rights, +visits} = m 148 cut_all(index, g, dir, k, lo, hi, n, ys, xs, lefts, VG.flush(st, n, rights, visits)) 149 150 def passed(+index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32, +n: U32, xs: Array, cc: VG.Cols & VG.Cols) -> VG.Edges: 151 (c1, c2) = cc 152 VG.Cols{ys, lefts, rights} = c2 153 merged(index, g, dir, k, lo, hi, n, ys, xs, merge(block_tree(block_depth(n), 0, n, fork_cols(c1)), VG.Merged{VG.KNil{}, lefts, rights, 0.0})) 154 155 def built(+index: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32, ys: Array, xs: Array, +n: U32) -> VG.Edges: 156 passed(index, g, dir, k, lo, hi, n, xs, fork_cols(VG.Cols{ys, VG.slots(n), VG.slots(n)})) 157 158 # VG.build on every thread: the stack runs on blocks, then the sweeps on 159 # chunks of equal work. 160 def visibility(+g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32, ys: Array, xs: VG.Xs, +n: U32) -> VG.Edges: 161 match xs: 162 case VG.Index{}: 163 built(True{}, g, dir, k, lo, hi, ys, VG.no_xs(), n) 164 case VG.At{a}: 165 built(False{}, g, dir, k, lo, hi, ys, a, n) 166 167 def chosen(is_visibility: Bool, +g: VG.Visibility, +dir: VG.Direction, +k: VG.Weight, +lo: F32, +hi: F32, +limit: U32, ys: Array, xs: VG.Xs, +n: U32) -> VG.Edges: 168 match is_visibility: 169 case True{}: 170 visibility(g, dir, k, lo, hi, ys, xs, n) 171 case False{}: 172 VG.penetrable(g, dir, k, lo, hi, limit, ys, xs, n) 173 174 # VG.build on every thread. 175 def build(o: VG.Options, s: VG.Series) -> VG.Graph: 176 VG.Options{g, +dir, k, lo, hi, +limit} = o 177 VG.Series{+n, ys, xs} = s 178 VG.Graph{n, dir, chosen(U32.is_zero(limit), g, dir, k, lo, hi, limit, ys, xs, n)} 179 180 # Forks down the build's tree of EJoin into VG.degrees.into, which 181 # forks nothing, so the runtime spreads the tallies over every thread. 182 def degrees_tree(depth: Nat, es: VG.Edges, dd: Array & Array) -> Array: 183 match depth es: 184 case 1n++p VG.EJoin{a, b}: 185 (d1, d2) = dd 186 x y = degrees_tree(p, a, Array.fork(U32, d1)) degrees_tree(p, b, Array.fork(U32, d2)) 187 Array.join(U32, x, y) 188 case 0n VG.EJoin{a, b}: 189 (d1, d2) = dd 190 x y = VG.degrees.into(a, d1) VG.degrees.into(b, d2) 191 Array.join(U32, x, y) 192 case _ _: 193 (d1, d2) = dd 194 VG.degrees.into(es, Array.join(U32, d1, d2)) 195 196 # VG.degrees on every thread. 197 def degrees(g: VG.Graph) -> Array: 198 VG.Graph{+n, dir, es} = g 199 degrees_tree(fork_depth(), es, Array.fork(U32, VG.slots(n)))