nx_pathstripe.nx source
↩ module page · 111 lines · 4358 B
1// nx_pathstripe.nx -- the RELIABILITY pillar: multi-path STRIPING with
2// seamless protection switching (SMPTE 2022-7 / IEEE 802.1CB FRER,
3// re-implemented from spec). Promotes nx_mediapath from best-path
4// SELECTION to best-K STRIPING: each packet is sent over the top-K
5// quality paths with a global sequence number; the receiver DEDUPES and
6// REORDERS by seq, so a loss on any one path is instantly covered by
7// another copy -- ZERO recovery time (no retransmit RTT, unlike ARQ).
8// This is the critical-grade exceed beyond SRT 1+1, matching RIST any-N.
9//
10// Pure/deterministic receiver core -> KAT-able offline. Integer-only,
11// sovereign (composes nx_mediapath + nx_linkqual; raw syscalls only).
12// Combines with nx_fec_xor: stripe FEC repair symbols across paths so two
13// lossy paths reconstruct one clean stream (next increment).
14//
15// license_tier: ORIGINAL
16
17import "nx_mediapath.nx"
18
19// ---- sender side: pick the top-K paths by measured quality ----
20// Fills out_idx[0..k) with the k highest-lq_score path indices (desc).
21// Generalizes mp_select_best (best-of-1) to best-of-K striping. Returns
22// the number actually chosen (<= k, <= n).
23func stripe_select_topk(flows: *i64, n: i64, k: i64, out_idx: *i64) -> i64 {
24 let used: *i64 = sys_mmap(n * 8 + 16) as *i64
25 var z: i64 = 0
26 while z < n { used[z] = 0; z = z + 1 }
27 var chosen: i64 = 0
28 while chosen < k {
29 if chosen >= n { return chosen }
30 var best: i64 = 0 - 1
31 var best_sc: i64 = 0 - 1
32 var i: i64 = 0
33 while i < n {
34 if used[i] == 0 {
35 let fl: *LinkFlow = (flows[i]) as *LinkFlow
36 let sc: i64 = lq_score(fl)
37 if sc > best_sc { best_sc = sc; best = i }
38 }
39 i = i + 1
40 }
41 if best == (0 - 1) { return chosen }
42 used[best] = 1
43 out_idx[chosen] = best
44 chosen = chosen + 1
45 }
46 return chosen
47}
48// adaptive K: more paths when loss is high (composes fec_scheme intent).
49func stripe_k_for_loss(loss_pct: i64, n_paths: i64) -> i64 {
50 var k: i64 = 1
51 if loss_pct > 0 { k = 2 } // dual-stripe = seamless 1+1
52 if loss_pct >= 10 { k = 3 } // triple for heavy loss
53 if k > n_paths { k = n_paths }
54 return k
55}
56
57// ---- receiver side: dedup + reorder window (seamless switch) ----
58// slots[] = caller-allocated W-entry circular bitmap (1 = received, awaiting
59// in-order delivery). base = next-expected seq. delivered/lost counters.
60struct StripeRx { base: i64, delivered: i64, lost: i64 }
61const STRIPE_RX_BYTES: i64 = 24
62
63func stripe_rx_init(rx: *StripeRx, slots: *i64, w: i64) -> i64 {
64 rx.base = 0; rx.delivered = 0; rx.lost = 0
65 var i: i64 = 0
66 while i < w { slots[i] = 0; i = i + 1 }
67 return 0
68}
69// a copy of `seq` arrived (on any path). Returns 1 if newly accepted,
70// 0 if a duplicate (already received/delivered) or outside the window.
71func stripe_rx_arrive(rx: *StripeRx, slots: *i64, w: i64, seq: i64) -> i64 {
72 if seq < rx.base { return 0 } // already delivered -> dup/late
73 if seq >= rx.base + w { return 0 } // beyond reorder window
74 let sl: i64 = seq % w
75 if slots[sl] == 1 { return 0 } // duplicate copy from another path
76 slots[sl] = 1
77 return 1
78}
79// deliver all in-order packets now ready; append delivered seqs to log;
80// returns count delivered this call.
81func stripe_rx_drain(rx: *StripeRx, slots: *i64, w: i64, log: *i64) -> i64 {
82 var c: i64 = 0
83 var go: i64 = 1
84 while go == 1 {
85 let sl: i64 = rx.base % w
86 if slots[sl] == 1 {
87 log[rx.delivered] = rx.base
88 slots[sl] = 0
89 rx.base = rx.base + 1
90 rx.delivered = rx.delivered + 1
91 c = c + 1
92 } else { go = 0 }
93 }
94 return c
95}
96// end of stream: advance to `n`, delivering any buffered + counting any seq
97// never received (lost on ALL paths) -- honest, never blocks forever.
98func stripe_rx_finalize(rx: *StripeRx, slots: *i64, w: i64, n: i64, log: *i64) -> i64 {
99 while rx.base < n {
100 let sl: i64 = rx.base % w
101 if slots[sl] == 1 {
102 log[rx.delivered] = rx.base
103 slots[sl] = 0
104 rx.delivered = rx.delivered + 1
105 } else {
106 rx.lost = rx.lost + 1
107 }
108 rx.base = rx.base + 1
109 }
110 return rx.lost
111}