feat: use tokio fully for async

Author: tristan-f-r

Cargo.lock

@@ -10,21 +10,23 @@ edition = "2021"
 
 [features]
 "xxhash" = ["dep:twox-hash"]
-"rayon" = ["dep:rayon", "xxhash", "dep:sysinfo", "dep:moka"]
+"rayon" = ["xxhash", "dep:sysinfo", "dep:moka", "dep:tokio", "dep:tokio-util"]
 "js" = ["moka/js"]
 
 [dependencies]
 # dfdx = { git = "https://github.com/coreylowman/dfdx.git", rev = "4722a99", optional = true }
 moka = { version = "0.12", optional = true, features = ["future"] }
 rand = { version = "0.8", optional = true }
-rayon = { version = "1.8", optional = true }
 sysinfo = { version = "0.30", optional = true }
 twox-hash = { version = "1.6", optional = true }
 itertools = { version = "0.13" }
 futures = "0.3.30"
 thiserror = "1.0"
 castaway = "0.2.3"
 fxhash = "0.2.1"
+smallvec = "1.13.2"
+tokio-util = { version = "0.7.13", optional = true }
+tokio = { version = "1.43.0", optional = true }
 
 [package.metadata.docs.rs]
 all-features = true

crates/game-solver/Cargo.toml

@@ -16,7 +16,11 @@ pub mod transposition;
 
 use core::panic;
 #[cfg(feature = "rayon")]
+use tokio_util::sync::CancellationToken;
+#[cfg(feature = "rayon")]
 use std::hash::BuildHasher;
+#[cfg(feature = "rayon")]
+use std::sync::Arc;
 use std::sync::atomic::Ordering;
 
 use game::{upper_bound, GameState};
@@ -32,6 +36,8 @@ use thiserror::Error;
 pub enum GameSolveError<T: Game> {
     #[error("could not make a move")]
     MoveError(T::MoveError),
+    #[error("solving was cancelled")]
+    Cancelled,
 }
 
 /// Runs the two-player minimax variant on a zero-sum game.
@@ -304,42 +310,61 @@ pub type CollectedMoves<T> = Vec<Result<(<T as Game>::Move, isize), GameSolveErr
 ///
 /// A vector of tuples of the form `(move, score)`.
 #[cfg(feature = "rayon")]
