Merge pull request #234 from Imberflur/send-dispatcher

Add a dispatcher that can be sent between threads

.github/workflows/ci.yml

@@ -73,4 +73,6 @@ jobs:
         # Miri currently reports leaks in some tests so we disable that check
         # here (might be due to ptr-int-ptr in crossbeam-epoch so might be
         # resolved in future versions of that crate).
-        run: MIRIFLAGS="-Zmiri-ignore-leaks" cargo miri test
+        #
+        # crossbeam-epoch doesn't pass with stacked borrows https://github.com/crossbeam-rs/crossbeam/issues/545
+        run: MIRIFLAGS="-Zmiri-ignore-leaks -Zmiri-tree-borrows" cargo miri test

src/dispatch/dispatcher.rs

@@ -1,6 +1,10 @@
 use smallvec::SmallVec;
 
-use crate::{dispatch::stage::Stage, system::RunNow, world::World};
+use crate::{
+    dispatch::{stage::Stage, SendDispatcher},
+    system::RunNow,
+    world::World,
+};
 
 /// This wrapper is used to share a replaceable ThreadPool with other
 /// dispatchers. Useful with batch dispatchers.
@@ -10,19 +14,15 @@ pub type ThreadPoolWrapper = Option<::std::sync::Arc<::rayon::ThreadPool>>;
 /// The dispatcher struct, allowing
 /// systems to be executed in parallel.
 pub struct Dispatcher<'a, 'b> {
-    stages: Vec<Stage<'a>>,
+    inner: SendDispatcher<'a>,
     thread_local: ThreadLocal<'b>,
-    #[cfg(feature = "parallel")]
-    thread_pool: ::std::sync::Arc<::std::sync::RwLock<ThreadPoolWrapper>>,
 }
 
 impl<'a, 'b> Dispatcher<'a, 'b> {
     /// Sets up all the systems which means they are gonna add default values
     /// for the resources they need.
     pub fn setup(&mut self, world: &mut World) {
-        for stage in &mut self.stages {
-            stage.setup(world);
-        }
+        self.inner.setup(world);
 
         for sys in &mut self.thread_local {
             sys.setup(world);
@@ -34,9 +34,7 @@ impl<'a, 'b> Dispatcher<'a, 'b> {
     /// / or resources from the `World` which are associated with external
     /// resources.
     pub fn dispose(self, world: &mut World) {
-        for stage in self.stages {
-            stage.dispose(world);
-        }
+        self.inner.dispose(world);
 
         for sys in self.thread_local {
             sys.dispose(world);
@@ -56,12 +54,7 @@ impl<'a, 'b> Dispatcher<'a, 'b> {
     /// Please note that this method assumes that no resource
     /// is currently borrowed. If that's the case, it panics.
     pub fn dispatch(&mut self, world: &World) {
-        #[cfg(feature = "parallel")]
-        self.dispatch_par(world);
-
-        #[cfg(not(feature = "parallel"))]
-        self.dispatch_seq(world);
-
+        self.inner.dispatch(world);
         self.dispatch_thread_local(world);
     }
 
@@ -77,18 +70,7 @@ impl<'a, 'b> Dispatcher<'a, 'b> {
     /// is currently borrowed. If that's the case, it panics.
     #[cfg(feature = "parallel")]
     pub fn dispatch_par(&mut self, world: &World) {
-        let stages = &mut self.stages;
-
-        self.thread_pool
-            .read()
-            .unwrap()
-            .as_ref()
-            .unwrap()
-            .install(move || {
-                for stage in stages {
-                    stage.execute(world);
-                }
-            });
+        self.inner.dispatch_par(world);
     }
 
     /// Dispatches the systems (except thread local systems) sequentially.
@@ -99,9 +81,7 @@ impl<'a, 'b> Dispatcher<'a, 'b> {
     /// Please note that this method assumes that no resource
     /// is currently borrowed. If that's the case, it panics.
     pub fn dispatch_seq(&mut self, world: &World) {
-        for stage in &mut self.stages {
-            stage.execute_seq(world);
-        }
+        self.inner.dispatch_seq(world);
     }
 
     /// Dispatch only thread local systems sequentially.
@@ -114,16 +94,28 @@ impl<'a, 'b> Dispatcher<'a, 'b> {
         }
     }
 
+    /// Converts this to a [`SendDispatcher`].
+    ///
+    /// Fails and returns the original distpatcher if it contains thread local systems.
+    pub fn try_into_sendable(self) -> Result<SendDispatcher<'a>, Self> {
+        let Dispatcher {
+            inner: _,
+            thread_local,
+        } = &self;
+
+        if thread_local.is_empty() {
+            Ok(self.inner)
+        } else {
+            Err(self)
+        }
+    }
+
     /// This method returns the largest amount of threads this dispatcher
     /// can make use of. This is mainly for debugging purposes so you can see
     /// how well your systems can make use of multi-threading.
     #[cfg(feature = "parallel")]
     pub fn max_threads(&self) -> usize {
-        self.stages
-            .iter()
-            .map(Stage::max_threads)
-            .max()
-            .unwrap_or(0)
+        self.inner.max_threads()
     }
 }
 
@@ -154,9 +146,11 @@ pub fn new_dispatcher<'a, 'b>(
     thread_pool: ::std::sync::Arc<::std::sync::RwLock<ThreadPoolWrapper>>,
 ) -> Dispatcher<'a, 'b> {
     Dispatcher {
-        stages,
+        inner: SendDispatcher {
+            stages,
+            thread_pool,
+        },
         thread_local,
-        thread_pool,
     }
 }
 
@@ -166,7 +160,7 @@ pub fn new_dispatcher<'a, 'b>(
     thread_local: ThreadLocal<'b>,
 ) -> Dispatcher<'a, 'b> {
     Dispatcher {
-        stages,
+        inner: SendDispatcher { stages },
         thread_local,
     }
 }

src/dispatch/mod.rs

@@ -9,6 +9,7 @@ pub use self::{
     },
     builder::DispatcherBuilder,
     dispatcher::Dispatcher,
+    send_dispatcher::SendDispatcher,
 };
 
 #[cfg(feature = "parallel")]
@@ -18,5 +19,6 @@ mod builder;
 mod dispatcher;
 #[cfg(feature = "parallel")]
 mod par_seq;
+mod send_dispatcher;
 mod stage;
 mod util;

src/dispatch/send_dispatcher.rs

@@ -0,0 +1,128 @@
+#[cfg(feature = "parallel")]
+use crate::dispatch::dispatcher::ThreadPoolWrapper;
+use crate::{dispatch::stage::Stage, system::RunNow, world::World};
+
+/// `Send`able version of [`Dispatcher`](crate::dispatch::Dispatcher).
+///
+/// Can't hold thread local systems.
+///
+/// Create using [`Dispatcher::try_into_sendable`](crate::dispatch::Dispatcher::try_into_sendable).
+pub struct SendDispatcher<'a> {
+    pub(super) stages: Vec<Stage<'a>>,
+    #[cfg(feature = "parallel")]
+    pub(super) thread_pool: ::std::sync::Arc<::std::sync::RwLock<ThreadPoolWrapper>>,
+}
+
+impl<'a> SendDispatcher<'a> {
+    /// Sets up all the systems which means they are gonna add default values
+    /// for the resources they need.
+    pub fn setup(&mut self, world: &mut World) {
+        for stage in &mut self.stages {
+            stage.setup(world);
+        }
+    }
+
+    /// Calls the `dispose` method of all systems and allows them to release
+    /// external resources. It is common this method removes components and
+    /// / or resources from the `World` which are associated with external
+    /// resources.
+    pub fn dispose(self, world: &mut World) {
+        for stage in self.stages {
+            stage.dispose(world);
+        }
+    }
+
+    /// Dispatch all the systems with given resources and context
+    /// and then run thread local systems.
+    ///
+    /// This function automatically redirects to
+    ///
+    /// * [SendDispatcher::dispatch_par] in case it is supported
+    /// * [SendDispatcher::dispatch_seq] otherwise
+    ///
+    /// and runs `dispatch_thread_local` afterwards.
+    ///
+    /// Please note that this method assumes that no resource
+    /// is currently borrowed. If that's the case, it panics.
+    pub fn dispatch(&mut self, world: &World) {
+        #[cfg(feature = "parallel")]
+        self.dispatch_par(world);
+
+        #[cfg(not(feature = "parallel"))]
+        self.dispatch_seq(world);
+    }
+
+    /// Dispatches the systems (except thread local systems)
+    /// in parallel given the resources to operate on.
+    ///
+    /// This operation blocks the
+    /// executing thread.
+    ///
+    /// Only available with "parallel" feature enabled.
+    ///
+    /// Please note that this method assumes that no resource
+    /// is currently borrowed. If that's the case, it panics.
+    #[cfg(feature = "parallel")]
+    pub fn dispatch_par(&mut self, world: &World) {
+        let stages = &mut self.stages;
+
+        self.thread_pool
+            .read()
+            .unwrap()
+            .as_ref()
+            .unwrap()
+            .install(move || {
+                for stage in stages {
+                    stage.execute(world);
+                }
+            });
+    }
+
+    /// Dispatches the systems (except thread local systems) sequentially.
+    ///
+    /// This is useful if parallel overhead is
+    /// too big or the platform does not support multithreading.
+    ///
+    /// Please note that this method assumes that no resource
+    /// is currently borrowed. If that's the case, it panics.
+    pub fn dispatch_seq(&mut self, world: &World) {
+        for stage in &mut self.stages {
+            stage.execute_seq(world);
+        }
+    }
+
+    /// This method returns the largest amount of threads this dispatcher
+    /// can make use of. This is mainly for debugging purposes so you can see
+    /// how well your systems can make use of multi-threading.
+    #[cfg(feature = "parallel")]
+    pub fn max_threads(&self) -> usize {
+        self.stages
+            .iter()
+            .map(Stage::max_threads)
+            .max()
+            .unwrap_or(0)
+    }
+}
+
+impl<'a, 'b> RunNow<'a> for SendDispatcher<'b> {
+    fn run_now(&mut self, world: &World) {
+        self.dispatch(world);
+    }
+
+    fn setup(&mut self, world: &mut World) {
+        self.setup(world);
+    }
+
+    fn dispose(self: Box<Self>, world: &mut World) {
+        (*self).dispose(world);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn send_dispatcher_is_send() {
+        fn is_send<T: Send>() {}
+        is_send::<super::SendDispatcher>();
+    }
+}

src/lib.rs

@@ -110,7 +110,7 @@ pub use crate::dispatch::{Par, ParSeq, RunWithPool, Seq};
 pub use crate::{
     dispatch::{
         BatchAccessor, BatchController, BatchUncheckedWorld, Dispatcher, DispatcherBuilder,
-        MultiDispatchController, MultiDispatcher,
+        MultiDispatchController, MultiDispatcher, SendDispatcher,
     },
     meta::{CastFrom, MetaIter, MetaIterMut, MetaTable},
     system::{

src/world/res_downcast/mod.rs

@@ -1,4 +1,4 @@
-//! Code is based on https://github.com/chris-morgan/mopa
+//! Code is based on <https://github.com/chris-morgan/mopa>
 //! with the macro inlined for `Resource`. License files can be found in the
 //! directory of this source file, see COPYRIGHT, LICENSE-APACHE and
 //! LICENSE-MIT.