Mob islands: villagers draw their own random under the parallel tick, and ordered waits sleep

A mob's first draw on the ordered level random made every lane wait for all earlier mobs, and in a farm-shaped
level all of those first draws were villager brains. With HORIZON_PARALLEL_TICK on, a villager's brain now draws
from a private random seeded from its own entity random (serial and grouped alike, so grouped still equals serial).
Ordered waits spin, yield, then sleep on a condvar. Adds pm_shaped_bench.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>

crates/horizon-client/src/game/creatures/mob_islands.rs

@@ -72,6 +72,46 @@ thread_local! {
     static FAIL_TURN: Cell<Option<usize>> = const { Cell::new(None) };
 }
 
+/// TEMP-DIAG: what each lane did in a grouped tick, and what the server thread did around them.
+#[cfg(test)]
+pub(super) mod diag {
+    use std::sync::Mutex;
+    #[derive(Clone, Debug, Default)]
+    pub struct Lane {
+        pub server: bool,
+        pub mobs: usize,
+        pub busy_ns: u64,
+        pub cpu_ns: u64,
+        pub waited_ns: u64,
+    }
+    #[derive(Clone, Debug, Default)]
+    pub struct Tick {
+        pub setup_ns: u64,
+        pub threads_ns: u64,
+        pub join_wait_ns: u64,
+        pub merge_ns: u64,
+        pub groups: Vec<usize>,
+        pub lanes: Vec<Lane>,
+    }
+    pub static LAST: Mutex<Tick> = Mutex::new(Tick {
+        setup_ns: 0,
+        threads_ns: 0,
+        join_wait_ns: 0,
+        merge_ns: 0,
+        groups: Vec::new(),
+        lanes: Vec::new(),
+    });
+    /// The last plan's (ticking mobs, island weights heaviest first).
+    pub static ISLANDS: Mutex<(usize, Vec<usize>)> = Mutex::new((0, Vec::new()));
+    pub static PLAN_NS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
+    pub fn thread_cpu_ns() -> u64 {
+        std::fs::read_to_string("/proc/thread-self/schedstat")
+            .ok()
+            .and_then(|s| s.split_whitespace().next().and_then(|v| v.parse().ok()))
+            .unwrap_or(0)
+    }
+}
+
 #[cfg(test)]
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum FailAt {
@@ -183,9 +223,17 @@ impl Islands<'_> {
             .chain(start.guests.iter().map(|g| &g.1))
             .map(|p| (p.uuid, p.position))
             .collect();
-        horizon_client::spans::time("islands_plan", || {
+        #[cfg(test)]
+        let began = std::time::Instant::now();
+        let planned = horizon_client::spans::time("islands_plan", || {
             islands(&reaches(mobs, snapshot, turns, &players)?, lanes)
-        })
+        });
+        #[cfg(test)]
+        diag::PLAN_NS.store(
+            began.elapsed().as_nanos() as u64,
+            std::sync::atomic::Ordering::Relaxed,
+        );
+        planned
     }
 
     /// The loop's turns in the plan's groups, each on its own thread, put back together as the
@@ -208,6 +256,14 @@ impl Islands<'_> {
             turns.cached_difficulty = None;
             (serial, seeds)
         });
+        #[cfg(test)]
+        let setup_began = std::time::Instant::now();
+        #[cfg(test)]
+        {
+            let mut d = diag::LAST.lock().unwrap();
+            *d = diag::Tick::default();
+            d.groups = plan.groups.iter().map(Vec::len).collect();
+        }
         let n = mobs.len();
         let lanes = plan.groups.len();
         let draws = OrderedDraws::new(&start.random, lanes);
@@ -252,6 +308,10 @@ impl Islands<'_> {
         #[cfg(test)]
         RAN.with(|r| r.set(lanes));
         let mut board = Board::new(snapshot, &groups, Arc::clone(&draws));
+        #[cfg(test)]
+        {
+            diag::LAST.lock().unwrap().setup_ns = setup_began.elapsed().as_nanos() as u64;
+        }
         self.run_threads(
             &mut groups,
             &board,
@@ -259,6 +319,8 @@ impl Islands<'_> {
             shared,
             serial.as_ref().map(|s| &s.1),
         );
