alex/beta-minecraft-rust

Minecraft Beta 1.7.3 Rewrite
root / crates / server / src / chunks.rs

use std::sync::mpsc::{self, Receiver, Sender};
use std::time::{Duration, Instant};

use common::math;
use common::protocol::Packet;
use common::world::chunk::{HEIGHT, WIDTH};
use common::world::{ChunkPos, TileEntity};
use rustc_hash::{FxHashMap, FxHashSet};

use crate::level::{Level, Prefetched, tile_position};
use crate::player::ServerPlayer;

/// Time per tick spent loading and populating chunks.
const LOAD_BUDGET: Duration = Duration::from_millis(30);

/// Fetches in flight, deep enough that chunks population spills into are usually ready.
fn max_in_flight() -> usize {
    rayon::current_num_threads() * 8
}

/// Loads chunks around players in a fixed nearest-first order, so population is reproducible,
/// while worker threads read or generate them ahead. Streams chunks to players once their
/// neighborhood is loaded, then keeps them up to date with block changes.
pub struct ChunkLoader {
    view: ViewArea,
    /// Offsets within the load area, nearest first.
    offsets: Vec<(i32, i32)>,
    /// Offsets within the view area, nearest first.
    sent_offsets: Vec<(i32, i32)>,
    /// Per player center, how many of the nearest offsets are loaded. They stay loaded while
    /// the center does, as unloading keeps a wider margin.
    loaded: FxHashMap<ChunkPos, usize>,
    /// The centers and loaded chunk count after the last unload sweep, while saving.
    swept: Option<(Vec<ChunkPos>, usize)>,
    generating: FxHashSet<ChunkPos>,
    /// In-flight fetches of chunks unloaded since, whose disk data may be outdated.
    stale: FxHashSet<ChunkPos>,
    /// The player served first this tick, rotating so no player starves the others.
    turn: usize,
    finished_sender: Sender<(ChunkPos, Prefetched)>,
    finished: Receiver<(ChunkPos, Prefetched)>,
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum ViewShape {
    #[default]
    Square,
    Disk,
}

impl ViewShape {
    /// Whether a chunk offset lies within `radius`. Disks use r² + r, about (r + ½)², so no
    /// single chunk sticks out on the axes.
    #[inline]
    pub fn contains(self, dx: i32, dz: i32, radius: i32) -> bool {
        match self {
            Self::Square => dx.abs().max(dz.abs()) <= radius,
            Self::Disk => dx * dx + dz * dz <= radius * radius + radius,
        }
    }
}

/// The chunks around a player that are sent to its client.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ViewArea {
    /// In chunks.
    pub radius: i32,
    pub shape: ViewShape,
}

impl ViewArea {
    /// The area loaded around a player: the view area with the neighbors of every sent chunk,
    /// which must be populated first.
    fn load_radius(self) -> i32 {
        let ring = match self.shape {
            ViewShape::Square => 1,
            ViewShape::Disk => 2,
        };
        (self.radius + ring).max(10)
    }

    /// Offsets within `radius`, nearest first.
    fn offsets(self, radius: i32) -> Vec<(i32, i32)> {
        let mut offsets: Vec<_> = (-radius..=radius)
            .flat_map(|dx| (-radius..=radius).map(move |dz| (dx, dz)))
            .filter(|&(dx, dz)| self.shape.contains(dx, dz, radius))
            .collect();
        offsets.sort_by_key(|&(dx, dz)| dx * dx + dz * dz);
        offsets
    }
}

/// How far streaming to a player got: every chunk before `sent` in the nearest-first order
/// around the area's center was sent.
#[derive(Clone, Copy, Debug, Default)]
pub struct StreamCursor {
    area: Option<(ChunkPos, ViewArea)>,
    sent: usize,
}

/// The distance in chunks along the farther axis.
fn distance(a: ChunkPos, b: ChunkPos) -> i32 {
    (a.x - b.x).abs().max((a.z - b.z).abs())
}

