nx_jobserver.nx source

1// nx_jobserver.nx -- PERSISTENT sovereign job-server: ONE long-lived Nishi process that runs queued commands, 2// so the agent loop stops spawning a fresh wsl.exe + sh -c per command (the per-command churn that eats memory 3// and triggers the .wslconfig crashes). Single-slot file queue (no shell, no quoting): 4// - poll <dir>/nxjob.req (line1 = elf path, lines2.. = one arg each, VERBATIM) 5// - on arrival: consume it (unlink first so it never re-runs), fork/exec the organ DIRECTLY with a WATCHDOG 6// (kill -9 at <watchdog> seconds so a runaway can't hang/eat memory), capture stdout+stderr -> nxjob.out, 7// write the exit code -> nxjob.res (the DONE signal, written LAST). The caller writes nxjob.req atomically 8// (write tmp -> rename) and waits for nxjob.res. One process, file-driven, watchdogged, NO bash/PowerShell/sh. 9// usage: nx_jobserver <queuedir> <pollbudget> <watchdog-sec> (pollbudget bounds the run for testing; a real 10// daemon passes a large budget or is restarted by the supervisor). exit codes in res: 124=watchdog-killed, 11// 127=execve-failed, 125=empty job, else the child's code. 12// Sovereign: nx_syscalls only (fork/execve/wait4/dup3/clock_nanosleep/unlinkat). license_tier: ORIGINAL 13import "nx_syscalls.nx" 14import "nx_itoa_lib.nx" // shared MSB-first emitter (zero-alloc) 15const K_MAGIC_1000000: i64 = 1000000 16const K_MAGIC_1024: i64 = 1024 17 18func js_w(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 } 19// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer 20// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the 21// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls). 22// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign. 23func js_wn(v: i64) -> i64 { nxi_out(v); return 0 } 24func js_atoi(s: *u8) -> i64 { var v: i64 = 0; var i: i64 = 0; while s[i] != (0 as u8) { let c: i64 = s[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 } return v } 25// build dir + "/" + name -> out (null-terminated) 26func js_path(dir: *u8, name: *u8, out: *u8) -> i64 { 27 var o: i64 = 0; var i: i64 = 0 28 while dir[i] != (0 as u8) { out[o] = dir[i]; o = o + 1; i = i + 1 } 29 out[o] = 47 as u8; o = o + 1 30 i = 0 31 while name[i] != (0 as u8) { out[o] = name[i]; o = o + 1; i = i + 1 } 32 out[o] = 0 as u8 33 return o 34} 35func js_unlink(path: *u8) -> i64 { __syscall(263, AT_FDCWD, path, 0, 0, 0, 0) return 0 } 36func js_sleep_ms(ms: i64) -> i64 { 37 let ts: *i64 = sys_mmap(16) as *i64 38 ts[0] = ms / 1000 39 ts[1] = (ms % 1000) * K_MAGIC_1000000 40 return __syscall(SYS_CLOCK_NANOSLEEP, 0, 0, ts, 0, 0, 0) 41} 42// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer 43// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the 44// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls). 45// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign. 46func js_wn_fd(fd: i64, v: i64) -> i64 { nxi_fd(fd, v); return 0 } 47 48// parse a job buffer -> argv, fork/exec the organ with output->out_path and a watchdog. returns exit code. 49func js_run_job(reqbuf: *u8, n: i64, out_path: *u8, wdog_sec: i64) -> i64 { 50 let buf: *u8 = sys_mmap(n + 8) 51 var ci: i64 = 0 52 while ci < n { buf[ci] = reqbuf[ci]; ci = ci + 1 } 53 buf[n] = 10 as u8 54 let cargv: *i64 = sys_mmap(8 * 128) as *i64 55 var ac: i64 = 0; var i: i64 = 0; var start: i64 = 0 56 while i <= n { 57 if (buf[i] as i64) == 10 { 58 var end: i64 = i 59 if end > start { if (buf[end - 1] as i64) == 13 { end = end - 1 } } 60 if end > start { buf[end] = 0 as u8; if ac < 127 { cargv[ac] = ((buf as i64) + start) as i64; ac = ac + 1 } } 61 start = i + 1 62 } 63 i = i + 1 64 } 65 cargv[ac] = 0 66 if ac == 0 { return 125 } 67 let out_fd: i64 = sys_openat_wr(out_path, 0x180) 68 if out_fd < 0 { return 126 } 69 let pid: i64 = sys_fork() 70 if pid == 0 { 71 sys_dup3(out_fd, 1, 0) 72 sys_dup3(out_fd, 2, 0) 73 let envp: *i64 = sys_mmap(16) as *i64 74 envp[0] = "PATH=/usr/bin:/bin" as *u8 as i64 75 envp[1] = 0 76 sys_execve(cargv[0] as *u8, cargv, envp) 77 sys_exit(127) 78 } 79 let st: *i64 = sys_mmap(16) as *i64 80 var ticks: i64 = 0 81 let deadline: i64 = wdog_sec * 5 // 200ms ticks 82 var rc: i64 = 124 83 var waiting: i64 = 1 84 while waiting == 1 { 85 let w: i64 = sys_wait4(pid, st, 1) // WNOHANG 86 if w == pid { 87 waiting = 0 88 if (st[0] % 128) != 0 { rc = 128 + (st[0] % 128) } else { rc = (st[0] >> 8) & 0xff } 89 } else { 90 ticks = ticks + 1 91 if ticks >= deadline { 92 waiting = 0 93 nx_kill(pid, 9) 94 sys_wait4(pid, st, 0) 95 rc = 124 96 } else { js_sleep_ms(200) } 97 } 98 } 99 sys_close(out_fd) 100 return rc 101} 102 103func main(argc: i64, argv: *i64) -> i64 { 104 if argc < 4 { sys_write(2, "usage: nx_jobserver <queuedir> <pollbudget> <watchdog-sec>\n" as *u8, 58); return 1 } 105 let dir: *u8 = argv[1] as *u8 106 let budget: i64 = js_atoi(argv[2] as *u8) 107 let wdog: i64 = js_atoi(argv[3] as *u8) 108 let req_path: *u8 = sys_mmap(K_MAGIC_1024); js_path(dir, "nxjob.req" as *u8, req_path) 109 let out_path: *u8 = sys_mmap(K_MAGIC_1024); js_path(dir, "nxjob.out" as *u8, out_path) 110 let res_path: *u8 = sys_mmap(K_MAGIC_1024); js_path(dir, "nxjob.res" as *u8, res_path) 111 let stop_path: *u8 = sys_mmap(K_MAGIC_1024); js_path(dir, "nxjob.stop" as *u8, stop_path) 112 js_w("nx_jobserver: watching " as *u8); js_w(dir); js_w(" (ONE persistent process; stop via nxjob.stop; budget=0 means forever)\n" as *u8) 113 var polls: i64 = 0 114 var ran: i64 = 0 115 var keep: i64 = 1 116 while keep == 1 { 117 let lenbox: *i64 = sys_mmap(16) as *i64; lenbox[0] = 0 118 let buf: *u8 = sys_read_file(req_path, lenbox) 119 if (buf as i64) != 0 { 120 if lenbox[0] > 0 { 121 js_unlink(req_path) 122 let code: i64 = js_run_job(buf, lenbox[0], out_path, wdog) 123 let rfd: i64 = sys_openat_wr(res_path, 0x180) 124 if rfd >= 0 { js_wn_fd(rfd, code); sys_write(rfd, "\n" as *u8, 1); sys_close(rfd) } 125 ran = ran + 1 126 js_w(" JOB " as *u8); js_wn(ran); js_w(" done exit=" as *u8); js_wn(code); js_w("\n" as *u8) 127 } 128 } 129 // stop signal (a real daemon needs a clean shutdown, not just a budget) 130 let sbox: *i64 = sys_mmap(16) as *i64; sbox[0] = 0 131 let sf: *u8 = sys_read_file(stop_path, sbox) 132 if (sf as i64) != 0 { keep = 0; js_unlink(stop_path) } 133 if keep == 1 { if budget > 0 { polls = polls + 1; if polls >= budget { keep = 0 } } } 134 if keep == 1 { js_sleep_ms(200) } 135 } 136 js_w("nx_jobserver: exit (jobs run=" as *u8); js_wn(ran); js_w(")\n" as *u8) 137 return 0 138}