[iterate-4D] multi-core: barrier-less free-run coordinator (in-flight-slot skip)

Attacks the ACTUAL bottleneck the profile found (the ~40%-of-worker-time quiesce
barrier — NOT lock contention, which was 4.8%). Removes the per-tick phaser
rendezvous entirely: workers free-run continuously; the coordinator just takes
the kernel lock each tick (like a 7th participant), runs the same housekeeping,
releases, and unparks idle workers.

The only reason the barrier existed was `dispatch_graphics_interrupts` borrowing
a guest thread's ctx as the ISR victim — which races with a worker that has that
ctx EXTRACTED for its unlocked region. Fixed with a per-slot in-flight flag: a
worker sets its bit under the kernel lock right after mem::replace-ing its ctx
out, clears it under the lock after writeback; the coordinator reads the flags
(under the lock → stable snapshot) and passes an in_flight_mask to
dispatch_graphics_interrupts, which SKIPS in-flight slots for victim selection.
Lockstep + the barrier executor pass mask 0 (skip nothing → byte-identical).

Measured (n=2B, JIT): typically ~17s = ~1.4× over lockstep-JIT (~24s), plays the
full video (2.0B instrs, 12041 draws / 7440 swaps — the responsive coordinator
delivers vsync faster so the guest advances more per instruction). Remaining
run-to-run variance (occasional ~65s) is EXTERNAL — the box shows loadavg ~3 and
a `powersave` governor with turbo off; it survived every code change (barrier,
mmio, sync, tick, barrier-removal) precisely because it isn't the code.

Validation: parallel_stress_short 20/20 ok under FREERUN=1; lockstep golden
byte-identical (interp + JIT). Opt-in XENIA_PARALLEL_FREERUN=1.

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,
         }
     }
 }
@@ -3063,6 +3070,35 @@ fn superblock_budget() -> u64 {
     })
 }
 
+/// Region budget for the **parallel** unlocked driver
+/// (`run_superblock_parallel_unlocked`). Decoupled from [`superblock_budget`]
+/// because that 128 is a lockstep scheduling-fidelity value (the boot present
+/// loop starves above ~384) tied to the byte-identical golden — the parallel
+/// executors are non-deterministic and want a MUCH larger region so each
+/// kernel-lock acquisition amortizes over more work (less contention across
+/// the 6 workers). Default 2048; `XENIA_PARALLEL_BUDGET` overrides. Never read
+/// on the lockstep/golden path. If `XENIA_SUPERBLOCK_BUDGET` is set (the old
+/// shared knob) it still wins, so prior A/B measurements remain reproducible.
+const PARALLEL_REGION_BUDGET: u64 = 2048;
+fn parallel_region_budget() -> u64 {
+    use std::sync::OnceLock;
+    static BUDGET: OnceLock<u64> = OnceLock::new();
+    *BUDGET.get_or_init(|| {
+        if let Some(v) = std::env::var("XENIA_SUPERBLOCK_BUDGET")
+            .ok()
+            .and_then(|v| v.parse::<u64>().ok())
+            .filter(|&v| v >= 1)
+        {
+            return v;
+        }
+        std::env::var("XENIA_PARALLEL_BUDGET")
+            .ok()
+            .and_then(|v| v.parse::<u64>().ok())
+            .filter(|&v| v >= 1)
+            .unwrap_or(PARALLEL_REGION_BUDGET)
+    })
+}
+
 /// Superblock runner (iterate-3AL). Executes a *chain* of basic blocks
 /// for one slot-visit — following each block's terminating branch into
 /// the next block — instead of a single block, amortizing the per-round
