Stop spawning tasks on the global rayon pool.

Also adds some missing parallelism to entity sync.

server/src/lib.rs

@@ -83,7 +83,7 @@ use common::{
     terrain::{TerrainChunk, TerrainChunkSize},
     vol::RectRasterableVol,
 };
-use common_ecs::run_now;
+use common_ecs::{dispatch, run_now};
 use common_net::{
     msg::{
         ClientType, DisconnectReason, ServerGeneral, ServerInfo, ServerInit, ServerMsg, WorldMapMsg,
@@ -100,7 +100,7 @@ use persistence::{
 };
 use prometheus::Registry;
 use prometheus_hyper::Server as PrometheusServer;
-use specs::{join::Join, Builder, Entity as EcsEntity, Entity, SystemData, WorldExt};
+use specs::{join::Join, Builder, DispatcherBuilder, Entity as EcsEntity, Entity, SystemData, WorldExt};
 use std::{
     i32,
     ops::{Deref, DerefMut},
@@ -741,7 +741,7 @@ impl Server {
         let before_sync = Instant::now();
 
         // 6) Synchronise clients with the new state of the world.
-        sys::run_sync_systems(self.state.ecs_mut());
+        sys::run_sync_systems(&mut self.state);
 
         let before_world_tick = Instant::now();
 
@@ -755,7 +755,13 @@ impl Server {
         // perform client disconnections have been processed. This ensures that any
         // items on the ground are deleted.
         if let Some(DisconnectType::WithoutPersistence) = disconnect_type {
-            run_now::<terrain::Sys>(self.state.ecs_mut());
+            // NOTE: We build a dispatcher to avoid running in the global thread pool, although
+            // that's not exactly our biggest concern.
+            let mut dispatch_builder =
+                DispatcherBuilder::new().with_pool(Arc::clone(&self.state.thread_pool()));
+            dispatch::<terrain::Sys>(&mut dispatch_builder, &[]);
+            let mut dispatcher = dispatch_builder.build();
+            dispatcher.dispatch(&self.state.ecs());
         }
 
         // Prevent anchor entity chains which are not currently supported
@@ -995,6 +1001,10 @@ impl Server {
         let end_of_server_tick = Instant::now();
 
         // 8) Update Metrics
+        //
+        // NOTE: This system may not use parallelism currently, since it would execute in the
+        // global pool and uses run_now!  If we want to parallelize it, we should explicitly do so
+        // within our thread pool.
         run_now::<sys::metrics::Sys>(self.state.ecs());
 
         {

server/src/sys/mod.rs

@@ -16,11 +16,14 @@ pub mod terrain_sync;
 pub mod waypoint;
 pub mod wiring;
 
+use common_base::span;
 use common_ecs::{dispatch, run_now, System};
+use common_state::State;
 use common_systems::{melee, projectile};
-use specs::DispatcherBuilder;
+use specs::{DispatcherBuilder, WorldExt};
 use std::{
     marker::PhantomData,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -42,15 +45,34 @@ pub fn add_server_systems(dispatch_builder: &mut DispatcherBuilder) {
     dispatch::<chunk_send::Sys>(dispatch_builder, &[]);
 }
 
-pub fn run_sync_systems(ecs: &mut specs::World) {
+pub fn run_sync_systems(state: &mut State) {
+    span!(_guard, "run_sync_systems");
+
+    // Create and run a dispatcher for ecs systems that synchronize state.
+    span!(guard, "create dispatcher");
+    let mut dispatch_builder =
+        DispatcherBuilder::new().with_pool(Arc::clone(&state.thread_pool()));
+
     // Setup for entity sync
-    // If I'm not mistaken, these two could be ran in parallel
+    dispatch::<sentinel::Sys>(&mut dispatch_builder, &[]);
-    run_now::<sentinel::Sys>(ecs);
+    dispatch::<subscription::Sys>(&mut dispatch_builder, &[]);
-    run_now::<subscription::Sys>(ecs);
 
     // Sync
-    run_now::<terrain_sync::Sys>(ecs);
+    dispatch::<terrain_sync::Sys>(&mut dispatch_builder, &[]);
-    run_now::<entity_sync::Sys>(ecs);
+    dispatch::<entity_sync::Sys>(&mut dispatch_builder, &[&sentinel::Sys::sys_name(), &subscription::Sys::sys_name()]);
+
+    // This dispatches all the systems in parallel.
+    let mut dispatcher = dispatch_builder.build();
+    drop(guard);
+
+    let ecs = state.ecs_mut();
+    span!(guard, "run systems");
+    dispatcher.dispatch(ecs);
+    drop(guard);
+
+    span!(guard, "maintain ecs");
+    ecs.maintain();
+    drop(guard);
 }
 
 /// Used to schedule systems to run at an interval