nx_dual_real_llm_compose_test.nx source

1// nx_dual_real_llm_compose_test.nx -- TWO v2 LLM actors concurrent on 2// REAL transformer forward, with wall-clock measurement. 3// 4// Substrate-honest version of [[nx_dual_companion_compose_test]] which 5// used synthetic 4-vocab manual matmul. THIS smoke runs two 6// independent v2 LLM actors that each drive the FULL Llama-class 7// runner (embedding + RoPE + RMSNorm + multi-head attention + SwiGLU 8// FFN + output projection + sample) cooperatively interleaved through 9// one nx_session. Each generates 5 tokens autoregressively. Output 10// from both actors flows through one multiplexer to one listener. 11// 12// Substrate invariants verified under DOUBLED real-transformer load: 13// * Both actors complete cleanly across 5 iterations each 14// * 10 LLM_TOKEN messages delivered (5 from each) 15// * All 10 tokens in [0, vocab) 16// * Counters scale: 30 actor steps total (5 iters * 3 phases * 2 17// actors = 30) 18// * Wall-clock measured around the whole concurrent workload 19// * Multiplexer correctly orders chunks from both actors by priority 20// 21// HONEST SCOPE: cooperative interleaving under qemu-riscv64, not 22// physical-parallel kernel execution. Phase E required for that. 23// But the substrate-honest claim PROVABLE today: one session hosts 24// two independent REAL transformer-forward actors without state 25// cross-contamination, message-bus collision, or arbiter starvation. 26 27import "nx_syscalls.nx" 28import "nx_tier.nx" 29import "nx_clock.nx" 30import "nx_actor.nx" 31import "nx_message.nx" 32import "nx_stream_multiplexer.nx" 33import "nx_session.nx" 34import "nx_bench_companion.nx" 35import "nx_bench_companion_timed.nx" 36import "nx_gguf_fixture_tiny.nx" 37import "nx_actor_role_llm_v2.nx" 38 39func main() -> i64 { 40 let now: nx_size = 1000000 41 let N_ITERS_PER_ACTOR: nx_int = 5 42 43 // ===== Two independent fixtures (different seeds, same shape) ===== 44 let fixA: *NxGgufFixtureBundle = nx_gft_build_tiny_llama(0x11111111) 45 let fixB: *NxGgufFixtureBundle = nx_gft_build_tiny_llama(0x22222222) 46 if nx_gft_is_built(fixA) != 1 { return 1 } 47 if nx_gft_is_built(fixB) != 1 { return 2 } 48 49 // ===== Shared session ===== 50 let s: *NxSession = nx_session_new_default(now) 51 if nx_session_is_ready(s) != 1 { return 3 } 52 53 let NARR: nx_int = 8001 // NARRATOR -- priority 80 54 let NPC: nx_int = 8002 // NPC -- priority 60 55 let LIS: nx_int = 8099 56 nx_session_spawn_actor(s, NARR, 1, 80, 0, now) 57 nx_session_spawn_actor(s, NPC, 1, 60, 0, now) 58 nx_session_spawn_actor(s, LIS, 5, 50, 0, now) 59 nx_session_subscribe(s, LIS, NX_MS_KIND_LLM_TOKEN) 60 61 // ===== Two LLM actor contexts on independent prompts ===== 62 let prompt_n: *u8 = sys_mmap(1) 63 prompt_n[0] = 0x61 // 'a' for NARRATOR 64 let ctxN: *NxLlmV2ActorCtx = nx_lv_actor_new( 65 fixA.spec, fixA.gguf_buf, fixA.hdr, fixA.bpe, 66 prompt_n, 1, 67 1024, 4, fixA.prng_state, 10000, 724) 68 if (ctxN as i64) == 0 { return 4 } 69 70 let prompt_p: *u8 = sys_mmap(1) 71 prompt_p[0] = 0x63 // 'c' for NPC 72 let ctxP: *NxLlmV2ActorCtx = nx_lv_actor_new( 73 fixB.spec, fixB.gguf_buf, fixB.hdr, fixB.bpe, 74 prompt_p, 1, 75 1024, 4, fixB.prng_state, 10000, 724) 76 if (ctxP as i64) == 0 { return 5 } 77 78 // ===== Wall-clock start ===== 79 let base: *NxBenchReport = nx_bc_report_new() 80 let timed: *NxBenchTimedReport = nx_bctm_new(base) 81 timed.wallclock_start_ns = nx_clock_monotonic_ns() 82 83 // ===== Cooperatively interleave both actors across 5 iterations ===== 84 // 85 // Schedule pattern: NARR.run NARR.emit reset | NPC.run NPC.emit reset 86 // 5 times. Each "run+emit" is 3 cooperative steps (INIT, RUN, EMIT). 87 // Total = 5 iters * 3 phases * 2 actors = 30 actor steps. 88 let hist_n: *i64 = (sys_mmap(N_ITERS_PER_ACTOR * 8)) as *i64 89 let hist_p: *i64 = (sys_mmap(N_ITERS_PER_ACTOR * 8)) as *i64 90 var iter: nx_int = 0 91 var tick: nx_size = now + 100 92 while iter < N_ITERS_PER_ACTOR { 93 // ----- NARRATOR cooperative slice ----- 94 nx_lv_actor_step(ctxN, s.scheduler, s.bus, NARR, tick) 95 tick = tick + 50 96 nx_lv_actor_step(ctxN, s.scheduler, s.bus, NARR, tick) 97 if ctxN.runner_result < 0 { return 10 + iter } 98 tick = tick + 50 99 nx_lv_actor_step(ctxN, s.scheduler, s.bus, NARR, tick) 100 if ctxN.current_phase != NX_LV_PHASE_DONE { return 20 + iter } 101 let tokN: nx_int = ctxN.runner_result 102 hist_n[iter] = tokN as i64 103 tick = tick + 50 104 105 // ----- NPC cooperative slice ----- 106 nx_lv_actor_step(ctxP, s.scheduler, s.bus, NPC, tick) 107 tick = tick + 50 108 nx_lv_actor_step(ctxP, s.scheduler, s.bus, NPC, tick) 109 if ctxP.runner_result < 0 { return 30 + iter } 110 tick = tick + 50 111 nx_lv_actor_step(ctxP, s.scheduler, s.bus, NPC, tick) 112 if ctxP.current_phase != NX_LV_PHASE_DONE { return 40 + iter } 113 let tokP: nx_int = ctxP.runner_result 114 hist_p[iter] = tokP as i64 115 tick = tick + 50 116 117 // ----- Autoregressive reset for next round (unless final) ----- 118 if iter < (N_ITERS_PER_ACTOR - 1) { 119 let nb_n: nx_int = nx_gft_vocab_byte(fixA, tokN) 120 prompt_n[0] = nb_n as u8 121 if nx_lv_actor_reset_for_next_token(ctxN, s.scheduler, NARR, prompt_n, 1) != NX_LV_V_STEPPED { return 50 + iter } 122 let nb_p: nx_int = nx_gft_vocab_byte(fixB, tokP) 123 prompt_p[0] = nb_p as u8 124 if nx_lv_actor_reset_for_next_token(ctxP, s.scheduler, NPC, prompt_p, 1) != NX_LV_V_STEPPED { return 60 + iter } 125 } 126 iter = iter + 1 127 } 128 129 // ===== Wall-clock end + capture substrate counters ===== 130 if nx_bctm_capture_finish(timed, s, 11) != NX_BCTM_OK { return 70 } 131 132 // ===== Substrate invariants under DUAL real-transformer load ===== 133 // 134 // 30 actor steps (5 iters * 3 phases * 2 actors). 135 if base.total_actor_steps != 30 { return 71 } 136 // Each step accounts 1000us logical runtime. 137 if base.cumulative_runtime_us != 30000 { return 72 } 138 // Both LLM actors COMPLETED on their last iteration; LIS stays READY. 139 if base.actors_completed != 2 { return 73 } 140 if base.actors_failed != 0 { return 74 } 141 if base.n_actors != 3 { return 75 } 142 143 // Listener received N_ITERS * 2 = 10 LLM_TOKEN messages 144 if nx_ms_pending(s.bus, LIS) != 10 { return 76 } 145 146 // ===== Wall-clock invariants ===== 147 if timed.wallclock_elapsed_us < 0 { return 77 } 148 if nx_bctm_verdict(timed) != NX_BCTM_OK { return 78 } 149 // Monotonic: end > start 150 if timed.wallclock_end_ns <= timed.wallclock_start_ns { return 79 } 151 152 // ===== Drain + verify every token ===== 153 var drained: nx_int = 0 154 var seen_narr: nx_int = 0 155 var seen_npc: nx_int = 0 156 while drained < 10 { 157 let m: *NxMessage = nx_ms_receive(s.bus, LIS) 158 if (m as i64) == 0 { return 80 + drained } 159 if m.kind != NX_MS_KIND_LLM_TOKEN { return 90 + drained } 160 let recv_tok: nx_int = m.payload_handle as nx_int 161 if recv_tok < 0 { return 100 + drained } 162 if recv_tok >= 4 { return 110 + drained } 163 if m.sender_actor_id == NARR { seen_narr = seen_narr + 1 } 164 if m.sender_actor_id == NPC { seen_npc = seen_npc + 1 } 165 drained = drained + 1 166 } 167 if seen_narr != N_ITERS_PER_ACTOR { return 120 } 168 if seen_npc != N_ITERS_PER_ACTOR { return 121 } 169 170 // ===== Push tokens into multiplexer with stream classes ===== 171 // 172 // Simulate the parallel-companion output pipeline: NARRATOR's tokens 173 // -> LLM_TEXT stream, NPC's tokens -> LLM_TEXT stream, mux delivers 174 // them in priority + FIFO order. 175 var pi: nx_int = 0 176 while pi < N_ITERS_PER_ACTOR { 177 nx_sm_push(s.multiplexer, NARR, NX_SM_CLASS_LLM_TEXT, 178 100 + pi, 8, hist_n[pi] as nx_size, tick) 179 nx_sm_push(s.multiplexer, NPC, NX_SM_CLASS_LLM_TEXT, 180 200 + pi, 8, hist_p[pi] as nx_size, tick) 181 pi = pi + 1 182 } 183 if nx_sm_pending_count(s.multiplexer) != 10 { return 122 } 184 185 return 0 186}