scripts.benchmark_prolog_clips

Reference › scripts › benchmark_prolog_clips

A standalone micro-benchmark that compares the CLIPS rule engine with SWI-Prolog tabling for transitive closure. A human runs it, and tests/scripts/test_benchmark_prolog_clips.py imports it.

Invocation: --nodes (default 100), --edges (200), --queries (50), --seed (42), --output FILE for a JSON report and --clips-binary (default /usr/bin/clips). It exits 2 if there are fewer than two nodes or fewer edges than nodes - 1, because the generated graph must contain a connected chain.

_generate_dag builds a seeded random directed acyclic graph. _benchmark_swi loads the edges into SWI-Prolog with a tabled reachability rule and times the queries. _write_clips_file writes the same graph as a CLIPS program and _benchmark_clips runs it with the CLIPS binary. A CLIPS failure is reported as a warning and does not stop the run: the report holds both result blocks, the graph size and the warning.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/benchmark_prolog_clips.py

#!/usr/bin/env python3
"""Standalone CLIPS-vs-SWI-Prolog-tabling micro-benchmark.

Generates a deterministic DAG, measures:

- SWI-Prolog (janus-swi) consulting knowledge.pl and querying the already-tabled
  depends_on_transitive/2 predicate.
- CLIPS 6.x running an equivalent transitive-closure rule set.

Reports both subprocess wall-clock (real adoption cost) and CLIPS's internal
(run) inference time (fair like-for-like vs SWI query time), as requested in
the review of plans/prolog-tabling-restraint-plan.md.

Usage::

    PYTHONPATH=/path/to/project/src python scripts/benchmark_prolog_clips.py \
        --nodes 100 --edges 200 --queries 50 --output /tmp/bench.json

If the ``clips`` binary is not found, the script emits SWI-only results and a
warning instead of failing.
"""

from __future__ import annotations

import argparse
import json
import random
import re
import subprocess
import sys
import tempfile
import time
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
PROLOG_DIR = REPO_ROOT / "src" / "orchestrata_ultimate" / "prolog"


def _generate_dag(num_nodes: int, num_edges: int, seed: int) -> list[tuple[str, str]]:
    """Return a list of (src, dst) edges forming a DAG (src index < dst index)."""
    rng = random.Random(seed)
    nodes = [f"node_{i}" for i in range(num_nodes)]

    # Guarantee a single long chain so every node is reachable from node_0.
    edges: set[tuple[str, str]] = {
        (nodes[i], nodes[i + 1]) for i in range(num_nodes - 1)
    }

    while len(edges) < num_edges:
        i = rng.randint(0, num_nodes - 2)
        j = rng.randint(i + 1, num_nodes - 1)
        edges.add((nodes[i], nodes[j]))

    return sorted(edges)


def _write_clips_file(path: Path, edges: list[tuple[str, str]]) -> None:
    # CLIPS 6.21 compatibility (the distro build on the dev/test fleet):
    # find-all-facts is a 6.3+ function, variables bound by top-level `bind`
    # do not persist across commands, and (declare (salience ...)) crashes
    # this build — so time and count via defglobals instead. The counting
    # rule fires exactly once per DISTINCT reach fact (the Rete ignores
    # duplicate assertions), so no salience ordering is needed: every reach
    # fact is counted once regardless of firing order.
    lines = [
        "(deftemplate edge (slot src) (slot dst))",
        "(deftemplate reach (slot src) (slot dst))",
        "",
        "(defglobal ?*reach-count* = 0)",
        "(defglobal ?*start* = 0)",
        "",
        "(deffacts edges",
    ]
    for src, dst in edges:
        lines.append(f"  (edge (src {src}) (dst {dst}))")
    lines.extend(
        [
            ")",
            "",
            f"; Single-source reachability from {edges[0][0]} (matches SWI query).",
            "(defrule reach-direct",
            f"  (edge (src {edges[0][0]}) (dst ?y))",
            "  =>",
            f"  (assert (reach (src {edges[0][0]}) (dst ?y))))",
            "",
            "(defrule reach-transitive",
            f"  (reach (src {edges[0][0]}) (dst ?y))",
            "  (edge (src ?y) (dst ?z))",
            "  =>",
            f"  (assert (reach (src {edges[0][0]}) (dst ?z))))",
            "",
            "(defrule count-reach",
            f"  (reach (src {edges[0][0]}) (dst ?))",
            "  =>",
            "  (bind ?*reach-count* (+ ?*reach-count* 1)))",
            "",
            "(reset)",
            "(bind ?*start* (time))",
            "(run)",
            '(printout t "inference_seconds=" (- (time) ?*start*) crlf)',
            '(printout t "reach_count=" ?*reach-count* crlf)',
            "(exit)",
        ]
    )
    path.write_text("\n".join(lines))


