commit 46562ae97ab8c0142e5930c97d3115bd8743418a
parent 65178051fd1bbb812c06eba5a065631e2769d3d8
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Fri, 18 Sep 2026 13:21:28 -0400
flow: add the manager shape for work that discovers itself
each needs its work known up front, so a traversal had to be run in rounds:
walk a level, collect what it found, walk the next. Every round ends at a
barrier where the pool waits for its slowest member, and starts its threads
again.
manage keeps one queue instead. Workers take the next item the moment one
exists, and what they discover goes back through a procedure serialised by the
queue's own lock, so deciding what to explore next needs no reasoning about
order. The threads are started once for the whole traversal.
Same rule as each: a worker owns one state, so the work needs no locks.
Diffstat:
3 files changed, 276 insertions(+), 2 deletions(-)
diff --git a/flow/flow.odin b/flow/flow.odin
@@ -4,8 +4,9 @@ Package flow holds this project's concurrency shapes.
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
+ 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
diff --git a/flow/manage.odin b/flow/manage.odin
@@ -0,0 +1,128 @@
+package flow
+
+import "core:sync"
+import "core:thread"
+
+/*
+Run work over a set that grows as the work discovers more of it.
+
+`work` runs on many threads at once; `found` runs on one at a time, and is the only
+thing that puts items in the queue. Workers take the next item the moment one exists
+rather than in rounds, so a shape only learned by walking it never stalls waiting for
+a round to end.
+
+Each worker owns one slot of `states`, so `work` needs no locks, exactly as in `each`.
+What a worker discovers goes in its own slot; `found` moves it into the queue and is
+serialised, so deciding what to explore next needs no reasoning about order.
+
+Either procedure returning false ends the run: no further items are handed out, and
+workers finish the item in hand. Report the reason through the state.
+
+`len(states)` sets the width, capped by what the machine can use. The threads are
+started once for the whole traversal rather than once per round.
+*/
+manage :: proc(
+ seed: []$I,
+ states: []$S,
+ work: proc(item: I, state: ^S) -> bool,
+ found: proc(state: ^S, queue: ^[dynamic]I) -> bool,
+ load := Load.Io,
+) {
+ if len(seed) == 0 || len(states) == 0 {
+ return
+ }
+ q: Queue(I, S)
+ q.states = states
+ q.work = work
+ q.found = found
+ q.items = make([dynamic]I, context.allocator)
+ defer delete(q.items)
+ 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)
+ }
+}
+
+@(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,
+ found: proc(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
+ if !ok || !q.found(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,145 @@
+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(v: ^Visit, queue: ^[dynamic]int) -> bool {
+ for c in v.found {
+ append(queue, c)
+ }
+ clear(&v.found)
+ return true
+}
+
+@(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.temp_allocator),
+ found = make([dynamic]int, context.temp_allocator),
+ stop = stop,
+ }
+ }
+ return v
+}
+
+@(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)
+ 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)
+ 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)
+ 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_work_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)
+ 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)
+ manage([]int{}, v, descend, hand_over)
+ count, _ := totals(v)
+ testing.expect_value(t, count, 0)
+}
