Round 2: Refactor to futures, attaching tasks as components.

Cargo.toml

@@ -26,8 +26,8 @@ bevy_openal = { path = "../bevy_openal" }
 bevy_rapier2d = "0.10"
 bevy_tts = { path = "../bevy_tts" }
 coord_2d = "0.3"
-crossbeam-channel = "0.5"
 derive_more = "0.99"
+futures-lite = "1"
 gilrs = "0.8"
 mapgen = { git = "https://github.com/ndarilek/mapgen.rs" }
 maze_generator = "1"

src/lib.rs

@@ -8,10 +8,10 @@ pub use bevy_rapier2d;
 pub use bevy_tts;
 #[macro_use]
 pub mod core;
-pub use crossbeam_channel;
 pub use derive_more;
 pub mod error;
 pub mod exploration;
+pub use futures_lite;
 pub use gilrs;
 pub mod log;
 pub mod map;

src/pathfinding.rs

@@ -1,13 +1,16 @@
 use std::collections::HashMap;
 
-use bevy::{prelude::*, tasks::prelude::*};
+use bevy::{
+    prelude::*,
+    tasks::{prelude::*, Task},
+};
 use bevy_rapier2d::{
     na::UnitComplex,
     prelude::*,
     rapier::data::{ComponentSet, ComponentSetOption, Index},
 };
-use crossbeam_channel::{unbounded, Receiver, Sender};
 use derive_more::{Deref, DerefMut};