+        #[cfg(test)]
+        let merge_began = std::time::Instant::now();
         // Each mob's entries, in list order: the turns' ranges in its group.
         let mut place = vec![(0u32, 0u32); n];
         for (gi, g) in groups.iter().enumerate() {
@@ -326,6 +388,10 @@ impl Islands<'_> {
         (merged.keep, merged.pushes, merged.snapshot) = (keep, pushes, frozen);
         (merged.dead, merged.corpses, merged.gone) = (dead, corpses, gone);
         merged.guest_events = guest_events;
+        #[cfg(test)]
+        {
+            diag::LAST.lock().unwrap().merge_ns = merge_began.elapsed().as_nanos() as u64;
+        }
         if let Some((serial, seeds)) = serial {
             let diff = difference(&serial, &merged)
                 .or_else(|| (!seeds.complete()).then(|| "new RNG allocation counts differ".into()));
@@ -378,6 +444,11 @@ impl Islands<'_> {
                 difficulty,
                 cached_difficulty: None,
             };
+            #[cfg(test)]
+            let (began, cpu0) = {
+                horizon_core::random::take_waited();
+                (std::time::Instant::now(), diag::thread_cpu_ns())
+            };
             if let Some(seeds) = seeds {
                 horizon_client::random_source::replay_seeds(Arc::clone(seeds), || {
                     g.run(&mut turns, shared, Some(&foreign));
@@ -385,6 +456,14 @@ impl Islands<'_> {
             } else {
                 g.run(&mut turns, shared, Some(&foreign));
             }
+            #[cfg(test)]
+            diag::LAST.lock().unwrap().lanes.push(diag::Lane {
+                server: g.global[0] == 0,
+                mobs: g.mobs.len(),
+                busy_ns: began.elapsed().as_nanos() as u64,
+                cpu_ns: diag::thread_cpu_ns() - cpu0,
+                waited_ns: horizon_core::random::take_waited().as_nanos() as u64,
+            });
         };
         let tickings: Vec<_> = groups.iter().map(|_| (self.ticking)()).collect();
         let mut parts = groups.iter_mut().zip(tickings);
@@ -444,8 +523,12 @@ impl Islands<'_> {
                     .expect("a mob island's thread starts");
                 handles.push(handle);
             }
+            #[cfg(test)]
+            let own_began = std::time::Instant::now();
             run(first, first_ticking);
             drop(first_done);
+            #[cfg(test)]
+            let own_done = std::time::Instant::now();
             let mut panic = None;
             for handle in handles {
                 match handle.join() {
@@ -454,6 +537,12 @@ impl Islands<'_> {
                     Err(_) => {}
                 }
             }
+            #[cfg(test)]
+            {
+                let mut d = diag::LAST.lock().unwrap();
+                d.join_wait_ns = own_done.elapsed().as_nanos() as u64;
+                d.threads_ns = own_began.elapsed().as_nanos() as u64;
+            }
             if let Some(payload) = panic {
                 std::panic::resume_unwind(payload);
             }
@@ -987,6 +1076,12 @@ fn islands(reaches: &[Reach], lanes: usize) -> Option<Plan> {
         islands[k].1.push(i);
         islands[k].0 += usize::from(r.ticking);
     }
+    #[cfg(test)]
+    {
+        let mut w: Vec<usize> = islands.iter().map(|i| i.0).collect();
+        w.sort_unstable_by(|a, b| b.cmp(a));
+        *diag::ISLANDS.lock().unwrap() = (ticking, w);
+    }
     let largest = islands.iter().map(|i| i.0).max().unwrap_or(0);
     if largest * 4 > ticking * 3 {
         return None;
@@ -2072,6 +2167,31 @@ mod tests {
         }
     }
 
+    /// With the villagers on their own random (`parallel::villagers_own_random`), villagers in one
+    /// herd tick in groups beside the other herds and leave the mobs and the level random as the
+    /// serial loop does, since the serial loop draws the same way while the switch is on.
+    #[test]
+    #[ignore = "needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn villagers_on_their_own_random_tick_in_groups_as_serial() {
+        struct Reset;
+        impl Drop for Reset {
+            fn drop(&mut self) {
+                horizon_client::parallel::OWN_RANDOM_TEST
+                    .store(false, std::sync::atomic::Ordering::Relaxed);
+            }
+        }
+        let _reset = Reset;
+        horizon_client::parallel::OWN_RANDOM_TEST.store(true, std::sync::atomic::Ordering::Relaxed);
+        let mut game = separated_herds_game();
+        for k in 0..12 {
+            let x = 4.0 + f64::from(k % 6) * 2.0;
+            let z = 14.0 + f64::from(k / 6) * 2.0;
+            assert!(game.run_command(&format!("summon villager {x:.1} 66 {z:.1}")));
+        }
+        let groups = grouped_as_serial(&mut game, 30, "villagers", |_| {});
+        assert_eq!(groups, groups_for(3));
+    }
+
     /// The raiders of one raid order their hurts across it (`raids.rs`): one in each of two herds
     /// joins them, and the herd between ticks in a group of its own.
     #[test]
