Merge remote-tracking branch 'origin/develop'

pyk/.gitignore

@@ -12,4 +12,5 @@ __pycache__/
 *.debug-log
 
 .idea/
+.vscode/
 .DS_Store

pyk/Makefile

@@ -105,6 +105,9 @@ pyupgrade: poetry-install
 	$(POETRY_RUN) pyupgrade --py310-plus $(SRC_FILES)
 
 
+pr: format check pyupgrade
+	git rebase -i develop
+
 # Documentation
 
 DOCS_API_DIR   := docs/api

pyk/src/pyk/cterm/cterm.py

@@ -24,7 +24,7 @@
 from ..prelude.k import GENERATED_TOP_CELL, K
 from ..prelude.kbool import andBool, orBool
 from ..prelude.ml import is_bottom, is_top, mlAnd, mlBottom, mlEquals, mlEqualsTrue, mlImplies, mlTop
-from ..utils import unique
+from ..utils import not_none, unique
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator
@@ -203,6 +203,8 @@ def anti_unify(
             - ``csubst2``: Constrained substitution to apply to `cterm` to obtain `other`.
         """
         new_config, self_subst, other_subst = anti_unify(self.config, other.config, kdef=kdef)
+        # todo: It's not able to distinguish between constraints in different cterms,
+        #  because variable names may be used inconsistently in different cterms.
         common_constraints = [constraint for constraint in self.constraints if constraint in other.constraints]
         self_unique_constraints = [
             ml_pred_to_bool(constraint) for constraint in self.constraints if constraint not in other.constraints
@@ -434,3 +436,30 @@ def cterm_build_rule(
         keep_vars,
         defunc_with=defunc_with,
     )
+
+
+def cterms_anti_unify(
+    cterms: Iterable[CTerm], keep_values: bool = False, kdef: KDefinition | None = None
+) -> tuple[CTerm, list[CSubst]]:
+    """Given many `CTerm` instances, find a more general `CTerm` which can instantiate to all.
+
+    Args:
+        cterms: `CTerm`s to consider for finding a more general `CTerm` with this one.
+        keep_values: do not discard information about abstracted variables in returned result.
+        kdef (optional): `KDefinition` to make analysis more precise.
+
+    Returns:
+        A tuple ``(cterm, csubsts)`` where
+
+        - ``cterm``: More general `CTerm` than any of the input `CTerm`s.
+        - ``csubsts``: List of `CSubst` which, when applied to `cterm`, yield the input `CTerm`s.
+    """
+    # TODO: optimize this function, reduce useless auto-generated variables.
+    cterms = list(cterms)
+    if not cterms:
+        raise ValueError('Anti-unification failed, no CTerms provided')
+    merged_cterm = cterms[0]
+    for cterm in cterms[1:]:
+        merged_cterm = merged_cterm.anti_unify(cterm, keep_values, kdef)[0]
+    csubsts = [not_none(cterm_match(merged_cterm, cterm)) for cterm in cterms]
+    return merged_cterm, csubsts

pyk/src/pyk/kcfg/kcfg.py

@@ -942,12 +942,20 @@ def contains_merged_edge(self, edge: MergedEdge) -> bool:
             return edge == other
         return False
 
-    def create_merged_edge(self, source_id: NodeIdLike, target_id: NodeIdLike, edges: Iterable[Edge]) -> MergedEdge:
+    def create_merged_edge(
+        self, source_id: NodeIdLike, target_id: NodeIdLike, edges: Iterable[Edge | MergedEdge]
+    ) -> MergedEdge:
         if len(list(edges)) == 0:
             raise ValueError(f'Cannot build KCFG MergedEdge with no edges: {edges}')
         source = self.node(source_id)
         target = self.node(target_id)
-        merged_edge = KCFG.MergedEdge(source, target, tuple(edges))
+        flatten_edges: list[KCFG.Edge] = []
+        for edge in edges:
+            if isinstance(edge, KCFG.MergedEdge):
+                flatten_edges.extend(edge.edges)
+            else:
+                flatten_edges.append(edge)
+        merged_edge = KCFG.MergedEdge(source, target, tuple(flatten_edges))
         self.add_successor(merged_edge)
         return merged_edge
 
@@ -959,6 +967,13 @@ def remove_merged_edge(self, source_id: NodeIdLike, target_id: NodeIdLike) -> No
             raise ValueError(f'MergedEdge does not exist: {source_id} -> {target_id}')
         self._merged_edges.pop(source_id)
 
+    def general_edges(
+        self, *, source_id: NodeIdLike | None = None, target_id: NodeIdLike | None = None
+    ) -> list[Edge | MergedEdge]:
+        return self.edges(source_id=source_id, target_id=target_id) + self.merged_edges(
+            source_id=source_id, target_id=target_id
+        )
+
     def cover(self, source_id: NodeIdLike, target_id: NodeIdLike) -> Cover | None:
         source_id = self._resolve(source_id)
         target_id = self._resolve(target_id)

pyk/src/pyk/kcfg/minimize.py

@@ -3,20 +3,32 @@
 from functools import reduce
 from typing import TYPE_CHECKING
 
-from ..cterm import CTerm
-from ..utils import not_none, single
+from pyk.cterm import CTerm
+from pyk.cterm.cterm import cterms_anti_unify
+from pyk.utils import not_none, partition, single
+
+from .semantics import DefaultSemantics
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 
+    from pyk.kast.outer import KDefinition
+
     from .kcfg import KCFG, NodeIdLike
+    from .semantics import KCFGSemantics
 
 
 class KCFGMinimizer:
     kcfg: KCFG
+    semantics: KCFGSemantics
+    kdef: KDefinition | None
 
-    def __init__(self, kcfg: KCFG) -> None:
+    def __init__(self, kcfg: KCFG, heuristics: KCFGSemantics | None = None, kdef: KDefinition | None = None) -> None:
+        if heuristics is None:
+            heuristics = DefaultSemantics()
         self.kcfg = kcfg
+        self.semantics = heuristics
+        self.kdef = kdef
 
     def lift_edge(self, b_id: NodeIdLike) -> None:
         """Lift an edge up another edge directly preceding it.
@@ -188,6 +200,94 @@ def _fold_lift(result: bool, finder_lifter: tuple[Callable, Callable]) -> bool:
             _fold_lift, [(self.kcfg.edges, self.lift_split_edge), (self.kcfg.splits, self.lift_split_split)], False
         )
 
