commit 08b3ecbe7ebc37c4ecfbbcaab7d11b7a0e240490
parent 84ba2311339b9b36decc453e70010e7586a90e97
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Wed, 23 Sep 2026 21:30:43 -0300
flow: add the worker-pool package from sonar
Work that splits into independent items wants many threads without the
locks that make threads hard. Every shape here obeys one rule: a worker
owns one slot of the caller's states and never shares it, and the caller
merges the slots once the run is over. width says how many workers a
piece of work deserves, each claims items from a shared counter until
they run out, and manage does the same for work that discovers more of
itself as it goes, with a manager that queues what each finished item
reveals. The package is the copy sonar now imports; the tests cover
every item reached once, early stops, empty input and failed items being
requeued.
Diffstat:
| A | flow/flow.odin | | | 152 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | flow/flow_test.odin | | | 193 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | flow/manage.odin | | | 147 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | flow/manage_test.odin | | | 214 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
4 files changed, 706 insertions(+), 0 deletions(-)
diff --git a/flow/flow.odin b/flow/flow.odin
@@ -0,0 +1,152 @@
+/*
+Package flow holds concurrency shapes for work that splits into independent items.
+
+Every shape obeys one rule: a worker owns its state and never shares it. Locks are
+then unnecessary, and the caller merges the states once the run is over.
+
+ width how many workers a piece of work deserves
+ each claim items from a shared counter until they run out
+ manage the same, for work that discovers more of itself as it goes
+*/
+package flow
+
+import "core:os"
+import "core:sync"
+import "core:thread"
+
+/*
+Run `work` over every item, across workers that claim one item at a time.
+
+Each worker owns one slot of `states`, so `work` needs no locks. Merge the slots
+afterwards, tolerating untouched ones. Anything else a worker touches must be read
+only, or written at disjoint addresses.
+
+Returning false stops the run; claimed items still finish.
+
+`len(states)` sets the width, capped by `width(len(items), load)`; pass the `load`
+the slice was sized with. One slot runs inline. Workers get a fresh context, so any
+allocator the work needs belongs in `State`.
+
+An item should cost more than the thirty microseconds it takes to start a thread.
+*/
+each :: proc(
+ items: []$I,
+ states: []$S,
+ work: proc(item: I, state: ^S) -> bool,
+ load := Load.Mixed,
+) {
+ if len(items) == 0 || len(states) == 0 {
+ return
+ }
+
+ shared := Run(I, S) {
+ items = items,
+ states = states,
+ work = work,
+ }
+
+ pool := min(len(states), width(len(items), load))
+
+ if pool == 1 {
+ claim_loop(&shared, 0)
+ return
+ }
+
+ threads := make([]^thread.Thread, pool - 1, context.temp_allocator)
+ defer delete(threads, context.temp_allocator)
+
+ started := 0
+ for i in 0 ..< len(threads) {
+ t := thread.create_and_start_with_poly_data(Arg(I, S){&shared, i + 1}, worker_entry)
+ if t == nil {
+ break
+ }
+ threads[i] = t
+ started += 1
+ }
+
+ // The calling thread takes slot 0 rather than idling while the others work.
+ claim_loop(&shared, 0)
+
+ thread.join_multiple(..threads[:started])
+ for t in threads[:started] {
+ thread.destroy(t)
+ }
+}
+
+// How a piece of work divides between waiting and computing, which is the part of
+// the width decision that only the caller knows.
+Load :: enum {
+ Cpu, // computing throughout: more workers than cores only makes them compete
+ Mixed, // alternates between the two, the common case
+ Io, // mostly parked in a device call, using no core while it waits
+}
+
+/*
+How many workers `items` pieces of work deserve. Size a state slice with it.
+
+Computing work wants one worker per core; work parked in a device call wants several
+times that, since it holds no core while it waits. `load` picks between them.
+
+The answer starts from the core count, so no input size can run it away; `items` and
+`limit` only reduce it. `limit` is also the cap when each worker needs a large buffer.
+
+The multipliers are starting points, not measured.
+*/
+width :: proc(items: int, load := Load.Mixed, limit := 0) -> int {
+ cores := os.get_processor_core_count()
+ if cores < 1 {
+ cores = 1
+ }
+ n: int
+ switch load {
+ case .Cpu:
+ n = cores
+ case .Mixed:
+ n = cores * 2
+ case .Io:
+ n = cores * 4
+ }
+ if limit > 0 && n > limit {
+ n = limit
+ }
+ // One worker is the floor: a caller with no work still needs a runnable answer.
+ return min(n, max(items, 1))
+}
+
+@(private)
+Run :: struct($I: typeid, $S: typeid) {
+ items: []I,
+ states: []S,
+ work: proc(item: I, state: ^S) -> bool,
+ next: int,
+ stop: b32,
+}
+
+@(private)
+Arg :: struct($I: typeid, $S: typeid) {
+ shared: ^Run(I, S),
+ index: int,
+}
+
+@(private)
+worker_entry :: proc(arg: Arg($I, $S)) {
+ claim_loop(arg.shared, arg.index)
+}
+
+@(private)
+claim_loop :: proc(shared: ^Run($I, $S), index: int) {
+ state := &shared.states[index]
+ for !sync.atomic_load(&shared.stop) {
+ // atomic_add returns the value from before the add, so this claims index i
+ // and leaves the next one for whoever gets here first.
+ i := sync.atomic_add(&shared.next, 1)
+ if i >= len(shared.items) {
+ return
+ }
+ if !shared.work(shared.items[i], state) {
+ sync.atomic_store(&shared.stop, true)
+ return
+ }
+ }
+}
diff --git a/flow/flow_test.odin b/flow/flow_test.odin
@@ -0,0 +1,193 @@
+package flow
+
+import "core:testing"
+
+// Counting alone cannot tell "every item once" from "one item twice and another
+// never", so each worker also sums the items it saw. The two together pin the run
+// down: the count proves how many were handled and the sum proves which.
+@(private = "file")
+Tally :: struct {
+ handled: int,
+ sum: int,
+}
+
+@(private = "file")
+ITEMS :: 100_000
+
+@(private = "file")
+count_up :: proc(item: int, tally: ^Tally) -> bool {
+ tally.handled += 1
+ tally.sum += item
+ return true
+}
+
+@(private = "file")
+sequence :: proc() -> []int {
+ items := make([]int, ITEMS, context.temp_allocator)
+ for i in 0 ..< ITEMS {
+ items[i] = i
+ }
+ return items
+}
+
+@(private = "file")
+busy :: proc(tallies: []Tally) -> (n: int) {
+ for tally in tallies {
+ if tally.handled > 0 {
+ n += 1
+ }
+ }
+ return
+}
+
+@(private = "file")
+totals :: proc(tallies: []Tally) -> (handled, sum: int) {
+ for t in tallies {
+ handled += t.handled
+ sum += t.sum
+ }
+ return
+}
+
+@(test)
+test_each_handles_every_item_exactly_once :: proc(t: ^testing.T) {
+ tallies := make([]Tally, 8, context.temp_allocator)
+ each(sequence(), tallies, count_up)
+
+ handled, sum := totals(tallies)
+ testing.expect_value(t, handled, ITEMS)
+ testing.expect_value(t, sum, ITEMS * (ITEMS - 1) / 2)
+}
+
+@(test)
+test_each_shares_the_work_out :: proc(t: ^testing.T) {
+ // Each item has to cost appreciably more than starting a thread, or the calling
+ // thread finishes the whole run before the others are scheduled and the split
+ // says nothing. That is a property of the work, not of the claiming.
+ tallies := make([]Tally, 4, context.temp_allocator)
+ items := make([]int, 32, context.temp_allocator)
+ each(
+ items,
+ tallies,
+ proc(item: int, tally: ^Tally) -> bool {
+ acc := 0
+ for i in 0 ..< 1_000_000 {
+ acc += i ~ item
+ }
+ tally.handled += 1
+ tally.sum += acc & 1 // consume acc so the loop cannot be optimised away
+ return true
+ },
+ )
+
+ testing.expect(t, busy(tallies) > 1, "work stayed on a single worker")
+}
+
+@(test)
+test_each_with_one_slot_runs_inline :: proc(t: ^testing.T) {
+ tallies := make([]Tally, 1, context.temp_allocator)
+ each(sequence(), tallies, count_up)
+
+ handled, sum := totals(tallies)
+ testing.expect_value(t, handled, ITEMS)
+ testing.expect_value(t, sum, ITEMS * (ITEMS - 1) / 2)
+}
+
+@(test)
+test_each_stops_when_work_returns_false :: proc(t: ^testing.T) {
+ // One slot keeps this deterministic: with several workers a few more items
+ // finish after the decision to stop, which is the documented behaviour.
+ tallies := make([]Tally, 1, context.temp_allocator)
+ each(sequence(), tallies, proc(item: int, tally: ^Tally) -> bool {
+ if item == 10 {
+ return false
+ }
+ tally.handled += 1
+ return true
+ })
+ testing.expect_value(t, tallies[0].handled, 10)
+}
+
+@(test)
+test_each_stops_early_across_workers :: proc(t: ^testing.T) {
+ tallies := make([]Tally, 8, context.temp_allocator)
+ each(sequence(), tallies, proc(item: int, tally: ^Tally) -> bool {
+ if item > 100 {
+ return false
+ }
+ tally.handled += 1
+ return true
+ })
+
+ handled, _ := totals(tallies)
+ testing.expect(t, handled > 0, "no item was handled before the stop")
+ testing.expect(t, handled < ITEMS, "stopping did not cut the run short")
+}
+
+@(test)
+test_each_tolerates_empty_input :: proc(t: ^testing.T) {
+ tallies := make([]Tally, 4, context.temp_allocator)
+ each([]int{}, tallies, count_up)
+ handled, _ := totals(tallies)
+ testing.expect_value(t, handled, 0)
+
+ // No slots means no worker can own state, so there is nothing to run on.
+ each(sequence(), []Tally{}, count_up)
+}
+
+@(test)
+test_width_never_exceeds_the_work :: proc(t: ^testing.T) {
+ // However wide the machine, three items can only keep three workers busy.
+ testing.expect_value(t, width(3, .Io), 3)
+ testing.expect_value(t, width(1, .Io), 1)
+ // No work still has to give a runnable answer rather than zero.
+ testing.expect_value(t, width(0), 1)
+}
+
+@(test)
+test_width_respects_the_limit :: proc(t: ^testing.T) {
+ testing.expect_value(t, width(1000, .Io, limit = 4), 4)
+ // The limit is a ceiling, not a target: fewer items still win.
+ testing.expect_value(t, width(2, .Io, limit = 4), 2)
+}
+
+@(test)
+test_width_grows_with_waiting :: proc(t: ^testing.T) {
+ // Plenty of work, so the load is the only thing deciding the answer.
+ cpu := width(10_000, .Cpu)
+ mixed := width(10_000, .Mixed)
+ io := width(10_000, .Io)
+ testing.expect(t, cpu >= 1)
+ testing.expect(t, mixed > cpu, "mixed work should outnumber cpu bound work")
+ testing.expect(t, io > mixed, "waiting work should outnumber mixed work")
+ // Derived from the core count, so a huge input cannot produce a huge width.
+ testing.expect(t, io < 10_000, "width ran away with the input")
+}
+
+@(test)
+test_each_ignores_a_pathological_width :: proc(t: ^testing.T) {
+ // Asking for thousands of workers is a mistake, not an instruction. The run has
+ // to stay correct and the pool has to stay within what the machine can use.
+ tallies := make([]Tally, 4000, context.temp_allocator)
+ each(sequence(), tallies, count_up)
+
+ handled, sum := totals(tallies)
+ testing.expect_value(t, handled, ITEMS)
+ testing.expect_value(t, sum, ITEMS * (ITEMS - 1) / 2)
+ testing.expect(
+ t,
+ busy(tallies) <= width(ITEMS, .Mixed),
+ "the pool grew past the default ceiling",
+ )
+}
+
+@(test)
+test_each_caps_by_the_load_it_is_given :: proc(t: ^testing.T) {
+ // Cpu is the narrowest tier, so it has to hold the pool below the default.
+ tallies := make([]Tally, 4000, context.temp_allocator)
+ each(sequence(), tallies, count_up, .Cpu)
+
+ handled, _ := totals(tallies)
+ testing.expect_value(t, handled, ITEMS)
+ testing.expect(t, busy(tallies) <= width(ITEMS, .Cpu), "the load did not reach the ceiling")
+}
diff --git a/flow/manage.odin b/flow/manage.odin
@@ -0,0 +1,147 @@
+package flow
+
+import "core:sync"
+import "core:thread"
+
+/*
+Run work over a set that grows as the work discovers more of it.
+
+`seed` starts the queue; every later item arrives through `manager`, which runs one
+at a time. `work` runs on many threads at once, taking the next item the moment one
+exists rather than in rounds.
+
+Each worker owns one slot of `states`, so `work` needs no locks, exactly as in `each`.
+`manager` sees that slot beside the item that filled it, and queues whatever comes
+next, that item included.
+
+`work` returning false marks the item failed, not the run. `manager` returning false
+ends the run; workers finish the item in hand and report through their state.
+
+Every item reaches `work` or `discard`, never both and never neither. A stopped run
+leaves items queued that nothing else can reach, so an item owning memory needs
+`discard` to release it.
+
+`len(states)` sets the width, capped by what the machine can use.
+*/
+manage :: proc(
+ seed: []$I,
+ states: []$S,
+ work: proc(item: I, state: ^S) -> bool,
+ manager: proc(item: I, ok: bool, state: ^S, queue: ^[dynamic]I) -> bool,
+ discard: proc(item: I) = nil,
+ load := Load.Io,
+) {
+ if len(seed) == 0 || len(states) == 0 {
+ return
+ }
+ q: Queue(I, S)
+ q.states = states
+ q.work = work
+ q.manager = manager
+ q.items = make([dynamic]I, context.allocator)
+ defer delete(q.items)
+ // Runs before the delete above, while the queue still holds what was never taken.
+ defer sweep(&q, discard)
+ append(&q.items, ..seed)
+
+ pool := min(len(states), width(1 << 30, load))
+ if pool == 1 {
+ drain(&q, 0)
+ return
+ }
+
+ threads := make([]^thread.Thread, pool - 1, context.temp_allocator)
+ defer delete(threads, context.temp_allocator)
+ started := 0
+ for i in 0 ..< len(threads) {
+ t := thread.create_and_start_with_poly_data(Hand(I, S){&q, i + 1}, hand_entry)
+ if t == nil {
+ break
+ }
+ threads[i] = t
+ started += 1
+ }
+ drain(&q, 0)
+ thread.join_multiple(..threads[:started])
+ for t in threads[:started] {
+ thread.destroy(t)
+ }
+}
+
+// Release what the run never took, once every worker has stopped.
+@(private)
+sweep :: proc(q: ^Queue($I, $S), discard: proc(item: I)) {
+ if discard == nil {
+ return
+ }
+ for item in q.items[q.head:] {
+ discard(item)
+ }
+ q.head = len(q.items)
+}
+
+@(private)
+Queue :: struct($I: typeid, $S: typeid) {
+ items: [dynamic]I,
+ head: int,
+ active: int, // workers holding an item, which may yet produce more
+ over: bool,
+ mutex: sync.Mutex,
+ wake: sync.Cond,
+ states: []S,
+ work: proc(item: I, state: ^S) -> bool,
+ manager: proc(item: I, ok: bool, state: ^S, queue: ^[dynamic]I) -> bool,
+}
+
+@(private)
+Hand :: struct($I: typeid, $S: typeid) {
+ queue: ^Queue(I, S),
+ index: int,
+}
+
+@(private)
+hand_entry :: proc(h: Hand($I, $S)) {
+ drain(h.queue, h.index)
+}
+
+@(private)
+drain :: proc(q: ^Queue($I, $S), index: int) {
+ state := &q.states[index]
+ for {
+ sync.mutex_lock(&q.mutex)
+ // Wait while the queue is empty but someone still holds an item, since that
+ // worker may yet discover more. Empty with nobody working means finished.
+ for q.head >= len(q.items) && q.active > 0 && !q.over {
+ sync.cond_wait(&q.wake, &q.mutex)
+ }
+ if q.over || q.head >= len(q.items) {
+ q.over = true
+ sync.cond_broadcast(&q.wake)
+ sync.mutex_unlock(&q.mutex)
+ return
+ }
+ item := q.items[q.head]
+ q.head += 1
+ q.active += 1
+ // Reclaim the consumed prefix once it dominates, or a deep traversal keeps
+ // every item it has ever seen.
+ if q.head > 1024 && q.head * 2 > len(q.items) {
+ n := copy(q.items[:], q.items[q.head:])
+ resize(&q.items, n)
+ q.head = 0
+ }
+ sync.mutex_unlock(&q.mutex)
+
+ ok := q.work(item, state)
+
+ sync.mutex_lock(&q.mutex)
+ q.active -= 1
+ // The manager runs for a failed item too: deciding what a failure means is
+ // the whole of its job.
+ if !q.manager(item, ok, state, &q.items) {
+ q.over = true
+ }
+ sync.cond_broadcast(&q.wake)
+ sync.mutex_unlock(&q.mutex)
+ }
+}
diff --git a/flow/manage_test.odin b/flow/manage_test.odin
@@ -0,0 +1,214 @@
+package flow
+
+import "core:testing"
+
+// A binary tree flattened into indices: node i holds 2i+1 and 2i+2. Walking it is
+// the shape manage is for, since a worker only learns of a node by visiting its
+// parent, and it is deterministic enough to check exactly.
+@(private = "file")
+NODES :: 20_000
+
+@(private = "file")
+Visit :: struct {
+ seen: [dynamic]int,
+ found: [dynamic]int,
+ stop: int, // visit this node and the run ends; -1 for never
+ sink: int, // keeps the busy loop below from being optimised away
+}
+
+@(private = "file")
+descend :: proc(item: int, v: ^Visit) -> bool {
+ append(&v.seen, item)
+ if item == v.stop {
+ return false
+ }
+ for c in ([]int{2 * item + 1, 2 * item + 2}) {
+ if c < NODES {
+ append(&v.found, c)
+ }
+ }
+ return true
+}
+
+@(private = "file")
+hand_over :: proc(item: int, ok: bool, v: ^Visit, queue: ^[dynamic]int) -> bool {
+ for c in v.found {
+ append(queue, c)
+ }
+ clear(&v.found)
+ return ok
+}
+
+/*
+Worker states, each holding what its worker saw.
+
+The dynamic arrays are deliberately not on the temp allocator. A dynamic array
+remembers the allocator it was made with, and these are appended to from every
+worker at once, so a per-thread arena belonging to whichever thread built them would
+be grown from all of them without a lock. That miscounted a node roughly once in
+twelve runs. `delete` them with `release`.
+*/
+@(private = "file")
+visitors :: proc(n: int, stop := -1) -> []Visit {
+ v := make([]Visit, n, context.temp_allocator)
+ for i in 0 ..< n {
+ v[i] = Visit {
+ seen = make([dynamic]int, context.allocator),
+ found = make([dynamic]int, context.allocator),
+ stop = stop,
+ }
+ }
+ return v
+}
+
+@(private = "file")
+release :: proc(v: []Visit) {
+ for s in v {
+ delete(s.seen)
+ delete(s.found)
+ }
+}
+
+@(private = "file")
+totals :: proc(v: []Visit) -> (count, sum: int) {
+ for s in v {
+ for i in s.seen {
+ count += 1
+ sum += i
+ }
+ }
+ return
+}
+
+@(test)
+test_manage_reaches_every_node_once :: proc(t: ^testing.T) {
+ v := visitors(8)
+ defer release(v)
+ manage([]int{0}, v, descend, hand_over)
+
+ // The count proves how many were visited and the sum proves which, so together
+ // they rule out one node twice and another never.
+ count, sum := totals(v)
+ testing.expect_value(t, count, NODES)
+ testing.expect_value(t, sum, NODES * (NODES - 1) / 2)
+}
+
+@(test)
+test_manage_agrees_with_one_worker :: proc(t: ^testing.T) {
+ // A single worker runs inline with no threads at all, which is the yardstick the
+ // concurrent run has to match.
+ one := visitors(1)
+ defer release(one)
+ manage([]int{0}, one, descend, hand_over)
+ count, sum := totals(one)
+ testing.expect_value(t, count, NODES)
+ testing.expect_value(t, sum, NODES * (NODES - 1) / 2)
+}
+
+@(private = "file")
+HEAVY :: 400
+
+// A node has to cost appreciably more than starting a thread, or the calling thread
+// drains the queue before the others are scheduled and the split says nothing. That
+// is a property of the work, not of the traversal.
+@(private = "file")
+descend_slowly :: proc(item: int, v: ^Visit) -> bool {
+ acc := 0
+ for i in 0 ..< 400_000 {
+ acc += i ~ item
+ }
+ v.sink += acc & 1
+ append(&v.seen, item)
+ for c in ([]int{2 * item + 1, 2 * item + 2}) {
+ if c < HEAVY {
+ append(&v.found, c)
+ }
+ }
+ return true
+}
+
+@(test)
+test_manage_spreads_across_workers :: proc(t: ^testing.T) {
+ v := visitors(4)
+ defer release(v)
+ manage([]int{0}, v, descend_slowly, hand_over)
+
+ count, sum := totals(v)
+ testing.expect_value(t, count, HEAVY)
+ testing.expect_value(t, sum, HEAVY * (HEAVY - 1) / 2)
+
+ busy := 0
+ for s in v {
+ if len(s.seen) > 0 {
+ busy += 1
+ }
+ }
+ testing.expect(t, busy > 1, "the traversal stayed on a single worker")
+}
+
+@(test)
+test_manage_stops_when_the_manager_returns_false :: proc(t: ^testing.T) {
+ // One worker keeps this exact: with several, those already holding an item
+ // finish it, which is the documented behaviour.
+ one := visitors(1, stop = 0)
+ defer release(one)
+ manage([]int{0}, one, descend, hand_over)
+ count, _ := totals(one)
+ testing.expect_value(t, count, 1)
+}
+
+@(test)
+test_manage_tolerates_an_empty_seed :: proc(t: ^testing.T) {
+ v := visitors(4)
+ defer release(v)
+ manage([]int{}, v, descend, hand_over)
+ count, _ := totals(v)
+ testing.expect_value(t, count, 0)
+}
+
+// A failed item is the manager's to judge, and the judgement it cannot make without
+// being told which item failed is to try that one again.
+@(private = "file")
+Attempt :: struct {
+ id: int,
+ tries: int,
+}
+
+@(private = "file")
+Attempt_Log :: struct {
+ seen: [dynamic]Attempt,
+}
+
+@(private = "file")
+refuse_twice :: proc(item: Attempt, f: ^Attempt_Log) -> bool {
+ append(&f.seen, item)
+ return !(item.id == 1 && item.tries < 2)
+}
+
+@(private = "file")
+retry :: proc(item: Attempt, ok: bool, f: ^Attempt_Log, queue: ^[dynamic]Attempt) -> bool {
+ if !ok {
+ append(queue, Attempt{id = item.id, tries = item.tries + 1})
+ }
+ return true
+}
+
+@(test)
+test_manage_requeues_a_failed_item :: proc(t: ^testing.T) {
+ // One worker keeps the count exact; the point is the shape, not the width.
+ f := make([]Attempt_Log, 1, context.temp_allocator)
+ f[0].seen = make([dynamic]Attempt, context.temp_allocator)
+ seed := []Attempt{{id = 0}, {id = 1}, {id = 2}}
+ manage(seed, f, refuse_twice, retry)
+
+ // Three items, one of them attempted three times, and the run carried on past
+ // the failures rather than ending at the first.
+ testing.expect_value(t, len(f[0].seen), 5)
+ tries := 0
+ for a in f[0].seen {
+ if a.id == 1 {
+ tries += 1
+ }
+ }
+ testing.expect_value(t, tries, 3)
+}