impl ChunkLoader {
    pub fn new(view: ViewArea) -> Self {
        let (finished_sender, finished) = mpsc::channel();
        Self {
            view,
            offsets: view.offsets(view.load_radius()),
            sent_offsets: view.offsets(view.radius),
            loaded: FxHashMap::default(),
            swept: None,
            generating: FxHashSet::default(),
            stale: FxHashSet::default(),
            turn: 0,
            finished_sender,
            finished,
        }
    }

    /// `PlayerManager.getMaxTrackingDistance`.
    pub fn tracking_distance(&self) -> i32 {
        self.view.radius * 16 - 16
    }

    pub fn set_view(&mut self, view: ViewArea) {
        if view != self.view {
            self.view = view;
            self.offsets = view.offsets(view.load_radius());
            self.sent_offsets = view.offsets(view.radius);
            self.loaded.clear();
        }
    }

    /// Loads, streams and unloads chunks for all players.
    pub fn update(&mut self, level: &mut Level, players: &mut [ServerPlayer]) {
        for (pos, prefetched) in self.finished.try_iter() {
            self.generating.remove(&pos);
            if !self.stale.remove(&pos) {
                level.offer_prefetched(pos, prefetched);
            }
        }
        let centers: Vec<ChunkPos> = players
            .iter()
            .map(|player| {
                let pos = player.position(level);
                ChunkPos::containing(math::floor_f64(pos.x), math::floor_f64(pos.z))
            })
            .collect();
        let keep = (self.view.radius + 3).max(11);
        level.retain_prefetched(|pos| centers.iter().any(|&center| distance(pos, center) <= keep));
        self.loaded.retain(|center, _| centers.contains(center));
        let turn = self.turn % centers.len().max(1);
        self.turn = turn + 1;
        let mut rotated = centers.clone();
        rotated.rotate_left(turn);
        let share = max_in_flight().div_ceil(rotated.len().max(1));
        for &center in &rotated {
            let start = self.loaded.get(&center).copied().unwrap_or(0);
            self.prefetch(level, center, start, share);
        }

        // Each area loads nearest first and waits at its first unfetched chunk
        let deadline = Instant::now() + LOAD_BUDGET;
        'load: for &center in &rotated {
            let loaded = self.loaded.entry(center).or_default();
            for (i, &(dx, dz)) in self.offsets.iter().enumerate().skip(*loaded) {
                let pos = center.offset(dx, dz);
                if !level.is_loaded(pos) {
                    if Instant::now() > deadline {
                        break 'load;
                    }
                    if !level.is_prefetched(pos) {
                        break;
                    }
                    level.load_chunk(pos);
                }
                *loaded = i + 1;
            }
        }
        if level.check_storage().is_err() {
            return;
        }
        level.update_light();

        let updates = level.take_updates();
        for player in players.iter_mut() {
            for (pos, batch) in &updates {
                if player.sent_chunks.contains(pos)
                    && let Some(chunk) = level.chunk(*pos)
                {
                    for packet in batch.packets(chunk) {
                        player.send(packet);
                    }
                }
            }
        }

        for (player, &center) in players.iter_mut().zip(&centers) {
            let area = Some((center, self.view));
            if player.stream.area != area {
                player.stream = StreamCursor { area, sent: 0 };
                let ViewArea { radius, shape } = self.view;
                let dropped: Vec<_> = (player.sent_chunks)
                    .extract_if(|&pos| {
                        !shape.contains(pos.x - center.x, pos.z - center.z, radius + 1)
                    })
                    .collect();
                for pos in dropped {
                    player.send(Packet::PreChunk {
                        x: pos.x,
                        z: pos.z,
                        load: false,
                    });
                }
            }
            for (i, &(dx, dz)) in self
                .sent_offsets
                .iter()
                .enumerate()
                .skip(player.stream.sent)
            {
                // `EntityPlayerMP.onUpdate` waits while a socket still holds terrain
                if player.terrain_backlog() >= 4 {
                    break;
                }
                let pos = center.offset(dx, dz);
                if !player.sent_chunks.contains(&pos) && !send_chunk(level, player, pos) {
                    continue;
                }
                if i == player.stream.sent {
                    player.stream.sent += 1;
                }
            }
        }