-pub fn par_move_scores_with_hasher<
-    T: Game<Player = impl TwoPlayer + Sync + 'static> + Eq + Hash + Sync + Send + 'static,
+pub async fn par_move_scores_with_hasher<
+    T: Game<Player = impl TwoPlayer + Sync + Send + 'static> + Eq + Hash + Sync + Send + 'static,
     S,
 >(
     game: &T,
-    stats: Option<&Stats<T::Player>>
+    stats: Option<Arc<Stats<T::Player>>>,
+    cancellation_token: Option<CancellationToken>
 ) -> CollectedMoves<T>
 where
     T::Move: Sync + Send,
     T::MoveError: Sync + Send,
     S: BuildHasher + Default + Sync + Send + Clone + 'static,
 {
+    use itertools::Itertools;
+
     use crate::transposition::TranspositionCache;
-    use rayon::prelude::*;
     use std::sync::Arc;
 
-    // we need to collect it first as we cant parallelize an already non-parallel iterator
-    let all_moves = game.possible_moves().collect::<Vec<_>>();
-    let hashmap = Arc::new(TranspositionCache::<T, S>::new());
-
-    all_moves
-        .par_iter()
-        .map(move |m| {
+    let result = game.possible_moves().map(|m| {
+        let m = m.clone();
+        let game = game.clone();
+        let cancellation_token = cancellation_token.clone();
+        let stats = stats.clone();
+        
+        tokio::spawn(async move {
+            let hashmap = Arc::new(TranspositionCache::<T, S>::new());
             let mut board = game.clone();
             board
-                .make_move(m)
+                .make_move(&m)
                 .map_err(|err| GameSolveError::MoveError::<T>(err))?;
             // We flip the sign of the score because we want the score from the
             // perspective of the player pla`ying the move, not the player whose turn it is.
             let mut map = Arc::clone(&hashmap);
-            Ok((
-                (*m).clone(),
-                -solve(&board, &mut map, stats)?,
-            ))
+
+            let handle = tokio::spawn(async move {
+                solve(&board, &mut map, stats.as_deref()).map(|score| -score)
+            });
+
+            if let Some(cancellation_token) = cancellation_token {
+                tokio::select! {
+                    _ = cancellation_token.cancelled() => {
+                        Err(GameSolveError::Cancelled)
+                    },
+                    result = handle => {
+                        result.unwrap().map(|result| (m, result))
+                    }
+                }
+            } else {
+                handle.await.unwrap().map(|x| (m, x))
+            }
         })
-        .collect::<Vec<_>>()
+    })
+    .collect::<Vec<_>>();
+
+    futures::future::join_all(result).await.into_iter().map(|result| result.unwrap()).collect_vec()
 }
 
 /// Parallelized version of `move_scores`. (faster by a large margin)
@@ -353,21 +378,22 @@ where
 ///
 /// A vector of tuples of the form `(move, score)`.
 #[cfg(feature = "rayon")]
-pub fn par_move_scores<
-    T: Game<Player = impl TwoPlayer + Sync + 'static> + Eq + Hash + Sync + Send + 'static,
+pub async fn par_move_scores<
+    T: Game<Player = impl TwoPlayer + Sync + Send + 'static> + Eq + Hash + Sync + Send + 'static,
 >(
     game: &T,
-    stats: Option<&Stats<T::Player>>
+    stats: Option<Arc<Stats<T::Player>>>,
+    cancellation_token: Option<CancellationToken>
 ) -> CollectedMoves<T>
 where
     T::Move: Sync + Send,
     T::MoveError: Sync + Send,
 {
     if cfg!(feature = "xxhash") {
         use twox_hash::RandomXxHashBuilder64;
-        par_move_scores_with_hasher::<T, RandomXxHashBuilder64>(game, stats)
+        par_move_scores_with_hasher::<T, RandomXxHashBuilder64>(game, stats, cancellation_token).await
     } else {
         use std::collections::hash_map::RandomState;
-        par_move_scores_with_hasher::<T, RandomState>(game, stats)
+        par_move_scores_with_hasher::<T, RandomState>(game, stats, cancellation_token).await
     }
 }

crates/game-solver/src/lib.rs

@@ -2,7 +2,7 @@ use std::fmt::Debug;
 use std::hash::Hash;
 use std::marker::PhantomData;
 
-use fxhash::FxHashSet;
+use smallvec::SmallVec;
 
 /// We handle loopy games with a custom struct, `LoopyTracker`, which is a
 /// HashSet of some state T. This is used to keep track of the states that
@@ -12,9 +12,9 @@ use fxhash::FxHashSet;
 /// 
 /// We say `T` is the primary type, and `S` is some representation of `T` without the `LoopyTracker`.
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct LoopyTracker<S: Eq + Hash, T: Eq + Hash> {
-    visited: FxHashSet<T>,
+    visited: SmallVec<[T; 3]>,
     _phantom: PhantomData<S>,
 }
 
@@ -25,55 +25,47 @@ pub trait Loopy<S: Hash + Eq> where Self: Eq + Hash + Sized {
     fn without_tracker(&self) -> S;
 }
 
-impl<S: Eq + Hash, T: Eq + Hash> LoopyTracker<S, T> {
+impl<S: Eq + Hash, T: Eq + Hash + Loopy<S>> LoopyTracker<S, T> {
     /// Create a new `LoopyTracker`.
     pub fn new() -> Self {
         Self {
-            visited: FxHashSet::default(),
+            visited: SmallVec::new(),
             _phantom: PhantomData,
         }
     }
 
     /// Check if a state has been visited.
     pub fn has_visited(&self, state: &T) -> bool {
-        self.visited.contains(state)
+        self.visited.contains(&state)
     }
 
     /// Mark a state as visited.
     pub fn mark_visited(&mut self, state: T) {
-        self.visited.insert(state);
+        self.visited.push(state);
     }
 
     /// The number of states visited.
-    pub fn age(&self) -> usize {
+    pub fn halfmoves(&self) -> usize {
         self.visited.len()
     }
-}
 
-impl<S: Eq + Hash, T: Eq + Hash> Default for LoopyTracker<S, T> {
-    fn default() -> Self {
-        Self::new()
+    /// This should be called when an irreversible move is made,
+    /// in place of `Self::mark_visited`.
+    pub fn clear(&mut self) {
+        self.visited.clear();
     }
 }
 
-impl<S: Eq + Hash, T: Eq + Hash + Loopy<S>> PartialEq for LoopyTracker<S, T> {
-    fn eq(&self, other: &Self) -> bool {
-        for item in self.visited.iter() {
-            if !other.visited.contains(item) {
-                return false;
-            }
-        }
-
-        return true;
+impl<S: Eq + Hash, T: Eq + Hash + Loopy<S>> Default for LoopyTracker<S, T> {
+    fn default() -> Self {
+        Self::new()
     }
 }
 
-impl<S: Eq + Hash, T: Eq + Hash + Loopy<S>> Eq for LoopyTracker<S, T> {}
-
-impl<S: Eq + Hash, T: Eq + Hash + Loopy<S>> Hash for LoopyTracker<S, T> {
-    fn hash<H: std::hash::Hasher>(&self, hasher: &mut H) {
-        for item in self.visited.iter() {
-            item.without_tracker().hash(hasher);
-        }
-    }
-}
+// impl<S: Eq + Hash, T: Eq + Hash + Loopy<S>> Hash for LoopyTracker<S, T> {
+//     fn hash<H: std::hash::Hasher>(&self, hasher: &mut H) {
+//         for item in self.visited.iter() {
+//             item.hash(hasher);
+//         }
+//     }
+// }

crates/game-solver/src/loopy.rs

@@ -13,5 +13,5 @@ dialoguer = "0.11.0"
 clearscreen = "4.0.1"
 owo-colors = "4.1.0"
 tokio-util = "0.7.13"
-tokio = { version = "1.43.0", features = ["rt", "macros", "rt-multi-thread"] }
+tokio = { version = "1.43.0", features = ["rt", "macros", "rt-multi-thread", "signal"] }
 ratatui = "0.29.0"

crates/games-cli/Cargo.toml

@@ -1,13 +1,14 @@
 use std::{
-    fmt::Display, future::IntoFuture, sync::{
+    fmt::Display,
+    future::IntoFuture,
+    sync::{
         atomic::{AtomicU64, AtomicUsize, Ordering},
         Arc,
-    }, time::Duration
+    },
+    time::Duration,
 };
 
 use anyhow::Result;
-use tokio::select;
-use tokio_util::sync::CancellationToken;
 use core::hash::Hash;
 use game_solver::{
     game::Game,
@@ -29,6 +30,8 @@ use ratatui::{
     DefaultTerminal, Frame,
 };
 use std::fmt::Debug;
+use tokio::select;
+use tokio_util::sync::CancellationToken;
 
 use super::report::{scores::show_scores, stats::show_stats};
 
@@ -196,12 +199,8 @@ where
     let internal_stats = stats.clone();
 
     let game_thread = tokio::spawn(async move {
-        let game_solving_thread = tokio::spawn(async move {
-            par_move_scores(
-                &internal_game,
-                Some(internal_stats.as_ref()),
-            )
-        }).into_future();
+        let game_solving_thread =
+            par_move_scores(&internal_game, Some(internal_stats), Some(exit.clone()));
 
         select! {
             score = game_solving_thread => {
@@ -218,7 +217,7 @@ where
 
     show_stats::<T>(&stats);
     match move_scores {
-        Some(move_scores) => show_scores(&game, move_scores?),
+        Some(move_scores) => show_scores(&game, move_scores),
         None => eprintln!("Game solving was cancelled!"),
     }
 

crates/games-cli/src/human.rs

@@ -33,7 +33,7 @@ pub async fn play<
     match game.state() {
         GameState::Playable => {
             if plain {
-                robotic_output(game);
+                robotic_output(game).await;
             } else {
                 human_output(game).await.unwrap();
             }

crates/games-cli/src/lib.rs

@@ -15,6 +15,9 @@ where
             GameSolveError::MoveError(err) => {
                 eprintln!("Error making move: {:?}", err);
             },
+            GameSolveError::Cancelled => {
+                eprintln!("Game solving was cancelled!");
+            },
         }
         vec![]
     });