[iterate-4D] multi-core: coarse-grained parallel-safe superblock driver (Phase A+B)

The --parallel worker did a full 7-party phaser barrier + kernel-lock dance
around every single ~13-instruction interpreter block, making it 20x SLOWER
than lockstep (2.7 vs 55 MIPS at n=400M) and JIT-less. Replace the single-block
unlocked window with run_superblock_parallel_unlocked: a whole straight-line
region on the extracted ctx + per-worker caches (block + JIT), stopping at the
first import/halt/mmio/sync/budget boundary for the locked epilogue to handle.
Touches zero KernelState in the lock-free window (thunk band cached once under
the lock via new KernelState::thunk_addr_band). Same JIT seam as run_superblock,
so XENIA_JIT unset = interp, set = JIT — one driver covers Phase A and B.

Measured (--gpu-inline):
- recovered the parallel path 16-30x (2.7 -> 43-95 MIPS)
- video-phase n=2B parallel-JIT budget=8192: 21.0s vs lockstep-JIT 23.8s (+13%),
  but violently budget-fragile (4096 = 79.9s) -> the hard per-round barrier +
  load imbalance caps it well below the 4.3x runnable-width ceiling. Phase C
  (free-running workers) needed for the real multiplier.

Determinism: lockstep path untouched; 6-config n=200M golden byte-identical
(incl. config6 JIT+chain+budget=1 == interp+budget=1). xenia-jit 24 tests green;
parallel_stress_short 20/20 ok.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

crates/xenia-app/src/main.rs

@@ -2513,6 +2513,12 @@ struct WorkerCtx {
     /// the `step_block` call in `run_superblock`; produces byte-identical
     /// state so goldens are unaffected.
     jit_cache: Option<xenia_jit::JitCache>,
+    /// Cached import-thunk address band, lazily populated (once) from
+    /// `KernelState::thunk_addr_band` under the kernel lock. Lets the
+    /// parallel-mode unlocked superblock driver run its chain-break check
+    /// without referencing `KernelState` (which is behind the kernel mutex)
+    /// during the lock-free window. The band is an init constant.
+    thunk_band: Option<(u32, u32)>,
 }
 
 impl WorkerCtx {
@@ -2530,6 +2536,7 @@ impl WorkerCtx {
             decode_cache: xenia_cpu::decoder::DecodeCache::new(),
             force_per_instr,
             jit_cache,
+            thunk_band: None,
         }
     }
 }
@@ -3286,6 +3293,116 @@ fn next_pc_breaks_chain(
         || !mem.is_mapped(next_pc)
 }
 
