Automatically add subproofs to parallelized execution once they become available

kevm-pyk/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "kevm-pyk"
-version = "1.0.293"
+version = "1.0.294"
 description = ""
 authors = [
     "Runtime Verification, Inc. <[email protected]>",
@@ -17,7 +17,8 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.10"
 pathos = "*"
-pyk = { git = "https://github.com/runtimeverification/pyk.git", tag="v0.1.434" }
+pyk = { path = "/home/noah/dev/pyk", develop=true }
+pyk = { git = "https://github.com/runtimeverification/pyk.git", tag="noah/subproof-split" }
 tomlkit = "^0.11.6"
 xdg-base-dirs = "^6.0.0"
 

kevm-pyk/src/kevm_pyk/__init__.py

@@ -6,4 +6,4 @@
     from typing import Final
 
 
-VERSION: Final = '1.0.293'
+VERSION: Final = '1.0.294'

kevm-pyk/src/kevm_pyk/__main__.py

@@ -495,7 +495,7 @@ def exec_run(
         kore_pgm = kevm.kast_to_kore(kast_pgm, sort=KSort('EthereumSimulation'))
         kore_pattern = kore_pgm_to_kore(kore_pgm, SORT_ETHEREUM_SIMULATION, schedule, mode, chainid)
 
-    kevm.run_kore(
+    kevm.run(
         kore_pattern,
         depth=depth,
         term=True,

kevm-pyk/src/kevm_pyk/utils.py

@@ -82,6 +82,7 @@ def kevm_prove(
     break_on_calls: bool = True,
     extract_branches: Callable[[CTerm], Iterable[KInner]] | None = None,
     abstract_node: Callable[[CTerm], CTerm] | None = None,
+    max_branches: int | None = None,
 ) -> bool:
     proof = proof
     terminal_rules = ['EVM.halt']
@@ -122,11 +123,15 @@ def kevm_prove(
                 execute_depth=max_depth,
                 terminal_rules=terminal_rules,
                 cut_point_rules=cut_point_rules,
+                max_branches=max_branches,
             )
             assert isinstance(proof, APRProof)
             if proof.passed:
                 _LOGGER.info(f'Proof passed: {proof.id}')
                 return True
+            elif proof.is_proof_pending:
+                _LOGGER.info(f'Proof pending: {proof.id}')
+                return True
             else:
                 _LOGGER.error(f'Proof failed: {proof.id}')
                 return False

kevm-pyk/src/kontrol/__main__.py

@@ -162,6 +162,7 @@ def exec_foundry_prove(
     counterexample_info: bool = False,
     trace_rewrites: bool = False,
     auto_abstract_gas: bool = False,
+    max_branches: int | None = None,
     **kwargs: Any,
 ) -> None:
     _ignore_arg(kwargs, 'main_module', f'--main-module: {kwargs["main_module"]}')
@@ -198,16 +199,18 @@ def exec_foundry_prove(
         smt_retry_limit=smt_retry_limit,
         trace_rewrites=trace_rewrites,
         auto_abstract_gas=auto_abstract_gas,
+        max_branches=max_branches,
     )
     failed = 0
-    for pid, r in results.items():
-        passed, failure_log = r
-        if passed:
-            print(f'PROOF PASSED: {pid}')
-        else:
-            failed += 1
-            print(f'PROOF FAILED: {pid}')
-            if failure_info and failure_log is not None:
+    for proof in results:
+        if proof.passed:
+            print(f'PROOF PASSED: {proof.id}')
+        elif proof.is_proof_pending:
+            print(f'PROOF PENDING: {proof.id}')
+        elif proof.failed:
+            print(f'PROOF FAILED: {proof.id}')
+            if failure_info and proof.failure_info is not None:
+                failure_log = proof.failure_info.print()
                 failure_log += Foundry.help_info()
                 for line in failure_log:
                     print(line)
@@ -517,6 +520,13 @@ def _parse_test_id_tuple(value: str) -> tuple[str, str | None]:
         type=int,
         help='Enables bounded model checking. Specifies the maximum depth to unroll all loops to.',
     )
+    foundry_prove_args.add_argument(
+        '--max-branches',
+        dest='max_branches',
+        default=None,
+        type=int,
+        help='Enables subproof splitting when the number of pending nodes exceeds max_branches',
+    )
     foundry_prove_args.add_argument(
         '--use-booster',
         dest='use_booster',

kevm-pyk/src/kontrol/foundry.py

@@ -6,19 +6,20 @@
 import re
 import shutil
 import sys
+from concurrent.futures import ThreadPoolExecutor, wait
 from functools import cached_property
 from os import listdir
 from pathlib import Path
 from subprocess import CalledProcessError
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Callable
 
 import tomlkit
-from pathos.pools import ProcessPool  # type: ignore
 from pyk.cterm import CTerm
 from pyk.kast.inner import KApply, KSequence, KSort, KToken, KVariable, Subst
 from pyk.kast.manip import free_vars, minimize_term
 from pyk.kast.outer import KDefinition, KFlatModule, KImport, KRequire
 from pyk.kcfg import KCFG
+from pyk.kore.rpc import kore_server
 from pyk.prelude.bytes import bytesToken
 from pyk.prelude.k import GENERATED_TOP_CELL
 from pyk.prelude.kbool import FALSE, notBool
@@ -52,6 +53,8 @@
     from pyk.kcfg import KCFGExplore
     from pyk.kcfg.kcfg import NodeIdLike
     from pyk.kcfg.tui import KCFGElem
+    from pyk.kore.pool import Future
+    from pyk.kore.rpc import KoreServer
     from pyk.proof.show import NodePrinter
     from pyk.utils import BugReport
 
@@ -413,10 +416,13 @@ def help_info() -> list[str]:
         return res_lines
 
     def proofs_with_test(self, test: str) -> list[Proof]:
+        #          print([pid.split(':')[0] for pid in listdir(self.proofs_dir)])
+        #          print(single(self._escape_brackets([test])))
         proofs = [
             self.get_optional_proof(pid)
             for pid in listdir(self.proofs_dir)
-            if re.search(single(self._escape_brackets([test])), pid.split(':')[0])
+            #              if re.search(single(self._escape_brackets([test])), pid.split(':')[0])
+            if test == pid.split(':')[0]
         ]
         return [proof for proof in proofs if proof is not None]
 
@@ -627,12 +633,12 @@ def foundry_prove(
     use_booster: bool = False,
     smt_timeout: int | None = None,
     smt_retry_limit: int | None = None,
-    failure_info: bool = True,
     counterexample_info: bool = False,
     trace_rewrites: bool = False,
     auto_abstract_gas: bool = False,
     port: int | None = None,
-) -> dict[tuple[str, str | None], tuple[bool, list[str] | None]]:
+    max_branches: int | None = None,
+) -> list[APRProof]:
     if workers <= 0:
         raise ValueError(f'Must have at least one worker, found: --workers {workers}')
     if max_iterations is not None and max_iterations < 0:
@@ -715,15 +721,51 @@ def foundry_prove(
         assert id is not None
         tests[i] = (test, id)
 
-    def _init_and_run_proof(_init_problem: tuple[str, str, str | None]) -> tuple[bool, list[str] | None]:
+    def _run_proof(_init_problem: tuple[int, Proof]) -> Proof:
+        port, proof = _init_problem
+
+        llvm_definition_dir = foundry.llvm_library if use_booster else None
+        start_server = port is None
+
+        with legacy_explore(
+            foundry.kevm,
+            kcfg_semantics=KEVMSemantics(auto_abstract_gas=auto_abstract_gas),
+            id=proof.id,
+            bug_report=bug_report,
+            kore_rpc_command=kore_rpc_command,
+            llvm_definition_dir=llvm_definition_dir,
+            smt_timeout=smt_timeout,
+            smt_retry_limit=smt_retry_limit,
+            trace_rewrites=trace_rewrites,
+            start_server=start_server,
+            port=port,
+        ) as kcfg_explore:
+            kevm_prove(
+                foundry.kevm,
+                proof,
+                kcfg_explore,
+                max_depth=max_depth,
+                max_iterations=max_iterations,
+                break_every_step=break_every_step,
+                break_on_jumpi=break_on_jumpi,
+                break_on_calls=break_on_calls,
+                max_branches=max_branches,
+            )
+            return proof
+
+    def _init_and_run_proof(_init_problem: tuple[str, str, str | None]) -> APRProof:
         contract_name, method_sig, id = _init_problem
         contract = foundry.contracts[contract_name]
         method = contract.method_by_sig[method_sig]
-        id = ':' + id if id is not None else ''
-        test_id = f'{contract_name}.{method_sig}{id}'
+        id_prefix = ':' + id if id is not None else ''
+        test_id = f'{contract_name}.{method_sig}{id_prefix}'
+
         llvm_definition_dir = foundry.llvm_library if use_booster else None
 
-        start_server = port is None
+        def generate_subproof_name(proof: APRProof, node: int) -> str:
+            id_without_version = proof.id.split(':')[0]
+
+            return f'{id_without_version}_node_{node}:{id}' if id else f'{id_without_version}_node_{node}:{id}'
 
         with legacy_explore(
             foundry.kevm,
@@ -735,7 +777,6 @@ def _init_and_run_proof(_init_problem: tuple[str, str, str | None]) -> tuple[boo
             smt_timeout=smt_timeout,
             smt_retry_limit=smt_retry_limit,
             trace_rewrites=trace_rewrites,
-            start_server=start_server,
             port=port,
         ) as kcfg_explore:
             proof = _method_to_apr_proof(
@@ -747,9 +788,10 @@ def _init_and_run_proof(_init_problem: tuple[str, str, str | None]) -> tuple[boo
                 test_id,
                 simplify_init=simplify_init,
                 bmc_depth=bmc_depth,
+                generate_subproof_name=generate_subproof_name,
             )
 
-            passed = kevm_prove(
+            kevm_prove(
                 foundry.kevm,
                 proof,
                 kcfg_explore,
@@ -758,37 +800,87 @@ def _init_and_run_proof(_init_problem: tuple[str, str, str | None]) -> tuple[boo
                 break_every_step=break_every_step,
                 break_on_jumpi=break_on_jumpi,
                 break_on_calls=break_on_calls,
+                max_branches=max_branches,
             )
-            failure_log = None
-            if not passed:
-                failure_log = print_failure_info(proof, kcfg_explore, counterexample_info)
-            return passed, failure_log
-
-    def run_cfg_group(
-        tests: list[tuple[str, str | None]]
-    ) -> dict[tuple[str, str | None], tuple[bool, list[str] | None]]:
-        def _split_test(test: tuple[str, str | None]) -> tuple[str, str, str | None]:
-            test_name, id = test
-            contract, method = test_name.split('.')
-            return contract, method, id
-
-        init_problems = [_split_test(test) for test in tests]
-
-        _apr_proofs: list[tuple[bool, list[str] | None]]
-        if workers > 1:
-            with ProcessPool(ncpus=workers) as process_pool:
-                _apr_proofs = process_pool.map(_init_and_run_proof, init_problems)
-        else:
-            _apr_proofs = []
-            for init_problem in init_problems:
-                _apr_proofs.append(_init_and_run_proof(init_problem))
+            return proof
+
+    def run_cfg_group(tests: list[tuple[str, str | None]]) -> list[APRProof]:
+        llvm_definition_dir = foundry.llvm_library if use_booster else None
+
+        def create_server() -> KoreServer:
+            return kore_server(
+                definition_dir=foundry.kevm.definition_dir,
+                llvm_definition_dir=llvm_definition_dir,
+                module_name=foundry.kevm.main_module,
+                command=kore_rpc_command,
+                bug_report=bug_report,
+                smt_timeout=smt_timeout,
+                smt_retry_limit=smt_retry_limit,
+            )
+
+        class Task:
+            ...
+
+        class CreateAndRunTask(Task):
+            _test: str
+            _id: str | None
+
+            def __init__(self, test: str, id: str | None):
+                self._test = test
+                self._id = id
+
+        class RunTask(Task):
+            _proof: Proof
+            _port: int
+
+            def __init__(self, proof: Proof, port: int):
+                self._proof = proof
+                self._port = port
+
+        results = []
+        futures: dict[str, Future] = {}
+        executor = ThreadPoolExecutor(max_workers=workers)
+
+        tasks: list[Task] = [CreateAndRunTask(test, id) for test, id in tests]
+        base_proofs = {test: test for test, _ in tests}
+        servers = {}
+
+        for test, _ in tests:
+            servers[test] = create_server()
+
+        while len(tasks) > 0 or len(futures) > 0:
+            if len(tasks) > 0:
+                task = tasks.pop()
 
-        apr_proofs = dict(zip(tests, _apr_proofs, strict=True))
-        return apr_proofs
+                if type(task) is CreateAndRunTask:
+                    contract, method = task._test.split('.')
+
+                    futures[task._test] = executor.submit(_init_and_run_proof, (contract, method, task._id))
+
+                elif type(task) is RunTask:
+                    futures[task._proof.id] = executor.submit(_run_proof, (task._port, task._proof))
+
+            if len(futures) > 0:
+                done_futures = [(_proof, future) for (_proof, future) in futures.items() if future.done()]
+                futures = {_proof: future for (_proof, future) in futures.items() if not future.done()}
+                for _proof, future in done_futures:
+                    proof = future.result()
+                    results.append(proof)
+
+                    subproofs = proof.subproofs
+                    for subproof in subproofs:
+                        base_proofs[subproof.id] = base_proofs[_proof]
+                        port = servers[base_proofs[subproof.id]].port
+                        tasks.append(RunTask(subproof, port))
+
+            wait(futures.values())
+        executor.shutdown()
+
+        return results
 
     _LOGGER.info(f'Running setup functions in parallel: {list(setup_methods.values())}')
     results = run_cfg_group([(method, None) for method in setup_methods.values()])
-    failed = [setup_cfg for setup_cfg, passed in results.items() if not passed]
+    failed = [proof.id for proof in results if not proof.passed]
     if failed:
         raise ValueError(f'Running setUp method failed for {len(failed)} contracts: {failed}')
 
@@ -1167,6 +1259,7 @@ def _method_to_apr_proof(
     test_id: str,
     simplify_init: bool = True,
     bmc_depth: int | None = None,
+    generate_subproof_name: Callable[[APRProof, int], str] | None = None,
 ) -> APRProof | APRBMCProof:
     contract_name = contract.name
     method_sig = method.signature
@@ -1216,7 +1309,16 @@ def _method_to_apr_proof(
                 proof_dir=save_directory,
             )
         else:
-            apr_proof = APRProof(test_id, kcfg, [], init_node_id, target_node_id, {}, proof_dir=save_directory)
+            apr_proof = APRProof(
+                test_id,
+                kcfg,
+                [],
+                init_node_id,
+                target_node_id,
+                {},
+                proof_dir=save_directory,
+                generate_subproof_name=generate_subproof_name,
+            )
 
     apr_proof.write_proof_data()
     return apr_proof
@@ -1251,7 +1353,6 @@ def _method_to_cfg(
 
 
 def get_final_accounts_cell(proof_id: str, proof_dir: Path) -> tuple[KInner, Iterable[KInner]]:
-    print(proof_id, proof_dir)
     apr_proof = APRProof.read_proof_data(proof_dir, proof_id)
     target = apr_proof.kcfg.node(apr_proof.target)
     target_states = apr_proof.kcfg.covers(target_id=target.id)