manage_test.odin (5486B)
1 package flow 2 3 import "core:testing" 4 5 // A binary tree flattened into indices: node i holds 2i+1 and 2i+2. Walking it is 6 // the shape manage is for, since a worker only learns of a node by visiting its 7 // parent, and it is deterministic enough to check exactly. 8 @(private = "file") 9 NODES :: 20_000 10 11 @(private = "file") 12 Visit :: struct { 13 seen: [dynamic]int, 14 found: [dynamic]int, 15 stop: int, // visit this node and the run ends; -1 for never 16 sink: int, // keeps the busy loop below from being optimised away 17 } 18 19 @(private = "file") 20 descend :: proc(item: int, v: ^Visit) -> bool { 21 append(&v.seen, item) 22 if item == v.stop { 23 return false 24 } 25 for c in ([]int{2 * item + 1, 2 * item + 2}) { 26 if c < NODES { 27 append(&v.found, c) 28 } 29 } 30 return true 31 } 32 33 @(private = "file") 34 hand_over :: proc(item: int, ok: bool, v: ^Visit, queue: ^[dynamic]int) -> bool { 35 for c in v.found { 36 append(queue, c) 37 } 38 clear(&v.found) 39 return ok 40 } 41 42 /* 43 Worker states, each holding what its worker saw. 44 45 The dynamic arrays are deliberately not on the temp allocator. A dynamic array 46 remembers the allocator it was made with, and these are appended to from every 47 worker at once, so a per-thread arena belonging to whichever thread built them would 48 be grown from all of them without a lock. That miscounted a node roughly once in 49 twelve runs. `delete` them with `release`. 50 */ 51 @(private = "file") 52 visitors :: proc(n: int, stop := -1) -> []Visit { 53 v := make([]Visit, n, context.temp_allocator) 54 for i in 0 ..< n { 55 v[i] = Visit { 56 seen = make([dynamic]int, context.allocator), 57 found = make([dynamic]int, context.allocator), 58 stop = stop, 59 } 60 } 61 return v 62 } 63 64 @(private = "file") 65 release :: proc(v: []Visit) { 66 for s in v { 67 delete(s.seen) 68 delete(s.found) 69 } 70 } 71 72 @(private = "file") 73 totals :: proc(v: []Visit) -> (count, sum: int) { 74 for s in v { 75 for i in s.seen { 76 count += 1 77 sum += i 78 } 79 } 80 return 81 } 82 83 @(test) 84 test_manage_reaches_every_node_once :: proc(t: ^testing.T) { 85 v := visitors(8) 86 defer release(v) 87 manage([]int{0}, v, descend, hand_over) 88 89 // The count proves how many were visited and the sum proves which, so together 90 // they rule out one node twice and another never. 91 count, sum := totals(v) 92 testing.expect_value(t, count, NODES) 93 testing.expect_value(t, sum, NODES * (NODES - 1) / 2) 94 } 95 96 @(test) 97 test_manage_agrees_with_one_worker :: proc(t: ^testing.T) { 98 // A single worker runs inline with no threads at all, which is the yardstick the 99 // concurrent run has to match. 100 one := visitors(1) 101 defer release(one) 102 manage([]int{0}, one, descend, hand_over) 103 count, sum := totals(one) 104 testing.expect_value(t, count, NODES) 105 testing.expect_value(t, sum, NODES * (NODES - 1) / 2) 106 } 107 108 @(private = "file") 109 HEAVY :: 400 110 111 // A node has to cost appreciably more than starting a thread, or the calling thread 112 // drains the queue before the others are scheduled and the split says nothing. That 113 // is a property of the work, not of the traversal. 114 @(private = "file") 115 descend_slowly :: proc(item: int, v: ^Visit) -> bool { 116 acc := 0 117 for i in 0 ..< 400_000 { 118 acc += i ~ item 119 } 120 v.sink += acc & 1 121 append(&v.seen, item) 122 for c in ([]int{2 * item + 1, 2 * item + 2}) { 123 if c < HEAVY { 124 append(&v.found, c) 125 } 126 } 127 return true 128 } 129 130 @(test) 131 test_manage_spreads_across_workers :: proc(t: ^testing.T) { 132 v := visitors(4) 133 defer release(v) 134 manage([]int{0}, v, descend_slowly, hand_over) 135 136 count, sum := totals(v) 137 testing.expect_value(t, count, HEAVY) 138 testing.expect_value(t, sum, HEAVY * (HEAVY - 1) / 2) 139 140 busy := 0 141 for s in v { 142 if len(s.seen) > 0 { 143 busy += 1 144 } 145 } 146 testing.expect(t, busy > 1, "the traversal stayed on a single worker") 147 } 148 149 @(test) 150 test_manage_stops_when_the_manager_returns_false :: proc(t: ^testing.T) { 151 // One worker keeps this exact: with several, those already holding an item 152 // finish it, which is the documented behaviour. 153 one := visitors(1, stop = 0) 154 defer release(one) 155 manage([]int{0}, one, descend, hand_over) 156 count, _ := totals(one) 157 testing.expect_value(t, count, 1) 158 } 159 160 @(test) 161 test_manage_tolerates_an_empty_seed :: proc(t: ^testing.T) { 162 v := visitors(4) 163 defer release(v) 164 manage([]int{}, v, descend, hand_over) 165 count, _ := totals(v) 166 testing.expect_value(t, count, 0) 167 } 168 169 // A failed item is the manager's to judge, and the judgement it cannot make without 170 // being told which item failed is to try that one again. 171 @(private = "file") 172 Attempt :: struct { 173 id: int, 174 tries: int, 175 } 176 177 @(private = "file") 178 Attempt_Log :: struct { 179 seen: [dynamic]Attempt, 180 } 181 182 @(private = "file") 183 refuse_twice :: proc(item: Attempt, f: ^Attempt_Log) -> bool { 184 append(&f.seen, item) 185 return !(item.id == 1 && item.tries < 2) 186 } 187 188 @(private = "file") 189 retry :: proc(item: Attempt, ok: bool, f: ^Attempt_Log, queue: ^[dynamic]Attempt) -> bool { 190 if !ok { 191 append(queue, Attempt{id = item.id, tries = item.tries + 1}) 192 } 193 return true 194 } 195 196 @(test) 197 test_manage_requeues_a_failed_item :: proc(t: ^testing.T) { 198 // One worker keeps the count exact; the point is the shape, not the width. 199 f := make([]Attempt_Log, 1, context.temp_allocator) 200 f[0].seen = make([dynamic]Attempt, context.temp_allocator) 201 seed := []Attempt{{id = 0}, {id = 1}, {id = 2}} 202 manage(seed, f, refuse_twice, retry) 203 204 // Three items, one of them attempted three times, and the run carried on past 205 // the failures rather than ending at the first. 206 testing.expect_value(t, len(f[0].seen), 5) 207 tries := 0 208 for a in f[0].seen { 209 if a.id == 1 { 210 tries += 1 211 } 212 } 213 testing.expect_value(t, tries, 3) 214 }