+/// Chain-break predicate for the parallel-mode unlocked window. Same decision
+/// as `next_pc_breaks_chain`, but takes the pre-captured thunk band instead of
+/// referencing `KernelState` (which is behind the kernel mutex and MUST NOT be
+/// touched during the lock-free step window). `thunk_band` is the init-constant
+/// `(lo, hi)` from `KernelState::thunk_addr_band`, cached once under the lock.
+#[inline]
+fn next_pc_breaks_chain_nolock(
+    thunk_band: Option<(u32, u32)>,
+    thunk_map: &HashMap<u32, (ModuleId, u16, String)>,
+    mem: &xenia_memory::GuestMemory,
+    next_pc: u32,
+) -> bool {
+    const LR_HALT: u32 = xenia_cpu::context::LR_HALT_SENTINEL as u32;
+    let in_band = matches!(thunk_band, Some((lo, hi)) if next_pc >= lo && next_pc <= hi);
+    next_pc == LR_HALT
+        || (in_band && thunk_map.contains_key(&next_pc))
+        || !mem.is_mapped(next_pc)
+}
+
+/// Parallel-mode unlocked superblock driver (multi-core Phase A/B). Runs a
+/// straight-line chain of blocks on the extracted `ctx` + per-worker caches
+/// (`wc.block_cache` / `wc.jit_cache`) with NO kernel lock held, so multiple
+/// worker threads execute concurrently. Touches ONLY `ctx`, `mem` (guest RAM +
+/// the global MMIO counter), `wc` (owned by this worker), and the read-only
+/// `thunk_map` / cached `wc.thunk_band` — never `KernelState`.
+///
+/// Mirrors `run_superblock`'s chain loop and STOP conditions exactly
+/// (non-`Continue` result, sync-sensitive block, MMIO touched, budget spent,
+/// or a next-PC that needs full prologue dispatch), but STOPS AND RETURNS
+/// rather than handling imports/halts/mmio inline — the caller processes the
+/// stop under the kernel lock via `worker_epilogue`. This is the fix for the
+/// per-block-barrier granularity that made the old parallel path 20× slower
+/// than lockstep: one lock+barrier now amortizes over a whole region instead
+/// of a single ~13-instruction block.
+///
+/// The JIT seam is identical to `run_superblock`: with `wc.jit_cache = Some`
+/// (i.e. `XENIA_JIT` set) the chain runs native code; with `None` it runs the
+/// interpreter. So Phase A (interp) and Phase B (JIT) are the SAME driver.
+///
+/// The per-block-entry diagnostic probes (`fire_block_entry_probes` in
+/// `run_superblock`) are intentionally OMITTED: they read `KernelState`, and
+/// parallel mode already forbids the debugger / DB-writer / per-instr paths
+/// (asserted at `run_execution_parallel` entry), so there is nothing to fire.
+///
+/// Returns `(result, last_block_ptr, last_pc_before, total_executed)`.
+fn run_superblock_parallel_unlocked(
+    wc: &mut WorkerCtx,
+    mem: &xenia_memory::GuestMemory,
+    ctx: &mut xenia_cpu::PpcContext,
+    thunk_map: &HashMap<u32, (ModuleId, u16, String)>,
+    first_block_ptr: *const xenia_cpu::block_cache::DecodedBlock,
+    first_pc_before: u32,
+) -> (
+    xenia_cpu::interpreter::StepResult,
+    *const xenia_cpu::block_cache::DecodedBlock,
+    u32,
+    u64,
+) {
+    use xenia_cpu::interpreter::{step_block, StepResult};
+
+    let budget = superblock_budget();
+    let chain_allowed = budget > 1;
+    let thunk_band = wc.thunk_band;
+
+    let mut block_ptr = first_block_ptr;
+    let mut pc_before = first_pc_before;
+    let mut total_executed: u64 = 0;
+
+    loop {
+        let mmio_before = mem.mmio_access_count();
+        // Raw-pointer deref (not a `wc` borrow), so the `wc.jit_cache` /
+        // `wc.block_cache` mutable borrows below don't conflict — same
+        // discipline as `run_superblock`. `block` is only read (step +
+        // `sync_sensitive`) before the next `lookup_or_build` re-borrow.
+        let block = unsafe { &*block_ptr };
+        let _prof_t0 = xenia_gpu::prof::is_on().then(std::time::Instant::now);
+        let cycle_before = ctx.cycle_count;
+        let result = match wc.jit_cache.as_mut() {
+            Some(jit) => jit.run_or_compile(block, ctx, mem),
+            None => step_block(ctx, mem, block),
+        };
+        let executed = ctx.cycle_count.saturating_sub(cycle_before);
+        if let Some(t0) = _prof_t0 {
+            use xenia_gpu::prof;
+            prof::add(&prof::STEP_NS, t0.elapsed().as_nanos() as u64);
+            prof::add(&prof::STEP_INSTR, executed);
+            prof::add(&prof::STEP_CALLS, 1);
+            prof::maybe_report_by_instr();
+        }
+        total_executed = total_executed.saturating_add(executed);
+
+        if !chain_allowed
+            || !matches!(result, StepResult::Continue)
+            || block.sync_sensitive
+            || mem.mmio_access_count() != mmio_before
+            || total_executed >= budget
+        {
+            return (result, block_ptr, pc_before, total_executed);
+        }
+
+        let next_pc = ctx.pc;
+        if next_pc_breaks_chain_nolock(thunk_band, thunk_map, mem, next_pc) {
+            return (result, block_ptr, pc_before, total_executed);
+        }
+
+        pc_before = next_pc;
+        block_ptr = wc.block_cache.lookup_or_build(next_pc, mem) as *const _;
+    }
+}
+
 /// JIT-specialized superblock runner (used when `wc.jit_cache.is_some()`).
 ///
 /// Identical scheduling/accounting to `run_superblock`, with ONE optimization:
