`nx_ship_fleet.nx` source

↩ module page · 234 lines · 12244 B
1// nx_ship_fleet.nx -- INTELLIGENT (adaptive-concurrency) FLEET SHIPPER. Drains a list of targets by running
2// as many concurrent ships as the box can take RIGHT NOW -- never serially (too slow to drain a backlog),
3// never blindly (3 blind concurrent ships stormed the box on 2026-09-02). It composes the estate's existing
4// pieces into the DRIVER that was missing:
5//   SENSE   nx_sysload (sl_ncpu / sl_loadavg_milli / sl_freemem_mb / sl_active_conns_8443)
6//   GOVERN  nx_resource_governor rg_worker_budget -> the polite headroom budget, used here as a concurrency
7//           WIDTH (the governed drivers computed this budget and then fired ONE task per beat -- the budget
8//           was never used as a width, which is exactly why concurrent shipping was ungoverned)
9//   ADMIT   fork _offc/nx_build_admit.elf -> its exit code is the DIRECT I/O-storm signal (D-state witness),
10//           which the CPU-load budget alone cannot see (our storms are I/O on a degraded RAID, not CPU)
11//   AIMD    nx_shipfleet_lib -- +1 on progress, halve on a storm refusal, ceilinged by the polite budget
12//   POOL    fork _offc/nx_organ_ship.elf per target; reap with wait4(WNOHANG); requeue nothing blindly
13//
14// Research (Sept 2026): AIMD adaptive concurrency limits (Netflix concurrency-limits, Envoy
15// adaptive_concurrency filter, ThomWright/congestion-limiter) + HPA-style headroom budgeting. The field
16// infers congestion from latency; our exceed is a measured congestion signal (build_admit reads /proc).
17//
18//   usage: nx_ship_fleet <targets-file> [plan]
19//     <targets-file>  one target name per line; '#' and blank lines skipped
20//     plan            SENSE + print the width the controller WOULD use, fork NOTHING (safe live demo)
21// license_tier: ORIGINAL
22
23import "nx_syscalls.nx"
24import "nx_sysload.nx"
25import "nx_resource_governor.nx"
26import "nx_shipfleet_lib.nx"
27
28const SF_MEM_FLOOR_MB:   i64 = 256
29const SF_CEIL_MILLI:     i64 = 800     // back off at 0.8 core/cpu (rg default; shipping is background, not latency-sacred)
30const SF_BASE_BACKOFF_MS:i64 = 2000    // AIMD storm back-off base (rg_backoff_ms scales it up with overload)
31const SF_TICK_MS:        i64 = 250     // settle time between launch ticks (avoid busy-spin on /proc)
32const SF_MAX_STORM_TICKS:i64 = 120     // ~ bounded: consecutive storm/busy ticks before giving up (killable, deferred reported)
33const SF_MAXTARGETS:     i64 = 4096
34const SF_NAMECAP:        i64 = 128
35const SF_INFLIGHT_CAP:   i64 = 256
36const SF_READCAP:        i64 = 1048576
37const SF_SHIP_ELF:  *u8 = "_offc/nx_organ_ship.elf" as *u8
38const SF_ADMIT_ELF: *u8 = "_offc/nx_build_admit.elf" as *u8
39const SF_DEVNULL:   *u8 = "/dev/null\x00" as *u8
40
41func sf_slen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} return n }
42func sf_w(s: *u8) -> i64 { sys_write(1, s, sf_slen(s)); return 0 }
43func sf_n(v: i64) -> i64 {
  if v==0 { sys_write(1,"0" as *u8,1); return 0 }
  var m: i64=v; if m<0 { sys_write(1,"-" as *u8,1); m=0-m }
  let t: *u8=sys_mmap(24); var k: i64=0
  while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
  let o: *u8=sys_mmap(24); var w: i64=0; var q: i64=k-1
  while q>=0 { o[w]=t[q]; w=w+1; q=q-1 }
  sys_write(1,o,w); return 0
51}
52
53// read the whole targets file into buf (NUL-terminated); returns byte count, 0 on open fail.
54func sf_read(path: *u8, buf: *u8, cap: i64) -> i64 {
  let fd: i64 = sys_openat_rd(path); if fd<0 { return 0 }
  var n: i64=0; var r: i64=sys_read(fd, buf, cap-1)
  while r>0 { n=n+r; if n>=cap-1 { r=0 } else { r=sys_read(fd, buf+n, cap-1-n) } }
  sys_close(fd); buf[n]=0 as u8; return n
59}
60
61// the DIRECT congestion signal: fork build_admit 'check', exit 0 = GRANT (headroom), else REFUSE (storm/busy).
62// cannot-fork or a missing admit binary is treated as NO headroom (fail-safe: not knowing is not permission).
63func sf_admit_grant() -> i64 {
  let pid: i64 = sys_fork()
  if pid < 0 { return 0 }
  if pid == 0 {
      let dn: i64 = sys_openat_wr(SF_DEVNULL, 420)
      if dn>=0 { sys_dup3(dn,1,0); sys_dup3(dn,2,0) }
      let argv: *i64 = sys_mmap(64) as *i64
      argv[0]=SF_ADMIT_ELF as i64; argv[1]="check" as *u8 as i64; argv[2]=0
      let envp: *i64 = sys_mmap(16) as *i64; envp[0]="PATH=/usr/bin:/bin" as *u8 as i64; envp[1]=0
      sys_execve(SF_ADMIT_ELF, argv, envp); sys_exit(127)
  }
  let st: *i64 = sys_mmap(16) as *i64; sys_wait4(pid, st, 0)
  let ex: i64 = (st[0]>>8)&0xff
  if ex==0 { return 1 }
  return 0
78}
79
80// fork one ship for a target (detached -- parent tracks the pid); returns pid (>0) or -1 on fork fail.
81func sf_launch(target: *u8) -> i64 {
  let pid: i64 = sys_fork()
  if pid < 0 { return 0-1 }
  if pid == 0 {
      let dn: i64 = sys_openat_wr(SF_DEVNULL, 420)
      if dn>=0 { sys_dup3(dn,1,0); sys_dup3(dn,2,0) }
      let argv: *i64 = sys_mmap(64) as *i64
      argv[0]=SF_SHIP_ELF as i64; argv[1]=target as i64; argv[2]=0
      let envp: *i64 = sys_mmap(16) as *i64; envp[0]="PATH=/usr/bin:/bin" as *u8 as i64; envp[1]=0
      sys_execve(SF_SHIP_ELF, argv, envp); sys_exit(127)
  }
  return pid
93}
94
95// the polite governed WIDTH ceiling for this instant (0 if memory is below floor).
96func sf_rg_budget(ncpu: i64, load: i64, freemb: i64) -> i64 {
  if rg_memory_ok(freemb, SF_MEM_FLOOR_MB)==0 { return 0 }
  let reserve: i64 = rg_hw_class_reserve(ncpu)   // sensor->supercomputer class-aware reserve
  return rg_worker_budget(RG_POLITE, ncpu, load, reserve, SF_CEIL_MILLI)
100}
101
102func main(argc: i64, argv: *i64) -> i64 {
  if argc < 2 {
      sf_w("usage: nx_ship_fleet <targets-file> [plan]\n" as *u8)
      return 1
  }
  let path: *u8 = argv[1] as *u8
  var plan_only: i64 = 0
  if argc >= 3 { let a2: *u8 = argv[2] as *u8; if a2[0]==(112 as u8) { plan_only=1 } }   // 'p'lan
110
  // parse the target list.
  let buf: *u8 = sys_mmap(SF_READCAP)
  let n: i64 = sf_read(path, buf, SF_READCAP)
  if n <= 0 { sf_w("SHIP-FLEET no-targets file=" as *u8); sf_w(path); sf_w("\n" as *u8); return 1 }
  let names: *u8 = sys_mmap(SF_MAXTARGETS * SF_NAMECAP)
  var nt: i64 = 0
  var ls: i64 = 0; var i: i64 = 0
  while i <= n {
      var eol: i64 = 0
      if i==n { eol=1 } else { if buf[i]==(10 as u8) { eol=1 } }
      if eol==1 {
          if i>ls { if buf[ls]!=(35 as u8) {          // skip '#'
              if nt < SF_MAXTARGETS {
                  let dst: *u8 = (names + nt*SF_NAMECAP) as *u8
                  var k: i64 = 0; var p: i64 = ls
                  while p<i { if buf[p]!=(13 as u8) { if k < SF_NAMECAP-1 { dst[k]=buf[p]; k=k+1 } } p=p+1 }
                  dst[k]=0 as u8
                  if k>0 { nt=nt+1 }
              }
          } }
          ls=i+1
      }
      i=i+1
  }
  sf_w("=== nx_ship_fleet: adaptive-concurrency drain of " as *u8); sf_n(nt); sf_w(" targets from " as *u8); sf_w(path); sf_w(" ===\n" as *u8)
  if nt==0 { sf_w("SHIP-FLEET nothing-to-ship verdict=GREEN\n" as *u8); return 0 }
137
  // SENSE once for the header / plan mode.
  let ncpu0: i64=sl_ncpu(); let load0: i64=sl_loadavg_milli(); let free0: i64=sl_freemem_mb(); let conns0: i64=sl_active_conns_8443()
  let rgb0: i64 = sf_rg_budget(ncpu0, load0, free0)
  let admit0: i64 = sf_admit_grant()
  let w0: i64 = sf_effective_width(rgb0, rgb0, admit0, rg_serve_first(conns0))
  sf_w("  SENSE ncpu=" as *u8); sf_n(ncpu0); sf_w(" load=" as *u8); sf_n(load0); sf_w("milli free=" as *u8); sf_n(free0)
  sf_w("MB conns=" as *u8); sf_n(conns0); sf_w(" -> rg_budget=" as *u8); sf_n(rgb0)
  sf_w(" admit=" as *u8); if admit0==1 { sf_w("GRANT" as *u8) } else { sf_w("REFUSE" as *u8) }
  sf_w(" -> start width=" as *u8); sf_n(w0); sf_w("\n" as *u8)
  if plan_only==1 {
      sf_w("SHIP-FLEET plan-only (forked nothing) would_start_width=" as *u8); sf_n(w0)
      sf_w(" targets=" as *u8); sf_n(nt); sf_w(" verdict=GREEN\n" as *u8)
      return 0
  }
152
  // the adaptive concurrency loop.
  let pids:  *i64 = sys_mmap(SF_INFLIGHT_CAP*8) as *i64
  let ptgt:  *i64 = sys_mmap(SF_INFLIGHT_CAP*8) as *i64
  var next: i64=0; var inflight: i64=0; var done: i64=0; var failed: i64=0
  var maxseen: i64=0; var cap: i64=1; var storm_ticks: i64=0
  let st: *i64 = sys_mmap(16) as *i64
159
  while (next < nt) | (inflight > 0) {
      // reap every finished child (non-blocking).
      var reaping: i64 = 1
      while reaping==1 {
          let r: i64 = sys_wait4(0-1, st, 1)   // WNOHANG
          if r > 0 {
              let ex: i64 = (st[0]>>8)&0xff
              var j: i64=0; var found: i64=0-1
              while j<inflight { if pids[j]==r { found=j; j=inflight } else { j=j+1 } }
              if found>=0 {
                  let ti: i64 = ptgt[found]
                  if ex==0 { done=done+1; sf_w("  SHIPPED " as *u8) } else { failed=failed+1; sf_w("  FAILED  " as *u8) }
                  sf_w((names + ti*SF_NAMECAP) as *u8); sf_w(" exit=" as *u8); sf_n(ex)
                  sf_w(" (done=" as *u8); sf_n(done); sf_w(" failed=" as *u8); sf_n(failed); sf_w(")\n" as *u8)
                  let last: i64 = inflight-1
                  pids[found]=pids[last]; ptgt[found]=ptgt[last]
                  inflight=last
              }
          } else { reaping=0 }
      }
      if (next >= nt) & (inflight==0) { reaping=0; next=next }   // done; fall through to exit
181
      if (next < nt) | (inflight > 0) {
          // SENSE + GOVERN + ADMIT.
          let ncpu: i64=sl_ncpu(); let load: i64=sl_loadavg_milli(); let freemb: i64=sl_freemem_mb(); let conns: i64=sl_active_conns_8443()
          let rgb: i64 = sf_rg_budget(ncpu, load, freemb)
          let admit: i64 = sf_admit_grant()
          let width: i64 = sf_effective_width(rgb, cap, admit, rg_serve_first(conns))
          if width==0 {
              cap = sf_aimd_next(cap, rgb, SF_STORM)
              if inflight==0 {
                  storm_ticks = storm_ticks+1
                  if storm_ticks > SF_MAX_STORM_TICKS {
                      sf_w("  STORM-GIVEUP after " as *u8); sf_n(storm_ticks); sf_w(" idle storm ticks -- deferring the rest (killable, re-run to resume)\n" as *u8)
                      next = nt   // stop launching; nothing in flight -> loop exits
                  }
              }
              var bo: i64 = rg_backoff_ms(RG_POLITE, ncpu, load, SF_CEIL_MILLI, SF_BASE_BACKOFF_MS)
              if bo < SF_BASE_BACKOFF_MS { bo = SF_BASE_BACKOFF_MS }
              sys_sleep_ms(bo)
          } else {
              storm_ticks = 0
              let slots: i64 = sf_launch_slots(width, inflight)
              var launched: i64=0; var s: i64=0
              while (s < slots) & (next < nt) {
                  if inflight < SF_INFLIGHT_CAP {
                      let pid: i64 = sf_launch((names + next*SF_NAMECAP) as *u8)
                      if pid > 0 {
                          pids[inflight]=pid; ptgt[inflight]=next; inflight=inflight+1; next=next+1; launched=launched+1
                      } else { s = slots }   // fork failed -> stop launching this tick
                  } else { s = slots }
                  s=s+1
              }
              if inflight > maxseen { maxseen=inflight }
              if launched>0 {
                  cap = sf_aimd_next(cap, rgb, SF_OK)
                  sf_w("  tick launch=" as *u8); sf_n(launched); sf_w(" inflight=" as *u8); sf_n(inflight)
                  sf_w(" width=" as *u8); sf_n(width); sf_w(" cap=" as *u8); sf_n(cap); sf_w(" load=" as *u8); sf_n(load); sf_w("milli\n" as *u8)
              } else {
                  cap = sf_aimd_next(cap, rgb, SF_HOLD)
              }
              sys_sleep_ms(SF_TICK_MS)
          }
      }
  }
225
  let deferred: i64 = nt - done - failed
  sf_w("SHIP-FLEET targets=" as *u8); sf_n(nt); sf_w(" shipped=" as *u8); sf_n(done)
  sf_w(" failed=" as *u8); sf_n(failed); sf_w(" deferred=" as *u8); sf_n(deferred)
  sf_w(" max_concurrency=" as *u8); sf_n(maxseen); sf_w(" final_cap=" as *u8); sf_n(cap)
  sf_w(" verdict=" as *u8)
  if (failed==0) & (deferred==0) { sf_w("GREEN\n" as *u8); return 0 }
  sf_w("AMBER\n" as *u8)
  return 0
234}
nx_ship_fleet.nx source

`nx_ship_fleet.nx` source
