commit ac798c62c06ff9fc7044068c1731ec67c42d27bc
parent 68d04bca0de2d92746edf00037eb9802aab2806a
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Fri, 18 Sep 2026 10:21:24 -0400
flow: add a package for this project's concurrency shapes
Threading decisions were about to be made inline in the MFT reader, where the
next place needing them would rediscover the reasoning. The package rests on
one rule: a worker owns its state and never shares it, so the work needs no
locks and the caller merges the slots afterwards.
each claims items with one atomic increment, so an early finisher takes more
rather than waiting on a slow neighbour. width sizes the pool from the core
count, the load, and the item count; each takes that as a ceiling, so an
implausible width cannot become thousands of threads.
Diffstat:
| M | Makefile | | | 5 | +++-- |
| A | flow/flow.odin | | | 146 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | flow/flow_test.odin | | | 184 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
3 files changed, 333 insertions(+), 2 deletions(-)
diff --git a/Makefile b/Makefile
@@ -2,7 +2,7 @@
#
# make debug build with AddressSanitizer -> build/debug/sonar.exe
# make release optimised build -> build/release/sonar.exe
-# make test run the ntfs tests under ASan -> build/test/ntfs.exe
+# make test run the package tests under ASan -> build/test/
# make check type-check both modes without producing a binary
# make clean remove build/
#
@@ -20,7 +20,7 @@ FLAGS := -vet -strict-style
SAN ?= -sanitize:address
EXE := $(if $(filter Windows_NT,$(OS)),.exe,)
-SRC := $(wildcard *.odin) $(wildcard ntfs/*.odin) $(wildcard debug/*.odin)
+SRC := $(wildcard *.odin) $(wildcard ntfs/*.odin) $(wildcard debug/*.odin) $(wildcard flow/*.odin)
.PHONY: all debug release test check clean
@@ -38,6 +38,7 @@ $(BUILD)/release/sonar$(EXE): $(SRC) | $(BUILD)/release
test: | $(BUILD)/test
$(ODIN) test ntfs -debug $(SAN) $(FLAGS) -out:$(BUILD)/test/ntfs$(EXE)
+ $(ODIN) test flow -debug $(SAN) $(FLAGS) -out:$(BUILD)/test/flow$(EXE)
check:
$(ODIN) check . $(FLAGS)
diff --git a/flow/flow.odin b/flow/flow.odin
@@ -0,0 +1,146 @@
+/*
+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
+*/
+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 := Shared(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)
+Shared :: 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: ^Shared(I, S),
+ index: int,
+}
+
+@(private)
+worker_entry :: proc(arg: Arg($I, $S)) {
+ claim_loop(arg.shared, arg.index)
+}
+
+@(private)
+claim_loop :: proc(shared: ^Shared($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,184 @@
+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 < 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")
+}