+    def merge_nodes(self) -> bool:
+        """Merge targets of Split for cutting down the number of branches, using heuristics KCFGSemantics.is_mergeable.
+
+        Side Effect: The KCFG is rewritten by the following rewrite pattern,
+            - Match: A -|Split|-> A_i -|Edge|-> B_i
+            - Rewrite:
+                - if `B_x, B_y, ..., B_z are not mergeable` then unchanged
+                - if `B_x, B_y, ..., B_z are mergeable`, then
+                    - A -|Split|-> A_x or A_y or ... or A_z
+                    - A_x or A_y or ... or A_z -|Edge|-> B_x or B_y or ... or B_z
+                    - B_x or B_y or ... or B_z -|Split|-> B_x, B_y, ..., B_z
+
+        Specifically, when `B_merge = B_x or B_y or ... or B_z`
+        - `or`: fresh variables in places where the configurations differ
+        - `Edge` in A_merged -|Edge|-> B_merge: list of merged edges is from A_i -|Edge|-> B_i
+        - `Split` in B_merge -|Split|-> B_x, B_y, ..., B_z: subst for it is from A -|Split|-> A_1, A_2, ..., A_n
+        :param semantics: provides the is_mergeable heuristic
+        :return: whether any merge was performed
+        """
+
+        def _is_mergeable(x: KCFG.Edge | KCFG.MergedEdge, y: KCFG.Edge | KCFG.MergedEdge) -> bool:
+            return self.semantics.is_mergeable(x.target.cterm, y.target.cterm)
+
+        # ---- Match ----
+
+        # A -|Split|> Ai -|Edge/MergedEdge|> Mergeable Bi
+        sub_graphs: list[tuple[KCFG.Split, list[list[KCFG.Edge | KCFG.MergedEdge]]]] = []
+
+        for split in self.kcfg.splits():
+            _edges = [
+                single(self.kcfg.general_edges(source_id=ai))
+                for ai in split.target_ids
+                if self.kcfg.general_edges(source_id=ai)
+            ]
+            _partitions = partition(_edges, _is_mergeable)
+            if len(_partitions) < len(_edges):
+                sub_graphs.append((split, _partitions))
+
+        if not sub_graphs:
+            return False
+
+        # ---- Rewrite ----
+
+        for split, edge_partitions in sub_graphs:
+
+            # Remove the original sub-graphs
+            for p in edge_partitions:
+                if len(p) == 1:
+                    continue
+                for e in p:
+                    # TODO: remove the split and edges, then safely remove the nodes.
+                    self.kcfg.remove_node(e.source.id)
+
+            # Create A -|MergedEdge|-> Merged_Bi -|Split|-> Bi, if one edge partition covers all the splits
+            if len(edge_partitions) == 1:
+                merged_bi_cterm, merged_bi_subst = cterms_anti_unify(
+                    [edge.target.cterm for edge in edge_partitions[0]], keep_values=True, kdef=self.kdef
+                )
+                merged_bi = self.kcfg.create_node(merged_bi_cterm)
+                self.kcfg.create_merged_edge(split.source.id, merged_bi.id, edge_partitions[0])
+                self.kcfg.create_split(
+                    merged_bi.id, zip([e.target.id for e in edge_partitions[0]], merged_bi_subst, strict=True)
+                )
+                continue
+
+            # Create A -|Split|-> Others & Merged_Ai -|MergedEdge|-> Merged_Bi -|Split|-> Bi
+            _split_nodes: list[NodeIdLike] = []
+            for edge_partition in edge_partitions:
+                if len(edge_partition) == 1:
+                    _split_nodes.append(edge_partition[0].source.id)
+                    continue
+                merged_ai_cterm, _ = cterms_anti_unify(
+                    [ai2bi.source.cterm for ai2bi in edge_partition], keep_values=True, kdef=self.kdef
+                )
+                merged_bi_cterm, merged_bi_subst = cterms_anti_unify(
+                    [ai2bi.target.cterm for ai2bi in edge_partition], keep_values=True, kdef=self.kdef
+                )
+                merged_ai = self.kcfg.create_node(merged_ai_cterm)
+                _split_nodes.append(merged_ai.id)
+                merged_bi = self.kcfg.create_node(merged_bi_cterm)
+                self.kcfg.create_merged_edge(merged_ai.id, merged_bi.id, edge_partition)
+                self.kcfg.create_split(
+                    merged_bi.id, zip([ai2bi.target.id for ai2bi in edge_partition], merged_bi_subst, strict=True)
+                )
+            self.kcfg.create_split_by_nodes(split.source.id, _split_nodes)
+
+        return True
+
     def minimize(self) -> None:
         """Minimize KCFG by repeatedly performing the lifting transformations.
 
@@ -198,3 +298,7 @@ def minimize(self) -> None:
         while repeat:
             repeat = self.lift_edges()
             repeat = self.lift_splits() or repeat
+
+        repeat = True
+        while repeat:
+            repeat = self.merge_nodes()

pyk/src/pyk/kcfg/semantics.py

@@ -34,6 +34,11 @@ def custom_step(self, c: CTerm) -> KCFGExtendResult | None: ...
 
     """Implement a custom semantic step."""
 
+    @abstractmethod
+    def is_mergeable(self, c1: CTerm, c2: CTerm) -> bool: ...
+
+    """Check whether or not the two given ``CTerm``s are mergeable. Must be transitive, commutative, and reflexive."""
+
 
 class DefaultSemantics(KCFGSemantics):
     def is_terminal(self, c: CTerm) -> bool:
@@ -50,3 +55,6 @@ def can_make_custom_step(self, c: CTerm) -> bool:
 
     def custom_step(self, c: CTerm) -> KCFGExtendResult | None:
         return None
+
+    def is_mergeable(self, c1: CTerm, c2: CTerm) -> bool:
+        return False

pyk/src/pyk/utils.py

@@ -319,6 +319,36 @@ def repeat_last(iterable: Iterable[T]) -> Iterator[T]:
             yield last
 
 
+def partition(iterable: Iterable[T], pred: Callable[[T, T], bool]) -> list[list[T]]:
+    """Partition the iterable into sublists based on the given predicate.
+
+    predicate pred(_, _) should satisfy:
+    - pred(x, x)
+    - if pred(x, y) and pred(y, z) then pred(x, z);
+    - if pred(x, y) then pred(y, x);
+    """
+    groups: list[list[T]] = []
+    for item in iterable:
+        found = False
+        for group in groups:
+            group_matches = []
+            for group_item in group:
+                group_match = pred(group_item, item)
+                if group_match != pred(item, group_item):
+                    raise ValueError(f'Partitioning failed, predicate commutativity failed on: {(item, group_item)}')
+                group_matches.append(group_match)
+            if found and any(group_matches):
+                raise ValueError(f'Partitioning failed, item matched multiple groups: {item}')
+            if all(group_matches):
+                found = True
+                group.append(item)
+            elif any(group_matches):
+                raise ValueError(f'Partitioning failed, item matched only some elements of group: {(item, group)}')
+        if not found:
+            groups.append([item])
+    return groups
+
+
 def nonempty_str(x: Any) -> str:
     if x is None:
         raise ValueError('Expected nonempty string, found: null.')

pyk/src/tests/integration/ktool/test_imp.py

@@ -9,7 +9,7 @@
 
 from pyk.cli.pyk import ProveOptions
 from pyk.kast.inner import KApply, KSequence, KVariable
-from pyk.kcfg.semantics import KCFGSemantics
+from pyk.kcfg.semantics import DefaultSemantics
 from pyk.ktool.prove_rpc import ProveRpc
 from pyk.proof import ProofStatus
 from pyk.testing import KCFGExploreTest, KProveTest
@@ -25,12 +25,13 @@
     from pyk.kast.outer import KDefinition
     from pyk.kcfg import KCFGExplore
     from pyk.kcfg.kcfg import KCFGExtendResult
+    from pyk.kcfg.semantics import KCFGSemantics
     from pyk.ktool.kprove import KProve
 
 _LOGGER: Final = logging.getLogger(__name__)
 
 
-class ImpSemantics(KCFGSemantics):
+class ImpSemantics(DefaultSemantics):
     definition: KDefinition | None
 
     def __init__(self, definition: KDefinition | None = None):

pyk/src/tests/integration/proof/test_custom_step.py

@@ -10,7 +10,7 @@
 from pyk.kast.manip import set_cell
 from pyk.kcfg import KCFGExplore
 from pyk.kcfg.kcfg import Step
-from pyk.kcfg.semantics import KCFGSemantics
+from pyk.kcfg.semantics import DefaultSemantics
 from pyk.kcfg.show import KCFGShow
 from pyk.proof import APRProof, APRProver, ProofStatus
 from pyk.proof.show import APRProofNodePrinter
@@ -26,6 +26,7 @@
     from pyk.cterm import CTermSymbolic
     from pyk.kast.outer import KClaim
     from pyk.kcfg.kcfg import KCFGExtendResult
+    from pyk.kcfg.semantics import KCFGSemantics
     from pyk.ktool.kprove import KProve
     from pyk.utils import BugReport
 
@@ -46,7 +47,7 @@
 )
 
 
-class CustomStepSemanticsWithoutStep(KCFGSemantics):
+class CustomStepSemanticsWithoutStep(DefaultSemantics):
     def is_terminal(self, c: CTerm) -> bool:
         k_cell = c.cell('K_CELL')
         if (

pyk/src/tests/integration/proof/test_goto.py

@@ -7,7 +7,7 @@
 import pytest
 
 from pyk.kast.inner import KApply, KSequence
-from pyk.kcfg.semantics import KCFGSemantics
+from pyk.kcfg.semantics import DefaultSemantics
 from pyk.kcfg.show import KCFGShow
 from pyk.proof import APRProof, APRProver, ProofStatus
 from pyk.proof.show import APRProofNodePrinter
@@ -24,12 +24,13 @@
     from pyk.kast.outer import KDefinition
     from pyk.kcfg import KCFGExplore
     from pyk.kcfg.kcfg import KCFGExtendResult
+    from pyk.kcfg.semantics import KCFGSemantics
     from pyk.ktool.kprove import KProve
 
 _LOGGER: Final = logging.getLogger(__name__)
 
 
-class GotoSemantics(KCFGSemantics):
+class GotoSemantics(DefaultSemantics):
     def is_terminal(self, c: CTerm) -> bool:
         return False
 

pyk/src/tests/integration/proof/test_imp.py

@@ -10,7 +10,7 @@
 from pyk.cterm import CSubst, CTerm
 from pyk.kast.inner import KApply, KSequence, KSort, KToken, KVariable, Subst
 from pyk.kast.manip import minimize_term, sort_ac_collections
-from pyk.kcfg.semantics import KCFGSemantics
+from pyk.kcfg.semantics import DefaultSemantics
 from pyk.kcfg.show import KCFGShow
 from pyk.prelude.kbool import BOOL, FALSE, andBool, orBool
 from pyk.prelude.kint import intToken
@@ -34,6 +34,7 @@
     from pyk.kast.outer import KDefinition
     from pyk.kcfg import KCFGExplore
     from pyk.kcfg.kcfg import KCFGExtendResult
+    from pyk.kcfg.semantics import KCFGSemantics
     from pyk.ktool.kprint import KPrint, SymbolTable
     from pyk.ktool.kprove import KProve
     from pyk.proof import Prover
@@ -46,7 +47,7 @@ def proof_dir(tmp_path_factory: TempPathFactory) -> Path:
     return tmp_path_factory.mktemp('proofs')
 
 
-class ImpSemantics(KCFGSemantics):
+class ImpSemantics(DefaultSemantics):
     definition: KDefinition | None
 
     def __init__(self, definition: KDefinition | None = None):