nx_parallel_run.nx source

1// nx_parallel_run.nx -- the PARALLEL EXECUTION engine (operator: go fast + in parallel, 2// 32-min sequential beat is a Camry). Runs a LIST of INDEPENDENT organs K-wide 3// concurrently (K from nx_resources -- schedule to the iron), instead of one-at-a-time. 4// The JazzNet move on our OWN beat: 20 cores, 16 lanes, ~16x the throughput on the 5// parallelizable graders. Wave-based: launch K, wait the batch, launch the next K. 6// SAFE for INDEPENDENT steps only (read different logs / write different outputs) -- 7// the dependent chain (attempt->reconcile->ark) stays sequential elsewhere. 8// in: a conf of organ names (one per line, # comments); K from resources.log. 9// Durable: PARALLEL row (lanes, count, wall-ms) -> knowledge/status/parallel.log. 10// Exit = number of nonzero organs. argv[1]=conf argv[2]=k-override. license_tier: ORIGINAL 11import "nx_syscalls.nx" 12const K_MAGIC_1000000: i64 = 1000000 13const K_MAGIC_65536: i64 = 65536 14const K_MAGIC_65520: i64 = 65520 15func _p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 16func _fp(fd: i64, s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(fd,s,n); return 0 } 17func _fn(fd: i64, v: i64) -> i64 { let bb: *u8=sys_mmap(28); var m: i64=v; if m<0{m=0-m}; let t: *u8=sys_mmap(28); var k: i64=0; if m==0{t[0]=48 as u8;k=1}; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1}; var i: i64=0; while i<k{bb[i]=t[k-1-i];i=i+1}; sys_write(fd,bb,k); return 0 } 18func pr_ms() -> i64 { 19 let ts: *i64 = sys_mmap(16) as *i64 20 sys_clock_gettime_mono(ts) 21 return ts[0] * 1000 + ts[1] / K_MAGIC_1000000 22} 23func pr_read(path: *u8, buf: *u8, cap: i64) -> i64 { 24 let fd: i64 = sys_openat_rd(path) 25 if fd < 0 { return 0 } 26 var n: i64 = 0 27 var go: i64 = 1 28 while go == 1 { let base: i64 = buf as i64; let r: i64 = sys_read(fd, (base + n) as *u8, cap - n); if r <= 0 { go = 0 } else { n = n + r } if n >= cap { go = 0 } } 29 sys_close(fd) 30 return n 31} 32func pr_klanes() -> i64 { 33 let b: *u8 = sys_mmap(K_MAGIC_65536) 34 let n: i64 = pr_read("knowledge/status/resources.log" as *u8, b, K_MAGIC_65520) 35 // last "k_lanes=" value 36 var k: i64 = 4 37 var i: i64 = 0 38 let key: *u8 = "k_lanes=" as *u8 39 while i + 8 <= n { 40 var kk: i64 = 0 41 var hit: i64 = 1 42 while kk < 8 { if b[i+kk] != key[kk] { hit = 0; kk = 8 } else { kk = kk + 1 } } 43 if hit == 1 { var p: i64 = i + 8; var v: i64 = 0; while p < n { if b[p] < (48 as u8) { p = n } else { if b[p] > (57 as u8) { p = n } else { v = v*10 + ((b[p] as i64)-48); p = p + 1 } } } k = v } 44 i = i + 1 45 } 46 if k < 1 { k = 1 } 47 return k 48} 49// launch one organ via the sov runner (muted), PINNED to `core` (sovereign-driver 50// governor-off: no OS scheduler bounce / cache thrash, X-DRV-001), return pid 51func pr_launch(name: *u8, core: i64) -> i64 { 52 let pid: i64 = sys_fork() 53 if pid == 0 { 54 // sched_setaffinity(0, 128, &mask) -- pin self to one core (syscall 203) 55 let mask: *i64 = sys_mmap(128) as *i64 56 mask[0] = 1 << (core & 63) 57 __syscall(203, 0, 128, mask as i64, 0, 0, 0) 58 let dn: i64 = sys_openat_wr("/dev/null" as *u8, 0x1a4) 59 if dn >= 0 { sys_dup3(dn, 1, 0); sys_dup3(dn, 2, 0) } 60 let argv: *i64 = sys_mmap(32) as *i64 61 argv[0] = "_offc/nx_sov_build_run.elf" as *u8 as i64 62 argv[1] = name as i64 63 argv[2] = 0 64 let envp: *i64 = sys_mmap(16) as *i64 65 envp[0] = 0 66 sys_execve("_offc/nx_sov_build_run.elf" as *u8, argv, envp) 67 sys_exit(127) 68 } 69 return pid 70} 71func main(argc: i64, argv: *i64) -> i64 { 72 var cp: *u8 = "knowledge/registry/parallel_set.conf" as *u8 73 if argc >= 2 { cp = argv[1] as *u8 } 74 var klanes: i64 = pr_klanes() 75 if argc >= 3 { var v: i64 = 0; var s: *u8 = argv[2] as *u8; var i: i64 = 0; while s[i] >= (48 as u8) { if s[i] > (57 as u8) { i = i + K_MAGIC_1000000 } else { v = v*10 + ((s[i] as i64)-48); i = i + 1 } } if v > 0 { klanes = v } } 76 _p("=== PARALLEL RUN: K-wide independent execution (schedule to the iron) ===\n" as *u8) 77 let conf: *u8 = sys_mmap(K_MAGIC_65536) 78 let cn: i64 = pr_read(cp, conf, K_MAGIC_65520) 79 if cn <= 0 { _p(" parallel_set conf MISSING -- fail loud\n" as *u8); sys_exit(1); return 1 } 80 // collect organ names (nul-terminated) into a pool + ptr array 81 let pool: *u8 = sys_mmap(K_MAGIC_65536) 82 let names: *i64 = sys_mmap(8 * 512) as *i64 83 var nn: i64 = 0 84 var po: i64 = 0 85 var i: i64 = 0 86 while i < cn { 87 var le: i64 = i 88 var s: i64 = 1 89 while s == 1 { if le >= cn { s = 0 } else { if conf[le] == (10 as u8) { s = 0 } else { le = le + 1 } } } 90 if conf[i] != (35 as u8) { if le > i { 91 let name_start: i64 = po 92 var j: i64 = i 93 while j < le { if conf[j] != (32 as u8) { if conf[j] != (9 as u8) { pool[po] = conf[j]; po = po + 1 } } j = j + 1 } 94 if po > name_start { // a non-empty name was copied 95 names[nn] = (pool as i64) + name_start 96 pool[po] = 0 as u8; po = po + 1 97 nn = nn + 1 98 } 99 } } 100 i = le + 1 101 } 102 let t0: i64 = pr_ms() 103 let pids: *i64 = sys_mmap(8 * 64) as *i64 104 var bad: i64 = 0 105 var done: i64 = 0 106 while done < nn { 107 // launch a wave of up to klanes 108 var w: i64 = 0 109 while w < klanes { if done + w < nn { pids[w] = pr_launch(names[done + w] as *u8, w) } w = w + 1 } 110 // wait the wave 111 var wave: i64 = klanes 112 if done + wave > nn { wave = nn - done } 113 var x: i64 = 0 114 while x < wave { 115 let st: *i64 = sys_mmap(16) as *i64 116 sys_wait4(pids[x], st, 0) 117 if (st[0] >> 8) & 0xff != 0 { bad = bad + 1 } 118 x = x + 1 119 } 120 done = done + wave 121 } 122 let wall: i64 = pr_ms() - t0 123 let lfd: i64 = sys_openat_append("knowledge/status/parallel.log" as *u8, 0x1a4) 124 if lfd >= 0 { 125 _fp(lfd, "PARALLEL epoch=" as *u8); _fn(lfd, sys_now_realtime_sec()) 126 _fp(lfd, " organs=" as *u8); _fn(lfd, nn) 127 _fp(lfd, " k_lanes=" as *u8); _fn(lfd, klanes) 128 _fp(lfd, " wall_ms=" as *u8); _fn(lfd, wall) 129 _fp(lfd, " nonzero=" as *u8); _fn(lfd, bad) 130 _fp(lfd, " verdict=RAN\n" as *u8) 131 sys_close(lfd) 132 } 133 _p(" ran " as *u8); _fn(1, nn); _p(" organs " as *u8); _fn(1, klanes); _p("-wide in " as *u8); _fn(1, wall); _p("ms\n" as *u8) 134 sys_exit(bad) 135 return bad 136}