commit a266c75f41755ec3cb9350528d2440c31d04078c
parent 599c622fdbb6874898f7078730abe40199bc1ab5
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Wed, 23 Sep 2026 20:41:32 -0300
sonar: import flow from the jfm collection
The worker-pool shapes are not sonar-specific, so they now live in
~/Source/Personal/odin as jfm:flow beside the debug allocator, resolved by
path through the same JFM variable until sonar ships.
Diffstat:
7 files changed, 4 insertions(+), 710 deletions(-)
diff --git a/flow/flow.odin b/flow/flow.odin
@@ -1,152 +0,0 @@
-/*
-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
- 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 := 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
@@ -1,192 +0,0 @@
-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")
-}
diff --git a/flow/manage.odin b/flow/manage.odin
@@ -1,147 +0,0 @@
-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
@@ -1,214 +0,0 @@
-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")
-Flaky :: struct {
- seen: [dynamic]Attempt,
-}
-
-@(private = "file")
-refuse_twice :: proc(item: Attempt, f: ^Flaky) -> bool {
- append(&f.seen, item)
- return !(item.id == 1 && item.tries < 2)
-}
-
-@(private = "file")
-retry :: proc(item: Attempt, ok: bool, f: ^Flaky, 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([]Flaky, 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)
-}
diff --git a/justfile b/justfile
@@ -6,8 +6,8 @@
# just check type-check both modes without producing a binary
# just clean remove build/
#
-# The debug allocator comes from the jfm collection at ~/Source/Personal/odin
-# (override with JFM=<path>); it is resolved by path until sonar ships.
+# The debug allocator and flow come from the jfm collection at ~/Source/Personal/odin
+# (override with JFM=<path>); they are resolved by path until sonar ships.
#
# Debug and test builds carry -sanitize:address. The debug allocator poisons its
# guard bytes and freed blocks, so instrumented reads past the end or after free
@@ -39,7 +39,6 @@ release:
test:
mkdir -p build/test
{{odin}} test ntfs -debug {{san}} {{flags}} -out:build/test/ntfs{{exe}}
- {{odin}} test flow -debug {{san}} {{flags}} -out:build/test/flow{{exe}}
{{odin}} test scan -debug {{san}} {{flags}} -out:build/test/scan{{exe}}
{{odin}} test walk -debug {{san}} {{flags}} -out:build/test/walk{{exe}}
{{odin}} test . -debug {{san}} {{flags}} -out:build/test/main{{exe}}
diff --git a/ntfs/reader.odin b/ntfs/reader.odin
@@ -4,7 +4,7 @@ import "core:mem"
import "core:mem/virtual"
import "core:time"
-import "../flow"
+import "jfm:flow"
// Alignment for I/O buffers. Raw volume reads want sector alignment; 4 KiB covers
// every sector size in use.
diff --git a/walk/walk.odin b/walk/walk.odin
@@ -17,7 +17,7 @@ package walk
import "core:mem"
import "core:strings"
-import "../flow"
+import "jfm:flow"
import "../scan"
Config :: struct {
