code wiki / _hdl_build / nx_room_perf_gate.nx
`nx_room_perf_gate.nx` source

↩ module page · 385 lines · 15706 B
1import "nx_gate_base.nx"
2import "nx_connect.nx"   // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
3// nx_room_perf_gate.nx -- ENGINEER perf gate: MEASURE the live room, then gate it.
4//
5// Operator 2026-06-10: "we tested the platform and now we can all join but the
6// performance was poor."  Optimal/S-class doctrine: slow is a bug, and a bug
7// needs a MEASURED, re-runnable gate before any fix.  This gate runs the same
8// sovereign TLS clients as nx_room_live_gate against the PUBLIC stack and
9// measures what the user feels:
10//   P1 unloaded latency   -- A sends paced 64B timestamped frames; B reports
11//                            inter-client delivery latency (same-process clock,
12//                            so the numbers are exact) p50/max in usec
13//   P2 loaded throughput  -- a forked sender child streams 120 x 8KB timestamped
14//                            frames from A; the parent reads them at BOTH B and C;
15//                            delivered count, wall time, KB/s, loaded p50/max
16//   P4 slow-reader stall  -- C STOPS reading entirely; a child streams 96 x 8KB
17//                            from A; B must STILL receive all 96.  On the v2
18//                            relay (blocking serial broadcast) one slow peer
19//                            wedges the whole daemon -> this is the measured
20//                            root cause of "poor performance", RED until fixed.
21// Every byte crosses the open internet through the deployed daemon; nothing
22// loopback, nothing mocked.  Requires /tmp/mozilla_certdata.txt staged.
23// license_tier: ORIGINAL
24
25import "nx_syscalls.nx"
26import "nx_gate_emit_lib.nx"
27import "nx_x509_trust_store.nx"
28import "nx_trust_store_load_from_certdata.nx"
29import "nx_tls13_client_validate_certificate.nx"
30import "nx_tls13_client_session_run.nx"
31import "nx_dns_resolve_a_record.nx"
32import "nx_https_url_connect.nx"
33import "nx_websocket_frame.nx"
34import "nx_websocket_client_upgrade.nx"
35
36
37// Slot layout for the per-connection state array `c` (*i64, RG_SLOTS entries, allocated by this
38// file). The whole block was missing -- defined in NO file in the tree -- so nx_parse stopped at the
39// first use (RG_S) and both gates had been unbuildable. These are private indices, so any distinct
40// assignment is correct; RG_SLOTS is the allocation count.
41const RG_FD:      i64 = 0
42const RG_S:       i64 = 1
43const RG_ACC:     i64 = 2
44const RG_ACC_CAP: i64 = 3
45const RG_ACC_LEN: i64 = 4
46const RG_ACC_OFF: i64 = 5
47const RG_SLOTS:   i64 = 6
48
49
50func grow(name: *u8, ok: i64) -> i64 { if ok==1 { gw("  PASS " as *u8) } else { gw("  FAIL " as *u8) } gw(name); gw("
51" as *u8); return ok }
52func rg_tls_send(c: *i64, buf: *u8, n: i64) -> i64 {
  let s: *Tls13ClientSession = c[RG_S] as *Tls13ClientSession
  let rec: *u8 = sys_mmap(n + 64)
  let hdr: *u8 = rec
  let ct: *u8 = rec + NX_TLS13_RECORD_HEADER_LEN
  let tag: *u8 = rec + NX_TLS13_RECORD_HEADER_LEN + n + 1
  let v: i64 = nx_tls13_record_encrypt_v2(
      s.cipher_suite, s.client_app_traffic_key, s.client_app_iv,
      s.client_app_seq, buf, n, NX_TLS13_CT_APPLICATION_DATA, 0,
      hdr, ct, tag)
  s.client_app_seq = s.client_app_seq + 1
  if v != NX_TLS13_REC_VERDICT_OK { return 0 - 1 }
  let total: i64 = NX_TLS13_RECORD_HEADER_LEN + n + 1 + NX_TLS13_RECORD_TAG_LEN
  var off: i64 = 0
  while off < total {
      let w: i64 = sys_write(c[RG_FD], (rec as i64 + off) as *u8, total - off)
      if w <= 0 { return 0 - 1 }
      off = off + w
  }
  return 0
72}
73
74func rg_pump(c: *i64) -> i64 {
  let s: *Tls13ClientSession = c[RG_S] as *Tls13ClientSession
  let rec: *u8 = sys_mmap(16700)
  let total: i64 = nx_tls13_read_record_from_fd(c[RG_FD], rec, 16700)
  if total < 0 { return 0 - 1 }
  let ct_len: i64 = total - NX_TLS13_RECORD_HEADER_LEN - NX_TLS13_RECORD_TAG_LEN
  if ct_len <= 0 { return 0 - 1 }
  let plain: *u8 = sys_mmap(ct_len + 16)
  let ctp: *i64 = sys_mmap(16) as *i64
  let lnp: *i64 = sys_mmap(16) as *i64
  let v: i64 = nx_tls13_record_decrypt_v2(
      s.cipher_suite, s.server_app_traffic_key, s.server_app_iv,
      s.server_app_seq, rec, rec + NX_TLS13_RECORD_HEADER_LEN, ct_len,
      rec + total - NX_TLS13_RECORD_TAG_LEN, plain, ctp, lnp)
  s.server_app_seq = s.server_app_seq + 1
  if v != NX_TLS13_REC_VERDICT_OK { return 0 - 2 }
  if *ctp == NX_TLS13_CT_ALERT { return 0 - 3 }
  if *ctp != NX_TLS13_CT_APPLICATION_DATA { return 0 }
  let acc: *u8 = c[RG_ACC] as *u8
  var w: i64 = c[RG_ACC_LEN]
  if w + *lnp > c[RG_ACC_CAP] { return 0 - 4 }
  var i: i64 = 0
  while i < *lnp { acc[w + i] = plain[i]; i = i + 1 }
  c[RG_ACC_LEN] = w + *lnp
  return *lnp
99}
100
101func rg_next_frame(c: *i64, f: *WsFrame) -> i64 {
  var tries: i64 = 0
  while tries < 256 {
      let acc: *u8 = c[RG_ACC] as *u8
      let v: i64 = ws_parse_frame(acc, c[RG_ACC_LEN], c[RG_ACC_OFF], f)
      if v == 0 {
          c[RG_ACC_OFF] = c[RG_ACC_OFF] + f.frame_len
          return 0
      }
      let p: i64 = rg_pump(c)
      if p < 0 { return p }
      tries = tries + 1
  }
  return 0 - 9
115}
116
117// reset the accumulator so long phases never overflow the cap
118func rg_reset_acc(c: *i64) -> i64 {
  let acc: *u8 = c[RG_ACC] as *u8
  let keep: i64 = c[RG_ACC_LEN] - c[RG_ACC_OFF]
  var i: i64 = 0
  while i < keep { acc[i] = acc[c[RG_ACC_OFF] + i]; i = i + 1 }
  c[RG_ACC_LEN] = keep
  c[RG_ACC_OFF] = 0
  return 0
126}
127
128func rg_send_ws(c: *i64, opcode: i64, payload: *u8, n: i64, mask: i64) -> i64 {
  let buf: *u8 = sys_mmap(n + 32)
  let hl: i64 = ws_build_header(buf, n + 32, 1, opcode, 1, mask, n)
  if hl < 0 { return 0 - 1 }
  var i: i64 = 0
  while i < n { buf[hl + i] = payload[i]; i = i + 1 }
  ws_apply_mask(buf, hl, n, mask)
  return rg_tls_send(c, buf, hl + n)
136}
137
138func rg_join(c: *i64, ipv4: i64, sni: *u8, sni_n: i64, path: *u8, path_n: i64,
           store: *TrustStore, seed: i64) -> i64 {
  let fd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
  if fd < 0 { return 0 - 1 }
  sys_set_socket_timeout(fd, 15)
  let sa: *u8 = sys_mmap(16)
  nx_https_build_sockaddr(sa, ipv4, 443)
  if nx_connect_bounded(fd, sa, 16, NX_CONN_DEFAULT_MS) < 0 { sys_close(fd); return 0 - 2 }
  let cr: *u8 = sys_mmap(32)
  let priv: *u8 = sys_mmap(32)
  var i: i64 = 0
  while i < 32 { cr[i] = ((seed * 37 + i * 11 + 5) & 0xff) as u8; priv[i] = ((seed * 53 + i * 7 + 9) & 0xff) as u8; i = i + 1 }
  let vraw: *u8 = sys_mmap(64)
  let val_ctx: *TlsValidationContext = vraw as *TlsValidationContext
  val_ctx.store = store
  val_ctx.sni_host = sni
  val_ctx.sni_host_len = sni_n
  val_ctx.now_epoch = sys_now_realtime_sec()
  let r: i64 = nx_tls13_client_session_run(fd, sni, sni_n, cr, priv, val_ctx)
  if r < 0 { g_puts("    tls run verdict=" as *u8); g_pn(r); g_puts("\n" as *u8); sys_close(fd); return 0 - 3 }
  c[RG_FD] = fd
  c[RG_S] = r
  c[RG_ACC] = sys_mmap(4194304) as i64
  c[RG_ACC_LEN] = 0
  c[RG_ACC_OFF] = 0
  c[RG_ACC_CAP] = 4194304
  let req: *u8 = sys_mmap(2048)
  let key24: *u8 = sys_mmap(32)
  let rn: i64 = nx_ws_client_build_request(req, 2048, sni, sni_n, path, path_n, 443, 0, key24)
  if rn <= 0 { return 0 - 4 }
  if rg_tls_send(c, req, rn) < 0 { return 0 - 5 }
  var hend: i64 = 0 - 1
  var tries: i64 = 0
  while tries < 32 {
      let acc: *u8 = c[RG_ACC] as *u8
      var j: i64 = 0
      while j + 3 < c[RG_ACC_LEN] {
          var hit: i64 = 1
          if acc[j] != (13 as u8) { hit = 0 }
          if acc[j+1] != (10 as u8) { hit = 0 }
          if acc[j+2] != (13 as u8) { hit = 0 }
          if acc[j+3] != (10 as u8) { hit = 0 }
          if hit == 1 { hend = j + 4; j = c[RG_ACC_LEN] } else { j = j + 1 }
      }
      if hend > 0 { tries = 32 } else {
          if rg_pump(c) < 0 { return 0 - 6 }
          tries = tries + 1
      }
  }
  if hend < 0 { return 0 - 6 }
  let acc2: *u8 = c[RG_ACC] as *u8
  let uv: i64 = nx_ws_client_validate_response(acc2, hend, key24)
  if uv != NX_WSCU_OK { return 0 - 7 }
  c[RG_ACC_OFF] = hend
  return 0
193}
194
195// media payload: [kind][8B id][u32 seq LE][u64 sent_us LE][body...]
196func rg_media_ts(out: *u8, kind: i64, id: *u8, seq: i64, body_n: i64) -> i64 {
  out[0] = kind as u8
  var i: i64 = 0
  while i < 8 { out[1 + i] = id[i]; i = i + 1 }
  out[9] = (seq & 255) as u8
  out[10] = ((seq / 256) & 255) as u8
  out[11] = ((seq / 65536) & 255) as u8
  out[12] = ((seq / 16777216) & 255) as u8
  var t: i64 = sys_now_us()
  i = 0
  while i < 8 { out[13 + i] = (t & 255) as u8; t = t / 256; i = i + 1 }
  i = 0
  while i < body_n { out[21 + i] = ((seq * 31 + i * 7 + 3) & 0xff) as u8; i = i + 1 }
  return 21 + body_n
210}
211
212func rg_read_ts(acc: *u8, off: i64) -> i64 {
  var t: i64 = 0
  var i: i64 = 7
  while i >= 0 {
      t = t * 256 + (acc[off + 13 + i] & 0xff)
      i = i - 1
  }
  return t
220}
221
222// insertion-sort a small i64 array ascending
223func rg_sort(a: *i64, n: i64) -> i64 {
  var i: i64 = 1
  while i < n {
      let v: i64 = a[i]
      var j: i64 = i - 1
      var stop: i64 = 0
      while stop == 0 {
          if j < 0 { stop = 1 }
          if stop == 0 { if a[j] <= v { stop = 1 } }
          if stop == 0 { a[j + 1] = a[j]; j = j - 1 }
      }
      a[j + 1] = v
      i = i + 1
  }
  return 0
238}
239
240// receive ONE binary frame at c; returns latency usec (recv-send) or negative
241func rg_recv_lat(c: *i64, f: *WsFrame) -> i64 {
  if rg_next_frame(c, f) != 0 { return 0 - 1 }
  if f.opcode != WS_OP_BINARY { return 0 - 2 }
  let acc: *u8 = c[RG_ACC] as *u8
  let sent: i64 = rg_read_ts(acc, f.payload_off)
  return sys_now_us() - sent
247}
248
249// fork a sender child: streams `count` frames of `body_n` bytes from A
250func rg_fork_sender(ca: *i64, id: *u8, count: i64, body_n: i64) -> i64 {
  let pid: i64 = sys_fork()
  if pid != 0 { return pid }
  let frm: *u8 = sys_mmap(body_n + 64)
  var s: i64 = 0
  while s < count {
      let n: i64 = rg_media_ts(frm, 0x56, id, s, body_n)
      if rg_send_ws(ca, WS_OP_BINARY, frm, n, 0x5a5a0000 + s) != 0 { sys_exit(2) }
      s = s + 1
  }
  sys_exit(0)
  return 0
262}
263
264func rg_kill(pid: i64) -> i64 {
  __syscall(129, pid, 9, 0, 0, 0, 0)     // rv64 kill=129. The old comment called raw x86 62 a 'table-gap escape' -- it is not a gap, 62 IS an RV64 KEY and the backend translated it to lseek(8), so the kill never happened (debt idx 2277)
  let st: *i64 = sys_mmap(16) as *i64
  sys_wait4(pid, st, 0)
  return 0
269}
270
271func main() -> i64 {
  g_puts("nx_room_perf gate (MEASURED live room performance)\n" as *u8)
  var pass: i64 = 0
  var total: i64 = 0
275
  let cpath: *u8 = "/tmp/mozilla_certdata.txt" as *u8
  let lr: i64 = nx_trust_store_load_from_certdata(cpath, 300, 4194304)
  if lr <= 0 { g_puts("FAIL trust store load\n" as *u8); return 2 }
  let store: *TrustStore = lr as *TrustStore
  let host: *u8 = "nishifamily.com" as *u8
  let hn: i64 = 15
  let dr: *DnsResolveResult = nx_dns_resolve_default(host, hn, sys_now_realtime_sec())
  if dr.verdict != NX_DNS_R_OK { g_puts("FAIL dns\n" as *u8); return 2 }
  let path: *u8 = "/signal/perfroomgate" as *u8
  let pn: i64 = g_slen(path)
286
  let ca: *i64 = sys_mmap(RG_SLOTS * 8) as *i64
  let cb: *i64 = sys_mmap(RG_SLOTS * 8) as *i64
  let cc: *i64 = sys_mmap(RG_SLOTS * 8) as *i64
  let ja: i64 = rg_join(ca, dr.ipv4_packed, host, hn, path, pn, store, 11)
  let jb: i64 = rg_join(cb, dr.ipv4_packed, host, hn, path, pn, store, 12)
  let jc: i64 = rg_join(cc, dr.ipv4_packed, host, hn, path, pn, store, 13)
  pass = pass + g_check("3 users joined (TLS1.3 real-CA + wss 101, public)" as *u8,
      (ja == 0) + (jb == 0) + (jc == 0) == 3); total = total + 1
  if ja != 0 { return 1 }
  if jb != 0 { return 1 }
  if jc != 0 { return 1 }
298
  let fraw: *u8 = sys_mmap(128)
  let f: *WsFrame = fraw as *WsFrame
  let lats: *i64 = sys_mmap(2048) as *i64
302
  // ---- P1: unloaded latency, 20 paced 64B frames A -> B (C reads too) ----
  let frm: *u8 = sys_mmap(512)
  var got1: i64 = 0
  var s: i64 = 0
  while s < 20 {
      let n: i64 = rg_media_ts(frm, 0x56, "useraaaa" as *u8, s, 64)
      rg_send_ws(ca, WS_OP_BINARY, frm, n, 0x11110000 + s)
      let lb: i64 = rg_recv_lat(cb, f)
      if lb >= 0 { lats[got1] = lb; got1 = got1 + 1 }
      rg_next_frame(cc, f)
      sys_sleep_ms(30)
      s = s + 1
  }
  rg_sort(lats, got1)
  g_puts("PERF P1 unloaded: delivered=" as *u8); g_pn(got1)
  g_puts("/20 p50_us=" as *u8); g_pn(lats[got1 / 2])
  g_puts(" max_us=" as *u8); g_pn(lats[got1 - 1]); g_puts("\n" as *u8)
  pass = pass + g_check("P1: 20/20 paced frames delivered to B" as *u8, got1 == 20); total = total + 1
321
  // ---- P2: loaded throughput, forked sender streams 120 x 8KB ----
  rg_reset_acc(cb); rg_reset_acc(cc)
  let t0: i64 = sys_now_us()
  let pid2: i64 = rg_fork_sender(ca, "useraaaa" as *u8, 120, 8000)
  var gb: i64 = 0
  var gc: i64 = 0
  var lat_n: i64 = 0
  var k: i64 = 0
  var alive: i64 = 1
  while k < 120 {
      if alive == 1 {
          let lb: i64 = rg_recv_lat(cb, f)
          if lb >= 0 { gb = gb + 1; lats[lat_n] = lb; lat_n = lat_n + 1 } else { alive = 0 }
          if rg_next_frame(cc, f) == 0 { gc = gc + 1 } else { alive = 0 }
          if (k & 15) == 15 { rg_reset_acc(cb); rg_reset_acc(cc) }
      }
      k = k + 1
  }
  let t1: i64 = sys_now_us()
  rg_kill(pid2)
  rg_sort(lats, lat_n)
  let wall_ms: i64 = (t1 - t0) / 1000
  var kbps: i64 = 0
  if wall_ms > 0 { kbps = (gb * 8021) / wall_ms }
  g_puts("PERF P2 loaded: B=" as *u8); g_pn(gb)
  g_puts("/120 C=" as *u8); g_pn(gc)
  g_puts("/120 wall_ms=" as *u8); g_pn(wall_ms)
  g_puts(" delivered_KBps=" as *u8); g_pn(kbps)
  g_puts(" loaded_p50_us=" as *u8); g_pn(lats[lat_n / 2])
  g_puts(" loaded_max_us=" as *u8); g_pn(lats[lat_n - 1]); g_puts("\n" as *u8)
  pass = pass + g_check("P2: all 120 x 8KB delivered to BOTH B and C under load" as *u8,
      (gb == 120) + (gc == 120) == 2); total = total + 1
354
  // ---- P4: slow reader -- C stops reading; B must still get all 96 ----
  // A's parent-side TLS seq is STALE after P2's forked sender advanced it
  // in the child, so A is RETIRED (closed) and a fresh client D joins as
  // the firehose: room = B (reading), C (asleep), D (forked sender).
  sys_close(ca[RG_FD])
  let cd: *i64 = sys_mmap(RG_SLOTS * 8) as *i64
  let jd: i64 = rg_join(cd, dr.ipv4_packed, host, hn, path, pn, store, 14)
  pass = pass + g_check("P4 setup: fresh user D joined as firehose" as *u8, jd == 0); total = total + 1
  rg_reset_acc(cb); rg_reset_acc(cc)
  sys_set_socket_timeout(cb[RG_FD], 8)
  let pid4: i64 = rg_fork_sender(cd, "userdddd" as *u8, 400, 8000)
  var gb4: i64 = 0
  var k4: i64 = 0
  var alive4: i64 = 1
  while k4 < 400 {
      if alive4 == 1 {
          if rg_recv_lat(cb, f) >= 0 { gb4 = gb4 + 1 } else { alive4 = 0 }
          if (k4 & 15) == 15 { rg_reset_acc(cb) }
      }
      k4 = k4 + 1
  }
  rg_kill(pid4)
  g_puts("PERF P4 slow-reader: B_delivered=" as *u8); g_pn(gb4)
  g_puts("/400 = 3.2MB past a dead peer (v2 relay wedges when C's chain buffers fill)\n" as *u8)
  pass = pass + g_check("P4: slow peer C does NOT stall delivery to B (400/400)" as *u8, gb4 == 400); total = total + 1
380
  sys_close(cb[RG_FD]); sys_close(cc[RG_FD]); sys_close(cd[RG_FD])
  g_puts("---- room_perf gate: passed " as *u8); g_pn(pass); g_puts(" / " as *u8); g_pn(total); g_puts(" ----\n" as *u8)
  if pass == total { return 0 }
  return 1
385}
nx_room_perf_gate.nx source

`nx_room_perf_gate.nx` source
