nx_actor.nx source
↩ module page · 341 lines · 12036 B
1// nx_actor.nx -- cooperative green-thread driver for differentiated cells.
2//
3// Per CARDINAL [[feedback-parallel-companion-multimodal-dnd-real-time]]:
4// LLM streaming + image gen + video + audio TTS + game actions all need
5// to RUN CONCURRENT. This primitive is the cooperative scheduler --
6// each actor is a runnable unit with state (READY/RUNNING/WAITING) and
7// a current step. V1 ships pure cooperative scheduling (call ac_step
8// to drive); preemptive scheduling layers on top via [[nx_interrupt_
9// broker]] checkpoint-yield gates.
10//
11// Composes [[nx_meristem]] (cells differentiate into roles; actors wrap
12// roles into runnable units) + [[nx_resource_arbiter]] (actors compete
13// for cycles via the arbiter's CPU_MICROSECONDS resource) + [[nx_stream_
14// multiplexer]] (actor output enters the mux as chunks).
15
16import "nx_syscalls.nx"
17import "nx_tier.nx"
18
19// ===== Sealed enum: NxActorState ==================================
20
21const NX_AC_STATE_READY: nx_int = 0 // can run on next step
22const NX_AC_STATE_RUNNING: nx_int = 1 // currently executing
23const NX_AC_STATE_WAITING: nx_int = 2 // blocked on dependency
24const NX_AC_STATE_COMPLETED: nx_int = 3 // finished normally
25const NX_AC_STATE_FAILED: nx_int = 4 // unrecoverable error
26const NX_AC_STATE_PAUSED: nx_int = 5 // operator-initiated pause
27const NX_AC_STATE_N: nx_int = 6
28
29// ===== Sealed enum: NxWaitReason ==================================
30
31const NX_AC_WAIT_NONE: nx_int = 0
32const NX_AC_WAIT_RESOURCE: nx_int = 1 // arbiter denied
33const NX_AC_WAIT_MESSAGE: nx_int = 2 // peer actor input
34const NX_AC_WAIT_IO: nx_int = 3 // file / network
35const NX_AC_WAIT_TIMER: nx_int = 4
36const NX_AC_WAIT_OPERATOR: nx_int = 5 // human input
37const NX_AC_WAIT_N: nx_int = 6
38
39// ===== Sealed enum: NxActorVerdict ================================
40
41const NX_AC_V_STEPPED: nx_int = 0
42const NX_AC_V_YIELDED: nx_int = 1
43const NX_AC_V_BLOCKED: nx_int = 2
44const NX_AC_V_COMPLETED: nx_int = 3
45const NX_AC_V_FAILED: nx_int = 4
46const NX_AC_V_INVALID: nx_int = 5
47const NX_AC_V_NULL: nx_int = 6
48const NX_AC_V_N: nx_int = 7
49
50// ===== Struct: NxActor ============================================
51
52struct NxActor {
53 actor_id: nx_int,
54 state: nx_int,
55 role: nx_int, // matches nx_meristem NxCellRole
56 priority: nx_int, // 0..100
57 current_step: nx_int, // opaque program counter
58 wait_reason: nx_int, // when state == WAITING
59 cumulative_runtime_us: nx_size,
60 last_stepped_us: nx_size,
61 quanta_used_this_epoch: nx_int,
62 deadline_us: nx_size, // 0 = no deadline
63}
64
65const NX_AC_BYTES: nx_int = 80
66
67struct NxActorScheduler {
68 actors: *u8,
69 n_actors: nx_int,
70 capacity: nx_int,
71 epoch_count: nx_int,
72 epoch_quantum_us: nx_size, // wall time per epoch
73 quantum_per_actor: nx_int, // max steps per epoch
74 steps_executed_total: nx_int,
75 epoch_started_us: nx_size,
76}
77
78const NX_AC_S_BYTES: nx_int = 56
79
80// ===== Validators =================================================
81
82func nx_ac_state_is_valid(s: nx_int) -> nx_int {
83 if s < 0 { return 0 }
84 if s >= NX_AC_STATE_N { return 0 }
85 return 1
86}
87
88func nx_ac_wait_is_valid(w: nx_int) -> nx_int {
89 if w < 0 { return 0 }
90 if w >= NX_AC_WAIT_N { return 0 }
91 return 1
92}
93
94func nx_ac_v_is_valid(v: nx_int) -> nx_int {
95 if v < 0 { return 0 }
96 if v >= NX_AC_V_N { return 0 }
97 return 1
98}
99
100func nx_ac_state_is_runnable(s: nx_int) -> nx_int {
101 if s == NX_AC_STATE_READY { return 1 }
102 if s == NX_AC_STATE_RUNNING { return 1 }
103 return 0
104}
105
106func nx_ac_state_is_terminal(s: nx_int) -> nx_int {
107 if s == NX_AC_STATE_COMPLETED { return 1 }
108 if s == NX_AC_STATE_FAILED { return 1 }
109 return 0
110}
111
112// ===== Constructor (scheduler) ====================================
113
114func nx_ac_sched_new(capacity: nx_int,
115 epoch_quantum_us: nx_size,
116 quantum_per_actor: nx_int,
117 now_us: nx_size) -> *NxActorScheduler {
118 if capacity <= 0 { return 0 as *NxActorScheduler }
119 if quantum_per_actor <= 0 { return 0 as *NxActorScheduler }
120 let raw: *u8 = sys_mmap(NX_AC_S_BYTES)
121 let s: *NxActorScheduler = raw as *NxActorScheduler
122 s.actors = sys_mmap(capacity * NX_AC_BYTES)
123 s.n_actors = 0
124 s.capacity = capacity
125 s.epoch_count = 0
126 s.epoch_quantum_us = epoch_quantum_us
127 s.quantum_per_actor = quantum_per_actor
128 s.steps_executed_total = 0
129 s.epoch_started_us = now_us
130 return s
131}
132
133func _ac_at(s: *NxActorScheduler, idx: nx_int) -> *NxActor {
134 if idx < 0 { return 0 as *NxActor }
135 if idx >= s.n_actors { return 0 as *NxActor }
136 let off: nx_int = idx * NX_AC_BYTES
137 return (s.actors + off) as *NxActor
138}
139
140// ===== Spawn actor ================================================
141
142func nx_ac_spawn(s: *NxActorScheduler,
143 actor_id: nx_int,
144 role: nx_int,
145 priority: nx_int,
146 deadline_us: nx_size,
147 now_us: nx_size) -> nx_int {
148 if (s as i64) == 0 { return NX_AC_V_NULL }
149 if actor_id == 0 { return NX_AC_V_INVALID }
150 if s.n_actors >= s.capacity { return NX_AC_V_INVALID }
151 let off: nx_int = s.n_actors * NX_AC_BYTES
152 let a: *NxActor = (s.actors + off) as *NxActor
153 a.actor_id = actor_id
154 a.state = NX_AC_STATE_READY
155 a.role = role
156 a.priority = priority
157 a.current_step = 0
158 a.wait_reason = NX_AC_WAIT_NONE
159 a.cumulative_runtime_us = 0
160 a.last_stepped_us = now_us
161 a.quanta_used_this_epoch = 0
162 a.deadline_us = deadline_us
163 s.n_actors = s.n_actors + 1
164 return NX_AC_V_STEPPED
165}
166
167func nx_ac_find(s: *NxActorScheduler, actor_id: nx_int) -> *NxActor {
168 if (s as i64) == 0 { return 0 as *NxActor }
169 var i: nx_int = 0
170 while i < s.n_actors {
171 let a: *NxActor = _ac_at(s, i)
172 if a.actor_id == actor_id { return a }
173 i = i + 1
174 }
175 return 0 as *NxActor
176}
177
178// ===== State transitions ==========================================
179
180func nx_ac_set_waiting(s: *NxActorScheduler,
181 actor_id: nx_int,
182 wait_reason: nx_int) -> nx_int {
183 if (s as i64) == 0 { return NX_AC_V_NULL }
184 if nx_ac_wait_is_valid(wait_reason) == 0 { return NX_AC_V_INVALID }
185 let a: *NxActor = nx_ac_find(s, actor_id)
186 if (a as i64) == 0 { return NX_AC_V_INVALID }
187 if nx_ac_state_is_terminal(a.state) == 1 { return NX_AC_V_INVALID }
188 a.state = NX_AC_STATE_WAITING
189 a.wait_reason = wait_reason
190 return NX_AC_V_BLOCKED
191}
192
193func nx_ac_wake(s: *NxActorScheduler, actor_id: nx_int) -> nx_int {
194 if (s as i64) == 0 { return NX_AC_V_NULL }
195 let a: *NxActor = nx_ac_find(s, actor_id)
196 if (a as i64) == 0 { return NX_AC_V_INVALID }
197 if a.state != NX_AC_STATE_WAITING { return NX_AC_V_INVALID }
198 a.state = NX_AC_STATE_READY
199 a.wait_reason = NX_AC_WAIT_NONE
200 return NX_AC_V_STEPPED
201}
202
203func nx_ac_pause(s: *NxActorScheduler, actor_id: nx_int) -> nx_int {
204 if (s as i64) == 0 { return NX_AC_V_NULL }
205 let a: *NxActor = nx_ac_find(s, actor_id)
206 if (a as i64) == 0 { return NX_AC_V_INVALID }
207 if nx_ac_state_is_terminal(a.state) == 1 { return NX_AC_V_INVALID }
208 a.state = NX_AC_STATE_PAUSED
209 return NX_AC_V_BLOCKED
210}
211
212func nx_ac_resume(s: *NxActorScheduler, actor_id: nx_int) -> nx_int {
213 if (s as i64) == 0 { return NX_AC_V_NULL }
214 let a: *NxActor = nx_ac_find(s, actor_id)
215 if (a as i64) == 0 { return NX_AC_V_INVALID }
216 if a.state != NX_AC_STATE_PAUSED { return NX_AC_V_INVALID }
217 a.state = NX_AC_STATE_READY
218 return NX_AC_V_STEPPED
219}
220
221func nx_ac_complete(s: *NxActorScheduler, actor_id: nx_int) -> nx_int {
222 if (s as i64) == 0 { return NX_AC_V_NULL }
223 let a: *NxActor = nx_ac_find(s, actor_id)
224 if (a as i64) == 0 { return NX_AC_V_INVALID }
225 a.state = NX_AC_STATE_COMPLETED
226 return NX_AC_V_COMPLETED
227}
228
229func nx_ac_fail(s: *NxActorScheduler, actor_id: nx_int) -> nx_int {
230 if (s as i64) == 0 { return NX_AC_V_NULL }
231 let a: *NxActor = nx_ac_find(s, actor_id)
232 if (a as i64) == 0 { return NX_AC_V_INVALID }
233 a.state = NX_AC_STATE_FAILED
234 return NX_AC_V_FAILED
235}
236
237// ===== Step ========================================================
238//
239// Advances the actor's current_step by one and accounts for runtime.
240// Returns YIELDED if the actor still has work, COMPLETED/FAILED if
241// terminal. Caller bumps current_step beyond N to signal completion.
242
243func nx_ac_step(s: *NxActorScheduler,
244 actor_id: nx_int,
245 step_runtime_us: nx_size,
246 now_us: nx_size) -> nx_int {
247 if (s as i64) == 0 { return NX_AC_V_NULL }
248 let a: *NxActor = nx_ac_find(s, actor_id)
249 if (a as i64) == 0 { return NX_AC_V_INVALID }
250 if nx_ac_state_is_runnable(a.state) == 0 { return NX_AC_V_BLOCKED }
251 a.current_step = a.current_step + 1
252 a.cumulative_runtime_us = a.cumulative_runtime_us + step_runtime_us
253 a.last_stepped_us = now_us
254 a.quanta_used_this_epoch = a.quanta_used_this_epoch + 1
255 s.steps_executed_total = s.steps_executed_total + 1
256 // Deadline check: if past, FAIL.
257 if a.deadline_us > 0 {
258 if now_us > a.deadline_us {
259 a.state = NX_AC_STATE_FAILED
260 return NX_AC_V_FAILED
261 }
262 }
263 // Quantum exhaustion -> mark READY (yield) for next epoch
264 if a.quanta_used_this_epoch >= s.quantum_per_actor {
265 a.state = NX_AC_STATE_READY
266 return NX_AC_V_YIELDED
267 }
268 a.state = NX_AC_STATE_RUNNING
269 return NX_AC_V_STEPPED
270}
271
272// ===== Pick next actor =============================================
273//
274// Priority-first across runnable actors; tie-break by smallest
275// quanta_used_this_epoch (fairness among equal priorities).
276
277func nx_ac_pick_next(s: *NxActorScheduler) -> *NxActor {
278 if (s as i64) == 0 { return 0 as *NxActor }
279 var best: *NxActor = 0 as *NxActor
280 var best_score: nx_int = -1
281 var i: nx_int = 0
282 while i < s.n_actors {
283 let a: *NxActor = _ac_at(s, i)
284 if nx_ac_state_is_runnable(a.state) == 1 {
285 // Score: priority * 1000 - quanta_used (so higher pri + fewer quanta wins)
286 let score: nx_int = (a.priority * 1000) - a.quanta_used_this_epoch
287 if score > best_score {
288 best_score = score
289 best = a
290 }
291 }
292 i = i + 1
293 }
294 return best
295}
296
297// ===== Epoch tick =================================================
298
299func nx_ac_epoch_tick(s: *NxActorScheduler, now_us: nx_size) -> nx_int {
300 if (s as i64) == 0 { return NX_AC_V_NULL }
301 var i: nx_int = 0
302 while i < s.n_actors {
303 let a: *NxActor = _ac_at(s, i)
304 a.quanta_used_this_epoch = 0
305 i = i + 1
306 }
307 s.epoch_count = s.epoch_count + 1
308 s.epoch_started_us = now_us
309 return NX_AC_V_STEPPED
310}
311
312// ===== Aggregations ===============================================
313
314func nx_ac_count_by_state(s: *NxActorScheduler, state: nx_int) -> nx_int {
315 if (s as i64) == 0 { return 0 }
316 var count: nx_int = 0
317 var i: nx_int = 0
318 while i < s.n_actors {
319 let a: *NxActor = _ac_at(s, i)
320 if a.state == state { count = count + 1 }
321 i = i + 1
322 }
323 return count
324}
325
326func nx_ac_total_runtime_us(s: *NxActorScheduler) -> nx_size {
327 if (s as i64) == 0 { return 0 }
328 var sum: nx_size = 0
329 var i: nx_int = 0
330 while i < s.n_actors {
331 let a: *NxActor = _ac_at(s, i)
332 sum = sum + a.cumulative_runtime_us
333 i = i + 1
334 }
335 return sum
336}
337
338func nx_ac_actor_count(s: *NxActorScheduler) -> nx_int {
339 if (s as i64) == 0 { return 0 }
340 return s.n_actors
341}