Improved Concurrent metadata fetching (#29)

* chore: update rust-toolchain to latest stable release

* feat: concurrent metadata fetching

* more concurrency

* feat: sort_candidates is now async

* fix: fmt and clippy

* refactor: no more channels

* refactor: everything concurrent

* fix: tests and clippy

* feat: runtime agnostic impl

* fix: forgot public docs

* fix: remove outdated comment

---------

Co-authored-by: Adolfo Ochagavía <github@adolfo.ochagavia.nl>
Co-authored-by: Tim de Jager <tim@prefix.dev>

Author: 3 people

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "resolvo"
 version = "0.3.0"
-authors = ["Adolfo Ochagavía <github@adolfo.ochagavia.nl>", "Bas Zalmstra <zalmstra.bas@gmail.com>", "Tim de Jager <tdejager89@gmail.com>" ]
+authors = ["Adolfo Ochagavía <github@adolfo.ochagavia.nl>", "Bas Zalmstra <zalmstra.bas@gmail.com>", "Tim de Jager <tdejager89@gmail.com>"]
 description = "Fast package resolver written in Rust (CDCL based SAT solving)"
 keywords = ["dependency", "solver", "version"]
 categories = ["algorithms"]
@@ -10,17 +10,25 @@ repository = "https://github.com/mamba-org/resolvo"
 license = "BSD-3-Clause"
 edition = "2021"
 readme = "README.md"
+resolver = "2"
 
 [dependencies]
-itertools = "0.11.0"
+itertools = "0.12.1"
 petgraph = "0.6.4"
 tracing = "0.1.37"
 elsa = "1.9.0"
 bitvec = "1.0.1"
 serde = { version = "1.0", features = ["derive"], optional = true }
+futures = { version = "0.3.30", default-features = false, features = ["alloc"] }
+event-listener = "5.0.0"
+
+tokio = { version = "1.35.1", features = ["rt"], optional = true }
+async-std = { version = "1.12.0", default-features = false, features = ["alloc", "default"], optional = true }
 
 [dev-dependencies]
 insta = "1.31.0"
 indexmap = "2.0.0"
 proptest = "1.2.0"
 tracing-test = { version = "0.2.4", features = ["no-env-filter"] }
+tokio = { version = "1.35.1", features = ["time", "rt"] }
+resolvo = { path = ".", features = ["tokio"] }

rust-toolchain

@@ -1 +1 @@
-1.72.0
+1.75.0

src/lib.rs

@@ -14,6 +14,7 @@ pub(crate) mod internal;
 mod pool;
 pub mod problem;
 pub mod range;
+pub mod runtime;
 mod solvable;
 mod solver;
 
@@ -30,9 +31,10 @@ use std::{
     any::Any,
     fmt::{Debug, Display},
     hash::Hash,
+    rc::Rc,
 };
 
-/// The solver is based around the fact that for for every package name we are trying to find a
+/// The solver is based around the fact that for every package name we are trying to find a
 /// single variant. Variants are grouped by their respective package name. A package name is
 /// anything that we can compare and hash for uniqueness checks.
 ///
@@ -44,7 +46,7 @@ pub trait PackageName: Eq + Hash {}
 
 impl<N: Eq + Hash> PackageName for N {}
 
-/// A [`VersionSet`] is describes a set of "versions". The trait defines whether a given version
+/// A [`VersionSet`] describes a set of "versions". The trait defines whether a given version
 /// is part of the set or not.
 ///
 /// One could implement [`VersionSet`] for [`std::ops::Range<u32>`] where the implementation
@@ -61,21 +63,26 @@ pub trait VersionSet: Debug + Display + Clone + Eq + Hash {
 /// packages that are available in the system.
 pub trait DependencyProvider<VS: VersionSet, N: PackageName = String>: Sized {
     /// Returns the [`Pool`] that is used to allocate the Ids returned from this instance
-    fn pool(&self) -> &Pool<VS, N>;
+    fn pool(&self) -> Rc<Pool<VS, N>>;
 
     /// Sort the specified solvables based on which solvable to try first. The solver will
     /// iteratively try to select the highest version. If a conflict is found with the highest
     /// version the next version is tried. This continues until a solution is found.
-    fn sort_candidates(&self, solver: &SolverCache<VS, N, Self>, solvables: &mut [SolvableId]);
+    #[allow(async_fn_in_trait)]
+    async fn sort_candidates(
+        &self,
+        solver: &SolverCache<VS, N, Self>,
+        solvables: &mut [SolvableId],
+    );
 
-    /// Returns a list of solvables that should be considered when a package with the given name is
+    /// Obtains a list of solvables that should be considered when a package with the given name is
     /// requested.
-    ///
-    /// Returns `None` if no such package exist.
-    fn get_candidates(&self, name: NameId) -> Option<Candidates>;
+    #[allow(async_fn_in_trait)]
+    async fn get_candidates(&self, name: NameId) -> Option<Candidates>;
 
     /// Returns the dependencies for the specified solvable.
-    fn get_dependencies(&self, solvable: SolvableId) -> Dependencies;
+    #[allow(async_fn_in_trait)]
+    async fn get_dependencies(&self, solvable: SolvableId) -> Dependencies;
 
     /// Whether the solver should stop the dependency resolution algorithm.
     ///
@@ -126,6 +133,7 @@ pub struct Candidates {
 }
 
 /// Holds information about the dependencies of a package.
+#[derive(Debug, Clone)]
 pub enum Dependencies {
     /// The dependencies are known.
     Known(KnownDependencies),

src/problem.rs

@@ -4,18 +4,17 @@ use std::collections::{HashMap, HashSet};
 use std::fmt;
 use std::fmt::{Display, Formatter};
 use std::hash::Hash;
-
 use std::rc::Rc;
 
 use itertools::Itertools;
 use petgraph::graph::{DiGraph, EdgeIndex, EdgeReference, NodeIndex};
 use petgraph::visit::{Bfs, DfsPostOrder, EdgeRef};
 use petgraph::Direction;
 
-use crate::internal::id::StringId;
 use crate::{
-    internal::id::{ClauseId, SolvableId, VersionSetId},
+    internal::id::{ClauseId, SolvableId, StringId, VersionSetId},
     pool::Pool,
+    runtime::AsyncRuntime,
     solver::{clause::Clause, Solver},
     DependencyProvider, PackageName, SolvableDisplay, VersionSet,
 };
@@ -41,9 +40,9 @@ impl Problem {
     }
 
     /// Generates a graph representation of the problem (see [`ProblemGraph`] for details)
-    pub fn graph<VS: VersionSet, N: PackageName, D: DependencyProvider<VS, N>>(
+    pub fn graph<VS: VersionSet, N: PackageName, D: DependencyProvider<VS, N>, RT: AsyncRuntime>(
         &self,
-        solver: &Solver<VS, N, D>,
+        solver: &Solver<VS, N, D, RT>,
     ) -> ProblemGraph {
         let mut graph = DiGraph::<ProblemNode, ProblemEdge>::default();
         let mut nodes: HashMap<SolvableId, NodeIndex> = HashMap::default();
@@ -53,7 +52,7 @@ impl Problem {
         let unresolved_node = graph.add_node(ProblemNode::UnresolvedDependency);
 
         for clause_id in &self.clauses {
-            let clause = &solver.clauses[*clause_id].kind;
+            let clause = &solver.clauses.borrow()[*clause_id].kind;
             match clause {
                 Clause::InstallRoot => (),
                 Clause::Excluded(solvable, reason) => {
@@ -73,7 +72,7 @@ impl Problem {
                 &Clause::Requires(package_id, version_set_id) => {
                     let package_node = Self::add_node(&mut graph, &mut nodes, package_id);
 
-                    let candidates = solver.cache.get_or_cache_sorted_candidates(version_set_id).unwrap_or_else(|_| {
+                    let candidates = solver.async_runtime.block_on(solver.cache.get_or_cache_sorted_candidates(version_set_id)).unwrap_or_else(|_| {
                         unreachable!("The version set was used in the solver, so it must have been cached. Therefore cancellation is impossible here and we cannot get an `Err(...)`")
                     });
                     if candidates.is_empty() {
@@ -167,13 +166,15 @@ impl Problem {
         N: PackageName + Display,
         D: DependencyProvider<VS, N>,
         M: SolvableDisplay<VS, N>,
+        RT: AsyncRuntime,
     >(
         &self,
-        solver: &'a Solver<VS, N, D>,
+        solver: &'a Solver<VS, N, D, RT>,
+        pool: Rc<Pool<VS, N>>,
         merged_solvable_display: &'a M,
     ) -> DisplayUnsat<'a, VS, N, M> {
         let graph = self.graph(solver);
-        DisplayUnsat::new(graph, solver.pool(), merged_solvable_display)
+        DisplayUnsat::new(graph, pool, merged_solvable_display)
     }
 }
 
@@ -515,7 +516,7 @@ pub struct DisplayUnsat<'pool, VS: VersionSet, N: PackageName + Display, M: Solv
     merged_candidates: HashMap<SolvableId, Rc<MergedProblemNode>>,
     installable_set: HashSet<NodeIndex>,
     missing_set: HashSet<NodeIndex>,
-    pool: &'pool Pool<VS, N>,
+    pool: Rc<Pool<VS, N>>,
     merged_solvable_display: &'pool M,
 }
 
@@ -524,10 +525,10 @@ impl<'pool, VS: VersionSet, N: PackageName + Display, M: SolvableDisplay<VS, N>>
 {
     pub(crate) fn new(
         graph: ProblemGraph,
-        pool: &'pool Pool<VS, N>,
+        pool: Rc<Pool<VS, N>>,
         merged_solvable_display: &'pool M,
     ) -> Self {
-        let merged_candidates = graph.simplify(pool);
+        let merged_candidates = graph.simplify(&pool);
         let installable_set = graph.get_installable_set();
         let missing_set = graph.get_missing_set();
 
@@ -669,10 +670,10 @@ impl<'pool, VS: VersionSet, N: PackageName + Display, M: SolvableDisplay<VS, N>>
                     let version = if let Some(merged) = self.merged_candidates.get(&solvable_id) {
                         reported.extend(merged.ids.iter().cloned());
                         self.merged_solvable_display
-                            .display_candidates(self.pool, &merged.ids)
+                            .display_candidates(&self.pool, &merged.ids)
                     } else {
                         self.merged_solvable_display
-                            .display_candidates(self.pool, &[solvable_id])
+                            .display_candidates(&self.pool, &[solvable_id])
                     };
 
                     let excluded = graph
@@ -796,9 +797,9 @@ impl<VS: VersionSet, N: PackageName + Display, M: SolvableDisplay<VS, N>> fmt::D
                         writeln!(
                             f,
                             "{indent}{} {} is locked, but another version is required as reported above",
-                            locked.name.display(self.pool),
+                            locked.name.display(&self.pool),
                             self.merged_solvable_display
-                                .display_candidates(self.pool, &[solvable_id])
+                                .display_candidates(&self.pool, &[solvable_id])
                         )?;
                     }
                     ConflictCause::Excluded => continue,

src/range.rs

@@ -409,7 +409,7 @@ pub mod tests {
                     segments.push((start_bound, Unbounded));
                 }
 
-                return Range { segments }.check_invariants();
+                Range { segments }.check_invariants()
             })
     }
 

src/runtime.rs

@@ -0,0 +1,78 @@
+//! Solving in resolvo is a compute heavy operation. However, while computing the solver will
+//! request additional information from the [`crate::DependencyProvider`] and a dependency provider
+//! might want to perform multiple requests concurrently. To that end the
+//! [`crate::DependencyProvider`]s methods are async. The implementer can implement the async
+//! operations in any way they choose including with any runtime they choose.
+//! However, the solver itself is completely single threaded, but it still has to await the calls to
+//! the dependency provider. Using the [`AsyncRuntime`] allows the caller of the solver to choose
+//! how to await the futures.
+//!
+//! By default, the solver uses the [`NowOrNeverRuntime`] runtime which polls any future once. If
+//! the future yields (thus requiring an additional poll) the runtime panics. If the methods of
+//! [`crate::DependencyProvider`] do not yield (e.g. do not `.await`) this will suffice.
+//!
+//! Only if the [`crate::DependencyProvider`] implementation yields you will need to provide a
+//! [`AsyncRuntime`] to the solver.
+//!
+//! ## `tokio`
+//!
+//! The [`AsyncRuntime`] trait is implemented both for [`tokio::runtime::Handle`] and for
+//! [`tokio::runtime::Runtime`].
+//!
+//! ## `async-std`
+//!
+//! Use the [`AsyncStdRuntime`] struct to block on async methods from the
+//! [`crate::DependencyProvider`] using the `async-std` executor.
+
+use futures::FutureExt;
+use std::future::Future;
+
+/// A trait to wrap an async runtime.
+pub trait AsyncRuntime {
+    /// Runs the given future on the current thread, blocking until it is complete, and yielding its
+    /// resolved result.
+    fn block_on<F: Future>(&self, f: F) -> F::Output;
+}
+
+/// The simplest runtime possible evaluates and consumes the future, returning the resulting
+/// output if the future is ready after the first call to [`Future::poll`]. If the future does
+/// yield the runtime panics.
+///
+/// This assumes that the passed in future never yields. For purely blocking computations this
+/// is the preferred method since it also incurs very little overhead and doesn't require the
+/// inclusion of a heavy-weight runtime.
+#[derive(Default, Copy, Clone)]
+pub struct NowOrNeverRuntime;
+
+impl AsyncRuntime for NowOrNeverRuntime {
+    fn block_on<F: Future>(&self, f: F) -> F::Output {
+        f.now_or_never()
+            .expect("can only use non-yielding futures with the NowOrNeverRuntime")
+    }
+}
+
+#[cfg(feature = "tokio")]
+impl AsyncRuntime for tokio::runtime::Handle {
+    fn block_on<F: Future>(&self, f: F) -> F::Output {
+        self.block_on(f)
+    }
+}
+
+#[cfg(feature = "tokio")]
+impl AsyncRuntime for tokio::runtime::Runtime {
+    fn block_on<F: Future>(&self, f: F) -> F::Output {
+        self.block_on(f)
+    }
+}
+
+/// An implementation of [`AsyncRuntime`] that spawns and awaits any passed future on the current
+/// thread.
+#[cfg(feature = "async-std")]
+pub struct AsyncStdRuntime;
+
+#[cfg(feature = "async-std")]
+impl AsyncRuntime for AsyncStdRuntime {
+    fn block_on<F: Future>(&self, f: F) -> F::Output {
+        async_std::task::block_on(f)
+    }
+}