feat: Make query callable on multiple threads

crates/logic-eval/Cargo.toml

@@ -16,6 +16,13 @@ indexmap = { workspace = true }
 fxhash = { workspace = true }
 smallvec = { workspace = true }
 
+[dev-dependencies]
+criterion = "0.4"
+
+[[bench]]
+name = "query_threads"
+harness = false
+
 [[example]]
 name = "meal_recommendation"
 path = "examples/meal_recommendation.rs"

crates/logic-eval/README.md

@@ -47,7 +47,7 @@ Replace `src/main.rs` with:
 use logic_eval::{parse_str, Database, StrInterner};
 
 fn main() {
-    let mut db = Database::new();
+    let mut db = Database::default();
     let interner = StrInterner::new();
 
     let pantry = r#"
@@ -61,7 +61,6 @@ fn main() {
     "#;
 
     db.insert_dataset(parse_str(pantry, &interner).unwrap());
-    db.commit();
 
     let query = parse_str("can_make($Meal).", &interner).unwrap();
     let mut results = db.query(query);

crates/logic-eval/benches/query_threads.rs

@@ -0,0 +1,129 @@
+use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
+use logic_eval::{parse_str, ClauseDataset, Database, Expr, InternedStr, Name, StrInterner};
+
+const NODES: usize = 240;
+const QUERY_COUNT: usize = 64;
+
+type BenchName<'a> = Name<InternedStr<'a>>;
+
+fn node(index: usize) -> String {
+    format!("n{index}")
+}
+
+fn build_database<'a>(interner: &'a StrInterner) -> Database<BenchName<'a>> {
+    let mut source = String::new();
+
+    // A chain makes each ancestry query produce many answers. A few skip edges add branching so the
+    // engine does enough independent work for parallel throughput to show up in the benchmark.
+    for i in 0..NODES - 1 {
+        source.push_str(&format!("parent({}, {}).\n", node(i), node(i + 1)));
+    }
+    for i in 0..NODES - 3 {
+        if i % 7 == 0 {
+            source.push_str(&format!("parent({}, {}).\n", node(i), node(i + 3)));
+        }
+    }
+
+    source.push_str(
+        "
+        ancestor($X, $Y) :- parent($X, $Y).
+        ancestor($X, $Z) :- parent($X, $Y), ancestor($Y, $Z).
+        ",
+    );
+
+    let dataset: ClauseDataset<_> = parse_str(&source, interner).unwrap();
+    let mut db = Database::default();
+    db.insert_dataset(dataset);
+    db
+}
+
+fn build_queries<'a>(interner: &'a StrInterner) -> Vec<Expr<BenchName<'a>>> {
+    (0..QUERY_COUNT)
+        .map(|i| {
+            let root = i % (NODES / 3);
+            parse_str(&format!("ancestor({}, $Who)", node(root)), interner).unwrap()
+        })
+        .collect()
+}
+
+fn count_answers(db: &Database<BenchName<'_>>, query: Expr<BenchName<'_>>) -> usize {
+    let mut cx = db.query(query);
+    let mut count = 0;
+    while let Some(answer) = cx.prove_next() {
+        count += answer.count();
+    }
+    count
+}
+
+fn run_serial(db: &Database<BenchName<'_>>, queries: &[Expr<BenchName<'_>>]) -> usize {
+    queries
+        .iter()
+        .cloned()
+        .map(|query| count_answers(db, query))
+        .sum()
+}
+
+fn run_threaded(
+    db: &Database<BenchName<'_>>,
+    queries: &[Expr<BenchName<'_>>],
+    threads: usize,
+) -> usize {
+    let chunk_len = (queries.len() + threads - 1) / threads;
+
+    std::thread::scope(|scope| {
+        let handles = queries
+            .chunks(chunk_len)
+            .map(|chunk| {
+                scope.spawn(move || {
+                    chunk
+                        .iter()
+                        .cloned()
+                        .map(|query| count_answers(db, query))
+                        .sum::<usize>()
+                })
+            })
+            .collect::<Vec<_>>();
+
+        handles
+            .into_iter()
+            .map(|handle| handle.join().unwrap())
+            .sum()
+    })
+}
+
+fn benchmark_query_threads(c: &mut Criterion) {
+    let interner = StrInterner::new();
+    let db = build_database(&interner);
+    let queries = build_queries(&interner);
+    let expected = run_serial(&db, &queries);
+
+    let mut group = c.benchmark_group("logic_eval_query_threads");
+    group.sample_size(10);
+
+    group.bench_with_input(BenchmarkId::new("threads", 1), &1, |b, _| {
+        b.iter(|| {
+            let count = run_serial(&db, &queries);
+            assert_eq!(count, expected);
+            black_box(count)
+        });
+    });
+
+    for threads in [2, 4] {
+        group.bench_with_input(
+            BenchmarkId::new("threads", threads),
+            &threads,
+            |b, &threads| {
+                b.iter(|| {
+                    let count = run_threaded(&db, &queries, threads);
+                    assert_eq!(count, expected);
+                    black_box(count)
+                });
+            },
+        );
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, benchmark_query_threads);
+criterion_main!(benches);

crates/logic-eval/examples/access_control.rs

@@ -6,7 +6,7 @@
 use logic_eval::{parse_str, Database, StrInterner};
 
 fn main() {
-    let mut db = Database::new();
+    let mut db = Database::default();
     let interner = StrInterner::new();
 
     // Facts describe users, roles, and documents.
@@ -27,7 +27,6 @@ fn main() {
     "#;
 
     db.insert_dataset(parse_str(policy, &interner).unwrap());
-    db.commit();
 
     println!("Who can read the handbook?");
     {

crates/logic-eval/examples/dependency_graph.rs

@@ -5,7 +5,7 @@
 use logic_eval::{parse_str, Database, StrInterner};
 
 fn main() {
-    let mut db = Database::new();
+    let mut db = Database::default();
     let interner = StrInterner::new();
 
     // Direct dependencies are facts. The recursive rule finds indirect dependencies.
@@ -24,7 +24,6 @@ fn main() {
     "#;
 
     db.insert_dataset(parse_str(graph, &interner).unwrap());
-    db.commit();
 
     println!("app requires:");
     {

crates/logic-eval/examples/meal_recommendation.rs

@@ -6,7 +6,7 @@
 use logic_eval::{parse_str, Database, StrInterner};
 
 fn main() {
-    let mut db = Database::new();
+    let mut db = Database::default();
     let interner = StrInterner::new();
 
     // Facts describe what is true. Rules describe what can be inferred.
@@ -22,7 +22,6 @@ fn main() {
 
     // Load the facts and rules into the database before asking questions.
     db.insert_dataset(parse_str(pantry, &interner).unwrap());
-    db.commit();
 
     // $Meal is a variable. logic-eval will find every value that works.
     let query = parse_str("can_make($Meal).", &interner).unwrap();

crates/logic-eval/src/lib.rs

@@ -15,7 +15,7 @@ pub use parse::{
 pub use prove::{
     common::Atom,
     db::{ClauseIter, ClauseRef, Database},
-    prover::ProveCx,
+    proof_engine::QueryCx,
 };
 
 /// Re-exports of the interning types used by this crate.

crates/logic-eval/src/parse/repr.rs

@@ -1,5 +1,5 @@
 use crate::{
-    prove::{canonical as canon, prover::Integer},
+    prove::{canonical as canon, proof_engine::AtomId},
     Atom,
 };
 use std::{
@@ -124,7 +124,7 @@ impl<T> Clause<T> {
     }
 }
 
-impl Clause<Integer> {
+impl Clause<AtomId> {
     /// Returns `true` if the clause needs SLG resolution (tabling).
     ///
     /// If a clause has left or mid recursion, it must be handled by tabling.
@@ -145,7 +145,7 @@ impl Clause<Integer> {
 
         // === Internal helper functions ===
 
-        fn helper(expr: &Expr<Integer>, head: &Term<Integer>) -> bool {
+        fn helper(expr: &Expr<AtomId>, head: &Term<AtomId>) -> bool {
             match expr {
                 Expr::Term(term) => term == head,
                 Expr::Not(arg) => helper(arg, head),

crates/logic-eval/src/prove/canonical.rs

@@ -1,6 +1,6 @@
 use crate::{
     prove::{
-        prover::Integer,
+        proof_engine::AtomId,
         repr::{TermId, TermStorage, TermViewMut},
     },
     Atom, Expr, Map, Term,
@@ -9,22 +9,25 @@ use crate::{
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub(crate) struct CanonicalTermId(TermId);
 
-pub(crate) fn canonicalize_term_id(stor: &mut TermStorage<Integer>, id: TermId) -> CanonicalTermId {
-    let mut view = stor.get_term_mut(id);
+pub(crate) fn canonicalize_term_id(
+    query_storage: &mut TermStorage<AtomId>,
+    id: TermId,
+) -> CanonicalTermId {
+    let mut view = query_storage.get_term_mut(id);
     canonicalize_term_view(&mut view);
     CanonicalTermId(view.id())
 }
 
 /// e.g. f($X, $Y, $X) -> f($0, $1, $0)
-pub(crate) fn canonicalize_term(term: &mut Term<Integer>) {
+pub(crate) fn canonicalize_term(term: &mut Term<AtomId>) {
     let mut c = canonicalizer();
     term.replace_variables(|functor| *functor = c(*functor));
 }
 
 /// Applies [`canonicalize_term`] to each term without crossing term boundaries.
 ///
 /// e.g. f($X), g($Y, $X) -> f($0), g($0, $1) (not f($0), g($1, $0))
-pub(crate) fn canonicalize_expr_on_term(expr: &mut Expr<Integer>) {
+pub(crate) fn canonicalize_expr_on_term(expr: &mut Expr<AtomId>) {
     match expr {
         Expr::Term(term) => canonicalize_term(term),
         Expr::Not(arg) => canonicalize_expr_on_term(arg),
@@ -36,7 +39,7 @@ pub(crate) fn canonicalize_expr_on_term(expr: &mut Expr<Integer>) {
     }
 }
 
-pub(crate) fn canonicalize_term_view(view: &mut TermViewMut<'_, Integer>) {
+pub(crate) fn canonicalize_term_view(view: &mut TermViewMut<'_, AtomId>) {
     let mut c = canonicalizer();
     view.replace_with(|functor| {
         if functor.is_variable() {
@@ -47,12 +50,12 @@ pub(crate) fn canonicalize_term_view(view: &mut TermViewMut<'_, Integer>) {
     });
 }
 
-fn canonicalizer() -> impl FnMut(Integer) -> Integer {
+fn canonicalizer() -> impl FnMut(AtomId) -> AtomId {
     let mut map = Map::default();
-    move |functor: Integer| {
+    move |functor: AtomId| {
         if functor.is_variable() {
-            let next_int = map.len() as u32;
-            *map.entry(functor).or_insert(Integer::variable(next_int))
+            let next_id = map.len() as u32;
+            *map.entry(functor).or_insert(AtomId::variable(next_id))
         } else {
             functor
         }

crates/logic-eval/src/prove/common.rs

@@ -31,3 +31,36 @@ impl<'int> Atom for Name<any_intern::Interned<'int, str>> {
         self.starts_with(VAR_PREFIX)
     }
 }
+
+impl Atom for String {
+    fn is_variable(&self) -> bool {
+        self.starts_with(VAR_PREFIX)
+    }
+}
+
+impl<T> Atom for Box<T>
+where
+    T: Atom + AsRef<str>,
+{
+    fn is_variable(&self) -> bool {
+        (**self).as_ref().starts_with(VAR_PREFIX)
+    }
+}
+
+impl<T> Atom for std::rc::Rc<T>
+where
+    T: Atom + AsRef<str>,
+{
+    fn is_variable(&self) -> bool {
+        (**self).as_ref().starts_with(VAR_PREFIX)
+    }
+}
+
+impl<T> Atom for std::sync::Arc<T>
+where
+    T: Atom + AsRef<str>,
+{
+    fn is_variable(&self) -> bool {
+        (**self).as_ref().starts_with(VAR_PREFIX)
+    }
+}