diff --git a/pyk/src/pyk/cli/pyk.py b/pyk/src/pyk/cli/pyk.py index 8a23469ed8..75a9ecc1ed 100644 --- a/pyk/src/pyk/cli/pyk.py +++ b/pyk/src/pyk/cli/pyk.py @@ -316,6 +316,7 @@ class ProveOptions(LoggingOptions, SpecOptions, SaveDirOptions): kore_rpc_command: str | Iterable[str] | None max_depth: int | None max_iterations: int | None + step_timeout: int | None assume_defined: bool show_kcfg: bool haskell_logging: bool @@ -329,6 +330,7 @@ def default() -> dict[str, Any]: 'kore_rpc_command': None, 'max_depth': None, 'max_iterations': None, + 'step_timeout': None, 'assume_defined': False, 'show_kcfg': False, 'haskell_logging': False, @@ -515,6 +517,17 @@ def create_argument_parser() -> ArgumentParser: type=int, help='Maximum number of KCFG explorations to take in attempting to discharge proof.', ) + prove_args.add_argument( + '--step-timeout', + dest='step_timeout', + type=int, + default=None, + help=( + 'Per-step wall-clock budget in whole seconds (floored at 1). When a symbolic-execution step' + ' exceeds it, the step is interrupted, its execution depth is halved, and it is retried;' + ' proving stops once the depth cannot be reduced further. Omit to disable the timeout.' + ), + ) prove_args.add_argument( '--kore-rpc-command', dest='kore_rpc_command', diff --git a/pyk/src/pyk/cterm/symbolic.py b/pyk/src/pyk/cterm/symbolic.py index af8f9a4e36..66d178dcb3 100644 --- a/pyk/src/pyk/cterm/symbolic.py +++ b/pyk/src/pyk/cterm/symbolic.py @@ -127,6 +127,10 @@ def kast_to_kore(self, kinner: KInner) -> Pattern: def kore_to_kast(self, pattern: Pattern) -> KInner: return kore_to_kast(self._definition, pattern) + def interrupt(self) -> None: + """Abort a backend request currently in flight on another thread; see `KoreClient.interrupt`.""" + self._kore_client.interrupt() + def _haskell_logging_request(self, haskell_logging: bool | None) -> tuple[str, ...] | None: """Resolve the per-call on/off flag to the list of log entries to request. diff --git a/pyk/src/pyk/kcfg/explore.py b/pyk/src/pyk/kcfg/explore.py index 1eef51588a..81716fdd4b 100644 --- a/pyk/src/pyk/kcfg/explore.py +++ b/pyk/src/pyk/kcfg/explore.py @@ -59,6 +59,10 @@ def _pretty_printer(self) -> PrettyPrinter: def pretty_print(self, kinner: KInner) -> str: return self._pretty_printer.print(kinner) + def interrupt(self) -> None: + """Abort a backend request currently in flight on another thread; see `KoreClient.interrupt`.""" + self.cterm_symbolic.interrupt() + def _extract_rule_labels(self, _logs: tuple[LogEntry, ...]) -> list[str]: _rule_lines = [] for node_log in _logs: diff --git a/pyk/src/pyk/kore/rpc.py b/pyk/src/pyk/kore/rpc.py index e4e0355883..099c86839f 100644 --- a/pyk/src/pyk/kore/rpc.py +++ b/pyk/src/pyk/kore/rpc.py @@ -85,6 +85,15 @@ def __exit__(self, *args: Any) -> None: @abstractmethod def close(self) -> None: ... + def send_interrupt(self, data: str) -> None: + """Send `data` on the live connection without waiting for a reply. + + Used to deliver a `cancel` to a connection whose reply another thread is already + awaiting. Default: no-op. Override only for connections that can be written to + while a request is in flight. + """ + ... + @abstractmethod def _request(self, req: str) -> str: ... @@ -101,6 +110,7 @@ class TransportType(Enum): class SingleSocketTransport(Transport): _host: str _port: int + _timeout: int | None _sock: socket.socket _file: IO[str] @@ -113,6 +123,7 @@ def __init__( ): self._host = host self._port = port + self._timeout = timeout self._sock = self._create_connection(host, port, timeout) self._file = self._sock.makefile('r') @@ -141,6 +152,12 @@ def close(self) -> None: self._file.close() self._sock.close() + def send_interrupt(self, data: str) -> None: + # Write the cancel to the socket but don't read the reply: the thread already blocked in + # `_request`'s `readline` will read the server's "cancelled" reply. The socket stays open. + # The leading newline separates the cancel from the request bytes already on the stream. + self._sock.sendall(b'\n' + data.encode()) + def _request(self, req: str) -> str: self._sock.sendall(req.encode()) server_addr = self._description() @@ -252,6 +269,12 @@ def last_request_id(self) -> str | None: """The JSON-RPC id of the most recent request issued through this facade (``None`` if none yet).""" return self._last_request_id + def interrupt(self) -> None: + self._default_client.interrupt() + for clients in self._clients.values(): + for client in clients: + client.interrupt() + def request(self, method: str, **params: Any) -> dict[str, Any]: if method in self._clients: for client in self._clients[method]: @@ -316,6 +339,19 @@ def __exit__(self, *args: Any) -> None: def close(self) -> None: self._transport.close() + def interrupt(self) -> None: + # Send a `cancel` so the server aborts the in-flight request. The cancel gets no reply of its + # own; the thread awaiting that request gets a "cancelled" error instead. The id is only for + # logs, so we derive it from the last request rather than touching the requester's `_req_id`. + cancel_id = f'{self._last_request_id}-cancel' if self._last_request_id is not None else 'cancel' + payload = { + 'jsonrpc': self._JSON_RPC_VERSION, + 'id': cancel_id, + 'method': 'cancel', + 'params': {}, + } + self._transport.send_interrupt(json.dumps(payload)) + def request(self, method: str, **params: Any) -> dict[str, Any]: label = client_label.get() prefix = label if label is not None else str(id(self)) @@ -1014,6 +1050,15 @@ def last_request_id(self) -> str | None: """ return self._client.last_request_id + def interrupt(self) -> None: + """Abort an `execute`/`simplify`/… request running on another thread. + + Sends a `cancel` so the server stops computing; the interrupted call raises a + "cancelled" error and the connection stays usable. Works on the single-socket + transport only; a no-op for HTTP (one connection per request, nothing to cancel). + """ + self._client.interrupt() + def _request(self, method: str, **params: Any) -> dict[str, Any]: try: return self._client.request(method, **params) diff --git a/pyk/src/pyk/proof/proof.py b/pyk/src/pyk/proof/proof.py index 0d1eef04d2..bd89052223 100644 --- a/pyk/src/pyk/proof/proof.py +++ b/pyk/src/pyk/proof/proof.py @@ -3,7 +3,9 @@ import json import logging from abc import ABC, abstractmethod -from concurrent.futures import ThreadPoolExecutor, wait +from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import TimeoutError as FuturesTimeoutError +from concurrent.futures import wait from dataclasses import dataclass from enum import Enum from itertools import chain @@ -459,6 +461,24 @@ def step() -> Iterable[SR]: return step +class _StepTimedOut(Exception): + """A timed `step_proof` exhausted its `step_timeout` budget. Carries the budget for logging.""" + + budget: int + + def __init__(self, budget: int) -> None: + super().__init__(budget) + self.budget = budget + + +class _StepShrunk(_StepTimedOut): + """The timed-out step was shrunk; `advance_proof` should re-fetch steps and retry it smaller.""" + + +class _StepExhausted(_StepTimedOut): + """The timed-out step could not be shrunk further; `advance_proof` should stop.""" + + class Prover(ContextManager['Prover'], Generic[P, PS, SR]): """Abstract class which advances `Proof`s with `init_proof()` and `step_proof()`. @@ -499,6 +519,54 @@ def init_proof(self, proof: P) -> None: """ ... + #: Per-step wall-clock budget in whole seconds (minimum 1). When set, `advance_proof` runs each + #: step under this budget and, on timeout, interrupts it, calls `shrink_step`, and retries. + #: `None` (the default) disables the policy, so steps run synchronously with no time limit. A + #: prover that can do less work per step (e.g. `APRProver`, by lowering its execution depth) + #: should pair this with `shrink_step`. + step_timeout: int | None = None + + def shrink_step(self) -> bool: + """Reduce the amount of work a single `step_proof` does, after a step timed out. + + Return `True` if the step was made smaller (so it is worth retrying), or `False` if the + step is already at its minimum size (so `advance_proof` should stop). The default is a + no-op that returns `False`; see `step_timeout`. + """ + return False + + def interrupt(self) -> None: + """Abort a `step_proof` call currently running on another thread, as quickly as possible. + + Used by the timeout-and-shrink policy in `advance_proof` to abandon a step that has + exhausted its `step_timeout` budget. After this returns, a thread blocked in `step_proof` + must raise promptly and the prover must remain usable for subsequent steps. The default + implementation does nothing; provers backed by an interruptible resource (e.g. a Kore RPC + connection) should override it. + """ + ... + + def _execute_step(self, step: PS, executor: Executor | None) -> Iterable[SR]: + """Run one `step_proof`, honoring `step_timeout` when an `executor` is given. + + With no executor, runs the step synchronously with no time limit. With one, runs the step + under the `step_timeout` budget; on timeout it interrupts the step, asks the prover to + `shrink_step`, and raises `_StepShrunk` (a smaller step is ready to retry) or `_StepExhausted` + (already minimal, give up). Returns the step results when the step completes within budget. + """ + if executor is None: + return self.step_proof(step) + + budget = self.step_timeout + assert budget is not None + future = executor.submit(self.step_proof, step) + try: + return future.result(timeout=budget) + except FuturesTimeoutError: + self.interrupt() + wait([future]) # let the interrupted step unwind before we reuse the prover + raise (_StepShrunk if self.shrink_step() else _StepExhausted)(budget) from None + def advance_proof( self, proof: P, @@ -509,7 +577,8 @@ def advance_proof( ) -> None: """Advance a proof. - Performs loop `Proof.get_steps()` -> `Prover.step_proof()` -> `Proof.commit()`. + Performs loop `Proof.get_steps()` -> `Prover.step_proof()` (within `step_timeout`, else + `interrupt` + `shrink_step` and retry, or stop if it cannot shrink) -> `Proof.commit()`. Args: proof: proof to advance. @@ -522,25 +591,48 @@ def advance_proof( iterations = 0 _LOGGER.info(f'Initializing proof: {proof.id}') self.init_proof(proof) - while True: - steps = list(proof.get_steps()) - _LOGGER.info(f'Found {len(steps)} next steps for proof: {proof.id}') - if len(steps) == 0: - break - for step in steps: - if fail_fast and proof.failed: - _LOGGER.warning(f'Terminating proof early because fail_fast is set: {proof.id}') - proof.failure_info = self.failure_info(proof) - return - if max_iterations is not None and max_iterations <= iterations: - return - iterations += 1 - results = self.step_proof(step) - for result in results: - proof.commit(result) - if iterations % maintenance_rate == 0: - proof.write_proof_data() - callback(proof) + + # When `step_timeout` is set, run each step on a worker thread so it can be timed and + # interrupted; otherwise steps run synchronously and `executor` stays `None`. + executor = ThreadPoolExecutor(max_workers=1) if self.step_timeout is not None else None + try: + while True: + steps = list(proof.get_steps()) + _LOGGER.info(f'Found {len(steps)} next steps for proof: {proof.id}') + if len(steps) == 0: + break + shrank_step = False + for step in steps: + if fail_fast and proof.failed: + _LOGGER.warning(f'Terminating proof early because fail_fast is set: {proof.id}') + proof.failure_info = self.failure_info(proof) + return + if max_iterations is not None and max_iterations <= iterations: + return + try: + results = self._execute_step(step, executor) + except _StepShrunk as timeout: + _LOGGER.warning( + f'Proof {proof.id}: step exceeded {timeout.budget}s budget; shrinking and retrying.' + ) + shrank_step = True # re-fetch steps and retry the same node, now smaller + break + except _StepExhausted as timeout: + _LOGGER.warning( + f'Proof {proof.id}: step exceeded {timeout.budget}s budget and cannot shrink further; stopping.' + ) + return + iterations += 1 + for result in results: + proof.commit(result) + if iterations % maintenance_rate == 0: + proof.write_proof_data() + callback(proof) + if shrank_step: + continue + finally: + if executor is not None: + executor.shutdown(wait=False) if proof.failed: proof.failure_info = self.failure_info(proof) diff --git a/pyk/src/pyk/proof/prove_rpc.py b/pyk/src/pyk/proof/prove_rpc.py index df01ad30b4..9b852daaa6 100644 --- a/pyk/src/pyk/proof/prove_rpc.py +++ b/pyk/src/pyk/proof/prove_rpc.py @@ -52,6 +52,7 @@ def prove_rpc(self, options: ProveOptions) -> list[Proof]: max_depth=options.max_depth, save_directory=options.save_directory, max_iterations=options.max_iterations, + step_timeout=options.step_timeout, ) for claim in all_claims ] @@ -63,6 +64,7 @@ def _prove_claim_rpc( max_depth: int | None = None, save_directory: Path | None = None, max_iterations: int | None = None, + step_timeout: int | None = None, ) -> Proof: definition = self._kprove.definition @@ -90,7 +92,12 @@ def _prove_claim_rpc( prover = ImpliesProver(proof, kcfg_explore, assume_defined=assume_defined) else: assert type(proof) is APRProof - prover = APRProver(kcfg_explore, execute_depth=max_depth, assume_defined=assume_defined) + prover = APRProver( + kcfg_explore, + execute_depth=max_depth, + assume_defined=assume_defined, + step_timeout=step_timeout, + ) prover.advance_proof(proof, max_iterations=max_iterations) # type: ignore [arg-type] if proof.passed: diff --git a/pyk/src/pyk/proof/reachability.py b/pyk/src/pyk/proof/reachability.py index 0b4a66d2b3..b7c01d913d 100644 --- a/pyk/src/pyk/proof/reachability.py +++ b/pyk/src/pyk/proof/reachability.py @@ -725,6 +725,7 @@ class APRProver(Prover[APRProof, APRProofStep, APRProofResult]): kcfg_explore: KCFGExplore extra_module: KFlatModule | None optimize_kcfg: bool + step_timeout: int | None def __init__( self, @@ -738,6 +739,7 @@ def __init__( assume_defined: bool = False, extra_module: KFlatModule | None = None, optimize_kcfg: bool = False, + step_timeout: int | None = None, ) -> None: self.kcfg_explore = kcfg_explore @@ -751,10 +753,23 @@ def __init__( self.assume_defined = assume_defined self.extra_module = extra_module self.optimize_kcfg = optimize_kcfg + # Whole seconds, floored at 1; None disables the per-step timeout/shrink policy entirely. + self.step_timeout = max(1, step_timeout) if step_timeout is not None else None def close(self) -> None: self.kcfg_explore.cterm_symbolic._kore_client.close() + def shrink_step(self) -> bool: + # On step timeout, halve the execution depth (floor 1) so the next attempt does less work + # per `extend_cterm`. Returns False once `execute_depth` is unset or already at the minimum. + if self.execute_depth is None or self.execute_depth <= 1: + return False + self.execute_depth = max(1, self.execute_depth // 2) + return True + + def interrupt(self) -> None: + self.kcfg_explore.interrupt() + def init_proof(self, proof: APRProof) -> None: # Stamp proof.id on every subsequent kore-RPC request from this thread so # booster's `{request: ...}` log lines self-identify the originating diff --git a/pyk/src/tests/integration/kore/test_interrupt.py b/pyk/src/tests/integration/kore/test_interrupt.py new file mode 100644 index 0000000000..1a1b3d2e64 --- /dev/null +++ b/pyk/src/tests/integration/kore/test_interrupt.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import threading +import time +from string import Template +from typing import TYPE_CHECKING + +from pyk.kore.parser import KoreParser +from pyk.kore.rpc import DefaultError +from pyk.testing import KoreClientTest + +if TYPE_CHECKING: + from pyk.kore.rpc import KoreClient + from pyk.kore.syntax import Pattern + + +def term(n: int) -> Pattern: + template = Template( + r""" + Lbl'-LT-'generatedTop'-GT-'{}( + Lbl'-LT-'k'-GT-'{}( + kseq{}( + inj{SortInt{}, SortKItem{}}(\dv{SortInt{}}("$n")), + K:SortK{} + ) + ), + GCC:SortGeneratedCounterCell{} + ) + """ + ) + parser = KoreParser(template.substitute(n=n)) + pattern = parser.pattern() + assert parser.eof + return pattern + + +class TestInterrupt(KoreClientTest): + # The interrupt mechanism (cancel over the single-socket transport) is what `advance_proof`'s + # step-timeout policy relies on. `inc` never terminates, so an `execute` only ever returns by + # being interrupted -- which is exactly what this test asserts. + DISABLE_BOOSTER = True # exercise the legacy (pure haskell) kore-rpc server + + KOMPILE_DEFINITION = """ + module INTERRUPT-TEST + imports INT + rule [inc]: I:Int => I +Int 1 + endmodule + """ + KOMPILE_MAIN_MODULE = 'INTERRUPT-TEST' + KOMPILE_ARGS = {'syntax_module': 'INTERRUPT-TEST'} + + def test_interrupt_aborts_in_flight_request_and_keeps_connection(self, kore_client: KoreClient) -> None: + # Given: a non-terminating `execute` running on another thread. + box: dict = {} + + def run() -> None: + try: + kore_client.execute(term(0), max_depth=1_000_000_000) + except BaseException as e: # noqa: B036 - record whatever the interrupted call raises + box['exc'] = e + + thread = threading.Thread(target=run, daemon=True) + thread.start() + time.sleep(2.0) # let the step get well underway + assert thread.is_alive() # sanity: it is genuinely long-running, not terminating on its own + + # When: the in-flight request is interrupted. + kore_client.interrupt() + + # Then: the call is aborted promptly (rather than running ~1e9 steps to completion)... + thread.join(timeout=10.0) + assert not thread.is_alive(), 'execute() was not aborted by interrupt() within 10s' + exc = box.get('exc') + assert isinstance(exc, DefaultError) + assert exc.message == 'Request cancelled' + + # ...and the connection survives the cancel: a fresh request still succeeds on it. + result = kore_client.execute(term(0), max_depth=1) + assert result.depth == 1