nx_stream_picker.nx source
↩ module page · 143 lines · 6601 B
1// nx_stream_picker.nx -- streaming + endgame piece selection (bits-up, deterministic).
2//
3// module: nishi-core.torrent.stream_picker
4// depends: nx_piece_manager.nx (reuses nx_pm_rarest; DRY -- one canonical tie-break)
5// capability: CORE_COMPUTE
6// wired_status: FULLY_WIRED
7//
8// The modern "watch / play while you download" capability and the long-tail-killing
9// endgame mode, with NO wall-clock and NO RNG -- every selection is bit-for-bit
10// replayable (the determinism exceed thesis: no mainstream client can replay a swarm
11// session). The player advertises a playhead (the piece it is about to render); the
12// picker prioritizes a sliding deadline window so playback never starves, while still
13// feeding rarest-first pieces to the swarm outside the window (swarm-health is not
14// sacrificed for streaming -- BUILD intelligence, don't strip a feature). Endgame lets
15// several peers race the final outstanding blocks (cancel the losers), killing the one
16// slow peer who holds the last block -- the dominant source of completion-time variance.
17// Pipeline depth is the request-queue throughput lever, derived from the bandwidth-delay
18// product (no magic number: callers pass measured bandwidth/rtt and the ceiling config).
19
20import "nx_piece_manager.nx"
21
22const NX_SP_BLOCK_SIZE: i64 = 16384 // mirrors nx_piece_manager's request block
23const NX_SP_PIPELINE_MIN: i64 = 2 // floor: at least 2 in flight to hide per-request RTT
24const NX_SP_PIPELINE_MAX: i64 = 256 // ceiling: bound peer memory / cancel cost
25
26// block lifecycle states (caller-owned array, one slot per block)
27const NX_SP_BLK_NEEDED: i64 = 0 // not yet requested from any peer
28const NX_SP_BLK_REQUESTED: i64 = 1 // requested, not yet received (eligible to re-race in endgame)
29const NX_SP_BLK_RECEIVED: i64 = 2 // bytes in hand; never re-pick
30
31// ---- sequential (in-order) selection: lowest-index NEEDED piece; -1 when complete ----
32// The simplest stream mode: download strictly in playback order.
33func nx_sp_sequential(have: *i64, n: i64) -> i64 {
34 var i: i64 = 0
35 while i < n {
36 if have[i] == 0 { return i }
37 i = i + 1
38 }
39 return 0 - 1
40}
41
42// ---- streaming pick: deadline window first (smooth playback), then rarest-first ----
43// Window = [playhead, playhead+window_len) clamped to [0, n). Returns the lowest-index
44// NEEDED, AVAILABLE piece inside the window (earliest unmet deadline first). If the window
45// is already satisfied (or no needed piece there is available), falls back to canonical
46// rarest-first over the whole bitfield so we still contribute rare data to the swarm.
47// -1 when there is nothing to pick. Deterministic: lowest-index tie-break throughout.
48func nx_sp_stream_pick(availability: *i64, have: *i64, n: i64, playhead: i64, window_len: i64) -> i64 {
49 var i: i64 = playhead
50 if i < 0 { i = 0 }
51 var endw: i64 = playhead + window_len
52 if endw > n { endw = n }
53 while i < endw {
54 if have[i] == 0 {
55 if availability[i] > 0 { return i }
56 }
57 i = i + 1
58 }
59 return nx_pm_rarest(availability, have, n)
60}
61
62// ---- urgency: is `piece_index` inside the panic window just ahead of the playhead? ----
63// Panic pieces must be fetched before ANY rarest-first work or the player stalls; callers
64// use this to override swarm-health selection. A piece already behind the playhead is not
65// urgent (it has been rendered). Returns 1 if urgent.
66func nx_sp_is_urgent(piece_index: i64, playhead: i64, panic_len: i64) -> i64 {
67 if piece_index < playhead { return 0 }
68 if piece_index < playhead + panic_len { return 1 }
69 return 0
70}
71
72// ---- playback gate: count of contiguous pieces we HAVE starting at the playhead ----
73// The player can keep rendering while this stays >= its required lookahead; 0 => stall now.
74func nx_sp_contiguous_have(have: *i64, n: i64, playhead: i64) -> i64 {
75 var c: i64 = 0
76 var i: i64 = playhead
77 if i < 0 { i = 0 }
78 while i < n {
79 if have[i] == 0 { return c }
80 c = c + 1
81 i = i + 1
82 }
83 return c
84}
85
86// ---- can we (re)start playback at `playhead` without an immediate stall? ----
87func nx_sp_can_play(have: *i64, n: i64, playhead: i64, lookahead: i64) -> i64 {
88 if nx_sp_contiguous_have(have, n, playhead) >= lookahead { return 1 }
89 return 0
90}
91
92// ---- endgame detection: few enough blocks remain that we should race them on many peers ----
93// `outstanding_blocks` = blocks not yet received; `threshold` is a caller config (no magic).
94func nx_sp_in_endgame(outstanding_blocks: i64, threshold: i64) -> i64 {
95 if outstanding_blocks <= threshold { return 1 }
96 return 0
97}
98
99// ---- count blocks not yet RECEIVED (drives endgame entry) ----
100func nx_sp_blocks_outstanding(block_state: *i64, nblocks: i64) -> i64 {
101 var c: i64 = 0
102 var i: i64 = 0
103 while i < nblocks {
104 if block_state[i] != NX_SP_BLK_RECEIVED { c = c + 1 }
105 i = i + 1
106 }
107 return c
108}
109
110// ---- pick the next block to request ----
111// Normal mode (endgame==0): lowest-index NEEDED (unrequested) block.
112// Endgame mode (endgame==1): prefer NEEDED, but if none remain re-pick an already-REQUESTED
113// block so a second/third peer races it (caller cancels the losers on first PIECE). RECEIVED
114// blocks are never re-picked. -1 when nothing remains. Deterministic lowest-index order.
115func nx_sp_pick_block(block_state: *i64, nblocks: i64, endgame: i64) -> i64 {
116 var i: i64 = 0
117 while i < nblocks {
118 if block_state[i] == NX_SP_BLK_NEEDED { return i }
119 i = i + 1
120 }
121 if endgame == 1 {
122 i = 0
123 while i < nblocks {
124 if block_state[i] == NX_SP_BLK_REQUESTED { return i }
125 i = i + 1
126 }
127 }
128 return 0 - 1
129}
130
131// ---- request-pipeline depth from the bandwidth-delay product (throughput lever) ----
132// Outstanding 16 KiB requests to keep in flight so a fat / high-latency link stays saturated:
133// depth = ceil( bw_bytes_per_s * rtt_ms / 1000 / block_size ) clamped to [MIN, MAX].
134// Pure integer, deterministic. Larger depth on long links = sustained throughput (the request
135// queue libtorrent/rqbit tune). bw in bytes/sec, rtt in ms; block_size lets callers test odd sizes.
136func nx_sp_pipeline_depth(bw_bytes_per_s: i64, rtt_ms: i64, block_size: i64) -> i64 {
137 if block_size <= 0 { return NX_SP_PIPELINE_MIN }
138 let bdp: i64 = (bw_bytes_per_s * rtt_ms) / 1000
139 var depth: i64 = (bdp + block_size - 1) / block_size
140 if depth < NX_SP_PIPELINE_MIN { depth = NX_SP_PIPELINE_MIN }
141 if depth > NX_SP_PIPELINE_MAX { depth = NX_SP_PIPELINE_MAX }
142 return depth
143}