From b255f0ee0fd3ed552716d192b21b4642978d9809 Mon Sep 17 00:00:00 2001 From: Imbris Date: Sun, 26 Dec 2021 11:48:34 -0500 Subject: [PATCH] put Client::tick_network function behind feature, remove unnecessary cloning of deleted entity vecs in entity sync, move prepare_send hack that avoids locking during message serialization from send_fallible to send, add job.cpu_status par mode adjustment around parallel section --- .cargo/config | 2 +- client/Cargo.toml | 3 ++- client/src/lib.rs | 1 + common/src/region.rs | 8 -------- server/src/client.rs | 15 +++++++-------- server/src/sys/entity_sync.rs | 31 +++++++++++++++---------------- server/src/sys/sentinel.rs | 13 +++++++------ server/src/sys/subscription.rs | 6 +----- 8 files changed, 34 insertions(+), 45 deletions(-) diff --git a/.cargo/config b/.cargo/config index a738fb83a0..9aaa76436a 100644 --- a/.cargo/config +++ b/.cargo/config @@ -15,7 +15,7 @@ test-voxygen = "run --bin veloren-voxygen --no-default-features --features simd, tracy-voxygen = "-Zunstable-options run --bin veloren-voxygen --no-default-features --features tracy,simd,egui-ui --profile no_overflow" server = "run --bin veloren-server-cli" dbg-voxygen = "run --bin veloren-voxygen -Zunstable-options --profile debuginfo" -swarm = "run --bin swarm --features client/bin_bot --" +swarm = "run --bin swarm --features client/bin_bot,client/tick_network --" [env] diff --git a/client/Cargo.toml b/client/Cargo.toml index f78a1fa44d..d48889a0b3 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -9,6 +9,7 @@ simd = ["vek/platform_intrinsics"] plugins = ["common-state/plugins"] bin_bot = ["common-ecs", "serde", "ron", "clap", "structopt", "rustyline", "common-frontend", "async-channel"] tracy = ["common-base/tracy"] +tick_network = [] default = ["simd"] @@ -58,4 +59,4 @@ required-features = ["bin_bot"] [[bin]] name = "swarm" -required-features = ["bin_bot"] +required-features = ["bin_bot", "tick_network"] diff --git a/client/src/lib.rs b/client/src/lib.rs index 84c966691f..8fc0059cf3 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -2484,6 +2484,7 @@ impl Client { /// /// The game state is purposefully not simulated to reduce the overhead of running the client. /// This method is for use in testing a server with many clients connected. + #[cfg(feature = "tick_network")] pub fn tick_network( &mut self, dt: Duration, diff --git a/common/src/region.rs b/common/src/region.rs index 4437384f43..9e89579912 100644 --- a/common/src/region.rs +++ b/common/src/region.rs @@ -349,14 +349,6 @@ impl RegionMap { pub fn iter(&self) -> impl Iterator, &Region)> { self.regions.iter().map(|(key, r)| (*key, r)) } - - /// Returns a parallel iterator of (Position, Regions) - pub fn par_iter( - &self, - ) -> impl rayon::iter::IndexedParallelIterator, &Region)> { - use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; - self.regions.par_iter().map(|(key, r)| (*key, r)) - } } /// Note vd is in blocks in this case diff --git a/server/src/client.rs b/server/src/client.rs index f03e44d566..bc45cd45a0 100644 --- a/server/src/client.rs +++ b/server/src/client.rs @@ -83,7 +83,11 @@ impl Client { } pub(crate) fn send>(&self, msg: M) -> Result<(), StreamError> { - match msg.into() { + // TODO: hack to avoid locking stream mutex while serializing the message, + // remove this when the mutexes on the Streams are removed + let prepared = self.prepare(msg); + self.send_prepared(&prepared) + /*match msg.into() { ServerMsg::Info(m) => self.register_stream.lock().unwrap().send(m), ServerMsg::Init(m) => self.register_stream.lock().unwrap().send(m), ServerMsg::RegisterAnswer(m) => self.register_stream.lock().unwrap().send(m), @@ -133,15 +137,10 @@ impl Client { } }, ServerMsg::Ping(m) => self.ping_stream.lock().unwrap().send(m), - } + }*/ } - pub(crate) fn send_fallible>(&self, msg: M) { - // TODO: hack to avoid locking stream mutex while serializing the message, - // remove this when the mutexes on the Streams are removed - let prepared = self.prepare(msg); - let _ = self.send_prepared(&prepared); - } + pub(crate) fn send_fallible>(&self, msg: M) { let _ = self.send(msg); } pub(crate) fn send_prepared(&self, msg: &PreparedMsg) -> Result<(), StreamError> { match msg.stream_id { diff --git a/server/src/sys/entity_sync.rs b/server/src/sys/entity_sync.rs index 2b76f73bb6..9810b30954 100644 --- a/server/src/sys/entity_sync.rs +++ b/server/src/sys/entity_sync.rs @@ -59,7 +59,7 @@ impl<'a> System<'a> for Sys { const PHASE: Phase = Phase::Create; fn run( - _job: &mut Job, + job: &mut Job, ( entities, tick, @@ -108,14 +108,23 @@ impl<'a> System<'a> for Sys { // Sync physics and other components // via iterating through regions (in parallel) - use rayon::iter::ParallelIterator; + + // Pre-collect regions paired with deleted entity list so we can iterate over + // them in parallel below + let regions_and_deleted_entities = region_map + .iter() + .map(|(key, region)| (key, region, deleted_entities.take_deleted_in_region(key))) + .collect::>(); + + use rayon::iter::{IntoParallelIterator, ParallelIterator}; + job.cpu_stats.measure(common_ecs::ParMode::Rayon); common_base::prof_span!(guard, "regions"); - region_map.par_iter().for_each_init( + regions_and_deleted_entities.into_par_iter().for_each_init( || { common_base::prof_span!(guard, "entity sync rayon job"); guard }, - |_guard, (key, region)| { + |_guard, (key, region, deleted_entities_in_region)| { // Assemble subscriber list for this region by iterating through clients and // checking if they are subscribed to this region let mut subscribers = ( @@ -192,10 +201,7 @@ impl<'a> System<'a> for Sys { let (entity_sync_package, comp_sync_package) = trackers.create_sync_packages( &tracked_comps, region.entities(), - deleted_entities - .get_deleted_in_region(key) - .cloned() // TODO: quick hack to make this parallel, we can avoid this - .unwrap_or_default(), + deleted_entities_in_region, ); // We lazily initialize the the synchronization messages in case there are no // clients. @@ -300,14 +306,7 @@ impl<'a> System<'a> for Sys { }, ); drop(guard); - - // TODO: this is a quick hack to make the loop over regions above parallel, - // there might is probably a way to make it cleaner - // - // Remove delete entities for each region that we alread handled above - region_map - .iter() - .for_each(|(key, _)| drop(deleted_entities.take_deleted_in_region(key))); + job.cpu_stats.measure(common_ecs::ParMode::Single); // Update the last physics components for each entity for (_, &pos, vel, ori, last_pos, last_vel, last_ori) in ( diff --git a/server/src/sys/sentinel.rs b/server/src/sys/sentinel.rs index 96ee14cf53..74f77376d0 100644 --- a/server/src/sys/sentinel.rs +++ b/server/src/sys/sentinel.rs @@ -430,14 +430,15 @@ impl DeletedEntities { .push(uid.into()); } - pub fn take_deleted_in_region(&mut self, key: Vec2) -> Option> { - self.map.remove(&key) + pub fn take_deleted_in_region(&mut self, key: Vec2) -> Vec { + self.map.remove(&key).unwrap_or_default() } - pub fn get_deleted_in_region(&self, key: Vec2) -> Option<&Vec> { self.map.get(&key) } + pub fn get_deleted_in_region(&self, key: Vec2) -> &[u64] { + self.map.get(&key).map_or(&[], |v| v.as_slice()) + } - pub fn take_remaining_deleted(&mut self) -> Vec<(Vec2, Vec)> { - // TODO: don't allocate - self.map.drain().collect() + pub fn take_remaining_deleted(&mut self) -> impl Iterator, Vec)> + '_ { + self.map.drain() } } diff --git a/server/src/sys/subscription.rs b/server/src/sys/subscription.rs index 43a4329f00..714c762d16 100644 --- a/server/src/sys/subscription.rs +++ b/server/src/sys/subscription.rs @@ -161,11 +161,7 @@ impl<'a> System<'a> for Sys { } // Send deleted entities since they won't be processed for this client in entity // sync - for uid in deleted_entities - .get_deleted_in_region(key) - .iter() - .flat_map(|v| v.iter()) - { + for uid in deleted_entities.get_deleted_in_region(key).iter() { client.send_fallible(ServerGeneral::DeleteEntity(Uid(*uid))); } }