diff --git a/protocol/src/utils.rs b/protocol/src/utils.rs index 81939bb..746849b 100644 --- a/protocol/src/utils.rs +++ b/protocol/src/utils.rs @@ -4,13 +4,13 @@ use std::ops::{Index, IndexMut}; use std::slice::Iter; use array_init::array_init; -use nalgebra::Point3; +use nalgebra::{Point2, Point3}; use strum::EnumCount; use tokio::io; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use crate::{ReadCwData, WriteCwData}; -use crate::utils::constants::SIZE_BLOCK; +use crate::utils::constants::{SIZE_BLOCK, SIZE_ZONE}; use crate::utils::io_extensions::{ReadArbitrary, WriteArbitrary}; pub mod io_extensions; @@ -52,6 +52,13 @@ pub fn sound_position_of(position: Point3) -> Point3 { //todo: move to position.map(|scalar| scalar as f32 / SIZE_BLOCK as f32) } +#[must_use] +pub fn zone_of(position: Point3) -> Point2 { + position + .xy() + .map(|scalar| scalar.div_euclid(SIZE_ZONE) as i32) +} + ///ideally this would be done with a `#[derive()]` macro instead, ///but the boilerplate required for that is completely overkill for this use case #[macro_export] diff --git a/server/src/addon.rs b/server/src/addon.rs index f00112b..e9ce046 100644 --- a/server/src/addon.rs +++ b/server/src/addon.rs @@ -64,6 +64,7 @@ impl Addons { pub async fn start(&self) { self.listforge_api.run().await; self.discord_integration.run(); + self.models.prepare().await; freeze_time(); events::start(); } diff --git a/server/src/addon/command_manager/commands/test.rs b/server/src/addon/command_manager/commands/test.rs index 9d2ba2e..8b87501 100644 --- a/server/src/addon/command_manager/commands/test.rs +++ b/server/src/addon/command_manager/commands/test.rs @@ -1,21 +1,22 @@ use std::ops::{Div as _, Mul as _, Sub as _}; use std::str::SplitWhitespace; -use protocol::packet::creature_update::{Affiliation, Appearance, AppearanceFlag}; -use protocol::packet::{CreatureUpdate, StatusEffect}; -use protocol::packet::world_update::block::Kind; -use protocol::packet::status_effect; +use strum::IntoEnumIterator; +use tap::{Pipe, Tap}; + use protocol::nalgebra::{Point3, Vector3}; -use protocol::packet::WorldUpdate; +use protocol::packet::{CreatureUpdate, StatusEffect, WorldUpdate}; use protocol::packet::common::{CreatureId, Hitbox, Race}; +use protocol::packet::creature_update::{Affiliation, Appearance, AppearanceFlag}; +use protocol::packet::status_effect; use protocol::packet::world_update::{Block, WorldObject}; +use protocol::packet::world_update::block::Kind; use protocol::packet::world_update::block::Kind::*; +use protocol::packet::world_update::sound; use protocol::packet::world_update::world_object::Kind::{Crate, FireTrap}; use protocol::utils::constants::{SIZE_BLOCK, SIZE_ZONE}; use protocol::utils::flagset::FlagSet; -use strum::IntoEnumIterator; -use protocol::packet::world_update::sound; -use tap::{Pipe, Tap}; +use protocol::utils::zone_of; use crate::addon::{command_manager::{Command, CommandResult}, models, play_sound_at_player}; use crate::addon::command_manager::commands::Test; @@ -282,11 +283,13 @@ async fn model(params: &mut SplitWhitespace<'_>, server: &Server, caller: &Playe }; let mut blocks = models::parse_model(file); - let offset = caller + let position = caller .character .read() .await - .position + .position; + + let offset = position .div(SIZE_BLOCK) .cast() .coords; @@ -295,7 +298,7 @@ async fn model(params: &mut SplitWhitespace<'_>, server: &Server, caller: &Playe block.position += offset; } - server.broadcast(&WorldUpdate::from(blocks), None).await; + server.broadcast_near(&WorldUpdate::from(blocks), zone_of(position)).await; Ok(()) } diff --git a/server/src/addon/models.rs b/server/src/addon/models.rs index 6c737d8..641f851 100644 --- a/server/src/addon/models.rs +++ b/server/src/addon/models.rs @@ -1,58 +1,76 @@ use std::ops::Div as _; use std::collections::HashMap; use std::path; +use std::sync::{Arc, OnceLock}; use config::{Config, ConfigError}; use protocol::utils::constants::{SIZE_BLOCK, SIZE_ZONE}; use protocol::rgb::RGB8; +use protocol::packet::WorldUpdate; use protocol::packet::world_update::Block; use protocol::nalgebra::{Point2, Vector3}; use protocol::packet::world_update::block::Kind::*; -use tap::Pipe; +use protocol::utils::io_extensions::WritePacket; mod vox; mod zox; +const BLOCKS_PER_ZONE: i32 = (SIZE_ZONE / SIZE_BLOCK) as i32; + pub struct Models { - models: Vec<(Point2, Vec)> + blocks_by_zone: HashMap, Vec>, + packets_by_zone: OnceLock, Arc<[u8]>>> } impl Models { pub fn new(config: &Config) -> Result { - Self { - models: config - .get::>("models")? - .into_iter() - .map(|(filename, pos)| { - let pos: Vector3 = pos.into(); - let zone = pos - .xy() - .div(SIZE_ZONE) - .cast::() - .into(); + let mut blocks_by_zone: HashMap, Vec> = HashMap::new(); + + for (filename, pos) in config.get::>("models")? { + let pos: Vector3 = pos.into(); + let model_origin = pos + .div(SIZE_BLOCK) + .cast::(); + + for mut block in parse_model(&filename) { + block.position += model_origin; + blocks_by_zone + .entry(block.position.xy().map(|scalar| scalar.div_euclid(BLOCKS_PER_ZONE))) + .or_default() + .push(block); + } + } + + Ok(Self { + blocks_by_zone, + packets_by_zone: OnceLock::new() + }) + } + + pub async fn prepare(&self) { + let mut packets_by_zone = HashMap::with_capacity(self.blocks_by_zone.len()); - let mut blocks = parse_model(&filename); - let model_origin = pos - .div(SIZE_BLOCK) - .cast::(); + for (zone, blocks) in &self.blocks_by_zone { + let mut packet = vec![]; + packet + .write_packet(&WorldUpdate::from(blocks.clone())) + .await + .expect("failed to serialize a world update in-memory"); - for block in &mut blocks { - block.position += model_origin; - } + packets_by_zone.insert(*zone, packet.into()); + } - (zone, blocks) - }) - .collect() - }.pipe(Ok) + assert!( + self.packets_by_zone.set(packets_by_zone).is_ok(), + "models prepared twice" + ); } - pub fn blocks_in(&self, requested_zone: Point2) -> Vec { - self.models - .iter() - .filter(|(zone, _blocks)| *zone == requested_zone) - .flat_map(|(_zone, blocks)| blocks) - .cloned() - .collect() + pub fn packet_for(&self, requested_zone: Point2) -> Option> { + self.packets_by_zone + .get()? + .get(&requested_zone) + .map(Arc::clone) } } diff --git a/server/src/server.rs b/server/src/server.rs index c7ce673..08f3103 100644 --- a/server/src/server.rs +++ b/server/src/server.rs @@ -28,8 +28,9 @@ use protocol::packet::creature_update::Affiliation; use protocol::packet::world_update::loot::GroundItem; use protocol::packet::world_update::Sound; use protocol::packet::world_update::sound::Kind::*; -use protocol::utils::constants::{SIZE_BLOCK, SIZE_ZONE}; +use protocol::utils::constants::SIZE_BLOCK; use protocol::utils::io_extensions::{ReadPacket, WriteArbitrary, WritePacket}; +use protocol::utils::zone_of; use crate::addon::{Addons, announce_join_leave}; use crate::addon::pvp::map_head; @@ -196,30 +197,53 @@ impl Server { pub async fn broadcast(&self, packet: &Packet, player_to_skip: Option<&Player>) where Vec: WriteCwData//todo: specialization could obsolete this { - let mut data = vec![]; - - data.write_packet(packet).await.expect("failed to serialize a packet in-memory"); - - _ = self.players + let recipients = self.players .read() .await .iter() .filter(|player| !player_to_skip.is_some_and(|pts| ptr::eq(player.as_ref(), pts))) - .map(async |player| { - let mut writer = player - .writer - .write() - .await; - - writer.write_all(&data).await?; - writer.flush().await - }) + .map(Arc::clone) + .collect(); + + self.send_to_all(packet, recipients).await; + } + + pub async fn broadcast_near(&self, packet: &Packet, zone: Point2) + where Vec: WriteCwData + { + let candidates = self.players + .read() + .await + .iter() + .map(Arc::clone) + .collect::>(); + + let mut recipients = vec![]; + for player in candidates { + if player.is_ready_for(zone).await { + recipients.push(player); + } + } + + self.send_to_all(packet, recipients).await; + } + + async fn send_to_all(&self, packet: &Packet, recipients: Vec>) + where Vec: WriteCwData + { + let mut data = vec![]; + + data.write_packet(packet).await.expect("failed to serialize a packet in-memory"); + + _ = recipients + .iter() + .map(async |player| player.send_raw(&data).await) .pipe(join_all) .await; } pub async fn add_drop(&self, item: Item, position: Point3, rotation: f32) { - let zone = position.xy().map(|scalar| (scalar / SIZE_ZONE) as i32); + let zone = zone_of(position); let mut loot = self.loot.write().await; let zone_loot = loot.entry(zone).or_insert(vec![]); @@ -236,15 +260,15 @@ impl Server { zone_loot_copy[zone_loot.len() - 1].droptime = 500; drop(loot); - self.broadcast(&WorldUpdate { + self.broadcast_near(&WorldUpdate { loot: HashMap::from([(zone, zone_loot_copy)]), sounds: vec![Sound::at(position, Drop)], ..Default::default() - }, None).await; + }, zone).await; tokio::spawn(async move { sleep(Duration::from_millis(500)).await; - SERVER.broadcast(&WorldUpdate::from(Sound::at(position, DropItem)), None).await; + SERVER.broadcast_near(&WorldUpdate::from(Sound::at(position, DropItem)), zone).await; }); } @@ -267,7 +291,7 @@ impl Server { } drop(drops_guard); - self.broadcast(&WorldUpdate::from((zone, zone_drops_owned)), None).await; + self.broadcast_near(&WorldUpdate::from((zone, zone_drops_owned)), zone).await; Some(removed_drop.item) } diff --git a/server/src/server/handle_packet/creature_update.rs b/server/src/server/handle_packet/creature_update.rs index b0f8ecc..cf41bb7 100644 --- a/server/src/server/handle_packet/creature_update.rs +++ b/server/src/server/handle_packet/creature_update.rs @@ -1,6 +1,7 @@ use std::sync::atomic::Ordering; use protocol::packet::CreatureUpdate; +use protocol::utils::zone_of; use crate::addon::{anti_cheat, kill_feed, pvp}; use crate::addon::fix_cutoff_animations; @@ -26,6 +27,13 @@ impl HandlePacket for Server { character.update(&packet); let character = character.downgrade(); + // client only requests the zone that its player is located at + // entering a new zone is the only sign that neighboring ones need to be revealed + let current_zone = zone_of(character.position); + if zone_of(snapshot.position) != current_zone { + self.schedule_neighborhood_reveal(source.id, current_zone); + } + if !filter(&mut packet, &snapshot, &character) { return; } diff --git a/server/src/server/handle_packet/zone_request.rs b/server/src/server/handle_packet/zone_request.rs index 7595a68..a1040d7 100644 --- a/server/src/server/handle_packet/zone_request.rs +++ b/server/src/server/handle_packet/zone_request.rs @@ -1,22 +1,233 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use tokio::time::sleep; + +use protocol::nalgebra::{Point2, Vector2}; use protocol::packet::{AreaRequest, WorldUpdate}; use protocol::packet::area_request::Zone; -use protocol::packet::world_update::p48::P48sub; +use protocol::packet::common::CreatureId; +use protocol::utils::zone_of; use crate::server::handle_packet::HandlePacket; -use crate::server::player::Player; +use crate::server::player::addon_data::ZoneState; +use crate::server::player::{Player, ZONE_DATA_RADIUS, ZONE_RETENTION_RADIUS}; use crate::server::Server; +use crate::SERVER; + +const CENTER_SETTLE: Duration = Duration::from_secs(1); +const NEIGHBOR_SETTLE: Duration = Duration::from_secs(3); +const STALE_RETRY_GRACE: Duration = Duration::from_secs(2); +const CENTER_TIMEOUT: Duration = Duration::from_secs(15); +const CENTER_POLL: Duration = Duration::from_millis(200); impl HandlePacket> for Server { async fn handle_packet(&self, source: &Player, packet: AreaRequest) { - let p48sub = P48sub([0_u8; 16]); - let world_update = WorldUpdate { - //todo: filter to just this + adjacent zones - loot: self.loot.read().await.clone(), - blocks: self.addons.models.blocks_in(packet.0), - p48: [(packet.0, vec![p48sub])].into(), + let zone = packet.0; + + let Some(player) = self.find_player_by_id(source.id).await + else { return }; + + self.answer_request(&player, zone).await; + self.schedule_neighborhood_reveal(source.id, zone); + } +} + +impl Server { + // client's zone request proves lack of content, no matter what the server believes + // it always gets answered, otherwise client re-requests forever + async fn answer_request(&self, player: &Arc, zone: Point2) { + { + let mut addon_data = player.addon_data.write().await; + match addon_data.zone_states.get(&zone) { + Some(ZoneState::Pending) => return, + Some(ZoneState::Revealed(at)) if at.elapsed() < STALE_RETRY_GRACE => return, + _ => {} + } + // claimed in the same lock as the check + addon_data.zone_states.insert(zone, ZoneState::Pending); + } + + // loot alone is safe immediately + let settle = if self.addons.models.packet_for(zone).is_some() { + CENTER_SETTLE + } else { + Duration::ZERO + }; + + let player = Arc::clone(player); + tokio::spawn(async move { + SERVER.deliver_zone(&player, zone, settle).await; + }); + } + + async fn reveal_neighborhood(&self, player: &Arc, center: Point2) { + for zone in neighbors(center) { + let settled_in = zone_of(player.character.read().await.position); + if settled_in != center { + return; + } + + if !self.has_content(zone).await { + continue; + } + + { + let mut addon_data = player.addon_data.write().await; + if addon_data.zone_states.contains_key(&zone) { + continue; + } + addon_data.zone_states.insert(zone, ZoneState::Pending); + } + + self.deliver_zone(player, zone, NEIGHBOR_SETTLE).await; + } + } + + // everything waits out the settle, not just terrain: acknowledgment itself is zone-keyed + // client ignores zone update if sent too early (before client-side zone generation) + async fn deliver_zone(&self, player: &Arc, zone: Point2, settle: Duration) { + sleep(settle).await; + + let delivered = self.send_zone(player, zone).await; + + let mut addon_data = player.addon_data.write().await; + if delivered { + addon_data.zone_states.insert(zone, ZoneState::Revealed(Instant::now())); + } else { + // let a later reveal or request retry it + addon_data.zone_states.remove(&zone); + } + } + + async fn send_zone(&self, player: &Arc, zone: Point2) -> bool { + // a zone change that took this zone out of range drops the claim + let still_claimed = matches!( + player.addon_data.read().await.zone_states.get(&zone), + Some(ZoneState::Pending) + ); + if !still_claimed || !player.is_near(zone).await { + return false; + } + + if let Some(blocks) = self.addons.models.packet_for(zone) + && player.send_raw(&blocks).await.is_err() + { + return false; + } + + self.acknowledge(player, zone).await; + true + } + + // an entry for a zone (even an empty one) acknowledges discovery + // and stops the client from re-requesting zones that remain loaded client-side + // p48 is unsafe, so empty loot updates are sent for empty zones instead + async fn acknowledge(&self, player: &Player, zone: Point2) { + let loot_in_zone = self.loot + .read().await + .get(&zone) + .cloned() + .unwrap_or_default(); + + let acknowledgment = WorldUpdate { + loot: [(zone, loot_in_zone)].into(), ..Default::default() }; + player.send_ignoring(&acknowledgment).await; + } + + async fn center_ready(&self, player: &Arc, center: Point2) -> bool { + if self.addons.models.packet_for(center).is_none() { + return true; + } + + let deadline = Instant::now() + CENTER_TIMEOUT; + + while Instant::now() < deadline { + let state = player.addon_data.read().await.zone_states.get(¢er).copied(); + + match state { + Some(ZoneState::Revealed(_)) => return true, + // claim was dropped; player left range + None => return false, + _ => sleep(CENTER_POLL).await + } + } + + false + } + + async fn has_content(&self, zone: Point2) -> bool { + self.addons.models.packet_for(zone).is_some() + || self.loot.read().await.contains_key(&zone) + } - source.send_ignoring(&world_update).await; + pub async fn prune_zone_states(&self, player: &Player, center: Point2) { + player + .addon_data + .write() + .await + .zone_states + .retain(|zone, state| match state { + ZoneState::Awaited(_) | ZoneState::Pending => + chebyshev_distance(*zone, center) <= ZONE_DATA_RADIUS, + ZoneState::Revealed(_) => + chebyshev_distance(*zone, center) <= ZONE_RETENTION_RADIUS + }); } + + pub fn schedule_neighborhood_reveal(&self, player_id: CreatureId, center: Point2) { + tokio::spawn(async move { + let Some(player) = SERVER.find_player_by_id(player_id).await + else { return }; + + SERVER.prune_zone_states(&player, center).await; + + if SERVER.addons.models.packet_for(center).is_some() { + player + .addon_data + .write() + .await + .zone_states + .entry(center) + .or_insert(ZoneState::Awaited(Instant::now())); + } + + if !SERVER.center_ready(&player, center).await { + return; + } + + { + let mut addon_data = player.addon_data.write().await; + if addon_data.revealing_neighborhood { + return; + } + addon_data.revealing_neighborhood = true; + } + + SERVER.reveal_neighborhood(&player, center).await; + + player.addon_data.write().await.revealing_neighborhood = false; + }); + } +} + +fn neighbors(center: Point2) -> [Point2; 8] { + [ + center + Vector2::new( 0, -1), + center + Vector2::new( 0, 1), + center + Vector2::new(-1, 0), + center + Vector2::new( 1, 0), + center + Vector2::new(-1, -1), + center + Vector2::new(-1, 1), + center + Vector2::new( 1, -1), + center + Vector2::new( 1, 1) + ] +} + +fn chebyshev_distance(from: Point2, to: Point2) -> i32 { + let delta = from - to; + + delta.x.abs().max(delta.y.abs()) } \ No newline at end of file diff --git a/server/src/server/player.rs b/server/src/server/player.rs index bf59b6e..b6f530b 100644 --- a/server/src/server/player.rs +++ b/server/src/server/player.rs @@ -1,18 +1,25 @@ -mod addon_data; +pub mod addon_data; use std::net::SocketAddr; use std::sync::atomic::AtomicBool; -use tokio::io::{self, SimplexStream, WriteHalf}; +use tokio::io::{self, AsyncWriteExt as _, SimplexStream, WriteHalf}; use tokio::sync::{oneshot, RwLock}; +use protocol::nalgebra::Point2; use protocol::packet::{ChatMessageFromServer, FromServer}; use protocol::packet::common::CreatureId; use protocol::utils::io_extensions::WritePacket; +use protocol::utils::zone_of; use protocol::WriteCwData; use crate::server::creature::Creature; -use crate::server::player::addon_data::AddonData; +use crate::server::player::addon_data::{AddonData, ZoneState, AWAITED_GRACE}; + +// current and adjacent zones revealed to the player client-side (limited by max render distance) +pub const ZONE_DATA_RADIUS: i32 = 1; +// any terrain beyond this radius is guaranteed to be unloaded client-side +pub const ZONE_RETENTION_RADIUS: i32 = 3; #[derive(Debug)] pub struct Player { @@ -61,6 +68,36 @@ impl Player { let _ = self.send(packet).await; } + pub async fn send_raw(&self, data: &[u8]) -> io::Result<()> { + let mut writer = self.writer.write().await; + + writer.write_all(data).await?; + writer.flush().await + } + + pub async fn is_near(&self, zone: Point2) -> bool { + let distance = zone_of(self.character.read().await.position) - zone; + + distance.x.abs().max(distance.y.abs()) <= ZONE_DATA_RADIUS + } + + // client readiness for zone update; pending delivery = server knows client is not ready + // live updates must not jump the queue (another player dropping loot while client + // is waiting for a zone update from the server would result in block updates getting lost) + pub async fn is_ready_for(&self, zone: Point2) -> bool { + if !self.is_near(zone).await { + return false; + } + + let state = self.addon_data.read().await.zone_states.get(&zone).copied(); + + match state { + Some(ZoneState::Pending) => false, + Some(ZoneState::Awaited(since)) => since.elapsed() >= AWAITED_GRACE, + _ => true + } + } + pub async fn notify(&self, message: impl Into) { self.send_ignoring(&ChatMessageFromServer { source: CreatureId(0), diff --git a/server/src/server/player/addon_data.rs b/server/src/server/player/addon_data.rs index 0d87882..29cde55 100644 --- a/server/src/server/player/addon_data.rs +++ b/server/src/server/player/addon_data.rs @@ -1,10 +1,25 @@ -use std::time::Instant; +use std::collections::HashMap; +use std::time::{Duration, Instant}; + +use protocol::nalgebra::Point2; use crate::addon::anti_cheat::PlayerData; +// zone update safety valve +pub const AWAITED_GRACE: Duration = Duration::from_secs(30); + #[derive(Debug, Default)] pub struct AddonData { pub team: Option, pub anti_cheat_data: PlayerData, - pub last_attacker: Option<(Instant, String)> + pub last_attacker: Option<(Instant, String)>, + pub zone_states: HashMap, ZoneState>, + pub revealing_neighborhood: bool +} + +#[derive(Debug, Clone, Copy)] +pub enum ZoneState { + Awaited(Instant), + Pending, + Revealed(Instant) } \ No newline at end of file diff --git a/server/src/server/utils.rs b/server/src/server/utils.rs index 6964345..8052021 100644 --- a/server/src/server/utils.rs +++ b/server/src/server/utils.rs @@ -19,6 +19,7 @@ use protocol::packet::creature_update::Affiliation; use protocol::packet::creature_update::Affiliation::Pet; use protocol::packet::creature_update::Animation::Riding; use protocol::packet::world_update::Kill; +use protocol::utils::zone_of; use crate::server::player::Player; use crate::addon::kill_feed; @@ -55,6 +56,8 @@ impl Server { } pub async fn teleport(&self, player: &Player, destination: Point3) { + self.prune_zone_states(player, zone_of(destination)).await; + let server_creature = CreatureUpdate { id: CreatureId(0), position: Some(destination),