Add wakeup

Author: g41797

last_plan.md

@@ -181,7 +181,7 @@ Loop_Mailbox :: struct($T: typeid) {
 
 ---
 
-### Stage 4 — wakeup/ package (WakeUper interface + sema impl)  ← NEXT
+### Stage 4 — wakeup/ package (WakeUper interface + sema impl) ✓ DONE (Session 74)
 
 **Claim**: Separate the wakeup mechanism from the queue.
 
@@ -228,7 +228,7 @@ init_loop_mailbox :: proc(m: ^Loop_Mailbox($T), w: wakeup.WakeUper) -> Loop_Mail
 
 ---
 
-### Stage 5 — Pool WakeUper
+### Stage 5 — Pool WakeUper  ← NEXT
 
 **Purpose**:
 Add optional `wakeup.WakeUper` to the pool so event-loop callers can be notified
@@ -257,7 +257,7 @@ Pool :: struct($T: typeid) {
 **-vet workaround**: add `@(private) _PoolWaker :: wakeup.WakeUper` to `pool.odin`.
 
 **Files changed**:
-- `pool/pool.odin` — add `waker`, \`empty_was_returned\`; update `init`/`get`/`put`/`destroy`
+- `pool/pool.odin` — add `waker`, `empty_was_returned`; update `init`/`get`/`put`/`destroy`
 - `pool_tests/pool_test.odin` — add tests: WakeUper wakes on put, waker.close on destroy
 
 **Design doc**: `design/loop-mbox-enhancement.md` — Stage 5 section.
@@ -358,6 +358,25 @@ Details TBD when we reach this stage.
 
 ---
 
+## Edge Cases & Missing Tests Strategy
+
+To keep `queue_test.odin` and `wakeup_test.odin` clean as "runner of examples" and basic unit tests, we will create separate files for edge cases and stress tests.
+
+### 1. MPSC Edge Cases (`mpsc/edge_test.odin`)
+- **Concurrent Push Stress**: 10 threads pushing 10,000 items each while 1 thread pops. Verify total count and no lost items.
+- **Stall State Handling**: A test that attempts to hit the Vyukov stall window (where `pop` returns `nil` but `len > 0`) under high contention.
+- **Stub Recycling Verification**: Test specifically targeting the logic path where the stub sentinel is recycled when one item remains.
+- **Length Consistency**: Verify `length()` accurately reflects atomic increments/decrements even if `pop` temporarily stalls.
+- **Missed Test**: `test_pop_all_drains_to_zero` — ensures that repeated pops from a multi-item queue eventually return `nil` and `length()` is 0.
+
+### 2. WakeUper Edge Cases (`wakeup/edge_test.odin`)
+- **Concurrent Wake Signals**: 10 threads calling `wake()` at the same time on one `sema_wakeup`. Verify the semaphore count matches the number of calls.
+- **Wake after Close Protection**: Define and test the behavior when `wake()` is called after `close()`. (Recommendation: document as undefined/illegal, but provide a safe-fail if possible).
+- **Custom WakeUper**: Implement a simple dummy `WakeUper` in the test to verify the interface works without `sema_wakeup`.
+- **Missed Test**: `test_ctx_persistence` — check that the `rawptr` passed to `wake` and `close` is bit-for-bit identical to the one provided during creation.
+
+---
+
 ## Critical files
 
 | File | Stage | Action |
@@ -368,9 +387,11 @@ Details TBD when we reach this stage.
 | `mpsc/queue.odin` | 3 | NEW |
 | `mpsc/doc.odin` | 3 | NEW |
 | `mpsc/queue_test.odin` | 3 | NEW (unit tests) |
-| `wakeup/wakeup.odin` | 4 | NEW |
+| `mpsc/edge_test.odin` | 3 | NEW (edge + stress tests) |
+| `wakeup/wakeup.odin` | 4 | NEW + nil-check guards in _sema_wake/_sema_close |
 | `wakeup/doc.odin` | 4 | NEW |
 | `wakeup/wakeup_test.odin` | 4 | NEW (unit tests) |
+| `wakeup/edge_test.odin` | 4 | NEW (edge + concurrent tests) |
 | `pool/pool.odin` | 5 | Modify (add waker, empty_was_returned) |
 | `pool_tests/pool_test.odin` | 5 | Add WakeUper tests |
 | `mbox.odin` → `mbox/mbox.odin` | 6 | MOVE |

mpsc/edge_test.odin

@@ -0,0 +1,141 @@
+package mpsc
+
+import "core:testing"
+import "core:thread"
+
+// ----------------------------------------------------------------------------
+// Edge cases and stress tests
+// ----------------------------------------------------------------------------
+
+// test_stub_recycling_explicit exercises the stub-recycling path in pop.
+// That path runs when exactly one item remains (head == tail, next == nil).
+// Each push/pop cycle of a single item triggers it.
+@(test)
+test_stub_recycling_explicit :: proc(t: ^testing.T) {
+	q: Queue(_Test_Msg)
+	init(&q)
+	for i in 0 ..< 5 {
+		msg := _Test_Msg{data = i}
+		push(&q, &msg)
+		got := pop(&q)
+		testing.expectf(t, got != nil && got.data == i, "round %d: pop should return the pushed message", i)
+		testing.expectf(t, length(&q) == 0, "round %d: length should be 0 after pop", i)
+	}
+}
+
+// test_pop_all_drains_to_zero pushes N messages then pops until the queue is empty.
+// Verifies all messages are received and length reaches zero.
+@(test)
+test_pop_all_drains_to_zero :: proc(t: ^testing.T) {
+	q: Queue(_Test_Msg)
+	init(&q)
+
+	N :: 50
+	msgs: [N]_Test_Msg
+	for i in 0 ..< N {
+		msgs[i].data = i
+		push(&q, &msgs[i])
+	}
+
+	count := 0
+	for length(&q) > 0 || count < N {
+		if pop(&q) != nil {
+			count += 1
+		}
+		if length(&q) == 0 && count == N {
+			break
+		}
+	}
+
+	testing.expect(t, count == N, "should drain all pushed messages")
+	testing.expect(t, length(&q) == 0, "length should be 0 after full drain")
+}
+
+// _Stress_Ctx passes queue and message slice to each producer thread.
+@(private)
+_Stress_Ctx :: struct {
+	q:    ^Queue(_Test_Msg),
+	msgs: []_Test_Msg,
+}
+
+_STRESS_PRODUCERS      :: 10
+_STRESS_ITEMS_PER_PROD :: 1000
+
+// test_concurrent_push_stress runs _STRESS_PRODUCERS threads each pushing
+// _STRESS_ITEMS_PER_PROD messages. The main thread consumes all of them.
+// Verifies no messages are lost and length reaches zero.
+@(test)
+test_concurrent_push_stress :: proc(t: ^testing.T) {
+	q: Queue(_Test_Msg)
+	init(&q)
+
+	total :: _STRESS_PRODUCERS * _STRESS_ITEMS_PER_PROD
+
+	msgs := make([]_Test_Msg, total)
+	defer delete(msgs)
+
+	ctxs := make([]_Stress_Ctx, _STRESS_PRODUCERS)
+	defer delete(ctxs)
+
+	for i in 0 ..< _STRESS_PRODUCERS {
+		ctxs[i] = _Stress_Ctx {
+			q    = &q,
+			msgs = msgs[i * _STRESS_ITEMS_PER_PROD:(i + 1) * _STRESS_ITEMS_PER_PROD],
+		}
+	}
+
+	threads := make([dynamic]^thread.Thread, 0, _STRESS_PRODUCERS)
+	defer delete(threads)
+
+	for i in 0 ..< _STRESS_PRODUCERS {
+		th := thread.create_and_start_with_poly_data(
+			&ctxs[i],
+			proc(ctx: ^_Stress_Ctx) {
+				for j in 0 ..< len(ctx.msgs) {
+					push(ctx.q, &ctx.msgs[j])
+				}
+			},
+		)
+		append(&threads, th)
+	}
+
+	// Consume until all messages are received.
+	received := 0
+	for received < total {
+		if pop(&q) != nil {
+			received += 1
+		}
+	}
+
+	for th in threads {
+		thread.join(th)
+		thread.destroy(th)
+	}
+
+	testing.expect(t, received == total, "should receive all pushed messages")
+	testing.expect(t, length(&q) == 0, "length should be 0 after full drain")
+}
+
+// test_length_consistency verifies that after a concurrent stress run
+// the length counter reaches zero and matches the drained count.
+@(test)
+test_length_consistency :: proc(t: ^testing.T) {
+	q: Queue(_Test_Msg)
+	init(&q)
+
+	N :: 200
+	msgs: [N]_Test_Msg
+	for i in 0 ..< N {
+		push(&q, &msgs[i])
+	}
+
+	testing.expect(t, length(&q) == N, "length should equal number of pushes")
+
+	count := 0
+	for pop(&q) != nil {
+		count += 1
+	}
+
+	testing.expect(t, count == N, "should pop exactly N messages")
+	testing.expect(t, length(&q) == 0, "length should be 0 after draining")
+}

wakeup/doc.odin

@@ -0,0 +1,23 @@
+/*
+Package wakeup provides a wake-up callback interface for event loops.
+
+WakeUper is a value type with three fields: ctx, wake, close.
+It is copyable. The zero value is valid — callers must check wake != nil before calling.
+
+sema_wakeup returns a WakeUper backed by a semaphore.
+Use it in non-nbio loops and unit tests.
+
+Usage:
+
+	w := wakeup.sema_wakeup()
+	defer w.close(w.ctx)
+
+	// From another thread or goroutine:
+	w.wake(w.ctx)
+
+	// In the event loop — wait for wake:
+	// (depends on your loop mechanism)
+
+Call close when done. close frees internal resources.
+*/
+package wakeup

wakeup/edge_test.odin

@@ -0,0 +1,97 @@
+package wakeup
+
+import "core:sync"
+import "core:testing"
+import "core:thread"
+import "core:time"
+
+// ----------------------------------------------------------------------------
+// Edge cases and concurrent tests
+// ----------------------------------------------------------------------------
+
+// test_wake_with_nil_ctx verifies that _sema_wake and _sema_close
+// are safe no-ops when ctx is nil.
+@(test)
+test_wake_with_nil_ctx :: proc(t: ^testing.T) {
+	_sema_wake(nil)  // must not crash
+	_sema_close(nil) // must not crash
+	testing.expect(t, true, "nil ctx calls should not crash")
+}
+
+_CONCURRENT_WAKERS :: 10
+
+// test_concurrent_wake_signals starts _CONCURRENT_WAKERS threads that each
+// call wake once. Verifies the semaphore receives all signals.
+@(test)
+test_concurrent_wake_signals :: proc(t: ^testing.T) {
+	w := sema_wakeup()
+
+	threads := make([dynamic]^thread.Thread, 0, _CONCURRENT_WAKERS)
+	defer delete(threads)
+
+	for _ in 0 ..< _CONCURRENT_WAKERS {
+		th := thread.create_and_start_with_poly_data(
+			&w,
+			proc(w: ^WakeUper) {
+				w.wake(w.ctx)
+			},
+		)
+		append(&threads, th)
+	}
+
+	for th in threads {
+		thread.join(th)
+		thread.destroy(th)
+	}
+
+	state := (^_Sema_State)(w.ctx)
+	count := 0
+	for sync.sema_wait_with_timeout(&state.sema, 10 * time.Millisecond) {
+		count += 1
+	}
+
+	testing.expect(t, count == _CONCURRENT_WAKERS, "all wake signals should be received")
+	w.close(w.ctx)
+}
+
+// _Custom_State holds flags set by a custom WakeUper's wake and close procs.
+@(private)
+_Custom_State :: struct {
+	fired:  bool,
+	closed: bool,
+}
+
+// test_custom_wakeup builds a WakeUper from raw procs (no semaphore).
+// Verifies the interface works with any implementation.
+@(test)
+test_custom_wakeup :: proc(t: ^testing.T) {
+	state := _Custom_State{}
+
+	w := WakeUper {
+		ctx   = rawptr(&state),
+		wake  = proc(ctx: rawptr) {(^_Custom_State)(ctx).fired = true},
+		close = proc(ctx: rawptr) {(^_Custom_State)(ctx).closed = true},
+	}
+
+	testing.expect(t, !state.fired, "should not be fired before wake")
+	testing.expect(t, !state.closed, "should not be closed before close")
+
+	w.wake(w.ctx)
+	testing.expect(t, state.fired, "should be fired after wake")
+
+	w.close(w.ctx)
+	testing.expect(t, state.closed, "should be closed after close")
+}
+
+// test_ctx_persistence verifies that ctx stored in WakeUper matches
+// the allocated _Sema_State address and that the stored allocator is valid.
+@(test)
+test_ctx_persistence :: proc(t: ^testing.T) {
+	w := sema_wakeup()
+	state := (^_Sema_State)(w.ctx)
+
+	testing.expect(t, rawptr(state) == w.ctx, "ctx should point to _Sema_State")
+	testing.expect(t, state.allocator.procedure != nil, "stored allocator should be valid")
+
+	w.close(w.ctx)
+}

wakeup/wakeup.odin

@@ -0,0 +1,52 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 g41797
+// SPDX-License-Identifier: MIT
+
+package wakeup
+
+import "core:mem"
+import "core:sync"
+
+// WakeUper is a wake-up callback. Value type — copyable.
+// wake and close do NOT use #contextless — callers may use context.logger.
+//
+// Call wake to signal the consumer that work is available.
+// Call close when done — it frees internal resources.
+// The zero value is valid. Callers must check wake != nil before calling.
+WakeUper :: struct {
+	ctx:   rawptr,
+	wake:  proc(rawptr),
+	close: proc(rawptr),
+}
+
+@(private)
+_Sema_State :: struct {
+	sema:      sync.Sema, // zero value is valid — no init or destroy needed
+	allocator: mem.Allocator,
+}
+
+// sema_wakeup returns a WakeUper backed by a semaphore.
+// Useful for non-nbio loops and unit tests.
+// Call waker.close(waker.ctx) when done to free resources.
+sema_wakeup :: proc(allocator := context.allocator) -> WakeUper {
+	state := new(_Sema_State, allocator)
+	state.allocator = allocator
+	return WakeUper{ctx = rawptr(state), wake = _sema_wake, close = _sema_close}
+}
+
+@(private)
+_sema_wake :: proc(ctx: rawptr) {
+	if ctx == nil {
+		return
+	}
+	state := (^_Sema_State)(ctx)
+	sync.sema_post(&state.sema)
+}
+
+@(private)
+_sema_close :: proc(ctx: rawptr) {
+	if ctx == nil {
+		return
+	}
+	state := (^_Sema_State)(ctx)
+	free(state, state.allocator)
+}