Add a deterministic blocking-task pool (limited multi-threading)

[mystenmark]

What

Adds limited multi-threading to the simulator: spawn_blocking closures now run on real OS threads, serialized by a ThreadQuantum so that at most one thread (the main executor or one pool thread) runs at any instant. The executor grants each pool thread a turn between polling rounds in seeded-random order, so execution stays deterministic without intercepting synchronization primitives.

Blocking code that must wait for progress made elsewhere in the simulation calls msim::task::yield_blocking() in its wait loop instead of blocking the OS thread — under the quantum an OS-level block would never end the thread's turn and would hang the simulator.

Node kill/restart, panic propagation, and kill_current_node() are all handled on pool threads: killed tasks unwind at their next yield point (via a deferred abort delivered as an Abort turn), and PanicWrapper restarts are routed back to the main loop through the pool.

Why

Enables simulating code that genuinely needs to yield to another thread — e.g. synchronous execution that blocks mid-transaction waiting on a condition another task must satisfy. Previously such code could only be modeled by running it inline on the single sim thread, where any real wait deadlocks the simulator.

Design sketch

The original high-level design. The implementation refines several details — a Turn enum instead of Option<u32> so kill() can request a deferred abort at a yield point, routing kill_current_node() panics/restarts back to the main loop, per-thread condvars, lazy thread startup, and joining the pool on shutdown — but the core idea (one thread runs at a time, woken in deterministic order) is unchanged.

The runtime spawns a fixed number of extra "blocking pool" threads for code that can yield execution to other threads. Normally no code needs to yield — we poll() ready futures one at a time and they always make progress immediately, because there is no other thread that could be holding a lock they require. spawn_blocking() sends tasks to a blocking_queue; the run loop enforces that only one thing happens at a time, and wakes the blocking threads in deterministic order:

// main run loop
pub fn block_on<F: Future>(&self, future: F) -> F::Output {
    let mut task = self.spawn_on_main_task(future);
    let waker = futures::task::noop_waker();
    let mut cx = Context::from_waker(&waker);
    loop {
        self.run_all_ready();
        if let Poll::Ready(val) = Pin::new(&mut task).poll(&mut cx) {
            return val;
        }
        self.quantum.wake_all();
        let going = self.time.advance_to_next_event();
        assert!(going, "no events, the task will block forever");
    }
}

// blocking pool thread loop (a fixed number are created at startup)
run_blocking_thread(thread_id: u32) {
    self.quantum.start(thread_id);
    loop {
        // Blocking threads yield before starting each task; tasks themselves can also
        // explicitly yield, generally while waiting on a notification. Rather than
        // mocking every tokio channel, a special-purpose one calls:
        //   #[cfg(msim)] ThreadQuantum::with(|q| q.yield());
        self.quantum.yield();
        if let Ok(callable) = self.blocking_queue.recv_random(&self.rand) {
            callable();
        }
    }
}

struct ThreadQuantum {
    max_thread: u32,
    active_thread: Mutex<Option<u32>>,
    cv: CondVar,
}

impl ThreadQuantum {
    // a single static ThreadQuantum lets threads call yield() from anywhere
    fn with(cb: Fn(&ThreadQuantum));

    fn start(&self, thread_id: u32) {
        *self.active_thread.lock() = Some(thread_id);
    }

    fn yield(&self) {
        let l = self.active_thread.lock();
        let cur_thread = l.expect("current thread should be set");
        *l = None;
        // wait until we are woken
        self.cv.wait_while(l, |active| active != Some(cur_thread));
    }

    fn wake_all(&self) {
        for i in 0..self.max_thread {
            let l = self.active_thread.lock();
            *l = Some(i);
            self.cv.notify_all();                   // only the assigned thread wakes
            self.cv.wait_while(l, |a| a.is_some()); // wait until it calls yield()
        }
    }
}

Notes

• Because a simulation now spans multiple threads, per-test state that consumers kept in thread_local!s (fail-point registries, injection overrides) must become process-global, which requires at most one simulation per process. Accordingly MSIM_TEST_NUM and MSIM_TEST_CHECK_DETERMINISM are retired in the sim_test macro (both now fail loudly; the check_determinism attribute is accepted and ignored). Replacements: an external seed-search runner (one process per seed) and running a test twice in separate processes comparing log output.
• Default pool size is 32 (MSIM_BLOCKING_THREADS); it must exceed the max number of concurrently-parked blocking tasks in the workload.

Test plan

• RUSTFLAGS="--cfg msim" cargo test -p msim --lib — 35 tests incl. 10 new pool tests: result delivery, determinism across a fixed seed, yield-based cross-task waits, nested spawn_blocking, panic propagation, kill-unwinds-parked-task (Drop impls run), clean shutdown with a parked task, kill_current_node restart, and deadlock detection.
• Validated end-to-end against Sui via LOCAL_MSIM_PATH: sui-e2e-tests checkpoint/reconfiguration/address-balance suites (78 tests) and test_simulated_load_reconfig_with_crashes_and_delays pass, and repeated concurrent runs produce byte-identical simulation logs (determinism preserved).