@@ -2194,4 +2314,223 @@ mod tests {
         assert_ne!(of[0], of[200]);
         assert_eq!(of[200], of[300], "both small herds fill the lighter group");
     }
+
+    /// A level shaped like a busy survival server: many small chicken farms, bee and villager
+    /// clusters, sulfur cubes, and monsters scattered between them, all far apart. Prints, for
+    /// each lane setting (alternating tick by tick), the mob phase's wall time, CPU time and the
+    /// server thread's wait, and where the lanes spent their time.
+    ///
+    /// `PM_LANES=1,4 PM_TICKS=300 PM_WARM=60 PM_FARMS=40 PM_CHICKENS=18`, run with
+    /// `--ignored --exact --nocapture --test-threads=1`.
+    #[test]
+    #[ignore = "a measurement; needs prepared resources in HORIZON_TEST_RESOURCES"]
+    fn pm_shaped_bench() {
+        let var = |k: &str, d: usize| {
+            std::env::var(k)
+                .ok()
+                .and_then(|v| v.parse().ok())
+                .unwrap_or(d)
+        };
+        let (ticks, warm) = (var("PM_TICKS", 300), var("PM_WARM", 60));
+        let lane_settings: Vec<usize> = std::env::var("PM_LANES")
+            .unwrap_or_else(|_| "1,4".into())
+            .split(',')
+            .filter_map(|v| v.parse().ok())
+            .collect();
+        horizon_client::parallel::OWN_RANDOM_TEST
+            .store(var("PM_OWN", 0) == 1, std::sync::atomic::Ordering::Relaxed);
+        let farms = var("PM_FARMS", 40);
+        let per_farm = var("PM_CHICKENS", 18);
+        let mut game = super::super::super::native_machines::tests::server_game();
+        game.build_registries();
+        game.set_allow_commands(true);
+        let span = 800;
+        let mid = f64::from(span) / 2.0;
+        game.player = Some(
+            horizon_client::motion::PlayerMotion::new(game.motion, Vec3::new(mid, 65.0, mid))
+                .unwrap(),
+        );
+        for x in -3..=(span / 16 + 3) {
+            for z in -3..=(span / 16 + 3) {
+                game.world
+                    .chunks
+                    .insert((x, z), Arc::new(horizon_client::chunk::Chunk::empty(x, z)));
+            }
+        }
+        game.update_chunk_tickets();
+        let _ = game.run_command("difficulty normal");
+        let stone = game
+            .pack
+            .assets
+            .registry
+            .default_state("minecraft:stone")
+            .unwrap();
+        // Cells of 80 blocks, shuffled by a fixed generator; each cluster takes one.
+        let mut state = 0x2545_f491_4f6c_dd1d_u64;
+        let mut next = move || {
+            state ^= state << 13;
+            state ^= state >> 7;
+            state ^= state << 17;
+            state
+        };
+        let mut cells: Vec<(i32, i32)> = (0..9)
+            .flat_map(|x| (0..9).map(move |z| (x, z)))
+            .map(|(x, z)| (40 + x * 80, 40 + z * 80))
+            .collect();
+        for i in (1..cells.len()).rev() {
+            cells.swap(i, (next() % (i as u64 + 1)) as usize);
+        }
+        let mut cluster = |game: &mut Game<'static>, kind: &str, count: usize, extra: &str| {
+            let (cx, cz) = cells.pop().expect("a free cell");
+            for x in cx..cx + 12 {
+                for z in cz..cz + 12 {
+                    game.world.set(x, 64, z, stone).unwrap();
+                }
+            }
+            for k in 0..count {
+                let x = f64::from(cx) + 1.0 + (k % 10) as f64 * 1.0;
+                let z = f64::from(cz) + 1.0 + (k / 10) as f64 * 1.0;
+                let c = format!("summon {kind} {x:.2} 66 {z:.2} {extra}");
+                assert!(game.run_command(c.trim()), "{c}");
+            }
+        };
+        for _ in 0..farms {
+            cluster(&mut game, "chicken", per_farm, "");
+        }
+        for _ in 0..var("PM_BEE_GROUPS", 8) {
+            cluster(&mut game, "bee", var("PM_BEES", 95), "");
+        }
+        for _ in 0..var("PM_VILLAGES", 3) {
+            cluster(&mut game, "villager", var("PM_VILLAGERS", 90), "");
+        }
+        for _ in 0..var("PM_CUBE_GROUPS", 6) {
+            cluster(&mut game, "sulfur_cube", var("PM_CUBES", 24), "");
+        }
+        // Monsters scattered anywhere, one by one.
+        for k in 0..var("PM_MONSTERS", 120) {
+            let kind = ["husk", "husk", "skeleton", "spider", "creeper"][k % 5];
+            let x = 20.0 + (next() % (span as u64 - 40)) as f64;
+            let z = 20.0 + (next() % (span as u64 - 40)) as f64;
+            let (bx, bz) = (x as i32, z as i32);
+            for dx in 0..3 {
+                for dz in 0..3 {
+                    game.world.set(bx + dx, 64, bz + dz, stone).unwrap();
+                }
+            }
+            if !game.run_command(&format!("summon {kind} {x:.2} 66 {z:.2}")) {
+                eprintln!("summon {kind} {x:.2} 66 {z:.2} failed");
+            }
+        }
+        let _ = game.run_command("difficulty normal");
+        let mut by_kind = std::collections::BTreeMap::<String, usize>::new();
+        for m in &game.mobs {
+            *by_kind.entry(format!("{:?}", m.kind.species)).or_default() += 1;
+        }
+        eprintln!("mobs: {} {by_kind:?}", game.mobs.len());
+        struct Reset;
+        impl Drop for Reset {
+            fn drop(&mut self) {
+                LANES.with(|l| l.set(None));
+                horizon_client::parallel::OWN_RANDOM_TEST
+                    .store(false, std::sync::atomic::Ordering::Relaxed);
+            }
+        }
+        let _reset = Reset;
+        #[derive(Default)]
+        struct Sum {
+            n: usize,
+            grouped: usize,
+            wall: Vec<f64>,
+            cpu: f64,
+            join: f64,
+            plan: f64,
+            setup: f64,
+            merge: f64,
+            lanes: Vec<(f64, f64, f64, f64, f64)>,
+            groups: Vec<usize>,
+        }
+        let mut sums: Vec<Sum> = lane_settings.iter().map(|_| Sum::default()).collect();
+        for t in 0..warm + ticks {
+            let which = t % lane_settings.len();
+            LANES.with(|l| l.set(Some(lane_settings[which])));
+            RAN.with(|r| r.set(0));
+            diag::LAST.lock().unwrap().lanes.clear();
+            let cpu0 = diag::thread_cpu_ns();
+            let began = std::time::Instant::now();
+            game.tick_creature_entities().unwrap();
+            let wall = began.elapsed().as_secs_f64() * 1e3;
+            let mut cpu = (diag::thread_cpu_ns() - cpu0) as f64 / 1e6;
+            if t < warm {
+                continue;
+            }
+            let s = &mut sums[which];
+            s.n += 1;
+            s.wall.push(wall);
+            s.plan += diag::PLAN_NS.load(std::sync::atomic::Ordering::Relaxed) as f64 / 1e6;
+            let ran = RAN.with(Cell::get);
+            if ran > 0 {
+                s.grouped += 1;
+                let d = diag::LAST.lock().unwrap().clone();
+                s.join += d.join_wait_ns as f64 / 1e6;
+                s.setup += d.setup_ns as f64 / 1e6;
+                s.merge += d.merge_ns as f64 / 1e6;
+                for l in &d.lanes {
+                    if !l.server {
+                        cpu += l.cpu_ns as f64 / 1e6;
+                    }
+                }
+                let mut ls: Vec<_> = d.lanes.clone();
+                ls.sort_by_key(|l| (!l.server, std::cmp::Reverse(l.mobs)));
+                if s.lanes.len() < ls.len() {
+                    s.lanes.resize(ls.len(), (0.0, 0.0, 0.0, 0.0, 0.0));
+                }
+                for (acc, l) in s.lanes.iter_mut().zip(&ls) {
+                    acc.0 += l.busy_ns as f64 / 1e6;
+                    acc.1 += l.cpu_ns as f64 / 1e6;
+                    acc.2 += l.waited_ns as f64 / 1e6;
+                    acc.3 += l.mobs as f64;
+                    acc.4 += 1.0;
+                }
+                s.groups = d.groups.clone();
+            }
+            s.cpu += cpu;
+        }
+        let (ticking, weights) = diag::ISLANDS.lock().unwrap().clone();
+        eprintln!(
+            "ticking {ticking}, {} islands, heaviest {:?}",
+            weights.len(),
+            &weights[..weights.len().min(12)]
+        );
+        for (s, lanes) in sums.iter_mut().zip(&lane_settings) {
+            s.wall.sort_by(|a, b| a.total_cmp(b));
+            let q = |p: f64| s.wall[((s.wall.len() - 1) as f64 * p) as usize];
+            let n = s.n as f64;
+            let g = s.grouped.max(1) as f64;
+            eprintln!(
+                "lanes {lanes}: wall mean {:.2} p50 {:.2} p95 {:.2} | cpu {:.2} | grouped {}/{} | plan {:.2} setup {:.2} server-join-wait {:.2} merge {:.2} (per grouped tick)",
+                s.wall.iter().sum::<f64>() / n,
+                q(0.5),
+                q(0.95),
+                s.cpu / n,
+                s.grouped,
+                s.n,
+                s.plan / n,
+                s.setup / g,
+                s.join / g,
+                s.merge / g
+            );
+            eprintln!("  group sizes (last grouped tick) {:?}", s.groups);
+            for (i, l) in s.lanes.iter().enumerate() {
+                let k = l.4.max(1.0);
+                eprintln!(
+                    "  lane {i}{}: mobs {:.0} busy {:.2} ms, cpu {:.2} ms, waited on draws {:.2} ms",
+                    if i == 0 { " (server)" } else { "" },
+                    l.3 / k,
+                    l.0 / k,
+                    l.1 / k,
+                    l.2 / k
+                );
+            }
+        }
+    }
 }

