manage.odin (3918B)
1 package flow 2 3 import "core:sync" 4 import "core:thread" 5 6 /* 7 Run work over a set that grows as the work discovers more of it. 8 9 `seed` starts the queue; every later item arrives through `manager`, which runs one 10 at a time. `work` runs on many threads at once, taking the next item the moment one 11 exists rather than in rounds. 12 13 Each worker owns one slot of `states`, so `work` needs no locks, exactly as in `each`. 14 `manager` sees that slot beside the item that filled it, and queues whatever comes 15 next, that item included. 16 17 `work` returning false marks the item failed, not the run. `manager` returning false 18 ends the run; workers finish the item in hand and report through their state. 19 20 Every item reaches `work` or `discard`, never both and never neither. A stopped run 21 leaves items queued that nothing else can reach, so an item owning memory needs 22 `discard` to release it. 23 24 `len(states)` sets the width, capped by what the machine can use. 25 */ 26 manage :: proc( 27 seed: []$I, 28 states: []$S, 29 work: proc(item: I, state: ^S) -> bool, 30 manager: proc(item: I, ok: bool, state: ^S, queue: ^[dynamic]I) -> bool, 31 discard: proc(item: I) = nil, 32 load := Load.Io, 33 ) { 34 if len(seed) == 0 || len(states) == 0 { 35 return 36 } 37 q: Queue(I, S) 38 q.states = states 39 q.work = work 40 q.manager = manager 41 q.items = make([dynamic]I, context.allocator) 42 defer delete(q.items) 43 // Runs before the delete above, while the queue still holds what was never taken. 44 defer sweep(&q, discard) 45 append(&q.items, ..seed) 46 47 pool := min(len(states), width(1 << 30, load)) 48 if pool == 1 { 49 drain(&q, 0) 50 return 51 } 52 53 threads := make([]^thread.Thread, pool - 1, context.temp_allocator) 54 defer delete(threads, context.temp_allocator) 55 started := 0 56 for i in 0 ..< len(threads) { 57 t := thread.create_and_start_with_poly_data(Hand(I, S){&q, i + 1}, hand_entry) 58 if t == nil { 59 break 60 } 61 threads[i] = t 62 started += 1 63 } 64 drain(&q, 0) 65 thread.join_multiple(..threads[:started]) 66 for t in threads[:started] { 67 thread.destroy(t) 68 } 69 } 70 71 // Release what the run never took, once every worker has stopped. 72 @(private) 73 sweep :: proc(q: ^Queue($I, $S), discard: proc(item: I)) { 74 if discard == nil { 75 return 76 } 77 for item in q.items[q.head:] { 78 discard(item) 79 } 80 q.head = len(q.items) 81 } 82 83 @(private) 84 Queue :: struct($I: typeid, $S: typeid) { 85 items: [dynamic]I, 86 head: int, 87 active: int, // workers holding an item, which may yet produce more 88 over: bool, 89 mutex: sync.Mutex, 90 wake: sync.Cond, 91 states: []S, 92 work: proc(item: I, state: ^S) -> bool, 93 manager: proc(item: I, ok: bool, state: ^S, queue: ^[dynamic]I) -> bool, 94 } 95 96 @(private) 97 Hand :: struct($I: typeid, $S: typeid) { 98 queue: ^Queue(I, S), 99 index: int, 100 } 101 102 @(private) 103 hand_entry :: proc(h: Hand($I, $S)) { 104 drain(h.queue, h.index) 105 } 106 107 @(private) 108 drain :: proc(q: ^Queue($I, $S), index: int) { 109 state := &q.states[index] 110 for { 111 sync.mutex_lock(&q.mutex) 112 // Wait while the queue is empty but someone still holds an item, since that 113 // worker may yet discover more. Empty with nobody working means finished. 114 for q.head >= len(q.items) && q.active > 0 && !q.over { 115 sync.cond_wait(&q.wake, &q.mutex) 116 } 117 if q.over || q.head >= len(q.items) { 118 q.over = true 119 sync.cond_broadcast(&q.wake) 120 sync.mutex_unlock(&q.mutex) 121 return 122 } 123 item := q.items[q.head] 124 q.head += 1 125 q.active += 1 126 // Reclaim the consumed prefix once it dominates, or a deep traversal keeps 127 // every item it has ever seen. 128 if q.head > 1024 && q.head * 2 > len(q.items) { 129 n := copy(q.items[:], q.items[q.head:]) 130 resize(&q.items, n) 131 q.head = 0 132 } 133 sync.mutex_unlock(&q.mutex) 134 135 ok := q.work(item, state) 136 137 sync.mutex_lock(&q.mutex) 138 q.active -= 1 139 // The manager runs for a failed item too: deciding what a failure means is 140 // the whole of its job. 141 if !q.manager(item, ok, state, &q.items) { 142 q.over = true 143 } 144 sync.cond_broadcast(&q.wake) 145 sync.mutex_unlock(&q.mutex) 146 } 147 }