Async events improve

ryot/Cargo.toml

@@ -16,6 +16,7 @@ readme = "../README.md"
 async-std = "1.12.0"
 derive_more = "0.99.17"
 serde_repr = "0.1"
+crossbeam-channel = "0.5.12"
 
 bevy = { workspace = true, optional = true }
 bevy_asset_loader = { workspace = true, optional = true }
@@ -54,10 +55,10 @@ default = ["bevy"]
 lmdb = ["dep:heed", "dep:postcard"]
 compression = ["dep:zstd"]
 bevy = [
-    "dep:bevy",
-    "dep:bevy_common_assets",
-    "dep:bevy_asset_loader",
-    "dep:leafwing-input-manager",
+	"dep:bevy",
+	"dep:bevy_common_assets",
+	"dep:bevy_asset_loader",
+	"dep:leafwing-input-manager",
 ]
 egui = ["dep:egui"]
 

ryot/src/bevy_ryot/async_events.rs

@@ -1,21 +1,55 @@
 //! This module provides a way to send events between systems asynchronously.
 //! It's useful to send events between threads that perform asynchronous tasks, such as loading
 //! content from disk or loading sprites from a sprite sheet before rendering.
+use std::{hash::Hasher, marker::PhantomData};
+
 use bevy::app::App;
 use bevy::prelude::*;
-use std::sync::mpsc::{channel, Receiver, Sender};
-use std::sync::Mutex;
+use crossbeam_channel::{Receiver, Sender};
+
+// TODO: doc.
+#[derive(SystemSet)]
+pub struct AsyncEventSet<T: Event>(PhantomData<T>);
+
+impl<T: Event> std::fmt::Debug for AsyncEventSet<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "AsyncEventSet<{}>", &std::any::type_name::<T>())
+    }
+}
+
+impl<T: Event> std::hash::Hash for AsyncEventSet<T> {
+    fn hash<H: Hasher>(&self, _state: &mut H) {
+        // all systems of a given type are the same
+    }
+}
+
+impl<T: Event> Clone for AsyncEventSet<T> {
+    fn clone(&self) -> Self {
+        *self
+    }
+}
+
+impl<T: Event> Copy for AsyncEventSet<T> {}
+
+impl<T: Event> PartialEq for AsyncEventSet<T> {
+    #[inline]
+    fn eq(&self, _other: &Self) -> bool {
+        // all systems of a given type are the same
+        true
+    }
+}
+
+impl<T: Event> Eq for AsyncEventSet<T> {}
 
 /// A resource that emits asynchronous events of a given type.
-/// It's a bevy friendly wrapper around a `std::sync::mpsc::Sender`.
+/// It's a bevy friendly wrapper around a `crossbeam_channel::Sender`.
 #[derive(Resource, Deref, DerefMut)]
-pub struct EventSender<T>(pub Sender<T>);
+pub struct AsyncEventSender<T: Event>(pub Sender<T>);
 
 /// A resource that receives asynchronous events of a given type.
-/// It's a bevy friendly wrapper around a `std::sync::mpsc::Receiver`.
-/// It's wrapped in a `Mutex` to allow multiple systems to safely access it.
+/// It's a bevy friendly wrapper around a `crossbeam_channel::Receiver`.
 #[derive(Resource, Deref, DerefMut)]
-struct EventReceiver<T>(Mutex<Receiver<T>>);
+struct AsyncEventReceiver<T: Event>(Receiver<T>);
 
 /// A trait to add an asynchronous event to an App.
 pub trait AsyncEventApp {
@@ -27,27 +61,29 @@ pub trait AsyncEventApp {
 /// the system that sends events from the receiver to Bevy's event system.
 impl AsyncEventApp for App {
     fn add_async_event<T: Event>(&mut self) -> &mut Self {
-        if self.world.contains_resource::<EventReceiver<T>>() {
+        if self.world.contains_resource::<AsyncEventReceiver<T>>() {
             return self;
         }
 
-        let (sender, receiver) = channel::<T>();
+        let (sender, receiver) = crossbeam_channel::unbounded::<T>();
 
         self.add_event::<T>()
-            .add_systems(Update, channel_to_event::<T>)
-            .insert_resource(EventSender(sender))
-            .insert_resource(EventReceiver(Mutex::new(receiver)));
+            .add_systems(
+                PreUpdate,
+                unbounded_channel_to_event::<T>.in_set(AsyncEventSet::<T>(PhantomData)),
+            )
+            .insert_resource(AsyncEventSender(sender))
+            .insert_resource(AsyncEventReceiver(receiver));
 
         self
     }
 }
 
 /// A system that sends events from the receiver to Bevy's event system.
 /// Converts the asynchronous event into a bevy event and sends it to the event system.
-fn channel_to_event<T: Event>(receiver: Res<EventReceiver<T>>, mut writer: EventWriter<T>) {
-    // this should be the only system working with the receiver,
-    // thus we always expect to get this lock
-    let events = receiver.lock().expect("unable to acquire mutex lock");
-
-    writer.send_batch(events.try_iter());
+fn unbounded_channel_to_event<T: Event>(
+    receiver: Res<AsyncEventReceiver<T>>,
+    mut writer: EventWriter<T>,
+) {
+    writer.send_batch(receiver.try_iter());
 }

ryot_compass/src/bevy_compass/gui.rs

@@ -10,7 +10,7 @@ use crate::{ExportMap, LoadMap};
 #[cfg(not(target_arch = "wasm32"))]
 use bevy::app::AppExit;
 #[cfg(not(target_arch = "wasm32"))]
-use ryot::bevy_ryot::{AsyncEventApp, EventSender};
+use ryot::bevy_ryot::{AsyncEventApp, AsyncEventSender};
 
 use bevy::{prelude::*, render::camera::Viewport, winit::WinitWindows};
 use bevy_egui::{EguiContext, EguiContexts, EguiPlugin, EguiUserTextures};
@@ -81,7 +81,7 @@ fn ui_menu_system<C: ContentAssets>(
     #[cfg(not(target_arch = "wasm32"))] content_assets: Res<C>,
     #[cfg(not(target_arch = "wasm32"))] mut exit: EventWriter<AppExit>,
     #[cfg(not(target_arch = "wasm32"))] mut map_export_sender: EventWriter<ExportMap>,
-    #[cfg(not(target_arch = "wasm32"))] load_map_sender: Res<EventSender<LoadMap>>,
+    #[cfg(not(target_arch = "wasm32"))] load_map_sender: Res<AsyncEventSender<LoadMap>>,
     _windows: NonSend<WinitWindows>,
 ) {
     let Ok(mut cursor) = cursor_query.get_single_mut() else {