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

↩ module page · 213 lines · 11307 B
1// nx_pagerank_build.nx -- P1 STEP 2: build the web LINK GRAPH from out:<cid> edge rows (captured by
2// nx_web_crawl_step's outlink capture) and compute PageRank authority, storing pr:<cid> priors back into the
3// web shard. Node identity = cid(url) (the frontier/outlink hash) -> an out: key's cid is the source node,
4// its packed-i64 value are the target nodes. Interns cids to dense indices, runs pr_compute (nx_pagerank),
5// stores pr:<cid> = rank. Serve-time authority (step 3) = content_cid -> url:<cid> -> ci_hash(url) -> pr:<cid>.
6//   usage: nx_pagerank_build [domain=web] [iters=100]
7// license_tier: ORIGINAL
8import "nx_corpus_ingest.nx"   // dss_prefix / seg_store (ss_open/ss_hget/ss_begin/ss_add/ss_commit/ss_r32) / ci_hash
9import "nx_itoa_lib.nx"   // shared MSB-first emitter (zero-alloc)
10import "nx_pagerank.nx"        // pr_compute / pr_argmax / pr_sum / PR_SCALE
11const PB_MAGIC_16777216: i64 = 16777216
12const PB_MAGIC_1000000: i64 = 1000000
13
14// SCALE RUNG (2026-07-23): linear pb_intern + the 64B-string key dedup were O(N^2)/O(K^2) -- fine at the
15// 1925-node crawl graph, fatal at bulk-WARC scale (one CC file ~= millions of target nodes). Interning is
16// now an open-addressing hash table; source-key dedup is a hash SET on the source cid (same identity).
17const PB_MAXNODE: i64 = 16777216   // distinct nodes per run (16M; one ~40k-page WARC links ~1-4M targets)
18const PB_MAXEDGE: i64 = 33554432   // directed edges per run (32M)
19const PB_HSIZE:   i64 = 33554432   // hash slots = 2^25 (2x MAXNODE load-factor headroom)
20const PB_HSHIFT:  i64 = 38         // take the TOP 25 bits of the 63-bit mixed hash (63 - 25)
21const PB_HMUL:    i64 = 7046029254386353131  // odd 63-bit multiplicative-mix constant (Fibonacci-hash family)
22
23func pb_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
24// MIGRATED to the shared emitter (debt 1785563586). The old body mmapped a scratch buffer
25// per call and never freed it. At PAGE granularity that is 4096B leaked PER CALL -- the
26// defect that took 28.5GB of a 36GB host in nx_ts_lumadiff (2MB input, ~3.66M calls).
27// nxi_* is MSB-first, allocates NOTHING, and emits identical bytes including the sign.
28func pb_num(v: i64) -> i64 { nxi_out(v); return 0 }
29// cid out of an "out:<cid>" key (digits after "out:")
30func pb_key_cid(k: *u8, kl: i64) -> i64 {
  var v: i64 = 0; var i: i64 = 4
  while i < kl { let c: i64 = k[i] as i64; if c >= 48 { if c <= 57 { v = v * 10 + (c - 48) } } i = i + 1 }
  return v
34}
35// build "pr:<cid>" key (the authority prior row). Returns length.
36func pb_prkey(cid: i64, out: *u8) -> i64 {
  out[0] = 112 as u8   // p
  out[1] = 114 as u8   // r
  out[2] = 58 as u8    // :
  var m: i64 = cid
  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 { out[3 + i] = t[k - 1 - i]; i = i + 1 }
  out[3 + k] = 0 as u8
  return 3 + k
49}
50// slot for a cid in a PB_HSIZE open-addressing table (cid is always >0 per ci_hash; 0 = empty slot)
51func pb_hslot(cid: i64) -> i64 {
  var s: i64 = cid * PB_HMUL
  s = s & 0x7fffffffffffffff
  return (s >> PB_HSHIFT) & (PB_HSIZE - 1)
55}
56// intern cid -> dense node index via the hash table (O(1) expected). cidtab[slot]=cid, idxtab[slot]=index.
57// Returns the index, or -1 when the node cap is hit (caller counts + skips -- NEVER silent).
58func pb_intern(cid: i64, cidtab: *i64, idxtab: *i64, node: *i64, nn: *i64) -> i64 {
  var slot: i64 = pb_hslot(cid)
  var probes: i64 = 0
  while probes < PB_HSIZE {
      if cidtab[slot] == cid { return idxtab[slot] }
      if cidtab[slot] == 0 {
          let idx: i64 = nn[0]
          if idx >= PB_MAXNODE { return 0 - 1 }
          cidtab[slot] = cid
          idxtab[slot] = idx
          node[idx] = cid
          nn[0] = idx + 1
          return idx
      }
      slot = (slot + 1) & (PB_HSIZE - 1)
      probes = probes + 1
  }
  return 0 - 1
76}
77// hash-SET membership+insert for processed SOURCE cids (replaces the O(K^2) 64B-string dedup).
78// Returns 1 = was already present, 0 = inserted now (or table full -> treat as present to stay safe).
79func pb_seen(cid: i64, seentab: *i64) -> i64 {
  var slot: i64 = pb_hslot(cid)
  var probes: i64 = 0
  while probes < PB_HSIZE {
      if seentab[slot] == cid { return 1 }
      if seentab[slot] == 0 { seentab[slot] = cid; return 0 }
      slot = (slot + 1) & (PB_HSIZE - 1)
      probes = probes + 1
  }
  return 1
89}
90// walk every segment's key index for "out:"-prefixed keys (cross-segment dedup = pb_seen hash set on the
91// SOURCE cid -- same identity, O(1) vs the old O(K^2) string table), ss_hget latest value = packed i64
92// target cids, intern src+target cids to dense nodes, emit directed edges src->target. Returns M.
93func pb_collect(h: *i64, cidtab: *i64, idxtab: *i64, seentab: *i64, node: *i64, nn: *i64, efrom: *i64, eto: *i64, srcflag: *u8) -> i64 {
  if (h as i64) == 0 { return 0 }
  let ns: i64 = h[0]
  let pp: *i64 = sys_mmap(16) as *i64
  let pl: *i64 = sys_mmap(16) as *i64
  let kbuf: *u8 = sys_mmap(64)         // ONE reused key buffer (the old per-key mmap leaked a page per key)
  var M: i64 = 0
  var capped: i64 = 0
  var s: i64 = 0
  while s < ns {
      let kb: *u8 = h[1 + 8 * s] as *u8
      if h[2 + 8 * s] >= 8 {
          let m9: i64 = ss_r32(kb, 4)
          var e9: i64 = 0
          while e9 < m9 {
              let eo: i64 = 8 + 4 * m9 + ss_r32(kb, 8 + 4 * e9)
              if (kb[eo] as i64) == 1 {
                  let kl9: i64 = ss_r32(kb, eo + 1)
                  if kl9 >= 5 { if kl9 < 60 {
                      if kb[eo + 5] == (111 as u8) { if kb[eo + 6] == (117 as u8) { if kb[eo + 7] == (116 as u8) { if kb[eo + 8] == (58 as u8) {
                          var c: i64 = 0
                          while c < kl9 { kbuf[c] = kb[eo + 5 + c]; c = c + 1 }
                          kbuf[kl9] = 0 as u8
                          let scid: i64 = pb_key_cid(kbuf, kl9)
                          if pb_seen(scid, seentab) == 0 {
                              if ss_hget(h, kbuf, pp, pl) == 1 { if pl[0] >= 8 {
                                  let si: i64 = pb_intern(scid, cidtab, idxtab, node, nn)
                                  if si < 0 { capped = 1 } else {
                                      srcflag[si] = 1 as u8   // this node IS one of OUR documents (an out: source)
                                      let tv: *i64 = pp[0] as *i64
                                      let nt: i64 = pl[0] / 8
                                      var t: i64 = 0
                                      while t < nt {
                                          if M < PB_MAXEDGE {
                                              let ti: i64 = pb_intern(tv[t], cidtab, idxtab, node, nn)
                                              if ti < 0 { capped = 1 } else { efrom[M] = si; eto[M] = ti; M = M + 1 }
                                          } else { capped = 1 }
                                          t = t + 1
                                      }
                                  }
                              } }
                          }
                      } } } }
                  } }
              }
              e9 = e9 + 1
          }
      }
      s = s + 1
  }
  if capped == 1 { pb_puts("PB CAPPED: node/edge cap hit -- graph TRUNCATED (raise PB_MAXNODE/PB_MAXEDGE)\n" as *u8) }
  return M
145}
146// full build: open shard -> collect edges -> pr_compute -> store pr:<cid>. segid0 = first segid for the pr:
147// segment(s) (caller supplies a fresh one). Returns node count N (>=0), or -1 on empty/no shard.
148func pb_run(prefix: *u8, iters: i64, segid0: i64) -> i64 {
  let h: *i64 = ss_open2(prefix, 1)   // mmap-open: shard pages in on demand (low RSS) -> fits on the NAS, no OOM
  if (h as i64) == 0 { return 0 - 1 }
  pb_puts("  PR ss_open(mmap) OK segments=" as *u8); pb_num(h[0]); pb_puts("\n" as *u8)
  let node: *i64 = sys_mmap(8 * PB_MAXNODE)
  let cidtab: *i64 = sys_mmap(8 * PB_HSIZE)     // anon mmap = zero pages -> empty slots are cid 0
  let idxtab: *i64 = sys_mmap(8 * PB_HSIZE)
  let seentab: *i64 = sys_mmap(8 * PB_HSIZE)
  let efrom: *i64 = sys_mmap(8 * PB_MAXEDGE)
  let eto: *i64 = sys_mmap(8 * PB_MAXEDGE)
  let srcflag: *u8 = sys_mmap(PB_MAXNODE)       // 1 = node is one of OUR docs (out: source) -> gets a pr: row
  let nn: *i64 = sys_mmap(16) as *i64
  nn[0] = 0
  pb_puts("  PR arrays allocated, collecting edges...\n" as *u8)
  let M: i64 = pb_collect(h, cidtab, idxtab, seentab, node, nn, efrom, eto, srcflag)
  let N: i64 = nn[0]
  pb_puts("  PR collect done N=" as *u8); pb_num(N); pb_puts(" M=" as *u8); pb_num(M); pb_puts("\n" as *u8)
  if N <= 0 { return 0 }
  let rank: *i64 = sys_mmap(8 * N)
  pr_compute(N, M, efrom, eto, rank, iters)
  pb_puts("  PR compute done, persisting document priors...\n" as *u8)
  // persist pr:<node[i]> = rank[i] ONLY for our documents (srcflag=1). External outlink-target leaf nodes
  // carry authority FLOW during pr_compute but are never served, so storing their priors is pure waste --
  // at bulk scale that was ~10M phantom rows (the persist phase that crashed the VM, seq631). Filtering to
  // source docs cuts it ~100x (~150k rows) AND keeps the served shard small. Buffers hoisted (ss_add2
  // copies immediately; per-row mmap would breach the kernel map cap at scale).
  var segid: i64 = segid0
  let w: *i64 = ss_begin_cap(PB_MAGIC_16777216)   // 16MiB pr segments: ~700k pr rows/segment, manifest stays short
  let pkey: *u8 = sys_mmap(32)
  let rv: *i64 = sys_mmap(16) as *i64
  var stored: i64 = 0
  var i: i64 = 0
  while i < N {
      if srcflag[i] == (1 as u8) {
          pb_prkey(node[i], pkey)
          rv[0] = rank[i]
          if ss_add(w, 1, pkey, rv as *u8, 8) < 0 {
              ss_commit(prefix, w, segid)
              segid = segid + 1
              w[1] = 0
              ss_add(w, 1, pkey, rv as *u8, 8)
          }
          stored = stored + 1
      }
      i = i + 1
  }
  ss_commit(prefix, w, segid)
  pb_puts("  PR persisted document priors=" as *u8); pb_num(stored); pb_puts(" (of " as *u8); pb_num(N); pb_puts(" graph nodes)\n" as *u8)
  return N
197}
198func main(argc: i64, argv: *i64) -> i64 {
  var domain: *u8 = "web" as *u8
  var iters: i64 = 100
  if argc >= 2 { domain = argv[1] as *u8 }
  // fresh segid base from the monotonic clock (us) -- unique per run, never collides a live segment
  let ts: *i64 = sys_mmap(16) as *i64
  __syscall(228, 1, ts as i64, 0, 0, 0, 0)   // clock_gettime(CLOCK_MONOTONIC, ts)
  let segid0: i64 = ts[0] * PB_MAGIC_1000000 + ts[1] / 1000
  let prefix: *u8 = sys_mmap(512)
  dss_prefix(domain, prefix)
  pb_puts("nx_pagerank_build domain=" as *u8); pb_puts(domain); pb_puts("\n" as *u8)
  let N: i64 = pb_run(prefix, iters, segid0)
  if N < 0 { pb_puts("no shard\n" as *u8); return 1 }
  pb_puts("PageRank: nodes="); pb_num(N); pb_puts(" pr: rows stored (authority priors)\n" as *u8)
  return 0
213}
nx_pagerank_build.nx source

`nx_pagerank_build.nx` source
