_galx_bulk_ingest.nx source

1// _galx_bulk_ingest.nx -- HIGH-THROUGHPUT bulk corpus ingest for the FULL ~230k NAS image library. 2// The per-image path (_galx_prod_ingest -> nx_store_ingest_ingest_cid) commits ONE segment per image 3// AND does an O(store) cap-aware dedup scan per image -- fine for thousands, fatal for 230k (the scan 4// re-reads every seg file each time over slow drvfs). This driver instead: 5// (1) COMPUTES the CID via the SHARED nx_store_ingest_compute_cid (byte-identical to the per-image 6// path -- same GENREC -> same CID), so the bulk store is consistent with the existing one; 7// (2) BATCHES commits -- BATCH images accumulate in ONE ss_begin writer, committed as ONE segment 8// (230k images -> ~460 segments, not 230k), so browse + future dedup stay bounded; 9// (3) SKIPS the per-image dedup -- this is a fresh distinct set; any accidental duplicate (e.g. the 10// 4436 already ingested) is COLLAPSED losslessly by the final ss_compact_cap (latest-per-key). 11// segid = SEGBASE + start/BATCH + (batch# in this slice) -- globally unique iff start & count are 12// multiples of BATCH (the runner nx_galx_fill enforces this). One process per slice (the runner forks 13// it) bounds RSS = slice*~filesize, reclaimed on exit (no munmap). Appends <CID>\t<path> to the sidecar. 14// Usage: _galx_bulk_ingest <start> <count> [corpus_path]. license_tier: ORIGINAL 15import "nx_store_ingest.nx" 16import "nx_seg_store.nx" 17import "nx_canon_cid.nx" 18import "nx_syscalls.nx" 19 20const PREFIX: *u8 = "knowledge/store/galx-prod-" 21const SEGBASE: i64 = 5000000 22const BATCH: i64 = 500 23 24func bp(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 } 25func bn(v: i64) -> i64 { let b: *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;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{b[i]=t[k-1-i];i=i+1}; sys_write(1,b,k); return 0 } 26func b_atoi(s: *u8) -> i64 { var v: i64=0; var i: i64=0; while s[i]!=(0 as u8){ if s[i]>=(48 as u8){ if s[i]<=(57 as u8){ v=v*10+((s[i]-(48 as u8)) as i64) } } i=i+1 } return v } 27 28func main(argc: i64, argv: *i64) -> i64 { 29 if argc < 3 { bp("usage: _galx_bulk_ingest <start> <count> [corpus]\n" as *u8); sys_exit(2); return 2 } 30 let start: i64 = b_atoi(argv[1] as *u8) 31 let count: i64 = b_atoi(argv[2] as *u8) 32 let endl: i64 = start + count 33 var corpus: *u8 = "knowledge/status/galx_corpus_full.txt" as *u8 34 if argc >= 4 { corpus = argv[3] as *u8 } 35 var prefix: *u8 = PREFIX 36 if argc >= 5 { prefix = argv[4] as *u8 } // scratch namespace for gate tests (avoids racing the live store) 37 var segbase: i64 = SEGBASE 38 if argc >= 6 { segbase = b_atoi(argv[5] as *u8) } // override seg-id base to avoid colliding with prior runs' seg ids 39 40 let lszp: *i64 = sys_mmap(16) as *i64 41 let lst: *u8 = sys_read_file(corpus, lszp) 42 let lsz: i64 = lszp[0] 43 if (lst as i64) == 0 { bp("BULK FAIL: cannot read corpus\n" as *u8); sys_exit(1); return 1 } 44 45 let sfd: i64 = sys_openat_append("knowledge/status/galx_cid_paths.tsv" as *u8, 0x1a4) 46 if sfd < 0 { bp("BULK FAIL: sidecar unwritable\n" as *u8); sys_exit(1); return 1 } 47 48 let pathbuf: *u8 = sys_mmap(1024) 49 let cid: *u8 = sys_mmap(96) 50 let canon: *u8 = sys_mmap(8192) 51 let rk: *u8 = sys_mmap(96) 52 var wr: *i64 = ss_begin() 53 var inbatch: i64 = 0 54 var committed: i64 = 0 55 var newc: i64 = 0 56 var skip: i64 = 0 57 var cfail: i64 = 0 58 let segbatch0: i64 = start / BATCH 59 var ls: i64 = 0 60 var i: i64 = 0 61 var lineno: i64 = 0 62 while i <= lsz { 63 var atend: i64 = 0 64 if i == lsz { atend = 1 } 65 var isnl: i64 = 0 66 if atend == 0 { if lst[i] == (10 as u8) { isnl = 1 } } 67 if atend == 1 { isnl = 1 } 68 if isnl == 1 { 69 let llen: i64 = i - ls 70 if llen > 0 { 71 if lineno >= start { if lineno < endl { 72 var po: i64 = 0 73 while po < llen { pathbuf[po] = lst[ls + po]; po = po + 1 } 74 pathbuf[llen] = 0 as u8 75 let fszp: *i64 = sys_mmap(16) as *i64 76 let fbuf: *u8 = sys_read_file(pathbuf, fszp) 77 let flen: i64 = fszp[0] 78 if (fbuf as i64) == 0 { skip = skip + 1 } else { 79 var clen2: i64 = nx_store_ingest_compute_cid(fbuf, flen, cid, canon, rk) 80 // FALLBACK (rule 25 -- never strip): an image with NO embedded GENREC tEXt 81 // (~10% of the corpus: pipelines that didn't write generation params) still 82 // belongs in the gallery. Give it a CONTENT-addressed CID = cid_of(file bytes) 83 // and a minimal "noattr" record. It browses + serves like any image; its factor 84 // panel is honestly empty (no data to attribute), NOT hidden. 85 if clen2 < 0 { 86 cid_of(fbuf, flen, cid) 87 cid[69] = 0 as u8 88 rk[0] = 105 as u8; rk[1] = 109 as u8; rk[2] = 103 as u8; rk[3] = 58 as u8 89 var rj: i64 = 0 90 while rj < 69 { rk[4 + rj] = cid[rj]; rj = rj + 1 } 91 rk[73] = 0 as u8 92 canon[0] = 110 as u8; canon[1] = 111 as u8; canon[2] = 97 as u8 93 canon[3] = 116 as u8; canon[4] = 116 as u8; canon[5] = 114 as u8 94 clen2 = 6 95 } 96 if clen2 < 0 { skip = skip + 1 } else { 97 if ss_add2(wr, 1, rk, 73, canon, clen2) != 0 { 98 if ss_commit(prefix, wr, segbase + start + committed) != 0 { cfail = cfail + 1 } 99 committed = committed + 1 100 wr = ss_begin() 101 inbatch = 0 102 ss_add2(wr, 1, rk, 73, canon, clen2) 103 } 104 inbatch = inbatch + 1 105 newc = newc + 1 106 sys_write(sfd, cid, 69) 107 sys_write(sfd, "\t" as *u8, 1) 108 sys_write(sfd, pathbuf, llen) 109 sys_write(sfd, "\n" as *u8, 1) 110 if inbatch >= BATCH { 111 if ss_commit(prefix, wr, segbase + start + committed) != 0 { cfail = cfail + 1 } 112 committed = committed + 1 113 wr = ss_begin() 114 inbatch = 0 115 } 116 } 117 } 118 }} 119 lineno = lineno + 1 120 } 121 ls = i + 1 122 } 123 i = i + 1 124 } 125 if inbatch > 0 { 126 if ss_commit(prefix, wr, segbase + start + committed) != 0 { cfail = cfail + 1 } 127 committed = committed + 1 128 } 129 sys_close(sfd) 130 bp("BULK range [" as *u8); bn(start); bp("," as *u8); bn(endl); bp(") new=" as *u8); bn(newc); bp(" skip=" as *u8); bn(skip); bp(" segs=" as *u8); bn(committed); bp(" commit_fail=" as *u8); bn(cfail); bp("\n" as *u8) 131 sys_exit(0) 132 return 0 133}