crates/horizon-client/src/mobs/ai.rs

@@ -1956,7 +1956,7 @@ fn server_ai_step(
     }
     if m.species() == Species::Villager {
         // Villager.customServerAiStep: the brain, then the raid sound roll.
-        super::villager_brain::tick(m, w, level_random, events);
+        super::villager::tick_brain(m, w, level_random, events);
         if let Some(player) = super::villager::ext_mut(m).last_traded_player.take() {
             events.push(MobEvent::VillagerTrade { mob: m.id, player });
         }

crates/horizon-client/src/mobs/villager.rs

@@ -1079,6 +1079,22 @@ pub fn read_nbt(m: &mut Mob, tag: &horizon_world::nbt::Tag) {
     }
 }
 
+/// The villager's brain tick. While the parallel mob tick is on it draws from the villager's own random, so
+/// that no other mob's turn is waited for (`parallel::villagers_own_random`); the level's random otherwise.
+pub fn tick_brain(
+    m: &mut Mob,
+    w: &MobWorld,
+    level_random: &mut horizon_core::random::LegacyRandom,
+    events: &mut Vec<MobEvent>,
+) {
+    if crate::parallel::villagers_own_random() {
+        let mut own = horizon_core::random::LegacyRandom::new(m.random.next_long());
+        super::villager_brain::tick(m, w, &mut own, events);
+    } else {
+        super::villager_brain::tick(m, w, level_random, events);
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/horizon-client/src/parallel.rs

@@ -16,6 +16,24 @@ pub fn parallel_tick() -> u8 {
     })
 }
 