+use futures_lite::future;
 use pathfinding::prelude::*;
 
 use crate::{
@@ -16,6 +19,9 @@ use crate::{
     navigation::{RotationSpeed, Speed},
 };
 
+#[derive(Debug, Deref, DerefMut)]
+struct Calculating(Task<Option<Path>>);
+
 #[derive(Clone, Copy, Debug, Default, Deref, DerefMut, Eq, Hash, PartialEq, Reflect)]
 #[reflect(Component)]
 pub struct Destination(pub (i32, i32));
@@ -129,59 +135,46 @@ impl ComponentSetOption<ColliderPosition> for StaticColliderComponentsSet {
 }
 
 fn find_path_for_shape(
-    pool: &AsyncComputeTaskPool,
-    query_pipeline: QueryPipeline,
     initiator: Entity,
     start: &dyn PointLike,
     destination: &dyn PointLike,
+    query_pipeline: &QueryPipeline,
     map: &Map,
-    collider_query: &QueryPipelineColliderComponentsQuery,
+    collider_set: StaticColliderComponentsSet,
-    obstructions: &Query<&MapObstruction>,
     shape: &SharedShape,
-    channel: &Sender<Option<Path>>,
+) -> Option<Path> {
-) {
+    let path = astar(
-    let collider_set: StaticColliderComponentsSet = (collider_query, obstructions).into();
+        &start.i32(),
-    let start = start.i32();
+        |p| {
-    let destination = destination.i32();
+            let mut successors: Vec<((i32, i32), u32)> = vec![];
-    let map_clone = map.clone();
+            for tile in map.get_available_exits(p.0 as usize, p.1 as usize) {
-    let shape_clone = shape.clone();
+                let mut should_push = true;
-    let channel_clone = channel.clone();
+                let shape_pos = Isometry::new(vector![tile.0 as f32, tile.1 as f32], 0.);
-    pool.spawn(async move {
+                query_pipeline.intersections_with_shape(
-        let path = astar(
+                    &collider_set,
-            &start,
+                    &shape_pos,
-            |p| {
+                    &**shape,
-                let mut successors: Vec<((i32, i32), u32)> = vec![];
+                    InteractionGroups::all(),
-                for tile in map_clone.get_available_exits(p.0 as usize, p.1 as usize) {
+                    Some(&|v| v.entity() != initiator),
-                    let mut should_push = true;
+                    |_handle| {
-                    let shape_pos = Isometry::new(vector![tile.0 as f32, tile.1 as f32], 0.);
+                        should_push = false;
-                    query_pipeline.intersections_with_shape(
+                        false
-                        &collider_set,
+                    },
-                        &shape_pos,
+                );
-                        &*shape_clone,
+                if should_push {
-                        InteractionGroups::all(),
+                    successors.push(((tile.0 as i32, tile.1 as i32), (tile.2 * 100.) as u32));
-                        Some(&|v| v.entity() != initiator),
-                        |_handle| {
-                            should_push = false;
-                            false
-                        },
-                    );
-                    if should_push {
-                        successors.push(((tile.0 as i32, tile.1 as i32), (tile.2 * 100.) as u32));
-                    }
                 }
-                successors
+            }
-            },
+            successors
-            |p| (p.distance_squared(&destination) * 100.) as u32,
+        },
-            |p| *p == destination,
+        |p| (p.distance_squared(destination) * 100.) as u32,
-        );
+        |p| *p == destination.i32(),
-        let rv = if let Some(path) = path {
+    );
-            Some(Path(path.0))
+    if let Some(path) = path {
-        } else {
+        Some(Path(path.0))
-            None
+    } else {
-        };
+        None
-        channel_clone.send(rv).ok();
+    }
-    })
-    .detach();
 }
 
 fn calculate_path(
@@ -190,54 +183,68 @@ fn calculate_path(
     query_pipeline: Res<QueryPipeline>,
     obstructions: Query<&MapObstruction>,
     collider_query: QueryPipelineColliderComponentsQuery,
-    mut calculating: Local<HashMap<Entity, Receiver<Option<Path>>>>,
+    query: Query<
-    query: Query<(Entity, &Destination, &Coordinates, &ColliderShape), Changed<Destination>>,
+        (Entity, &Destination, &Coordinates, &ColliderShape),
-    destinations: Query<&Destination, Without<NoPath>>,
+        (Without<Calculating>, Without<NoPath>, Without<Path>),
+    >,
     map: Query<&Map>,
 ) {
-    let calculating_clone = calculating.clone();
-    for (entity, rx) in calculating_clone.iter() {
-        if destinations.get(*entity).is_ok() {
-            if let Ok(msg) = rx.try_recv() {
-                if let Some(path) = msg {
-                    commands.entity(*entity).insert(path);
-                } else {
-                    println!("No path, tagging {:?}", entity);
-                    commands.entity(*entity).insert(NoPath);
-                }
-                calculating.remove(&entity);
-            }
-        } else {
-            calculating.remove(&entity);
-        }
-    }
     for (entity, destination, coordinates, shape) in query.iter() {
-        if calculating.contains_key(&entity) {
-            continue;
-        }
         if coordinates.i32() == **destination {
             println!("Removing path1: {:?}", entity);
             commands
                 .entity(entity)
                 .remove::<Path>()
+                .remove::<Calculating>()
                 .remove::<Destination>();
             continue;
         }
-        let (tx, rx) = unbounded();
-        calculating.insert(entity, rx);
         for map in map.iter() {
-            find_path_for_shape(
+            let coordinates_clone = *coordinates;
-                &*pool,
+            let destination_clone = *destination;
-                query_pipeline.clone(),
+            let query_pipeline_clone = query_pipeline.clone();
-                entity,
+            let map_clone = map.clone();
-                coordinates,
+            let shape_clone = shape.clone();
-                destination,
+            let collider_set: StaticColliderComponentsSet = (&collider_query, &obstructions).into();
-                &map,
+            println!("Spawning pathfinding task for {:?}", entity);
-                &collider_query,
+            let task = pool.spawn(async move {
-                &obstructions,
+                find_path_for_shape(
-                shape,
+                    entity,
-                &tx,
+                    &coordinates_clone,
-            );
+                    &destination_clone,
+                    &query_pipeline_clone,
+                    &map_clone,
+                    collider_set,
+                    &shape_clone,
+                )
+            });
+            commands.entity(entity).insert(Calculating(task));
+            println!("Inserted");
+        }
+    }
+}
+
+fn poll_tasks(
+    mut commands: Commands,
+    mut query: Query<(Entity, &mut Calculating)>,
+    destinations: Query<&Destination>,
+) {
+    for (entity, mut calculating) in query.iter_mut() {
+        println!("Got a calculating task for {:?}", entity);
+        if destinations.get(entity).is_ok() {
+            if let Some(result) = future::block_on(future::poll_once(&mut **calculating)) {
+                println!("Got a result, setting and returning");
+                if let Some(path) = result {
+                    commands.entity(entity).insert(path);
+                } else {
+                    commands.entity(entity).insert(NoPath);
+                }
+                println!("Removing calculation task");
+                commands.entity(entity).remove::<Calculating>();
+            }
+        } else {
+            println!("Destination was removed, clearing task");
+            commands.entity(entity).remove::<Calculating>();
         }
     }
 }
@@ -312,6 +319,16 @@ fn negotiate_path(
     }
 }
 
+fn remove_calculating(
+    mut commands: Commands,
+    query: Query<Entity, (Changed<Destination>, With<Calculating>)>,
+) {
+    for entity in query.iter() {
+        println!("Probably shouldn't happen");
+        commands.entity(entity).remove::<Calculating>();
+    }
+}
+
 fn remove_no_path(
     mut commands: Commands,
     query: Query<Entity, (With<NoPath>, Changed<Destination>)>,
@@ -325,8 +342,10 @@ pub struct PathfindingPlugin;
 
 impl Plugin for PathfindingPlugin {
     fn build(&self, app: &mut AppBuilder) {
-        app.add_system_to_stage(CoreStage::PostUpdate, calculate_path.system())
+        app.add_system(calculate_path.system())
+            .add_system(poll_tasks.system())
             .add_system(negotiate_path.system())
+            .add_system_to_stage(CoreStage::PostUpdate, remove_calculating.system())
             .add_system(remove_no_path.system());
     }
 }