nx_par_pull.nx source
↩ module page · 120 lines · 7190 B
1// nx_par_pull.nx -- PARALLEL concurrent download (the max-speed S-class gap): fork one worker per file, each
2// running the resumable streaming pull over its OWN TLS-1.3 connection, all downloading AT ONCE. Staged to fast
3// local disk (/tmp = ext4, not the 9p NAS mount that bottlenecks + crashes WSL) -> both faster AND stable; a caller
4// then does one big sequential copy to the NAS. Reuses nx_https_get_stream + nx_https_fetch_follow's plumbing (same
5// as the sequential nx_apertus_shards_pull; the pull loop is duplicated here as the parallel driver's worker body).
6// Demonstrated live pulling several real Apertus files concurrently. license_tier: ORIGINAL expect_exit: 0
7import "nx_syscalls.nx"
8import "nx_trust_store_load_from_certdata.nx"
9import "nx_https_fetch_follow.nx"
10import "nx_https_get_stream.nx"
11const K_MAGIC_4096: i64 = 4096
12const K_MAGIC_2048: i64 = 2048
13const K_MAGIC_4194304: i64 = 4194304
14const K_MAGIC_17078479: i64 = 17078479
15const K_MAGIC_4192195: i64 = 4192195
16const K_MAGIC_7291: i64 = 7291
17const K_MAGIC_1024: i64 = 1024
18
19func w(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} sys_write(1,s,n); return 0 }
20func wn(v: i64) -> i64 { var m: i64=v; if m<0{w("-" as *u8);m=0-m} let t:*u8=sys_mmap(24); 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; let o:*u8=sys_mmap(24); while i<k{o[i]=t[k-1-i];i=i+1} sys_write(1,o,k); return 0 }
21func pp_cat(dst: *u8, off: i64, s: *u8) -> i64 { var o: i64=off; var i: i64=0; while s[i]!=(0 as u8){dst[o]=s[i];o=o+1;i=i+1} return o }
22func file_size(path: *u8) -> i64 { let fd: i64=sys_openat_rd(path); if fd<0 { return 0 } let sz: i64=sys_lseek(fd,0,2); sys_close(fd); if sz<0 { return 0 } return sz }
23
24// one attempt (connect+handshake+get_stream(range=have) -> dest). rc: >0 body, -9 range-ignored, <0 err.
25func pull_attempt(url0: *u8, store: *TrustStore, dest_path: *u8, have: i64) -> i64 {
26 let urlbuf: *u8 = sys_mmap(K_MAGIC_4096)
27 var ui: i64=0; while url0[ui]!=0 as u8 { urlbuf[ui]=url0[ui]; ui=ui+1 } urlbuf[ui]=0 as u8
28 var dest_fd: i64 = 0-1
29 if have>0 { dest_fd=sys_openat_append(dest_path,0x1a4) } else { dest_fd=sys_openat_wr(dest_path,0x1a4) }
30 if dest_fd<0 { return 0-100 }
31 let cr: *u8=sys_mmap(32); let priv: *u8=sys_mmap(32); let statusp: *i64=sys_mmap(16) as *i64; let loc: *u8=sys_mmap(K_MAGIC_4096)
32 var hop: i64=0
33 while hop<=8 {
34 let target_raw: *u8=sys_mmap(64); let target: *NxHttpsTarget=target_raw as *NxHttpsTarget
35 target.url=nx_url_new(); target.port=0
36 if nx_https_url_for_fetch(urlbuf,target)!=NX_HTTPS_URL_OK { sys_close(dest_fd); return 0-101 }
37 let fd_p: *i64=(sys_mmap(16)) as *i64
38 if nx_https_url_connect(target,urlbuf,sys_now_realtime_sec(),fd_p)!=NX_HTTPS_CONNECT_OK { sys_close(dest_fd); return 0-102 }
39 let fd: i64=fd_p[0]
40 var i: i64=0; while i<32 { cr[i]=(0xC0+i) as u8; priv[i]=(0xA0+i) as u8; i=i+1 }
41 let vc_raw: *u8=sys_mmap(64); let vc: *TlsValidationContext=vc_raw as *TlsValidationContext
42 vc.store=store; vc.sni_host=urlbuf+target.url.host_off; vc.sni_host_len=target.url.host_len; vc.now_epoch=sys_now_realtime_sec()
43 let sr: i64=nx_tls13_client_session_run(fd,urlbuf+target.url.host_off,target.url.host_len,cr,priv,vc)
44 if sr<=0 { sys_close(fd); sys_close(dest_fd); return 0-103 }
45 let session: *Tls13ClientSession=sr as *Tls13ClientSession
46 let path: *u8=sys_mmap(K_MAGIC_2048); let plen: i64=ff_path(urlbuf,target,path)
47 let rc: i64=nx_https_get_stream(session,fd,path,plen,urlbuf+target.url.host_off,target.url.host_len,have,dest_fd,statusp,loc,K_MAGIC_4096)
48 sys_close(fd)
49 if rc==0 { let resolved: *u8=sys_mmap(K_MAGIC_4096); ff_resolve_location(loc,urlbuf,target,resolved); var k: i64=0; while resolved[k]!=0 as u8 { urlbuf[k]=resolved[k]; k=k+1 } urlbuf[k]=0 as u8; hop=hop+1 }
50 else { sys_close(dest_fd); return rc }
51 }
52 sys_close(dest_fd); return 0-200
53}
54// resumable worker: retry until dest reaches target.
55func pull_one_file(url0: *u8, store: *TrustStore, dest_path: *u8, target: i64) -> i64 {
56 var tries: i64=0
57 while tries<20 {
58 let have: i64=file_size(dest_path)
59 if have>=target { return have }
60 let rc: i64=pull_attempt(url0,store,dest_path,have)
61 if rc==(0-9) { let tf: i64=sys_openat_wr(dest_path,0x1a4); if tf>=0 { sys_close(tf) } tries=tries+1 }
62 else { let h2: i64=file_size(dest_path); if h2>=target { return h2 } if h2>have { tries=0 } else { tries=tries+1 } }
63 }
64 return 0-1
65}
66
67func main() -> i64 {
68 w("=== nx_par_pull: PARALLEL concurrent download (fork per file, own TLS each) -> ext4 stage ===\n" as *u8)
69 let r: i64 = nx_trust_store_load_from_certdata("data/mozilla_certdata.txt" as *u8, 512, K_MAGIC_4194304)
70 if r<=0 { w("certdata load FAILED\n" as *u8); sys_exit(1); return 1 }
71 let store: *TrustStore = r as *TrustStore
72 let base: *u8 = "https://huggingface.co/adamo1139/Apertus-8B-Instruct-2509-ungated/resolve/main/" as *u8
73
74 let names: *i64=sys_mmap(8*8) as *i64; let sizes: *i64=sys_mmap(8*8) as *i64
75 names[0]="tokenizer.json" as *u8 as i64; sizes[0]=K_MAGIC_17078479
76 names[1]="Apertus_Tech_Report.pdf" as *u8 as i64; sizes[1]=K_MAGIC_4192195
77 names[2]="README.md" as *u8 as i64; sizes[2]=K_MAGIC_7291
78 names[3]="config.json" as *u8 as i64; sizes[3]=901
79 let nn: i64 = 4
80
81 // build url/dest arrays
82 let urls: *i64=sys_mmap(8*8) as *i64; let dests: *i64=sys_mmap(8*8) as *i64
83 var i: i64=0
84 while i<nn {
85 let nm: *u8=names[i] as *u8
86 let u: *u8=sys_mmap(K_MAGIC_1024); var uo: i64=pp_cat(u,0,base); uo=pp_cat(u,uo,nm); u[uo]=0 as u8; urls[i]=u as i64
87 let d: *u8=sys_mmap(K_MAGIC_1024); var doo: i64=pp_cat(d,0,"/tmp/par_" as *u8); doo=pp_cat(d,doo,nm); d[doo]=0 as u8; dests[i]=d as i64
88 i=i+1
89 }
90
91 // FORK one worker per file -> all download AT ONCE
92 let pids: *i64=sys_mmap(8*8) as *i64
93 i=0
94 while i<nn {
95 let pid: i64=sys_fork()
96 if pid==0 {
97 let got: i64=pull_one_file(urls[i] as *u8, store, dests[i] as *u8, sizes[i])
98 if got>=sizes[i] { sys_exit(0) }
99 sys_exit(1)
100 }
101 pids[i]=pid
102 w(" forked worker for " as *u8); w(names[i] as *u8); w(" (pid " as *u8); wn(pid); w(")\n" as *u8)
103 i=i+1
104 }
105 // parent waits ALL
106 var ok: i64=0
107 i=0
108 while i<nn {
109 let st: *i64=sys_mmap(16) as *i64
110 sys_wait4(pids[i], st, 0)
111 let ec: i64=(st[0]>>8)&0xff
112 let sz: i64=file_size(dests[i] as *u8)
113 w(" " as *u8); w(names[i] as *u8); w(" exit=" as *u8); wn(ec); w(" bytes=" as *u8); wn(sz); w("/" as *u8); wn(sizes[i])
114 if sz>=sizes[i] { w(" [OK]\n" as *u8); ok=ok+1 } else { w(" [INCOMPLETE]\n" as *u8) }
115 i=i+1
116 }
117 w("\n PARALLEL WORKERS COMPLETE: " as *u8); wn(ok); w("/" as *u8); wn(nn); w("\n" as *u8)
118 if ok==nn { w("=== GREEN (concurrent multi-file download works; ext4-staged; ready to copy to NAS) ===\n" as *u8); sys_exit(0); return 0 }
119 w("=== PARTIAL ===\n" as *u8); sys_exit(1); return 1
120}