        if !level.saving {
            self.swept = None;
            return;
        }
        // Chunks only become far as players move or more chunks load
        let count = level.loaded_count();
        if (self.swept.as_ref()).is_some_and(|(swept, n)| *swept == centers && *n == count) {
            return;
        }
        let far: Vec<ChunkPos> = level
            .loaded_chunks()
            .filter(|&pos| {
                !retained(level, pos) && centers.iter().all(|&center| distance(pos, center) > keep)
            })
            .collect();
        for pos in far {
            level.unload_chunk(pos);
            if self.generating.contains(&pos) {
                self.stale.insert(pos);
            }
        }
        self.swept = Some((centers, level.loaded_count()));
    }

    /// Queues fetching of up to `share` of the nearest chunks from offset `start` on that are
    /// neither loaded nor pending, within the shared limit of fetches in flight.
    fn prefetch(&mut self, level: &Level, center: ChunkPos, start: usize, share: usize) {
        let max_in_flight = max_in_flight();
        let mut queued = 0;
        for &(dx, dz) in &self.offsets[start..] {
            if self.generating.len() >= max_in_flight || queued >= share {
                break;
            }
            let pos = center.offset(dx, dz);
            if level.is_loaded(pos) || level.is_prefetched(pos) || !self.generating.insert(pos) {
                continue;
            }
            queued += 1;
            let generator = level.generator().clone();
            let storage = level.storage().cloned();
            let sender = self.finished_sender.clone();
            rayon::spawn(move || {
                let _ = sender.send((pos, Prefetched::fetch(&generator, storage.as_deref(), pos)));
            });
        }
    }
}

/// Sends a chunk once all its neighbors are loaded, as every population that writes into it has
/// run then. Returns whether it was sent.
fn send_chunk(level: &Level, player: &mut ServerPlayer, pos: ChunkPos) -> bool {
    if !(-1..=1).all(|dx| (-1..=1).all(|dz| level.is_loaded(pos.offset(dx, dz)))) {
        return false;
    }
    let Some(chunk) = level.chunk(pos) else {
        return false;
    };
    player.send(Packet::PreChunk {
        x: pos.x,
        z: pos.z,
        load: true,
    });
    let mut data = Vec::with_capacity(WIDTH * WIDTH * HEIGHT * 5 / 2);
    chunk.write_region(&mut data, 0..WIDTH, 0..HEIGHT, 0..WIDTH);
    player.send(Packet::MapChunk {
        x: pos.min_block_x(),
        y: 0,
        z: pos.min_block_z(),
        size: [15, 127, 15],
        data: data.into(),
    });
    // What a client needs to render a block: sign text, and the spawner mob and moving piston
    // over the local connection
    for (&index, tile_entity) in &chunk.tile_entities {
        let [x, y, z] = tile_position(pos, index);
        let packet = match tile_entity {
            TileEntity::Sign(lines) => Packet::UpdateSign {
                x,
                y: y as i16,
                z,
                lines: lines.clone(),
            },
            TileEntity::MobSpawner { .. } | TileEntity::Piston { .. } => Packet::TileEntity {
                x,
                y,
                z,
                data: Box::new(tile_entity.clone()),
            },
            _ => continue,
        };
        player.send(packet);
    }
    player.sent_chunks.insert(pos);
    true
}

/// `ChunkProviderServer.dropChunk`: dedicated servers keep the chunks around the spawn.
fn retained(level: &Level, pos: ChunkPos) -> bool {
    level.rules.dedicated_spawn.is_some_and(|[x, z]| {
        let range = -128..=128;
        range.contains(&(pos.x * 16 + 8 - x)) && range.contains(&(pos.z * 16 + 8 - z))
    })
}