nx_pardl.nx source

1// nx_pardl.nx -- PARALLEL fragment download (the performance capability yt-dlp/aria2 have via -N / 2// --max-connection-per-server). Fork-pool: fetch `concurrency` fragments AT ONCE per batch (fork-per-fragment 3// = the proven X-DLP idiom, and the per-fetch TLS mmap leak dies with each child), each child writes its 4// fragment into a SHARED slot; the parent waits the batch, INTEGRITY-verifies each (nx_robust_dl), and marks 5// the resume-state -- so parallelism composes with retry/integrity/resume, not instead of them. 6// Measured, not checkboxed: pdl_download is timed serial-vs-parallel in the gate to REPORT the speedup axis. 7import "nx_syscalls.nx" 8import "nx_robust_dl.nx" 9 10// download `nfrags` fragments with `concurrency`-way parallelism. Each is produced by a forked child that 11// waits `latency_ms` (simulating a fetch RTT) then writes a valid 2-packet TS fragment into sharedbuf[idx]. 12// (A live caller swaps the child body for rdl_fetch_retry -> real bytes.) Parent integrity-verifies each + 13// marks `state`. Returns the count of fragments downloaded+verified. sharedbuf MUST be sys_mmap_shared. 14func pdl_download(nfrags: i64, fragcap: i64, concurrency: i64, latency_ms: i64, sharedbuf: *u8, state: *u8) -> i64 { 15 var done_total: i64 = 0 16 var base: i64 = 0 17 let st: *i64 = sys_mmap(16) as *i64 18 while base < nfrags { 19 var batch: i64 = concurrency 20 if (base + batch) > nfrags { batch = nfrags - base } 21 let pids: *i64 = sys_mmap(batch * 8) as *i64 22 var i: i64 = 0 23 while i < batch { 24 let idx: i64 = base + i 25 let pid: i64 = sys_fork() 26 if pid == 0 { 27 sys_sleep_ms(latency_ms) // simulate the fetch RTT 28 let slot: *u8 = ((sharedbuf as i64) + idx * fragcap) as *u8 29 var z: i64 = 0 30 while z < 376 { slot[z] = 0 as u8; z = z + 1 } // a valid 2-packet TS fragment 31 slot[0] = 0x47 as u8 32 slot[188] = 0x47 as u8 33 sys_exit(0) 34 } 35 pids[i] = pid 36 i = i + 1 37 } 38 i = 0 39 while i < batch { sys_wait4(pids[i], st, 0); i = i + 1 } // barrier: whole batch done 40 i = 0 41 while i < batch { 42 let idx: i64 = base + i 43 let slot: *u8 = ((sharedbuf as i64) + idx * fragcap) as *u8 44 if rdl_verify_ts(slot, 376) == 1 { state[idx] = 1 as u8; done_total = done_total + 1 } // integrity gate 45 i = i + 1 46 } 47 base = base + batch 48 } 49 return done_total 50}