jm

Odin for scripts: small packages and a runner, on core: only
Log | Files | Refs | README

flow.odin (4032B)

      1 /*
      2 Package flow holds concurrency shapes for work that splits into independent items.
      3 
      4 Every shape obeys one rule: a worker owns its state and never shares it. Locks are
      5 then unnecessary, and the caller merges the states once the run is over.
      6 
      7 	width   how many workers a piece of work deserves
      8 	each    claim items from a shared counter until they run out
      9 	manage  the same, for work that discovers more of itself as it goes
     10 */
     11 package flow
     12 
     13 import "core:os"
     14 import "core:sync"
     15 import "core:thread"
     16 
     17 /*
     18 Run `work` over every item, across workers that claim one item at a time.
     19 
     20 Each worker owns one slot of `states`, so `work` needs no locks. Merge the slots
     21 afterwards, tolerating untouched ones. Anything else a worker touches must be read
     22 only, or written at disjoint addresses.
     23 
     24 Returning false stops the run; claimed items still finish.
     25 
     26 `len(states)` sets the width, capped by `width(len(items), load)`; pass the `load`
     27 the slice was sized with. One slot runs inline. Workers get a fresh context, so any
     28 allocator the work needs belongs in `State`.
     29 
     30 An item should cost more than the thirty microseconds it takes to start a thread.
     31 */
     32 each :: proc(
     33 	items: []$I,
     34 	states: []$S,
     35 	work: proc(item: I, state: ^S) -> bool,
     36 	load := Load.Mixed,
     37 ) {
     38 	if len(items) == 0 || len(states) == 0 {
     39 		return
     40 	}
     41 
     42 	shared := Run(I, S) {
     43 		items  = items,
     44 		states = states,
     45 		work   = work,
     46 	}
     47 
     48 	pool := min(len(states), width(len(items), load))
     49 
     50 	if pool == 1 {
     51 		claim_loop(&shared, 0)
     52 		return
     53 	}
     54 
     55 	threads := make([]^thread.Thread, pool - 1, context.temp_allocator)
     56 	defer delete(threads, context.temp_allocator)
     57 
     58 	started := 0
     59 	for i in 0 ..< len(threads) {
     60 		t := thread.create_and_start_with_poly_data(Arg(I, S){&shared, i + 1}, worker_entry)
     61 		if t == nil {
     62 			break
     63 		}
     64 		threads[i] = t
     65 		started += 1
     66 	}
     67 
     68 	// The calling thread takes slot 0 rather than idling while the others work.
     69 	claim_loop(&shared, 0)
     70 
     71 	thread.join_multiple(..threads[:started])
     72 	for t in threads[:started] {
     73 		thread.destroy(t)
     74 	}
     75 }
     76 
     77 // How a piece of work divides between waiting and computing, which is the part of
     78 // the width decision that only the caller knows.
     79 Load :: enum {
     80 	Cpu, // computing throughout: more workers than cores only makes them compete
     81 	Mixed, // alternates between the two, the common case
     82 	Io, // mostly parked in a device call, using no core while it waits
     83 }
     84 
     85 /*
     86 How many workers `items` pieces of work deserve. Size a state slice with it.
     87 
     88 Computing work wants one worker per core; work parked in a device call wants several
     89 times that, since it holds no core while it waits. `load` picks between them.
     90 
     91 The answer starts from the core count, so no input size can run it away; `items` and
     92 `limit` only reduce it. `limit` is also the cap when each worker needs a large buffer.
     93 
     94 The multipliers are starting points, not measured.
     95 */
     96 width :: proc(items: int, load := Load.Mixed, limit := 0) -> int {
     97 	cores := os.get_processor_core_count()
     98 	if cores < 1 {
     99 		cores = 1
    100 	}
    101 	n: int
    102 	switch load {
    103 	case .Cpu:
    104 		n = cores
    105 	case .Mixed:
    106 		n = cores * 2
    107 	case .Io:
    108 		n = cores * 4
    109 	}
    110 	if limit > 0 && n > limit {
    111 		n = limit
    112 	}
    113 	// One worker is the floor: a caller with no work still needs a runnable answer.
    114 	return min(n, max(items, 1))
    115 }
    116 
    117 @(private)
    118 Run :: struct($I: typeid, $S: typeid) {
    119 	items:  []I,
    120 	states: []S,
    121 	work:   proc(item: I, state: ^S) -> bool,
    122 	next:   int,
    123 	stop:   b32,
    124 }
    125 
    126 @(private)
    127 Arg :: struct($I: typeid, $S: typeid) {
    128 	shared: ^Run(I, S),
    129 	index:  int,
    130 }
    131 
    132 @(private)
    133 worker_entry :: proc(arg: Arg($I, $S)) {
    134 	claim_loop(arg.shared, arg.index)
    135 }
    136 
    137 @(private)
    138 claim_loop :: proc(shared: ^Run($I, $S), index: int) {
    139 	state := &shared.states[index]
    140 	for !sync.atomic_load(&shared.stop) {
    141 		// atomic_add returns the value from before the add, so this claims index i
    142 		// and leaves the next one for whoever gets here first.
    143 		i := sync.atomic_add(&shared.next, 1)
    144 		if i >= len(shared.items) {
    145 			return
    146 		}
    147 		if !shared.work(shared.items[i], state) {
    148 			sync.atomic_store(&shared.stop, true)
    149 			return
    150 		}
    151 	}
    152 }