_pe_dchan_reorder.nx source

1// _pe_dchan_reorder.nx -- X-ROOM-DC-003 rung-3 RELIABILITY DATA PLANE (the ACTION layer the spec 2// flagged as the emitter frontier: "seq-number arithmetic + payload reorder-buffer ... wired at 3// integration"). COMPOSES the organ-authored FSM control core _pe_dchan_arq (states/transitions = 4// WHEN to retransmit) with the data-plane ACTIONS (WHAT: in-order delivery under loss+reorder). 5// dp_recv: a sliding-window reorder buffer -- buffer out-of-order seqs, deliver contiguous from 6// base, dedup old seqs, drop out-of-window, cumulative-ack = base (next expected). This is the 7// N-party correctness layer (a 2-peer TCP room gets order free; N>2 / late-join needs this). 8// HONEST AUTHORSHIP (memory discipline): the FSM TABLE is organ-emitted (_pe_dchan_arq); these 9// ACTIONS are hand-built -- a DECLARED SCAFFOLD, emitter-of-emitters back-fill OWED (X-AUT-012 class). 10// Gate is self-validating: FSM send/ack/timeout-retransmit control + reorder/loss/dup data plane, 11// with a neg-control (a broken pass-through would deliver out-of-order -> RED). 12import "_pe_dchan_arq.nx" 13 14func dr_w(fd: i64, s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(fd, s, n); return 0 } 15func dr_n(fd: i64, v: i64) -> i64 { 16 let bb: *u8 = sys_mmap(28); var m: i64 = v; if m < 0 { sys_write(fd, "-" as *u8, 1); m = 0 - m } 17 let t: *u8 = sys_mmap(28); var k: i64 = 0; if m == 0 { t[0] = 48 as u8; k = 1 } 18 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 } 19 var i: i64 = 0; while i < k { bb[i] = t[k - 1 - i]; i = i + 1 } sys_write(fd, bb, k); return 0 20} 21// receive one (seq,payload) into the reorder buffer over window WIN. present[]/payl[] are WIN-sized 22// ring slots; base_box[0] = next-expected seq. On a contiguous run, deliver into got[] (append at 23// ng_box[0]). dedup seq<base, drop seq>=base+WIN. Returns the cumulative ack (= new base). 24func dp_recv(seq: i64, payload: i64, present: *i64, payl: *i64, base_box: *i64, WIN: i64, got: *i64, ng_box: *i64) -> i64 { 25 let base: i64 = base_box[0] 26 if seq < base { return base } // old/duplicate -> already delivered, re-ack 27 if seq >= base + WIN { return base } // out of window -> cannot buffer (sender backs off) 28 let slot: i64 = seq % WIN 29 present[slot] = 1 30 payl[slot] = payload 31 var b: i64 = base_box[0] 32 var go: i64 = 1 33 while go == 1 { 34 let bs: i64 = b % WIN 35 if present[bs] == 1 { 36 got[ng_box[0]] = payl[bs] // DELIVER in order 37 ng_box[0] = ng_box[0] + 1 38 present[bs] = 0 39 b = b + 1 40 } else { go = 0 } 41 } 42 base_box[0] = b 43 return b 44} 45func main(argc: i64, argv: *i64) -> i64 { 46 // ---- PART A: FSM CONTROL (the organ-authored core) -- a send/ack/timeout cycle ---- 47 // ESTABLISHED_IDLE(2) --app-send(2)--> AWAIT_ACK(3) --timeout(4)retransmit--> AWAIT_ACK(3) 48 // --data-ack(3)--> ESTABLISHED_IDLE(2) 49 var ctl_ok: i64 = 1 50 if _pe_dchan_arq_step(2, 2) != 3 { ctl_ok = 0 } // send -> AWAIT_ACK 51 if _pe_dchan_arq_step(3, 4) != 3 { ctl_ok = 0 } // timeout -> retransmit (self-loop) 52 if _pe_dchan_arq_step(3, 3) != 2 { ctl_ok = 0 } // ack -> ESTABLISHED_IDLE 53 if _pe_dchan_arq_step(2, 5) != 4 { ctl_ok = 0 } // close -> CLOSED 54 // a full connect+send+ack run ends ESTABLISHED_IDLE(2): open,synack,send,ack 55 let ev: *i64 = sys_mmap(8 * 4) as *i64 56 ev[0] = 0; ev[1] = 1; ev[2] = 2; ev[3] = 3 57 if _pe_dchan_arq_run(ev, 4, 0) != 2 { ctl_ok = 0 } 58 59 // ---- PART B: DATA PLANE -- reorder/loss/dup, must deliver 0..N-1 in order exactly once ---- 60 let WIN: i64 = 8 61 let N: i64 = 16 62 let present: *i64 = sys_mmap(WIN * 8) as *i64 63 let payl: *i64 = sys_mmap(WIN * 8) as *i64 64 let got: *i64 = sys_mmap(64 * 8) as *i64 65 let base_box: *i64 = sys_mmap(8) as *i64 66 let ng_box: *i64 = sys_mmap(8) as *i64 67 base_box[0] = 0; ng_box[0] = 0 68 // schedule: every adjacent pair swapped (reorder), dups (2,6 twice), gap-then-fill (11 before 10 69 // = loss-of-10-then-retransmit). payload == seq so we can assert delivered == 0..N-1. 70 let sched: *i64 = sys_mmap(40 * 8) as *i64 71 let SL: i64 = 19 72 sched[0]=1; sched[1]=0; sched[2]=3; sched[3]=2; sched[4]=2; sched[5]=5; sched[6]=4 73 sched[7]=7; sched[8]=6; sched[9]=6; sched[10]=9; sched[11]=8; sched[12]=11; sched[13]=10 74 sched[14]=13; sched[15]=12; sched[16]=15; sched[17]=14; sched[18]=14 75 var i: i64 = 0 76 while i < SL { dp_recv(sched[i], sched[i], present, payl, base_box, WIN, got, ng_box); i = i + 1 } 77 // assert: delivered count == N, and got[k]==k (in order, complete, exactly once) 78 var data_ok: i64 = 1 79 if ng_box[0] != N { data_ok = 0 } 80 var k: i64 = 0 81 while k < N { if got[k] != k { data_ok = 0 } k = k + 1 } 82 83 // ---- NEG control: out-of-window seq must NOT deliver (drop) ---- 84 let p2: *i64 = sys_mmap(WIN * 8) as *i64 85 let y2: *i64 = sys_mmap(WIN * 8) as *i64 86 let g2: *i64 = sys_mmap(8 * 8) as *i64 87 let bb2: *i64 = sys_mmap(8) as *i64 88 let ng2: *i64 = sys_mmap(8) as *i64 89 bb2[0] = 0; ng2[0] = 0 90 dp_recv(WIN + 5, 999, p2, y2, bb2, WIN, g2, ng2) // seq 13 with base 0, WIN 8 -> out of window 91 var neg_ok: i64 = 0 92 if ng2[0] == 0 { neg_ok = 1 } // nothing delivered = correctly dropped 93 94 let ok: i64 = ctl_ok * data_ok * neg_ok 95 dr_w(1, "DCHAN-RELIABILITY ctl_ok=" as *u8); dr_n(1, ctl_ok) 96 dr_w(1, " delivered=" as *u8); dr_n(1, ng_box[0]); dr_w(1, " data_ok=" as *u8); dr_n(1, data_ok) 97 dr_w(1, " neg_ok=" as *u8); dr_n(1, neg_ok); dr_w(1, "\n" as *u8) 98 let lf: i64 = sys_openat_append("knowledge/status/dchan_reliability.log" as *u8, 420) 99 if lf >= 0 { 100 dr_w(lf, "DCHAN-RELIABILITY X-ROOM-DC-003 ctl_ok=" as *u8); dr_n(lf, ctl_ok) 101 dr_w(lf, " delivered=" as *u8); dr_n(lf, ng_box[0]); dr_w(lf, " data_ok=" as *u8); dr_n(lf, data_ok) 102 dr_w(lf, " neg_ok=" as *u8); dr_n(lf, neg_ok) 103 if ok == 1 { dr_w(lf, " verdict=GREEN\n" as *u8) } else { dr_w(lf, " verdict=RED\n" as *u8) } 104 sys_close(lf) 105 } 106 if ok == 1 { sys_exit(0) } 107 sys_exit(1) 108 return 1 109}