nx_thread_pool_test.nx source
↩ module page · 84 lines · 3083 B
1// nx_thread_pool_test.nx -- end-to-end pool stress. Submit
2// TOTAL_TASKS tasks each bumping a shared atomic counter; wait
3// for completion; verify counter == TOTAL_TASKS exactly.
4//
5// Two passes:
6// T1: small pool (hw workers) + many tasks; verifies fan-out
7// T2: re-use the same pool for another batch; verifies pool
8// liveness across multiple submit waves
9
10import "nx_kernel_v2.nx"
11import "nx_log.nx"
12import "nx_atom.nx"
13import "nx_thread.nx"
14import "nx_thread_pool.nx"
15import "nx_hw.nx"
16
17const TOTAL_TASKS_PER_WAVE: i64 = 500
18
19// Each task receives a pointer-as-i64 to the shared counter and
20// bumps it once.
21func bump_counter_task(ctx: i64) -> i64 {
22 let counter: *i64 = ctx as *i64
23 nx_atom_faa_i64(counter, 1, NX_MO_SEQ_CST)
24 return 0
25}
26
27func main() -> nx_exit {
28 var n_workers: i64 = nx_hw_worker_count()
29 if n_workers < 2 { n_workers = 2 } // ensure contention even on 1-cpu qemu
30 println("=== nx_thread_pool stress smoke ===" as *u8)
31 println("Pool workers (max(hw, 2)):" as *u8); print_i64(n_workers); println("" as *u8)
32 println("Tasks per wave:" as *u8); print_i64(TOTAL_TASKS_PER_WAVE); println("" as *u8)
33
34 let pool: *NxThreadPool = nx_pool_new(n_workers, 64)
35 if pool.n_workers != n_workers { println("FAIL: pool size mismatch" as *u8); return 1 }
36 if nx_pool_n_alive(pool) != n_workers {
37 println("FAIL: workers_alive != n_workers after creation" as *u8); return 2
38 }
39
40 // ---- Wave 1 ----
41 let counter_raw: *u8 = sys_mmap(16)
42 let counter: *i64 = counter_raw as *i64
43 *counter = 0
44
45 var i: i64 = 0
46 while i < TOTAL_TASKS_PER_WAVE {
47 nx_pool_submit(pool, bump_counter_task, counter_raw as i64)
48 i = i + 1
49 }
50 if nx_pool_wait(pool, TOTAL_TASKS_PER_WAVE) != 0 {
51 println("FAIL: wait timeout T1" as *u8); return 3
52 }
53 let v1: i64 = nx_atom_load_i64(counter, NX_MO_SEQ_CST)
54 if v1 != TOTAL_TASKS_PER_WAVE {
55 println("FAIL T1: counter != tasks" as *u8); print_i64(v1); println("" as *u8); return 4
56 }
57 println("PASS T1: wave 1 ran exactly TOTAL_TASKS_PER_WAVE times via pool." as *u8)
58
59 // ---- Wave 2 (pool reused) ----
60 *counter = 0
61 var j: i64 = 0
62 while j < TOTAL_TASKS_PER_WAVE {
63 nx_pool_submit(pool, bump_counter_task, counter_raw as i64)
64 j = j + 1
65 }
66 if nx_pool_wait(pool, TOTAL_TASKS_PER_WAVE * 2) != 0 {
67 println("FAIL: wait timeout T2" as *u8); return 5
68 }
69 let v2: i64 = nx_atom_load_i64(counter, NX_MO_SEQ_CST)
70 if v2 != TOTAL_TASKS_PER_WAVE {
71 println("FAIL T2: counter != tasks" as *u8); return 6
72 }
73 println("PASS T2: pool re-used for wave 2 -- counter cleared and bumped exactly." as *u8)
74
75 // ---- Shutdown ----
76 nx_pool_shutdown(pool)
77 if nx_pool_n_alive(pool) != 0 {
78 println("FAIL: workers_alive != 0 after shutdown" as *u8); return 7
79 }
80 println("PASS shutdown: all workers exited cleanly." as *u8)
81 println("" as *u8)
82 println("Thread pool proven: typed-fn-ptr dispatch + MPMC queue + hw-sized workers." as *u8)
83 return 0
84}