nx_parallel_ingest.nx source

1// nx_parallel_ingest.nx -- ingest many exceeds/capabilities IN PARALLEL without running out 2// of resources. The mechanism is three guarantees: 3// 1. BOUNDED CONCURRENCY -- a worker pool of at most K live forks at any instant (never a 4// fork-bomb); items stream through the pool, K busy at a time, the rest queued. 5// 2. PROCESS ISOLATION -- each worker is a fork; its scratch memory lives in ITS address 6// space and is reclaimed by the kernel on exit, so the parent's memory never grows with 7// the work. Per-item RAM is O(1) in the parent, not O(n). 8// 3. COMPACT SHARED BANK -- results are fixed-size records in ONE shared mmap (MAP_SHARED, 9// 0x21, so children write what the parent reads); the bank is O(n) bytes, not O(n) machine code. 10// So total resource use = K*(worker RAM) + n*(record size), both bounded and tunable -- you 11// can ingest thousands of exceeds with a handful of cores and a few KB. license_tier: ORIGINAL 12 13import "nx_syscalls.nx" // __syscall, SYS_MMAP, sys_fork, sys_wait4, sys_exit 14import "nx_boolsynth.nx" // per-item work: synthesize a minimal circuit 15 16// a SHARED anonymous mapping (visible across fork) -- the result bank. 17func par_mmap_shared(size: i64) -> *u8 { 18 let r: i64 = __syscall(SYS_MMAP, 0, size, 3, 0x21, 0 - 1, 0) 19 return r as *u8 20} 21 22// the per-item work a worker does: synthesize the minimal boolean circuit for truth table i 23// (pure compute, all scratch in the worker's own address space). returns its gate count. 24func par_work(i: i64) -> i64 { 25 let op: *i64 = sys_mmap(8 * 12) as *i64 26 let a: *i64 = sys_mmap(8 * 12) as *i64 27 let b: *i64 = sys_mmap(8 * 12) as *i64 28 let L: i64 = bl_find(i, op, a, b, 6) 29 if L < 0 { return 99 } 30 return L 31} 32 33// BOUNDED-PARALLEL ingest of items 0..n: at most K live workers; each writes bank[i]. The 34// parent only ever holds K child PIDs + the bank -- memory stays flat as n grows. 35func par_ingest(n: i64, K: i64, bank: *u8) -> i64 { 36 var next: i64 = 0 37 var live: i64 = 0 38 let status: *i64 = sys_mmap(16) as *i64 39 var done: i64 = 0 40 while done == 0 { 41 var fill: i64 = 1 42 while fill == 1 { 43 if live >= K { fill = 0 } else { if next >= n { fill = 0 } else { 44 let pid: i64 = sys_fork() 45 if pid == 0 { 46 bank[next] = par_work(next) as u8 // CHILD: compute + write shared, then exit 47 sys_exit(0) 48 } 49 live = live + 1 50 next = next + 1 51 } } 52 } 53 if live > 0 { sys_wait4(0 - 1, status, 0); live = live - 1 } 54 if next >= n { if live <= 0 { done = 1 } } 55 } 56 return 0 57}