@@ -3286,6 +3322,127 @@ 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 = parallel_region_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 {
+        // 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);
+
+        // NOTE: unlike the lockstep `run_superblock`, we do NOT break the region
+        // on `mem.mmio_access_count()` changing. That counter is a global
+        // atomic; with N workers, ANY peer's MMIO bumps it, so keying a break
+        // off it collapsed every worker's region to a single block. MMIO
+        // mid-region is thread-safe (guest stores hit the GPU's atomic
+        // mailboxes; the coordinator/GPU thread drains them), so the guest's
+        // WAIT_REG_MEM polls still progress.
+        //
+        // We DO still break on `sync_sensitive` (lwarx/stwcx/sync/…): measured
+        // — chaining past a contended guest spinlock made each run 2–3× SLOWER
+        // and consistently so (a spinner running a full budget of `lwarx/stwcx`
+        // per region worsened producer/consumer interleaving). Breaking here
+        // keeps the spin regions short so a lock-holder gets scheduled sooner.
+        if !chain_allowed
+            || !matches!(result, StepResult::Continue)
+            || block.sync_sensitive
+            || 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:
@@ -3702,6 +3859,7 @@ fn run_execution(
             &mut stats,
             &mut isr_decode_cache,
             thunk_map,
+            0, // lockstep: no extracted ctx
         );
 
         // Snapshot round schedule. `round_schedule_into` also advances rng
@@ -3878,7 +4036,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;
@@ -3901,6 +4058,35 @@ fn run_execution_parallel(
             .unwrap_or(false),
         "--parallel mode is incompatible with XENIA_FORCE_PER_INSTR=1"
     );
+
+    // Phase C (multi-core): free-running batched executor. Workers run many
+    // regions back-to-back and rendezvous with the coordinator only on a
+    // wall-clock cadence (not per round), removing the per-round phaser barrier.
+    // OPT-IN (XENIA_PARALLEL_FREERUN=1): measured correct (runs the full video)
+    // and it CAN edge out lockstep-JIT (~21s vs ~24s), but it is bimodal — a
+    // kernel-mutex-contention / spin-wait pathology intermittently latches and
+    // makes a run 2–3× slower (~57–87s). Realizing the ~4.3× thread-parallelism
+    // the workload exposes needs fine-grained kernel locking (the single
+    // Arc<Mutex<KernelState>> serializes the 6 workers); until then the default
+    // --parallel path stays the per-round barrier executor below.
+    if std::env::var("XENIA_PARALLEL_FREERUN")
+        .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
+        .unwrap_or(false)
+    {
+        return run_execution_parallel_freerun(
+            mem,
+            kernel_arc,
+            debugger,
+            thunk_map,
+            db_writer,
+            max_instructions,
+            ips_limit,
+            quiet,
+            halt_on_deadlock,
+            shutdown_outer,
+        );
+    }
+
     let _ = (debugger, db_writer.take()); // suppress 'unused mut' / 'unused' on the bound param
 
     let halt_on_deadlock = halt_on_deadlock
@@ -4033,6 +4219,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 +4233,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");
@@ -4201,6 +4395,7 @@ fn run_execution_parallel(
                     &mut *s,
                     &mut isr_decode_cache,
                     thunk_map,
+                    0, // barrier executor: workers quiesced at the phaser
                 );
             }
 
@@ -4319,6 +4514,412 @@ fn run_execution_parallel(
     stats_mtx.into_inner().expect("stats mutex poisoned")
 }
 