+/// Whether a villager's brain draws from the villager's own random instead of the level's.
+///
+/// The level random is one stream, so a draw has to wait for every mob before it (`OrderedDraws`),
+/// and a villager's brain draws on nearly every turn (behaviour durations, one-in-N rolls).
+/// With the parallel tick on, those draws come from a stream seeded from the villager's own, so
+/// no villager waits on another mob and the lanes run apart. That is the one place the parallel
+/// tick departs from vanilla's draw order: the villagers' rolls have the same odds but not the
+/// same values a vanilla level would give them. The serial loop does the same while the parallel
+/// tick is on, so which path a tick takes never changes the result.
+pub fn villagers_own_random() -> bool {
+    OWN_RANDOM_TEST.load(Ordering::Relaxed) || parallel_tick() > 0
+}
+
+/// A test's switch for [`villagers_own_random`], process-wide because the lanes are threads.
+#[doc(hidden)]
+pub static OWN_RANDOM_TEST: std::sync::atomic::AtomicBool =
+    std::sync::atomic::AtomicBool::new(false);
+
 /// How many threads a phase may share its work among: `horizon-threads`' count (at most 8) while
 /// the parallel tick is on, else 1.
 pub fn lanes() -> usize {

crates/horizon-core/src/random.rs

@@ -5,7 +5,7 @@ mod owner;
 mod positional;
 mod worldgen;
 pub use hash::{seed_from_hash, seed_from_hash_utf16};
-pub use ordered::{LaneDone, OrderedDraws};
+pub use ordered::{LaneDone, OrderedDraws, take_waited};
 pub use owner::{EntityPositional, EntityRandom, LevelPositional, LevelRandom};
 pub use positional::{
     LegacyPositionalRandom, PositionalRandom, XoroshiroPositionalRandom, java_string_hash,

crates/horizon-core/src/random/ordered.rs

@@ -10,10 +10,31 @@
 //! lane has its own thread. An item that never draws or reseeds never waits.
 use super::LegacyRandom;
 use std::sync::{
-    Arc, Mutex, PoisonError,
+    Arc, Condvar, Mutex, PoisonError,
     atomic::{AtomicUsize, Ordering},
 };
 
+thread_local! {
+    /// Nanoseconds this thread has spent waiting for other lanes' items to finish.
+    static WAITED: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
+}
+
+/// The time this thread has spent waiting on other lanes since it last asked (for the first draw
+/// of an item, a lookup of a mob in another group, a new seed), and starts the count again.
+pub fn take_waited() -> std::time::Duration {
+    std::time::Duration::from_nanos(WAITED.with(|w| w.replace(0)))
+}
+
+/// Adds the time since it was made to this thread's wait count when dropped; made only once a
+/// wait is certain, so a wait that isn't needed costs nothing.
+struct Waiting(std::time::Instant);
+impl Drop for Waiting {
+    fn drop(&mut self) {
+        let ns = self.0.elapsed().as_nanos() as u64;
+        WAITED.with(|w| w.set(w.get().saturating_add(ns)));
+    }
+}
+
 /// The shared stream and where each lane is.
 #[derive(Debug)]
 pub struct OrderedDraws {
@@ -21,8 +42,17 @@ pub struct OrderedDraws {
     state: Mutex<(u64, Option<f64>)>,
     /// The item each lane is on; `usize::MAX` once it has finished.
     lanes: Box<[AtomicUsize]>,
+    /// Lanes asleep in `wait_until`.
+    sleepers: AtomicUsize,
+    parked: Mutex<()>,
+    wake_up: Condvar,
 }
 
+/// Spins a wait makes before it yields its core.
+const SPINS: u32 = 200;
+/// How long a wait yields its core before it sleeps.
+const YIELD_FOR: std::time::Duration = std::time::Duration::from_micros(20);
+
 /// A lane's hold on the shared stream, inside its [`LegacyRandom`].
 #[derive(Debug)]
 pub(super) struct Gate {
@@ -38,6 +68,9 @@ impl OrderedDraws {
         Arc::new(Self {
             state: Mutex::new((random.state, random.gaussian)),
             lanes: (0..lanes).map(|_| AtomicUsize::new(0)).collect(),
+            sleepers: AtomicUsize::new(0),
+            parked: Mutex::new(()),
+            wake_up: Condvar::new(),
         })
     }
     /// The random lane `lane` draws from: its draws wait for their turn.
@@ -60,34 +93,57 @@ impl OrderedDraws {
     }
     /// Wait until lane `lane` has finished item `item` (it is on a later one, or done).
     pub fn wait_past(&self, lane: usize, item: usize) {
-        let mut spins = 0u32;
-        while self.lanes[lane].load(Ordering::Acquire) <= item {
-            spins += 1;
-            if spins < 64 {
-                std::hint::spin_loop();
-            } else {
-                std::thread::yield_now();
-            }
-        }
+        self.wait_until(|| self.lanes[lane].load(Ordering::Acquire) > item);
     }
     fn wait_for_turn(&self, lane: usize, item: usize) {
-        let mut spins = 0u32;
-        loop {
-            let ready = self
-                .lanes
+        self.wait_until(|| {
+            self.lanes
                 .iter()
                 .enumerate()
-                .all(|(k, at)| k == lane || at.load(Ordering::Acquire) > item);
-            if ready {
+                .all(|(k, at)| k == lane || at.load(Ordering::Acquire) > item)
+        });
+    }
+    /// Returns once `ready` holds. A wait is mostly short, so it spins first. The lane waited on
+    /// may be a thread the machine has not given a core (the helpers compete with the server's
+    /// other threads), and a waiter that keeps its core then holds that lane up: after a moment
+    /// the waiter sleeps until a lane moves on ([`Self::wake`]).
+    fn wait_until(&self, ready: impl Fn() -> bool) {
+        if ready() {
+            return;
+        }
+        let _waiting = Waiting(std::time::Instant::now());
+        for _ in 0..SPINS {
+            std::hint::spin_loop();
+            if ready() {
                 return;
             }
-            spins += 1;
-            if spins < 64 {
-                std::hint::spin_loop();
-            } else {
-                std::thread::yield_now();
+        }
+        let until = std::time::Instant::now() + YIELD_FOR;
+        while std::time::Instant::now() < until {
+            std::thread::yield_now();
+            if ready() {
+                return;
             }
         }
+        self.sleepers.fetch_add(1, Ordering::SeqCst);
+        let mut guard = self.parked.lock().unwrap_or_else(PoisonError::into_inner);
+        while !ready() {
+            // The timeout only guards against a wake that was missed; `wake` takes this lock.
+            guard = self
+                .wake_up
+                .wait_timeout(guard, std::time::Duration::from_millis(1))
+                .unwrap_or_else(PoisonError::into_inner)
+                .0;
+        }
+        drop(guard);
+        self.sleepers.fetch_sub(1, Ordering::SeqCst);
+    }
+    /// A lane has moved on (or finished): wakes the lanes that sleep. Cheap while none does.
+    fn wake(&self) {
+        if self.sleepers.load(Ordering::SeqCst) > 0 {
+            drop(self.parked.lock().unwrap_or_else(PoisonError::into_inner));
+            self.wake_up.notify_all();
+        }
     }
 }
 
@@ -113,7 +169,8 @@ impl LegacyRandom {
     pub fn begin_item(&mut self, item: usize) {
         if let Some(gate) = self.gate.as_deref_mut() {
             gate.item = item;
-            gate.draws.lanes[gate.lane].store(item, Ordering::Release);
+            gate.draws.lanes[gate.lane].store(item, Ordering::SeqCst);
+            gate.draws.wake();
         }
     }
     /// The item's draws are done: the shared stream goes on from where it left it.
@@ -133,7 +190,8 @@ impl LegacyRandom {
     pub fn finish_lane(&mut self) {
         self.end_item();
         if let Some(gate) = self.gate.as_deref_mut() {
-            gate.draws.lanes[gate.lane].store(usize::MAX, Ordering::Release);
+            gate.draws.lanes[gate.lane].store(usize::MAX, Ordering::SeqCst);
+            gate.draws.wake();
         }
     }
     /// A gated random's [`LaneDone`], for its lane's thread to hold while it runs; None for an
@@ -157,7 +215,8 @@ pub struct LaneDone {
 
 impl Drop for LaneDone {
     fn drop(&mut self) {
-        self.draws.lanes[self.lane].store(usize::MAX, Ordering::Release);
+        self.draws.lanes[self.lane].store(usize::MAX, Ordering::SeqCst);
+        self.draws.wake();
     }
 }
 
@@ -182,6 +241,44 @@ mod tests {
         }
     }
 
+    /// A lane that waits long enough to sleep (the lane before it is slow) still draws after it
+    /// and gets the serial stream, and wakes when that lane moves on or finishes.
+    #[test]
+    fn a_long_wait_sleeps_and_wakes_in_order() {
+        let mut serial = LegacyRandom::new(5);
+        let expected = [serial.next_int() as u64, serial.next_int() as u64];
+        let got = bounded(|| {
+            let shared = OrderedDraws::new(&LegacyRandom::new(5), 2);
+            std::thread::scope(|s| {
+                let slow = {
+                    let shared = Arc::clone(&shared);
+                    s.spawn(move || {
+                        let mut random = shared.lane(0);
+                        random.begin_item(0);
+                        std::thread::sleep(std::time::Duration::from_millis(50));
+                        let first = random.next_int() as u64;
+                        random.end_item();
+                        random.finish_lane();
+                        first
+                    })
+                };
+                let fast = {
+                    let shared = Arc::clone(&shared);
+                    s.spawn(move || {
+                        let mut random = shared.lane(1);
+                        random.begin_item(1);
+                        let second = random.next_int() as u64;
+                        random.end_item();
+                        random.finish_lane();
+                        second
+                    })
+                };
+                [slow.join().unwrap(), fast.join().unwrap()]
+            })
+        });
+        assert_eq!(got, expected);
+    }
+
     #[test]
     fn reseeding_lanes_preserve_the_serial_stream() {
         let mut initial = LegacyRandom::new(42);

docs/mob-islands.md

@@ -115,6 +115,25 @@ pending vibration as mob ids, read and angered at any distance: every id in its
 The suspects it keeps by UUID resolve only to the host's player during the loop, serially as in
 groups (`LoopLookup`).
 
+## Villagers draw their own random
+
+The ordered level random (above) makes a mob's first draw wait for every mob before it in the
+list, on every group. A villager's brain draws on nearly every turn (behaviour durations in
+`Timed::begin`, one-in-N rolls such as `nextInt(20) != 0`), and the result of most rolls decides
+what the villager does in the same turn, so those draws cannot be put off. With
+`HORIZON_PARALLEL_TICK` on, the brain therefore draws from a `LegacyRandom` seeded from one
+`next_long` of the villager's own random (`parallel::villagers_own_random`, `mobs/ai.rs`), not
+from the level random. No villager waits for another mob's draws, and a measured world whose
+only level-random draws were villagers' had none left to wait for.
+
+This is a deliberate departure from vanilla's draw order, accepted for the parallel tick: the
+odds of every roll are vanilla's, the values are not the ones a vanilla level would give, and the
+level random is no longer advanced by villagers. The serial loop draws the same way while the
+switch is on, so the grouped result still equals the serial result and `verify` compares like
+with like. With the parallel tick off nothing changes, and the villager parity harnesses (which
+pin vanilla's exact draws) run with it off. A test turns the switch on with
+`parallel::OWN_RANDOM_TEST`.
+
 ## Native mods and the profiler
 
 What a mod can run in a mob's turn: `native_living_change_target`, `native_living_heal`,