commit 33cca93aefa426d04d9950c42baa0b5c2f9dbebf
parent 2c30a3334f018f8de99f3dca9e59f88c7a5a2143
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Fri, 18 Sep 2026 16:57:26 -0400
flow: let the manager judge a failed item
`manage` gave `work` one boolean for two unrelated answers: this item did not
work out, and abandon the whole run. Callers had to pick one. The walker picked
the first, reporting success for a directory it could not open, and so had no
way to report running out of room for the tree: it halted and returned .None,
losing a truncated scan silently.
`work` returning false now marks only the item. The manager is told which item
it was and what happened, and decides: skip it, queue it again, or stop. That
is flowmatic's split, which our version had collapsed.
Diffstat:
3 files changed, 90 insertions(+), 34 deletions(-)
diff --git a/flow/manage.odin b/flow/manage.odin
@@ -6,26 +6,24 @@ 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.
+`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`.
-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.
+`manager` sees that slot beside the item that filled it, and queues whatever comes
+next, that item included.
-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.
+`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.
-`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.
+`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,
- found: proc(state: ^S, queue: ^[dynamic]I) -> bool,
+ manager: proc(item: I, ok: bool, state: ^S, queue: ^[dynamic]I) -> bool,
load := Load.Io,
) {
if len(seed) == 0 || len(states) == 0 {
@@ -34,7 +32,7 @@ manage :: proc(
q: Queue(I, S)
q.states = states
q.work = work
- q.found = found
+ q.manager = manager
q.items = make([dynamic]I, context.allocator)
defer delete(q.items)
append(&q.items, ..seed)
@@ -65,15 +63,15 @@ manage :: proc(
@(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,
+ 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)
@@ -119,7 +117,9 @@ drain :: proc(q: ^Queue($I, $S), index: int) {
sync.mutex_lock(&q.mutex)
q.active -= 1
- if !ok || !q.found(state, &q.items) {
+ // 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)
diff --git a/flow/manage_test.odin b/flow/manage_test.odin
@@ -31,12 +31,12 @@ descend :: proc(item: int, v: ^Visit) -> bool {
}
@(private = "file")
-hand_over :: proc(v: ^Visit, queue: ^[dynamic]int) -> bool {
+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 true
+ return ok
}
@(private = "file")
@@ -127,7 +127,7 @@ test_manage_spreads_across_workers :: proc(t: ^testing.T) {
}
@(test)
-test_manage_stops_when_work_returns_false :: proc(t: ^testing.T) {
+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)
@@ -143,3 +143,50 @@ test_manage_tolerates_an_empty_seed :: proc(t: ^testing.T) {
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/walk/walk.odin b/walk/walk.odin
@@ -82,21 +82,30 @@ scan :: proc(root: string, t: ^scan.Tree, cfg := Config{}) -> Error {
seed := []Dir{{index = first, path = strings.clone(root, allocator)}}
flow.manage(seed, states, walk_dir, hand_over)
+ for &s in states {
+ if s.err != nil {
+ return s.err
+ }
+ }
if scan.cancelled(t) {
return .Cancelled
}
return .None
}
-// Move what a worker found into the queue. Serialised by `manage`, so this is the
-// one place the set of directories still to walk is touched.
+// Move what a worker found into the queue, and judge whatever it could not read.
+// Serialised by `manage`, so this is the one place the set of directories still to
+// walk is touched.
@(private)
-hand_over :: proc(w: ^Worker, queue: ^[dynamic]Dir) -> bool {
- for d in w.found {
- append(queue, d)
+hand_over :: proc(d: Dir, ok: bool, w: ^Worker, queue: ^[dynamic]Dir) -> bool {
+ for found in w.found {
+ append(queue, found)
}
clear(&w.found)
- return !scan.cancelled(w.writer.tree)
+ // A directory that would not open is worth nothing and is not worth retrying,
+ // but it is no reason to abandon the rest of the volume. Running out of room
+ // for the tree is, since everything after it would be missing silently.
+ return w.err == nil && !scan.cancelled(w.writer.tree)
}
// A directory waiting to be read, and the node already standing for it.
@@ -111,6 +120,7 @@ Worker :: struct {
writer: scan.Writer,
found: [dynamic]Dir,
follow: bool,
+ err: Error, // why this worker gave up, which only the manager acts on
allocator: mem.Allocator,
}
@@ -122,9 +132,7 @@ walk_dir :: proc(d: Dir, w: ^Worker) -> bool {
}
f, open_err := os.open(d.path)
if open_err != nil {
- // A directory we may not read is not a reason to abandon the scan; it is
- // simply worth nothing. Permission denied is normal on a live system.
- return true
+ return false
}
defer os.close(f)
@@ -139,6 +147,7 @@ walk_dir :: proc(d: Dir, w: ^Worker) -> bool {
}
index, claim_err := scan.claim(&w.writer, 1)
if claim_err != nil {
+ w.err = .Out_Of_Memory
return false
}
n := scan.node(w.writer.tree, index)