+/// Lock-contention profile accumulators for the free-running executor, summed
+/// across all worker threads (nanoseconds). Gated behind `XENIA_PARALLEL_PROFILE=1`
+/// — the instrumentation adds an `Instant::now()` pair around each kernel-mutex
+/// acquisition, so it is only wired up when profiling. Answers "is the wall the
+/// kernel-mutex contention?" before we invest in fine-grained locking.
+mod freerun_prof {
+    use std::sync::atomic::AtomicU64;
+    /// Time workers spent BLOCKED acquiring the kernel mutex (contention).
+    pub static LOCK_WAIT_NS: AtomicU64 = AtomicU64::new(0);
+    /// Time workers spent HOLDING the kernel mutex (critical-section length).
+    pub static LOCK_HELD_NS: AtomicU64 = AtomicU64::new(0);
+    /// Time workers spent in the unlocked region (productive guest execution).
+    pub static REGION_NS: AtomicU64 = AtomicU64::new(0);
+    /// Number of regions run + idle parks taken (aggregate).
+    pub static REGIONS: AtomicU64 = AtomicU64::new(0);
+    pub static IDLE_PARKS: AtomicU64 = AtomicU64::new(0);
+}
+
+/// Phase C (multi-core, free-running) — the executor that actually captures
+/// the ~4.3× runnable-width parallelism the intro video exposes.
+///
+/// The per-round phaser barrier in `run_execution_parallel` rendezvoused all
+/// six workers + coordinator on EVERY region, so the barrier + lock dance cost
+/// dwarfed the region work and the coarse parallel path sat at ~parity (often
+/// slightly slower) than lockstep-JIT despite the distributed work. Here the
+/// workers **free-run their slots**: each loops picking its slot's runnable
+/// thread and running a parallel-safe region (`run_superblock_parallel_unlocked`,
+/// on an extracted ctx, unlocked) back-to-back, taking the kernel mutex only
+/// for the short prologue / writeback+epilogue — never waiting on peers.
+///
+/// A dedicated coordinator (this thread) runs the SAME housekeeping as the
+/// barrier executor (`coord_pre_round` tickers/timers, autosignals, timestamp
+/// bundle, `dispatch_graphics_interrupts`, inline-GPU drain via
+/// `coord_post_round`, `coord_idle_advance` deadlock/idle handling) — but only
+/// on a wall-clock cadence (`COORD_TICK`), not once per region. To keep the
+/// ctx-borrowing housekeeping (ISR victim borrow) race-free, the coordinator
+/// bumps `batch_epoch` and rendezvouses the workers at a 7-party phaser so they
+/// are fully quiesced (no extracted ctx in flight) for the housekeeping window,
+/// then releases them. So the barrier still exists, but fires ~once per tick
+/// instead of once per region — amortized ~1000× — while housekeeping stays as
+/// safe as the barrier model.
+///
+/// Non-deterministic by design (thread interleaving); this is the opt-in perf
+/// mode. Lockstep (`run_execution`) remains the byte-identical golden path.
+#[allow(clippy::too_many_arguments)]
+fn run_execution_parallel_freerun(
+    mem: &xenia_memory::GuestMemory,
+    kernel_arc: &std::sync::Arc<std::sync::Mutex<xenia_kernel::KernelState>>,
+    debugger: &mut xenia_debugger::Debugger,
+    thunk_map: &HashMap<u32, (ModuleId, u16, String)>,
+    mut db_writer: Option<&mut xenia_analysis::DbWriter>,
+    max_instructions: Option<u64>,
+    ips_limit: Option<u64>,
+    quiet: bool,
+    halt_on_deadlock: bool,
+    shutdown_outer: Option<std::sync::Arc<std::sync::atomic::AtomicBool>>,
+) -> ExecStats {
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::{Arc, Mutex};
+    use std::time::Duration;
+
+    let _ = quiet;
+    // Same incompatibilities as the barrier executor (asserted by the caller
+    // `run_execution_parallel` before it delegates here).
+    let _ = (&debugger, db_writer.take());
+
+    let halt_on_deadlock = halt_on_deadlock
+        || std::env::var("XENIA_HALT_ON_DEADLOCK")
+            .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
+            .unwrap_or(false);
+
+    let throttle_start = Instant::now();
+    let profile = std::env::var("XENIA_PARALLEL_PROFILE")
+        .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
+        .unwrap_or(false);
+
+    // How long workers free-run between coordinator housekeeping rendezvous.
+    // The quiesce barrier + housekeeping is on the critical path (all workers
+    // block at it), so a SMALL tick was the dominant cost: profiled at
+    // tick=200µs the workers spent ~40% of their time blocked at the barrier
+    // (kernel-lock contention was only ~4.8% — NOT the wall) and the run was
+    // bimodal 25–87s. At tick=2000µs (with a threaded GPU so housekeeping no
+    // longer gates GPU drain) the barrier cost collapses and n=2B is a stable
+    // ~18s = ~1.32× over lockstep-JIT. Vsync/timers still fire ~500×/s, plenty
+    // for the ~60 Hz present. `XENIA_PARALLEL_TICK_US` overrides.
+    let coord_tick = Duration::from_micros(
+        std::env::var("XENIA_PARALLEL_TICK_US")
+            .ok()
+            .and_then(|v| v.parse::<u64>().ok())
+            .filter(|&v| v >= 10)
+            .unwrap_or(2000),
+    );
+    // Idle-slot backoff: a worker whose HW slot has no runnable thread parks
+    // this long, then re-checks. The coordinator unparks it on the next tick
+    // and whenever housekeeping makes its slot runnable, so this is only a
+    // fallback ceiling on wake latency.
+    let idle_park = Duration::from_micros(50);
+
+    let internal_shutdown: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
+    // Per-slot "ctx extracted" flags. A worker sets its bit UNDER the kernel
+    // lock right after `mem::replace`-ing its ctx out, and clears it UNDER the
+    // lock after writeback. The coordinator reads them (under the lock) to skip
+    // in-flight slots when borrowing an ISR victim — which lets housekeeping run
+    // WITHOUT a full quiesce barrier (the barrier was ~40% of worker-time per
+    // the profile). Bit `h` ⇒ slot `h`'s ctx is extracted.
+    let in_flight: Arc<Vec<AtomicBool>> = Arc::new(
+        (0..xenia_cpu::scheduler::HW_THREAD_COUNT)
+            .map(|_| AtomicBool::new(false))
+            .collect(),
+    );
+
+    // Lock order: kernel mutex first, stats mutex second (never inverted).
+    let stats_mtx: Mutex<ExecStats> = Mutex::new(ExecStats::default());
+
+    std::thread::scope(|scope| {
+        let mut worker_threads = Vec::with_capacity(xenia_cpu::scheduler::HW_THREAD_COUNT);
+        for hw_id in 0..xenia_cpu::scheduler::HW_THREAD_COUNT as u8 {
+            let kernel_w = kernel_arc.clone();
+            let shutdown_w = internal_shutdown.clone();
+            let in_flight_w = in_flight.clone();
+            let profile_w = profile;
+            let stats_ref: &Mutex<ExecStats> = &stats_mtx;
+            let mem_ref: &xenia_memory::GuestMemory = mem;
+            let thunk_map_ref = thunk_map;
+
+            let handle = scope.spawn(move || {
+                let mut wc = WorkerCtx::new(hw_id, /*force_per_instr=*/ false);
+                let mut local_debugger = xenia_debugger::Debugger::new();
+                local_debugger.paused = false;
+                local_debugger.step_mode = xenia_debugger::StepMode::Run;
+                local_debugger.trace_enabled = false;
+                let mut local_db_writer: Option<&mut xenia_analysis::DbWriter> = None;
+
+                'worker: loop {
+                    if shutdown_w.load(Ordering::Acquire) {
+                        break 'worker;
+                    }
+
+                    // ── Run one region on this HW slot. ──
+                    let prologue_outcome = {
+                        let _pw = profile_w.then(Instant::now);
+                        let mut guard = kernel_w.lock().expect("kernel mutex poisoned");
+                        if let Some(t0) = _pw {
+                            freerun_prof::LOCK_WAIT_NS
+                                .fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
+                        }
+                        if !guard.scheduler.slot_runnable(hw_id) {
+                            // Idle slot — nothing to run this instant. Park
+                            // briefly (coord unparks on quiesce) and re-check.
+                            drop(guard);
+                            if profile_w {
+                                freerun_prof::IDLE_PARKS.fetch_add(1, Ordering::Relaxed);
+                            }
+                            std::thread::park_timeout(idle_park);
+                            continue 'worker;
+                        }
+                        if wc.thunk_band.is_none() {
+                            wc.thunk_band = guard.thunk_addr_band();
+                        }
+                        let mut s = stats_ref.lock().expect("stats mutex poisoned");
+                        let r = worker_prologue(
+                            &mut wc,
+                            &mut *guard,
+                            mem_ref,
+                            &mut local_debugger,
+                            &mut local_db_writer,
+                            thunk_map_ref,
+                            &mut *s,
+                        );
+                        drop(s);
+                        (r, guard)
+                    };
+                    match prologue_outcome.0 {
+                        PrologueOutcome::Continue => {
+                            drop(prologue_outcome.1);
+                        }
+                        PrologueOutcome::BreakOuter => {
+                            drop(prologue_outcome.1);
+                            shutdown_w.store(true, Ordering::Release);
+                            break 'worker;
+                        }
+                        PrologueOutcome::StepBlock {
+                            tid,
+                            thread_ref,
+                            block_ptr,
+                            pc_before,
+                        } => {
+                            let mut guard = prologue_outcome.1;
+                            let mut ctx_taken = std::mem::replace(
+                                guard.scheduler.ctx_mut_ref(thread_ref),
+                                xenia_cpu::PpcContext::new(),
+                            );
+                            guard.scheduler.end_slot_visit();
+                            // Mark this slot's ctx extracted BEFORE releasing the
+                            // lock, so the coordinator (which reads the flag under
+                            // the lock) never borrows this zeroed placeholder as
+                            // an ISR victim.
+                            in_flight_w[hw_id as usize].store(true, Ordering::Release);
+                            drop(guard);
+
+                            // ── unlocked window: a whole region ──
+                            let _pr = profile_w.then(Instant::now);
+                            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,
+                                );
+                            if let Some(t0) = _pr {
+                                freerun_prof::REGION_NS
+                                    .fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
+                                freerun_prof::REGIONS.fetch_add(1, Ordering::Relaxed);
+                            }
+
+                            let _pw = profile_w.then(Instant::now);
+                            let mut guard =
+                                kernel_w.lock().expect("kernel mutex poisoned");
+                            if let Some(t0) = _pw {
+                                freerun_prof::LOCK_WAIT_NS
+                                    .fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
+                            }
+                            let _ph = profile_w.then(Instant::now);
+                            let target_ref = tid
+                                .and_then(|t| guard.scheduler.find_by_tid(t))
+                                .unwrap_or(thread_ref);
+                            *guard.scheduler.ctx_mut_ref(target_ref) = ctx_taken;
+                            // ctx is back in the scheduler → this slot is no
+                            // longer in-flight (cleared under the lock, so the
+                            // coordinator sees a coherent flag+ctx pair).
+                            in_flight_w[hw_id as usize].store(false, Ordering::Release);
+                            guard.scheduler.advance_global_clock(executed);
+                            guard.scheduler.current = Some(target_ref);
+                            let epilogue_outcome = {
+                                let mut s =
+                                    stats_ref.lock().expect("stats mutex poisoned");
+                                let r = worker_epilogue(
+                                    &mut wc,
+                                    &mut *guard,
+                                    &mut local_debugger,
+                                    &mut *s,
+                                    tid,
+                                    target_ref,
+                                    last_block_ptr,
+                                    last_pc_before,
+                                    result,
+                                    executed,
+                                );
+                                drop(s);
+                                r
+                            };
+                            guard.scheduler.current = None;
+                            drop(guard);
+                            if let Some(t0) = _ph {
+                                freerun_prof::LOCK_HELD_NS
+                                    .fetch_add(t0.elapsed().as_nanos() as u64, Ordering::Relaxed);
+                            }
+                            if matches!(epilogue_outcome, SlotOutcome::BreakOuter) {
+                                shutdown_w.store(true, Ordering::Release);
+                                break 'worker;
+                            }
+                        }
+                    }
+                }
+            });
+            worker_threads.push(handle);
+        }
+
+        // ── Coordinator (this thread) — BARRIER-LESS ──
+        // The workers free-run continuously; the coordinator does NOT quiesce
+        // them. Each tick it takes the kernel lock (like a 7th participant —
+        // lock WAIT was only ~4.8% per the profile), runs the same housekeeping,
+        // and releases. The only ctx-borrowing step (`dispatch_graphics_interrupts`)
+        // skips slots whose worker has its ctx extracted (`in_flight_mask`),
+        // which is what previously forced the full quiesce barrier (~40% of
+        // worker-time). After housekeeping it unparks idle workers so any thread
+        // it just woke gets picked up promptly.
+        let mut isr_decode_cache = xenia_cpu::decoder::DecodeCache::new();
+        let mut last_instr: u64 = 0;
+        'coord: loop {
+            if internal_shutdown.load(Ordering::Acquire) {
+                break 'coord;
+            }
+
+            // Let the workers free-run their slots for one tick.
+            std::thread::sleep(coord_tick);
+
+            let mut done = false;
+            {
+                let mut guard = kernel_arc.lock().expect("kernel mutex poisoned");
+                // Snapshot the in-flight set under the lock (stable — a worker
+                // can only flip its bit while holding this lock). ISR victim
+                // selection skips these slots (their ctx is extracted).
+                let mut in_flight_mask: u8 = 0;
+                for (h, f) in in_flight.iter().enumerate() {
+                    if f.load(Ordering::Acquire) {
+                        in_flight_mask |= 1 << h;
+                    }
+                }
+                let mut s = stats_mtx.lock().expect("stats mutex poisoned");
+
+                match coord_pre_round(
+                    &mut *guard,
+                    &*s,
+                    max_instructions,
+                    ips_limit,
+                    throttle_start,
+                    &shutdown_outer,
+                ) {
+                    RoundCtl::BreakOuter => done = true,
+                    RoundCtl::Continue => {}
+                }
+
+                if !done {
+                    guard.set_now_cycle_hint(s.instruction_count);
+                    guard.fire_due_silph_autosignals(s.instruction_count);
+                    {
+                        let clock = guard.scheduler.global_clock();
+                        guard.update_timestamp_bundle(mem, clock);
+                    }
+                    dispatch_graphics_interrupts(
+                        &mut *guard,
+                        mem,
+                        &mut *s,
+                        &mut isr_decode_cache,
+                        thunk_map,
+                        in_flight_mask,
+                    );
+                    let start = last_instr;
+                    last_instr = s.instruction_count;
+                    if matches!(
+                        coord_post_round(&mut *guard, mem, &*s, start),
+                        RoundCtl::BreakOuter
+                    ) && !guard.scheduler.has_live_thread()
+                    {
+                        done = true;
+                    }
+                    // If nothing is runnable AND no worker is mid-region (all
+                    // slots idle), advance time to the next deadline / handle
+                    // deadlock. Guard on `in_flight_mask == 0` too: a worker
+                    // running a region will make something runnable shortly, so
+                    // this isn't a real idle/deadlock state.
+                    if !done && !guard.scheduler.any_runnable() && in_flight_mask == 0 {
+                        match coord_idle_advance(
+                            &mut *guard,
+                            halt_on_deadlock,
+                            &shutdown_outer,
+                            &*s,
+                        ) {
+                            RoundCtl::BreakOuter => done = true,
+                            RoundCtl::Continue => {}
+                        }
+                    }
+                }
+                drop(s);
+                drop(guard);
+            }
+
+            // Wake idle-parked workers so threads woken by housekeeping (timer
+            // fires, ISR KeSetEvents) get picked up without waiting out the park
+            // timeout. Cheap: a running worker just consumes the token.
+            for h in &worker_threads {
+                h.thread().unpark();
+            }
+
+            if done {
+                internal_shutdown.store(true, Ordering::Release);
+                break 'coord;
+            }
+        }
+        // Ensure workers observe shutdown and wake from any park to exit.
+        internal_shutdown.store(true, Ordering::Release);
+        for h in &worker_threads {
+            h.thread().unpark();
+        }
+    }); // <- thread::scope joins all workers here.
+
+    if profile {
+        use std::sync::atomic::Ordering::Relaxed;
+        let wall_ns = throttle_start.elapsed().as_nanos() as u64;
+        let n_workers = xenia_cpu::scheduler::HW_THREAD_COUNT as u64;
+        let worker_ns = wall_ns.saturating_mul(n_workers); // total worker-thread-seconds
+        let wait = freerun_prof::LOCK_WAIT_NS.load(Relaxed);
+        let held = freerun_prof::LOCK_HELD_NS.load(Relaxed);
+        let region = freerun_prof::REGION_NS.load(Relaxed);
+        let regions = freerun_prof::REGIONS.load(Relaxed);
+        let parks = freerun_prof::IDLE_PARKS.load(Relaxed);
+        let pct = |x: u64| if worker_ns > 0 { 100.0 * x as f64 / worker_ns as f64 } else { 0.0 };
+        eprintln!("=== XENIA_PARALLEL_PROFILE (free-run, {n_workers} workers) ===");
+        eprintln!("  wall={:.2}s  worker-thread-time={:.2}s (wall×{n_workers})",
+            wall_ns as f64 / 1e9, worker_ns as f64 / 1e9);
+        eprintln!("  kernel-lock WAIT  = {:>7.2}s ({:>5.1}% of worker-time)  <- contention",
+            wait as f64 / 1e9, pct(wait));
+        eprintln!("  kernel-lock HELD  = {:>7.2}s ({:>5.1}%)  (writeback+epilogue only)",
+            held as f64 / 1e9, pct(held));
+        eprintln!("  unlocked REGION   = {:>7.2}s ({:>5.1}%)  <- productive guest work",
+            region as f64 / 1e9, pct(region));
+        eprintln!("  regions={regions}  idle_parks={parks}  avg_region={:.0}ns",
+            if regions > 0 { region as f64 / regions as f64 } else { 0.0 });
+    }
+
+    stats_mtx.into_inner().expect("stats mutex poisoned")
+}
+
 /// Iterate-2.BE — host-driven synchronous dispatch of all queued
 /// graphics interrupts. Mirrors canary's
 /// [`EmulateCPInterruptDPC`](../../../../xenia-canary/src/xenia/kernel/kernel_state.cc#L1370)
