nx_stream_multiplexer.nx source
↩ module page · 298 lines · 10990 B
1// nx_stream_multiplexer.nx -- interleaved concurrent modality streams.
2//
3// Per CARDINAL [[feedback-parallel-companion-multimodal-dnd-real-time]]:
4// LLM streaming + image gen + video + audio TTS + game actions must run
5// CONCURRENT on one RTX 5080 with sub-50ms preemption. The multiplexer
6// is the substrate that orders + interleaves output tokens from multiple
7// generators into one ordered output stream the operator perceives.
8//
9// V1 ships sealed stream-class enum (one per modality), per-stream
10// priority (LLM speech > game actions > image > video > audio fill),
11// chunk arrival timestamps, fairness budget (no starvation), and
12// next-chunk arbitration that respects priority + freshness + budget.
13
14import "nx_syscalls.nx"
15import "nx_tier.nx"
16
17// ===== Sealed enum: NxStreamClass =================================
18
19const NX_SM_CLASS_LLM_TEXT: nx_int = 0
20const NX_SM_CLASS_GAME_ACTION: nx_int = 1
21const NX_SM_CLASS_AUDIO_TTS: nx_int = 2
22const NX_SM_CLASS_IMAGE_TILE: nx_int = 3
23const NX_SM_CLASS_VIDEO_FRAME: nx_int = 4
24const NX_SM_CLASS_TOOL_RESULT: nx_int = 5
25const NX_SM_CLASS_SYSTEM_EVENT: nx_int = 6 // highest priority, always wins
26const NX_SM_CLASS_N: nx_int = 7
27
28// ===== Sealed enum: NxMuxVerdict ==================================
29
30const NX_SM_V_DELIVERED: nx_int = 0
31const NX_SM_V_EMPTY: nx_int = 1 // no chunks available
32const NX_SM_V_STARVED: nx_int = 2 // backed off a stream that hit budget
33const NX_SM_V_INVALID: nx_int = 3
34const NX_SM_V_NULL: nx_int = 4
35const NX_SM_V_N: nx_int = 5
36
37// ===== Struct: NxStreamChunk ======================================
38
39struct NxStreamChunk {
40 stream_id: nx_int,
41 stream_class: nx_int,
42 sequence_no: nx_int,
43 bytes: nx_size,
44 arrived_at_us: nx_size,
45 payload_handle: nx_size,
46}
47
48const NX_SM_C_BYTES: nx_int = 48
49
50struct NxStreamMultiplexer {
51 chunks: *u8, // pending queue
52 head: nx_int, // ring head
53 tail: nx_int, // ring tail
54 capacity: nx_int, // ring size
55 n_pending: nx_int,
56 // Per-class fairness budget (chunks served per epoch).
57 epoch_count: nx_int,
58 // Per-class served count this epoch (NX_SM_CLASS_N slots, index by class)
59 served_counts: *u8, // i64[NX_SM_CLASS_N]
60 // Per-class budgets (i64[NX_SM_CLASS_N])
61 budgets: *u8,
62 last_delivered_us: nx_size,
63 starvation_threshold_us: nx_size,
64}
65
66const NX_SM_BYTES: nx_int = 80
67
68// ===== Validators =================================================
69
70func nx_sm_class_is_valid(c: nx_int) -> nx_int {
71 if c < 0 { return 0 }
72 if c >= NX_SM_CLASS_N { return 0 }
73 return 1
74}
75
76func nx_sm_v_is_valid(v: nx_int) -> nx_int {
77 if v < 0 { return 0 }
78 if v >= NX_SM_V_N { return 0 }
79 return 1
80}
81
82// Priority -- HIGHER returned int = HIGHER priority. System events
83// always win; LLM text + game actions + audio TTS are the realtime
84// trio; image / video are bulk; tool result mid.
85func nx_sm_class_priority(c: nx_int) -> nx_int {
86 if c == NX_SM_CLASS_SYSTEM_EVENT { return 100 }
87 if c == NX_SM_CLASS_GAME_ACTION { return 90 }
88 if c == NX_SM_CLASS_LLM_TEXT { return 80 }
89 if c == NX_SM_CLASS_AUDIO_TTS { return 70 }
90 if c == NX_SM_CLASS_TOOL_RESULT { return 60 }
91 if c == NX_SM_CLASS_IMAGE_TILE { return 40 }
92 if c == NX_SM_CLASS_VIDEO_FRAME { return 30 }
93 return 0
94}
95
96// ===== Constructor ================================================
97
98func nx_sm_new(capacity: nx_int,
99 starvation_threshold_us: nx_size) -> *NxStreamMultiplexer {
100 if capacity <= 0 { return 0 as *NxStreamMultiplexer }
101 let raw: *u8 = sys_mmap(NX_SM_BYTES)
102 let m: *NxStreamMultiplexer = raw as *NxStreamMultiplexer
103 m.chunks = sys_mmap(capacity * NX_SM_C_BYTES)
104 m.head = 0
105 m.tail = 0
106 m.capacity = capacity
107 m.n_pending = 0
108 m.epoch_count = 0
109 m.served_counts = sys_mmap(NX_SM_CLASS_N * 8)
110 m.budgets = sys_mmap(NX_SM_CLASS_N * 8)
111 m.last_delivered_us = 0
112 m.starvation_threshold_us = starvation_threshold_us
113 // Initialize budgets: realtime classes get larger per-epoch budget.
114 let budgets_arr: *i64 = m.budgets as *i64
115 let served_arr: *i64 = m.served_counts as *i64
116 var i: nx_int = 0
117 while i < NX_SM_CLASS_N {
118 budgets_arr[i] = 4
119 served_arr[i] = 0
120 i = i + 1
121 }
122 budgets_arr[NX_SM_CLASS_SYSTEM_EVENT] = 16
123 budgets_arr[NX_SM_CLASS_LLM_TEXT] = 16
124 budgets_arr[NX_SM_CLASS_GAME_ACTION] = 16
125 budgets_arr[NX_SM_CLASS_AUDIO_TTS] = 16
126 budgets_arr[NX_SM_CLASS_IMAGE_TILE] = 2
127 budgets_arr[NX_SM_CLASS_VIDEO_FRAME] = 2
128 return m
129}
130
131func _sm_chunk_at(m: *NxStreamMultiplexer, idx: nx_int) -> *NxStreamChunk {
132 let off: nx_int = idx * NX_SM_C_BYTES
133 return (m.chunks + off) as *NxStreamChunk
134}
135
136// ===== Push chunk =================================================
137
138func nx_sm_push(m: *NxStreamMultiplexer,
139 stream_id: nx_int,
140 stream_class: nx_int,
141 sequence_no: nx_int,
142 bytes: nx_size,
143 payload_handle: nx_size,
144 now_us: nx_size) -> nx_int {
145 if (m as i64) == 0 { return NX_SM_V_NULL }
146 if nx_sm_class_is_valid(stream_class) == 0 { return NX_SM_V_INVALID }
147 if m.n_pending >= m.capacity { return NX_SM_V_INVALID }
148 let c: *NxStreamChunk = _sm_chunk_at(m, m.tail)
149 c.stream_id = stream_id
150 c.stream_class = stream_class
151 c.sequence_no = sequence_no
152 c.bytes = bytes
153 c.arrived_at_us = now_us
154 c.payload_handle = payload_handle
155 m.tail = (m.tail + 1) % m.capacity
156 m.n_pending = m.n_pending + 1
157 return NX_SM_V_DELIVERED
158}
159
160// ===== Peek (no consume) ==========================================
161//
162// Returns pointer to chosen chunk or null. Used by next() and by
163// preemption checks (e.g. nx_interrupt_broker).
164
165func nx_sm_peek_next(m: *NxStreamMultiplexer, now_us: nx_size) -> *NxStreamChunk {
166 if (m as i64) == 0 { return 0 as *NxStreamChunk }
167 if m.n_pending == 0 { return 0 as *NxStreamChunk }
168 let budgets_arr: *i64 = m.budgets as *i64
169 let served_arr: *i64 = m.served_counts as *i64
170 var best_idx: nx_int = -1
171 var best_score: nx_int = -1
172 var i: nx_int = 0
173 var slot: nx_int = m.head
174 while i < m.n_pending {
175 let c: *NxStreamChunk = _sm_chunk_at(m, slot)
176 let pri: nx_int = nx_sm_class_priority(c.stream_class)
177 let served: i64 = served_arr[c.stream_class]
178 let budget: i64 = budgets_arr[c.stream_class]
179 var score: nx_int = pri
180 // Backoff if this class is over-budget this epoch
181 if served >= budget { score = score - 50 }
182 // Boost if this chunk has been waiting past starvation threshold
183 let wait_us: nx_size = now_us - c.arrived_at_us
184 if wait_us >= m.starvation_threshold_us { score = score + 30 }
185 if score > best_score {
186 best_score = score
187 best_idx = slot
188 }
189 slot = (slot + 1) % m.capacity
190 i = i + 1
191 }
192 if best_idx < 0 { return 0 as *NxStreamChunk }
193 return _sm_chunk_at(m, best_idx)
194}
195
196// ===== Pop selected (delete from ring, compacting) ================
197
198func _sm_remove_at(m: *NxStreamMultiplexer, idx: nx_int) -> nx_int {
199 // Shift chunks: simple linear compaction since we picked arbitrary slot.
200 // V1 priority: correctness over O(1).
201 var i: nx_int = idx
202 var next_i: nx_int = (i + 1) % m.capacity
203 while next_i != m.tail {
204 let src: *NxStreamChunk = _sm_chunk_at(m, next_i)
205 let dst: *NxStreamChunk = _sm_chunk_at(m, i)
206 dst.stream_id = src.stream_id
207 dst.stream_class = src.stream_class
208 dst.sequence_no = src.sequence_no
209 dst.bytes = src.bytes
210 dst.arrived_at_us = src.arrived_at_us
211 dst.payload_handle = src.payload_handle
212 i = next_i
213 next_i = (next_i + 1) % m.capacity
214 }
215 if m.tail == 0 { m.tail = m.capacity - 1 }
216 if m.tail != 0 { m.tail = m.tail - 1 }
217 m.n_pending = m.n_pending - 1
218 return 0
219}
220
221// ===== Next: pick + consume + accounting ==========================
222
223func nx_sm_next(m: *NxStreamMultiplexer,
224 out_class: *i64,
225 out_seq: *i64,
226 out_stream_id: *i64,
227 now_us: nx_size) -> nx_int {
228 if (m as i64) == 0 { return NX_SM_V_NULL }
229 if m.n_pending == 0 { return NX_SM_V_EMPTY }
230 let chosen: *NxStreamChunk = nx_sm_peek_next(m, now_us)
231 if (chosen as i64) == 0 { return NX_SM_V_EMPTY }
232 // Find chosen's slot by linear scan (we need the index for removal).
233 var slot: nx_int = m.head
234 var found_slot: nx_int = -1
235 var i: nx_int = 0
236 while i < m.n_pending {
237 let c: *NxStreamChunk = _sm_chunk_at(m, slot)
238 if c.sequence_no == chosen.sequence_no {
239 if c.stream_id == chosen.stream_id {
240 found_slot = slot
241 i = m.n_pending // break
242 }
243 }
244 slot = (slot + 1) % m.capacity
245 i = i + 1
246 }
247 if found_slot < 0 { return NX_SM_V_INVALID }
248 // Copy out
249 let c_out: *NxStreamChunk = _sm_chunk_at(m, found_slot)
250 out_class[0] = c_out.stream_class as i64
251 out_seq[0] = c_out.sequence_no as i64
252 out_stream_id[0] = c_out.stream_id as i64
253 let chosen_class: nx_int = c_out.stream_class
254 // Accounting
255 let served_arr: *i64 = m.served_counts as *i64
256 served_arr[chosen_class] = served_arr[chosen_class] + 1
257 m.last_delivered_us = now_us
258 // Remove
259 _sm_remove_at(m, found_slot)
260 if found_slot == m.head {
261 if m.n_pending == 0 {
262 m.head = 0
263 m.tail = 0
264 }
265 }
266 return NX_SM_V_DELIVERED
267}
268
269// ===== Epoch tick =================================================
270//
271// Operator calls this every N microseconds to reset budgets. Without
272// this, served counts only grow and eventually every stream is starved.
273
274func nx_sm_epoch_tick(m: *NxStreamMultiplexer) -> nx_int {
275 if (m as i64) == 0 { return NX_SM_V_NULL }
276 let served_arr: *i64 = m.served_counts as *i64
277 var i: nx_int = 0
278 while i < NX_SM_CLASS_N {
279 served_arr[i] = 0
280 i = i + 1
281 }
282 m.epoch_count = m.epoch_count + 1
283 return NX_SM_V_DELIVERED
284}
285
286// ===== Accessors ==================================================
287
288func nx_sm_pending_count(m: *NxStreamMultiplexer) -> nx_int {
289 if (m as i64) == 0 { return 0 }
290 return m.n_pending
291}
292
293func nx_sm_served_count(m: *NxStreamMultiplexer, stream_class: nx_int) -> nx_int {
294 if (m as i64) == 0 { return 0 }
295 if nx_sm_class_is_valid(stream_class) == 0 { return 0 }
296 let served_arr: *i64 = m.served_counts as *i64
297 return served_arr[stream_class] as nx_int
298}