sonar

commit 6319df6d084a88159255796eaffd6a8d75c1b5eb
parent 8059c606ef926a920089a0a4fe36de15e610ad7b
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date:   Sun, 20 Sep 2026 07:40:30 -0300

flow: hand back the work a stopped run never took

A manager that ends the run leaves items queued, and the queue is local to
`manage`, so an item owning memory leaked with no way for the caller to reach it.
Stopping after 100 of a branching workload leaked 101 allocations, one per item
still waiting.

`discard` closes the contract: every item reaches `work` or `discard`, never both
and never neither. It defaults to nil, which is right for an item that owns
nothing. Sweeping is deferred so it covers the inline path and the pooled one at
once, and runs while the queue is still alive.

Diffstat:
M
flow/manage.odin
 | 
19
+++++++++++++++++++

1 file changed, 19 insertions(+), 0 deletions(-)
diff --git a/flow/manage.odin b/flow/manage.odin
@@ -17,6 +17,10 @@ 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(
@@ -24,6 +28,7 @@ manage :: proc(
 	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 {
@@ -35,6 +40,8 @@ manage :: proc(
 	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))
@@ -61,6 +68,18 @@ manage :: proc(
 	}
 }
 
+// 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,