fix a bug where racing asyncs might cause resource contention

Author: schell

.github/dependabot.yml

@@ -0,0 +1,8 @@
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    # Check for updates every Monday
+    schedule:
+      interval: "weekly"
+    open-pull-requests-limit: 10

crates/apecs/Cargo.toml

@@ -13,7 +13,8 @@ description = "An asyncronous and parallel entity-component system"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
-default = ["derive"]
+default = ["derive", "panic_on_async_holdup"]
+panic_on_async_holdup = []
 derive = []
 
 [lib]
@@ -40,4 +41,5 @@ smallvec = "^1.9"
 [dev-dependencies]
 async-timer = "^0.7"
 env_logger = "^0.9"
+futures-lite = "1.12.0"
 wasm-bindgen-test = "^0.3"

crates/apecs/src/lib.rs

@@ -9,7 +9,7 @@
 #![allow(clippy::type_complexity)]
 
 use ::anyhow::Context;
-use chan::oneshot;
+use chan::spsc;
 use internal::{FetchReadyResource, LoanManager, Resource, ResourceId};
 use rayon::iter::IntoParallelIterator;
 use std::{
@@ -336,14 +336,15 @@ impl<T> Gen<T> for NoDefault {
 /// fetch will err if the resource doesn't already exist.
 ///
 /// After a successful fetch, the resource will be automatically sent back to
-/// the world on drop. For this reason, when writing async systems you must take
-/// care to drop the fetched resources before your code crosses an await point.
+/// the world on drop. To make sure that your async functions don't hold fetched
+/// resources over await points, [`Facade`] uses [`visit`](Facade::visit) which
+/// fetches inside a syncronous closure.
 ///
 /// `Write` has two type parameters:
 /// * `T` - The type of the resource.
 /// * `G` - The method by which the resource can be generated if it doesn't
-///   exist. By default this is [`SomeDefault`], which denotes creating the
-///   resource using its default instance. Another option is [`NoDefault`] which
+///   already exist. By default this is [`SomeDefault`], which denotes creating the
+///   resource using its default implementation. Another option is [`NoDefault`] which
 ///   fails to generate the resource.
 ///
 /// ```rust
@@ -449,8 +450,7 @@ impl<T: IsResource, G: Gen<T>> Write<T, G> {
 /// fetch will err if the resource doesn't already exist.
 ///
 /// After a successful fetch, the resource will be automatically sent back to
-/// the world on drop. For this reason, when writing async systems you must take
-/// care to drop the fetched resources before your code crosses an await point.
+/// the world on drop.
 ///
 /// `Read` has two type parameters:
 /// * `T` - The type of the resource.
@@ -523,7 +523,7 @@ impl<T: IsResource, G: Gen<T>> Read<T, G> {
 pub(crate) struct Request {
     pub borrows: Vec<internal::Borrow>,
     pub construct: fn(&mut LoanManager<'_>) -> anyhow::Result<Resource>,
-    pub deploy_tx: oneshot::Sender<Resource>,
+    pub deploy_tx: spsc::Sender<Resource>,
 }
 
 impl std::fmt::Debug for Request {

crates/apecs/src/resource_manager.rs

@@ -169,6 +169,9 @@ impl ResourceManager {
     /// and unwrapping shared references.
     ///
     /// Returns `Ok(true)` if resources are still on loan.
+    ///
+    /// Errs if a system returns a resource that was not loaned. This doesn't
+    /// actually happen.
     pub fn try_unify_resources(&mut self, label: &str) -> anyhow::Result<bool> {
         log::trace!("try unify resources {}", label);
         while let Ok((rez_id, resource)) = self.exclusive_return_chan.1.try_recv() {
@@ -192,12 +195,10 @@ impl ResourceManager {
                         "duplicate resources"
                     ),
                     Err(arc_rez) => {
-                        log::warn!(
+                        log::error!(
                             "could not retreive borrowed resource {:?}, it is still borrowed by \
-                             {} - for better performance, try not to hold loaned resources over \
-                             an await point",
-                            id.name,
-                            label,
+                             '{}' - do not to hold loaned resources over an await point",
+                            id.name, label,
                         );
                         let _ = self.loaned_refs.insert(id, arc_rez);
                     }

crates/apecs/src/system.rs

@@ -368,29 +368,39 @@ impl<'a> IsBatch for AsyncBatch<'a> {
         resource_manager: &mut ResourceManager,
     ) -> anyhow::Result<()> {
         let mut loan_mngr = LoanManager(resource_manager);
-        let mut systems = self.take_systems();
+        let systems = self.take_systems();
         let mut data = VecDeque::new();
         for sys in systems.iter() {
             data.push_back(sys.prep(&mut loan_mngr)?);
         }
         drop(loan_mngr);
 
-        systems.retain_mut(|system| {
+        for system in systems.into_iter() {
             let data: Resource = data.pop_front().unwrap();
             if system.0.resource_request_rx.is_closed() {
                 log::trace!(
                     "system '{}' closed after requesting resources",
                     system.name()
                 );
-                return false;
+                continue;
             }
             // send the resources off, if need be
-            log::trace!("sending resources to async '{}'", system.name());
-            let _ = system.run(data).unwrap();
-            true
-        });
-
-        self.set_systems(systems);
+            if !system.1.deploy_tx.is_closed() {
+                log::trace!(
+                    "sending resource '{}' to async '{}'",
+                    data.type_name().unwrap_or("unknown"),
+                    system.name()
+                );
+                // UNWRAP: safe because we checked above that the channel is still open
+                system.1.deploy_tx.try_send(data).unwrap();
+            } else {
+                log::trace!(
+                    "cancelling send of resource '{}' to async '{}'",
+                    data.type_name().unwrap_or("unknown"),
+                    system.name()
+                );
+            }
+        }
 
         fn tick(executor: &async_executor::Executor<'static>) {
             while executor.try_tick() {
@@ -406,9 +416,10 @@ impl<'a> IsBatch for AsyncBatch<'a> {
             } else {
                 tick(extra);
             }
+
             let resources_still_loaned = resource_manager.try_unify_resources("async batch")?;
             if resources_still_loaned {
-                log::warn!(
+                panic!(
                     "an async system is holding onto resources over an await point! systems:{:#?}",
                     self.systems()
                         .iter()

crates/apecs/src/world.rs

@@ -51,7 +51,8 @@ impl Facade {
         // request the resources from the world
         // these requests are gathered in World::tick_async into an AsyncSchedule
         // and then run with the executor and resource maanger
-        let (deploy_tx, deploy_rx) = oneshot::oneshot();
+        let (deploy_tx, deploy_rx) = spsc::bounded(1);
+        // UNWRAP: safe because the request channel is unbounded
         self.resource_request_tx
             .try_send(Request {
                 borrows,
@@ -65,13 +66,11 @@ impl Facade {
                 },
                 deploy_tx,
             })
-            .context("could not send request for resources")?;
-        log::trace!("'{}' requested", std::any::type_name::<D>());
-
+            .unwrap();
         let rez: Resource = deploy_rx
+            .recv()
             .await
-            .map_err(|_| anyhow::anyhow!("could not fetch resources"))?;
-        log::trace!("'{}' received", std::any::type_name::<D>());
+            .unwrap();
         let box_d: Box<D> = rez.downcast().map_err(|rez| {
             anyhow::anyhow!(
                 "Facade could not downcast resource '{}' to '{}'",
@@ -1420,4 +1419,14 @@ mod test {
             assert_eq!(9, deleted_strings[0].id());
         }
     }
+
+    #[test]
+    fn spsc_drop_sanity() {
+        // ensure the spsc channel delivers messages when the sender was dropped after
+        // sending and before receiving
+        let (tx, rx) = spsc::bounded::<()>(1);
+        tx.try_send(()).unwrap();
+        drop(tx);
+        assert!(rx.try_recv().is_ok());
+    }
 }

crates/apecs/tests/regression.rs

@@ -0,0 +1,90 @@
+//! Tests for bugs we've encountered.
+use std::sync::Arc;
+
+use ::anyhow::Context;
+use apecs::{anyhow, chan::mpmc::Channel, ok, Facade, Read, ShouldContinue, World, Write};
+use futures_lite::StreamExt;
+
+#[test]
+fn can_race_asyncs() {
+    // ensures that resources are dropped after being sent to
+    // an async that is caling Facade::visit, but then gets
+    // cancelled
+
+    //let _ = env_logger::builder()
+    //    .is_test(true)
+    //    .filter_level(log::LevelFilter::Trace)
+    //    .try_init();
+
+    // each tick this increments a counter by 1
+    // when the counter reaches 3 it fires an event
+    fn ticker(
+        (mut chan, mut tick): (Write<Channel<()>>, Write<usize>),
+    ) -> anyhow::Result<ShouldContinue> {
+        *tick += 1;
+        println!("ticked {}", *tick);
+        if *tick == 3 {
+            chan.try_send(()).unwrap();
+        }
+        ok()
+    }
+
+    // loops acquiring a unit resource and reading it
+    async fn loser(facade: &mut Facade) -> anyhow::Result<()> {
+        loop {
+            println!("loser awaiting Read<()>");
+            facade
+                .visit(|unit: Read<()>| {
+                    println!("loser got Read<()>");
+                    let () = *unit;
+                    Ok(())
+                })
+                .await?;
+        }
+    }
+
+    // races the losing async against awaiting an event from ticker
+    // after ticker wins the race it should be able to access the unit
+    // resource
+    async fn system(mut facade: Facade) -> anyhow::Result<()> {
+        let mut rx = facade
+            .visit(|chan: Read<Channel<()>>| Ok(chan.new_receiver()))
+            .await?;
+
+        {
+            futures_lite::future::or(
+                async {
+                    rx.next().await.context("impossible")?;
+                    anyhow::Ok(())
+                },
+                async {
+                    loser(&mut facade).await?;
+                    panic!("this was supposed to lose the race");
+                },
+            )
+                .await?;
+        }
+
+        println!("race is over");
+
+        facade
+            .visit(|unit: Read<()>| {
+                let () = *unit;
+                Ok(())
+            })
+            .await?;
+
+        Ok(())
+    }
+
+    let mut world = World::default();
+    world
+        .with_system("ticker", ticker)
+        .unwrap()
+        .with_async_system("system", system);
+
+    world.tick();
+    world.tick();
+    world.tick();
+    world.tick();
+}