def _benchmark_swi(edges: list[tuple[str, str]], query_count: int) -> dict:
    try:
        import janus_swi as janus
    except ImportError as exc:
        raise RuntimeError("janus-swi is required for the SWI benchmark") from exc

    knowledge_pl = PROLOG_DIR / "knowledge.pl"
    if not knowledge_pl.exists():
        raise FileNotFoundError(f"knowledge.pl not found at {knowledge_pl}")

    start = time.perf_counter()
    janus.consult(str(knowledge_pl))
    for src, dst in edges:
        janus.query_once(f"assertz(knowledge:depends_on('{src}', '{dst}'))")
    setup_time = time.perf_counter() - start

    root = edges[0][0]
    goal = f"knowledge:depends_on_transitive('{root}', B)"

    start = time.perf_counter()
    cold_results = list(janus.query(goal))
    cold_time = time.perf_counter() - start

    start = time.perf_counter()
    for _ in range(query_count):
        list(janus.query(goal))
    warm_time = time.perf_counter() - start

    reach_count = len(cold_results)

    return {
        "setup_seconds": setup_time,
        "cold_query_seconds": cold_time,
        "warm_queries_total_seconds": warm_time,
        "warm_query_mean_seconds": warm_time / query_count if query_count else None,
        "reach_count": reach_count,
    }


def _benchmark_clips(edges: list[tuple[str, str]], clips_bin: Path) -> dict | None:
    if not clips_bin.exists():
        return None

    with tempfile.TemporaryDirectory() as tmp:
        clp_path = Path(tmp) / "bench.clp"
        _write_clips_file(clp_path, edges)

        start = time.perf_counter()
        proc = subprocess.run(
            [str(clips_bin), "-f", str(clp_path)],
            capture_output=True,
            text=True,
        )
        subprocess_seconds = time.perf_counter() - start

    if proc.returncode != 0:
        raise RuntimeError(f"CLIPS failed: {proc.stderr}")

    output = proc.stdout + proc.stderr
    inference_match = re.search(r"inference_seconds=([0-9.eE+-]+)", output)
    count_match = re.search(r"reach_count=(\d+)", output)

    return {
        "subprocess_seconds": subprocess_seconds,
        "inference_seconds": (
            float(inference_match.group(1)) if inference_match else None
        ),
        "reach_count": int(count_match.group(1)) if count_match else None,
        "raw_output": output,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Micro-benchmark: CLIPS vs SWI-Prolog tabling transitive closure"
    )
    parser.add_argument("--nodes", type=int, default=100)
    parser.add_argument("--edges", type=int, default=200)
    parser.add_argument("--queries", type=int, default=50)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--output", type=Path, default=None)
    parser.add_argument("--clips-binary", type=Path, default=Path("/usr/bin/clips"))
    args = parser.parse_args()

    if args.nodes < 2:
        print("--nodes must be at least 2 to form a chain", file=sys.stderr)
        return 2

    if args.edges < args.nodes - 1:
        print(
            "--edges must be at least --nodes - 1 to form a connected chain",
            file=sys.stderr,
        )
        return 2

    edges = _generate_dag(args.nodes, args.edges, args.seed)

    swi_result = _benchmark_swi(edges, args.queries)

    clips_result: dict | None = None
    clips_warning: str | None = None
    try:
        clips_result = _benchmark_clips(edges, args.clips_binary)
    except Exception as exc:  # noqa: BLE001
        clips_warning = f"CLIPS benchmark failed: {exc}"
        print(clips_warning, file=sys.stderr)

    report = {
        "nodes": args.nodes,
        "edges": len(edges),
        "queries": args.queries,
        "seed": args.seed,
        "swi_prolog": swi_result,
        "clips": clips_result,
        "clips_warning": clips_warning,
    }

    if clips_result and clips_result.get("inference_seconds"):
        swi_total = swi_result["cold_query_seconds"]
        report["speedups"] = {
            "swi_cold_vs_clips_subprocess": (
                subprocess_seconds / swi_total
                if (subprocess_seconds := clips_result["subprocess_seconds"])
                else None
            ),
            "swi_cold_vs_clips_inference": (
                clips_result["inference_seconds"] / swi_total
            ),
        }
    else:
        report["speedups"] = None

    print(json.dumps(report, indent=2))

    if args.output:
        args.output.write_text(json.dumps(report, indent=2))
        print(f"Wrote report to {args.output}", file=sys.stderr)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())