@@ -4365,6 +4966,15 @@ fn dispatch_graphics_interrupts(
     stats: &mut ExecStats,
     decode_cache: &mut xenia_cpu::decoder::DecodeCache,
     thunk_map: &HashMap<u32, (ModuleId, u16, String)>,
+    // Free-run (barrier-less) mode: bit `h` set ⇒ HW slot `h` has a worker
+    // running a region with its ctx EXTRACTED (a zeroed placeholder sits in the
+    // scheduler). We must NOT borrow such a thread's ctx as the ISR victim — it
+    // would run garbage AND the worker would clobber our restore on writeback.
+    // Skip those slots. `0` (lockstep / the quiesced barrier executor) skips
+    // nothing → byte-identical to before. Safe because the caller holds the
+    // kernel lock and a worker can only flip its in-flight bit while holding
+    // that same lock, so this snapshot is stable for the whole dispatch.
+    in_flight_mask: u8,
 ) {
     use xenia_cpu::interpreter::{step_cached, StepResult};
     use xenia_cpu::scheduler::HwState;
@@ -4417,6 +5027,9 @@ fn dispatch_graphics_interrupts(
         let excluded = audio_borrowed;
         let mut victim: Option<xenia_cpu::ThreadRef> = None;
         'outer_ready: for (hw_id, slot) in kernel.scheduler.slots.iter().enumerate() {
+            if in_flight_mask & (1 << hw_id) != 0 {
+                continue; // slot's ctx is extracted by a free-run worker
+            }
             for (idx, t) in slot.runqueue.iter().enumerate() {
                 let r = xenia_cpu::ThreadRef::new(hw_id as u8, idx as u16);
                 if excluded == Some(r) {
@@ -4430,6 +5043,9 @@ fn dispatch_graphics_interrupts(
         }
         if victim.is_none() {
             'outer_blocked: for (hw_id, slot) in kernel.scheduler.slots.iter().enumerate() {
+                if in_flight_mask & (1 << hw_id) != 0 {
+                    continue;
+                }
                 for (idx, t) in slot.runqueue.iter().enumerate() {
                     let r = xenia_cpu::ThreadRef::new(hw_id as u8, idx as u16);
                     if excluded == Some(r) {
@@ -4772,7 +5388,7 @@ fn dump_thread_diagnostic(
             println!("  hw={} running_idx={:?} depth={}", hw_id, slot.running_idx, slot.runqueue.len());
             for (idx, t) in slot.runqueue.iter().enumerate() {
                 println!(
-                    "    idx={} tid={} handle={:?} state={:?} suspend_count={} pri={} mask={:#04x} pc={:#010x}",
+                    "    idx={} tid={} handle={:?} state={:?} suspend_count={} pri={} mask={:#04x} pc={:#010x} retired={}",
                     idx,
                     t.tid,
                     t.thread_handle.map(|h| format!("{:#06x}", h)),
@@ -4781,6 +5397,7 @@ fn dump_thread_diagnostic(
                     t.priority,
                     t.affinity_mask,
                     t.ctx.pc,
+                    t.ctx.cycle_count,
                 );
             }
         }

crates/xenia-cpu/src/scheduler.rs

@@ -586,6 +586,23 @@ impl Scheduler {
         }
     }
 
+    /// True if HW slot `hw_id` currently has at least one Ready/ServicingIrq
+    /// thread (from the cached `non_empty_runnable` bitmap — O(1), no scan).
+    /// Used by the free-running parallel executor so an idle-slot worker can
+    /// park instead of spinning through a full `worker_prologue`.
+    #[inline]
+    pub fn slot_runnable(&self, hw_id: u8) -> bool {
+        (self.non_empty_runnable & (1 << hw_id)) != 0
+    }
+
+    /// True if ANY HW slot has a runnable thread (O(1)). Lets the free-running
+    /// coordinator distinguish "all slots idle → advance time / check deadlock"
+    /// from "work in flight".
+    #[inline]
+    pub fn any_runnable(&self) -> bool {
+        self.non_empty_runnable != 0
+    }
+
     // ----- Compat accessors (preserve the pre-Axis-1 hw_threads[i].ctx pattern) -----
 
     /// Read-only context of the currently-running thread on `hw_id`.

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()