nx_dispatch_lease.nx source

1// nx_dispatch_lease.nx -- WMS rung M4: the LEASING DISPATCHER / SCHEDULER. 2// 3// module: nishi-core.autonomy.dispatch_lease 4// capability: DISPATCH_LEASE_NEXT_READY_RUNG (frontier-aware pick + O_EXCL lease) 5// 6// WHAT THIS ADDS (the missing capability): nx_assign_next/an_pick are READ-ONLY -- 7// they NAME the next runnable rung but never CLAIM it, so two parallel workers both 8// see the same "next" and collide on one stream. M4 LEASES the pick under an 9// O_CREAT|O_EXCL lockfile (the kernel guarantees exactly one creator wins -- no 10// TOCTOU), so exactly one owner holds a given stream at a time. Generalizes the 11// conductor's single-arc drill into "keep the WHOLE machine moving": if the top 12// pick is already leased, the dispatcher scans down the w-ordering and leases the 13// NEXT ready stream instead of stalling. 14// 15// REUSE / lineage (compose, do not reinvent): 16// - an_load / an_pick / an_deps_done / an_id_at / an_find <- nx_assign_core (the 17// dep-graph + readiness + frontier-aware pick are ALREADY correct there) 18// - O_EXCL acquire/steal lock <- the rt_lock idiom from nx_registry_lock (INLINED 19// here as dl_lock/dl_unlock so this file keeps a SINGLE nx_syscalls surface and 20// dodges the double-import rc=6 trap -- nx_assign_core already pulls nx_syscalls 21// transitively, importing nx_registry_lock too would surface it twice) 22// - fa_cat / fa_catn / fa_appendz <- nx_framed_append (every status write is ONE 23// buffer -> ONE locked atomic fa_appendz; the torn-line bug we must NOT reintroduce) 24// 25// IMPORT DISCIPLINE: import nx_assign_core (surfaces nx_syscalls) + nx_framed_append 26// (also needs nx_syscalls). The prepass dedups nx_syscalls across imports (proven: 27// nx_assign_next imports BOTH nx_syscalls AND nx_assign_core and builds clean). We do 28// NOT import nx_registry_lock (its O_EXCL logic is inlined) -> only two import edges. 29// license_tier: ORIGINAL 30import "nx_assign_core.nx" 31import "nx_framed_append.nx" 32const DL_MAGIC_4096: i64 = 4096 33 34const DL_OEXCL: i64 = 193 // O_CREAT(0x40)|O_EXCL(0x80)|O_WRONLY(0x1) 35const DL_MODE: i64 = 420 // 0644 36const DL_STALE_SEC: i64 = 30 // steal a lease whose stamped epoch is older than this (crashed holder) 37const DL_REC_CAP: i64 = 256 // bounded status record size (no magic number) 38const DL_IDCAP: i64 = 32 // mirrors AN_IDCAP -- max business-key length 39 40// ---- string helpers (sovereign, local; mirror the substrate idioms) ---- 41func dl_streq(a: *u8, b: *u8) -> i64 { 42 var i: i64 = 0 43 while i < 64 { 44 if a[i] != b[i] { return 0 } 45 if a[i] == (0 as u8) { return 1 } 46 i = i + 1 47 } 48 return 1 49} 50 51// build "<dir>/<id>.lease" (NUL-terminated) into out; dir has NO trailing slash. 52// Returns out so callers can pass it straight into dl_lock. 53func dl_lease_path(dir: *u8, id: *u8, out: *u8) -> *u8 { 54 var o: i64 = 0 55 var i: i64 = 0 56 while dir[i] != (0 as u8) { out[o] = dir[i]; o = o + 1; i = i + 1 } 57 out[o] = 47 as u8; o = o + 1 // '/' 58 i = 0 59 while id[i] != (0 as u8) { out[o] = id[i]; o = o + 1; i = i + 1 } 60 out[o] = 46 as u8; o = o + 1 // '.' 61 out[o] = 108 as u8; o = o + 1 // 'l' 62 out[o] = 101 as u8; o = o + 1 // 'e' 63 out[o] = 97 as u8; o = o + 1 // 'a' 64 out[o] = 115 as u8; o = o + 1 // 's' 65 out[o] = 101 as u8; o = o + 1 // 'e' 66 out[o] = 0 as u8 67 return out 68} 69 70// ---- INLINED O_EXCL lease lock (rt_lock idiom; single nx_syscalls surface) ---- 71// stamp the holder's acquire-epoch into the lockfile (staleness channel for contenders) 72func dl_writeint_fd(fd: i64, v: i64) -> i64 { 73 let b: *u8 = sys_mmap(28) 74 var m: i64 = v 75 if m == 0 { b[0] = 48 as u8; sys_write(fd, b, 1); return 0 } 76 let t: *u8 = sys_mmap(28) 77 var k: i64 = 0 78 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 } 79 var i: i64 = 0 80 while i < k { b[i] = t[k - 1 - i]; i = i + 1 } 81 sys_write(fd, b, k) 82 return 0 83} 84func dl_readint(path: *u8) -> i64 { 85 let fd: i64 = sys_openat_rd(path) 86 if fd < 0 { return 0 } 87 let b: *u8 = sys_mmap(64) 88 let n: i64 = sys_read(fd, b, 63) 89 sys_close(fd) 90 var v: i64 = 0 91 var i: i64 = 0 92 while i < n { if b[i] >= (48 as u8) { if b[i] <= (57 as u8) { v = v * 10 + ((b[i] as i64) - 48) } } i = i + 1 } 93 return v 94} 95func dl_unlink(path: *u8) -> i64 { return __syscall(263, AT_FDCWD, path as i64, 0, 0, 0, 0) } 96 97// ACQUIRE a lease: O_CREAT|O_EXCL open of lockpath; exactly one creator wins. A 98// contender STEALS a lease whose stamped epoch is older than DL_STALE_SEC (a crashed 99// holder cannot wedge the machine). Bounded-spin once (this is a TRY: a held lease 100// returns -1 fast so the dispatcher can move to the next ready stream rather than 101// blocking on one collision). Returns the held fd (>=0) or -1 (already held). 102func dl_lock(lockpath: *u8) -> i64 { 103 let fd: i64 = __syscall(SYS_OPENAT, AT_FDCWD, lockpath as i64, DL_OEXCL, DL_MODE, 0, 0) 104 if fd >= 0 { 105 dl_writeint_fd(fd, sys_now_realtime_sec()) // stamp acquire-epoch 106 return fd 107 } 108 // already exists -- check staleness; steal a dead holder's lease then retry once. 109 let held: i64 = dl_readint(lockpath) 110 if held > 0 { 111 if sys_now_realtime_sec() - held > DL_STALE_SEC { 112 dl_unlink(lockpath) 113 let fd2: i64 = __syscall(SYS_OPENAT, AT_FDCWD, lockpath as i64, DL_OEXCL, DL_MODE, 0, 0) 114 if fd2 >= 0 { dl_writeint_fd(fd2, sys_now_realtime_sec()); return fd2 } 115 } 116 } 117 return 0 - 1 118} 119func dl_unlock(lockpath: *u8, fd: i64) -> i64 { sys_close(fd); dl_unlink(lockpath); return 0 } 120 121// ---- the M4 lease API (composes an_* + the inlined lock) ---- 122 123// Is stream r dispatchable? status==TODO(84) AND every dep DONE. 124func dl_ready(cx: *i64, r: i64) -> i64 { 125 let st: *i64 = cx[4] as *i64 126 if st[r] != 84 { return 0 } // 'T' TODO 127 return an_deps_done(cx, r) 128} 129 130// pick the next READY rung across ALL rows (frontier-aware): an_pick already enforces 131// highest-w + deps-done + tie smaller-size + file-order. Returns row idx or -1. 132func dl_pick_ready(cx: *i64) -> i64 { 133 return an_pick(cx) 134} 135 136// TRY-LEASE one stream by id under <dir>/<id>.lease. Returns the held lockfd (>=0) 137// iff THIS caller won the exclusive lease, or -1 if already held (DENIED). 138func dl_try_lease(dir: *u8, id: *u8) -> i64 { 139 let lp: *u8 = sys_mmap(512) 140 dl_lease_path(dir, id, lp) 141 return dl_lock(lp) 142} 143 144// RELEASE a held lease: close fd + unlink lease file (the stream is re-dispatchable). 145func dl_release(dir: *u8, id: *u8, fd: i64) -> i64 { 146 let lp: *u8 = sys_mmap(512) 147 dl_lease_path(dir, id, lp) 148 return dl_unlock(lp, fd) 149} 150 151// the LEASE-EXISTS probe (read-only): 1 if a lease lockfile is currently present. 152// Lets the keep-moving scan skip an already-leased stream WITHOUT consuming a steal. 153func dl_lease_held(dir: *u8, id: *u8) -> i64 { 154 let lp: *u8 = sys_mmap(512) 155 dl_lease_path(dir, id, lp) 156 let fd: i64 = sys_openat_rd(lp) 157 if fd < 0 { return 0 } 158 sys_close(fd) 159 return 1 160} 161 162// DISPATCH = pick a READY rung and LEASE it atomically. Frontier-aware + keep-moving: 163// 1. r = an_pick (top ready by w); if none ready -> -1. 164// 2. try-lease that id; if won -> return its id in id_out, fd in fd_out. 165// 3. if the top pick is already leased by another worker, scan the REMAINING ready 166// rows in w-descending order and lease the first un-leased one ("keep the whole 167// machine moving" rather than colliding or stalling). 168// Returns the held lockfd (>=0) with id_out NUL-terminated, or -1 if nothing ready / 169// every ready stream is already leased. A BLOCKED stream is NEVER returned (dl_ready 170// gates the scan) -- that is the structural neg-control guarantee. 171func dl_dispatch(cx: *i64, dir: *u8, id_out: *u8, fd_out: *i64) -> i64 { 172 let r: i64 = dl_pick_ready(cx) 173 if r < 0 { return 0 - 1 } 174 let topid: *u8 = an_id_at(cx, r) 175 let fd: i64 = dl_try_lease(dir, topid) 176 if fd >= 0 { 177 var i: i64 = 0 178 while topid[i] != (0 as u8) { id_out[i] = topid[i]; i = i + 1 } 179 id_out[i] = 0 as u8 180 fd_out[0] = fd 181 return fd 182 } 183 // top pick already leased -> keep moving: lease the next-best READY un-leased row. 184 let wv: *i64 = cx[3] as *i64 185 let sz: *i64 = cx[5] as *i64 186 var best: i64 = 0 - 1 187 var guard: i64 = 0 188 var done: i64 = 0 189 while done == 0 { 190 // find the best READY row that is NOT already leased and NOT already chosen-failed 191 best = 0 - 1 192 var rr: i64 = 0 193 while rr < cx[0] { 194 if dl_ready(cx, rr) == 1 { 195 if dl_lease_held(dir, an_id_at(cx, rr)) == 0 { 196 var take: i64 = 0 197 if best < 0 { take = 1 } else { 198 if wv[rr] > wv[best] { take = 1 } 199 if wv[rr] == wv[best] { if sz[rr] < sz[best] { take = 1 } } 200 } 201 if take == 1 { best = rr } 202 } 203 } 204 rr = rr + 1 205 } 206 if best < 0 { return 0 - 1 } // nothing ready+unleased remains 207 let bid: *u8 = an_id_at(cx, best) 208 let fd2: i64 = dl_try_lease(dir, bid) 209 if fd2 >= 0 { 210 var j: i64 = 0 211 while bid[j] != (0 as u8) { id_out[j] = bid[j]; j = j + 1 } 212 id_out[j] = 0 as u8 213 fd_out[0] = fd2 214 return fd2 215 } 216 // raced: someone leased `best` between our probe and our lock -> retry scan 217 guard = guard + 1 218 if guard > DL_MAGIC_4096 { done = 1 } // bounded give-up (no infinite spin) 219 } 220 return 0 - 1 221} 222 223// ---- single-write status emitter (the torn-line discipline) ---- 224// One record built into ONE buffer -> ONE locked atomic fa_appendz. Mirrors cn_emit_*. 225func dl_emit(path: *u8, verb: *u8, id: *u8, epoch: i64, ok: i64) -> i64 { 226 let buf: *u8 = sys_mmap(DL_REC_CAP + 16) 227 var o: i64 = 0 228 o = fa_cat(buf, o, "DISPATCH verb=\x00" as *u8) 229 o = fa_cat(buf, o, verb) 230 o = fa_cat(buf, o, " id=\x00" as *u8) 231 o = fa_cat(buf, o, id) 232 o = fa_cat(buf, o, " epoch=\x00" as *u8) 233 o = fa_catn(buf, o, epoch) 234 o = fa_cat(buf, o, " verdict=\x00" as *u8) 235 if ok == 1 { o = fa_cat(buf, o, "GREEN\x00" as *u8) } else { o = fa_cat(buf, o, "RED\x00" as *u8) } 236 return fa_appendz(path, buf, DL_REC_CAP) 237} 238 239func dl_p(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 240 241// thin CLI main: dispatch ONE ready rung over the live (or argv-overridden) queue, 242// leasing it under the (argv-overridden) lease dir, then RELEASE (CLI is a probe, not 243// a long-running worker -- it shows the pick + lease worked, then frees it). 244// argv[1] = queue path (default: knowledge/registry/assignment_queue.tsv) 245// argv[2] = lease dir (default: knowledge/status/leases) 246func main(argc: i64, argv: *i64) -> i64 { 247 var qp: *u8 = "knowledge/registry/assignment_queue.tsv" as *u8 248 var dir: *u8 = "knowledge/status/leases" as *u8 249 if argc >= 2 { qp = argv[1] as *u8 } 250 if argc >= 3 { dir = argv[2] as *u8 } 251 sys_mkdir(dir, 0x1ed) // ensure lease dir exists (0755); idempotent 252 let cx: *i64 = an_newcx() 253 let rows: i64 = an_load(qp, cx) 254 if rows <= 0 { 255 dl_p("DISPATCH verb=load id=NONE verdict=RED reason=queue-missing-or-empty\n" as *u8) 256 sys_exit(2) 257 return 2 258 } 259 let id_out: *u8 = sys_mmap(DL_IDCAP) 260 let fd_out: *i64 = sys_mmap(16) as *i64 261 let fd: i64 = dl_dispatch(cx, dir, id_out, fd_out) 262 if fd >= 0 { 263 dl_emit("knowledge/status/dispatch_lease.log\x00" as *u8, "leased\x00" as *u8, id_out, sys_now_realtime_sec(), 1) 264 dl_p("DISPATCH verb=leased id=" as *u8); dl_p(id_out); dl_p(" (CLI releases)\n" as *u8) 265 dl_release(dir, id_out, fd_out[0]) 266 sys_exit(0) 267 return 0 268 } 269 dl_emit("knowledge/status/dispatch_lease.log\x00" as *u8, "none\x00" as *u8, "NONE-READY\x00" as *u8, sys_now_realtime_sec(), 0) 270 dl_p("DISPATCH verb=none id=NONE-READY (all ready streams blocked or already leased)\n" as *u8) 271 sys_exit(1) 272 return 0 273}