@@ -3878,7 +3995,6 @@ fn run_execution_parallel(
 ) -> ExecStats {
     use std::sync::atomic::{AtomicBool, Ordering};
     use std::sync::{Arc, Mutex};
-    use xenia_cpu::interpreter::step_block;
     use xenia_cpu::{Phaser, PhaserOutcome};
 
     let _ = quiet;
@@ -4033,6 +4149,12 @@ fn run_execution_parallel(
                                 pc_before,
                             } => {
                                 let mut guard = prologue_outcome.1;
+                                // Cache the (init-constant) import-thunk band
+                                // once, under the lock, so the unlocked driver's
+                                // chain-break check never touches KernelState.
+                                if wc.thunk_band.is_none() {
+                                    wc.thunk_band = guard.thunk_addr_band();
+                                }
                                 // Snapshot ctx into a local; replace
                                 // the in-scheduler ctx with a fresh
                                 // (zeroed) PpcContext so peers can't
@@ -4041,28 +4163,30 @@ fn run_execution_parallel(
                                     guard.scheduler.ctx_mut_ref(thread_ref),
                                     xenia_cpu::PpcContext::new(),
                                 );
-                                let cycle_before = ctx_taken.cycle_count;
                                 // Clear scheduler.current so peers
                                 // don't see this slot as "running"
                                 // while the lock is unheld.
                                 guard.scheduler.end_slot_visit();
                                 drop(guard);
 
-                                // ── unlocked window ───────────────
-                                let block = unsafe { &*block_ptr };
-                                let _prof_t0 =
-                                    xenia_gpu::prof::is_on().then(std::time::Instant::now);
-                                let result = step_block(&mut ctx_taken, mem_ref, block);
-                                let executed = ctx_taken
-                                    .cycle_count
-                                    .saturating_sub(cycle_before);
-                                if let Some(t0) = _prof_t0 {
-                                    use xenia_gpu::prof;
-                                    prof::add(&prof::STEP_NS, t0.elapsed().as_nanos() as u64);
-                                    prof::add(&prof::STEP_INSTR, executed);
-                                    prof::add(&prof::STEP_CALLS, 1);
-                                    prof::maybe_report_by_instr();
-                                }
+                                // ── unlocked window (multi-core Phase A/B) ──
+                                // Run a whole parallel-safe superblock/region
+                                // (many blocks) instead of one block, so the
+                                // per-round barrier + kernel-lock dance amortizes
+                                // over the region. Stops at the first boundary
+                                // that needs locked handling (import/halt/mmio/
+                                // sync/budget); the epilogue processes it.
+                                let (result, last_block_ptr, last_pc_before, executed) =
+                                    run_superblock_parallel_unlocked(
+                                        &mut wc,
+                                        mem_ref,
+                                        &mut ctx_taken,
+                                        thunk_map_ref,
+                                        block_ptr,
+                                        pc_before,
+                                    );
+                                let block_ptr = last_block_ptr;
+                                let pc_before = last_pc_before;
                                 // ──────────────────────────────────
 
                                 let mut guard = kernel_w.lock().expect("kernel mutex poisoned");

crates/xenia-kernel/src/state.rs

@@ -648,6 +648,15 @@ impl KernelState {
         }
     }
 
+    /// The registered import-thunk address band `(lo, hi)`, set once at load
+    /// and never mutated during execution. Exposed so the parallel-mode
+    /// unlocked superblock driver can cache it and run its chain-break check
+    /// without touching `KernelState` (which is behind the kernel mutex).
+    #[inline]
+    pub fn thunk_addr_band(&self) -> Option<(u32, u32)> {
+        self.thunk_addr_band
+    }
+
     /// Resolve a `(module, ordinal)` to its registered thunk address.
     pub fn resolve_thunk(&self, module: ModuleId, ordinal: u16) -> Option<u32> {
         self.thunks_by_ordinal.get(&(module, ordinal)).copied()