diff --git a/CLAUDE.md b/CLAUDE.md index a0a2285c..a9da5d96 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -87,6 +87,12 @@ cd targets/sv-comp && ./scripts/svcomp test run --mode parallel --no-witness --c - **Duration:** a full valid-assert run takes about 40 min with the default 900 s timeout, and much longer with `--testcase-timeout-s 3600`. Only compare runs made with the same timeout. +- **Resource budget:** as in SV-COMP, the explorer and all executor JVMs share one memory and CPU + budget (`--memory-limit` MB, default 15000; `--cpu-limit` s, default 900; the harness passes + `--testcase-timeout-s` as the CPU limit). The executor heap is derived from the memory limit (a + lower cgroup limit wins). Close to a limit the explorer kills the executor run or interrupts the + Z3 query in progress; 20 s before the CPU limit it ends with UNKNOWN. Both rule out SAFE, see + `symbolic-explorer/svcomp/ResourceBudget.py` and `resources` in `stats.json`. - **Output:** results go to `targets/sv-comp/runs/run_/`. Per-task results are in `results/results__*.json`, and per-task logs (`explorer.log`, `Main_main_interprocedural.json`, `stats.json`) are in `logs//_/`. diff --git a/scripts/svcomp-package/run_swat.py b/scripts/svcomp-package/run_swat.py index dbeda389..052d9a88 100755 --- a/scripts/svcomp-package/run_swat.py +++ b/scripts/svcomp-package/run_swat.py @@ -11,6 +11,8 @@ # The explorer writes the violation witness (witness.yml, witness.graphml) here; BenchExec collects # it via the `**/witness.*` result-files pattern of the benchmark definition. WITNESS_DIR = os.path.join(SCRIPT_DIR, 'witness') +# The explorer's logs and testcase statistics (stats.json) +LOG_DIR = os.path.join(SCRIPT_DIR, 'logs') class ExecutionStatus(enum.Enum): @@ -104,7 +106,6 @@ def generate_command(args, property_file: str, property_path: str, input_paths: agent_path = os.path.join(SCRIPT_DIR, 'symbolic-executor', 'lib', 'symbolic-executor.jar') config_path = os.path.join(SCRIPT_DIR, config_file) library_path = os.path.join(SCRIPT_DIR, 'libs', 'java-library-path') - logging_dir = os.path.join(SCRIPT_DIR, 'logs') # Same interpreter (and PYTHONPATH, via the inherited environment) as run-swat.sh set up. base_command = [ sys.executable, "-u", os.path.join(SCRIPT_DIR, 'symbolic-explorer', 'SymbolicExplorer.py'), @@ -112,7 +113,7 @@ def generate_command(args, property_file: str, property_path: str, input_paths: "--agent", agent_path, "--config", config_path, "-z3", library_path, - "--logdir", logging_dir, + "--logdir", LOG_DIR, "--mode", "sv-comp", '--port', "8000", "--witness-dir", WITNESS_DIR, @@ -149,20 +150,6 @@ def compile_target(args, local_target_dir): log_output(output) -def extract_last_round(out, marker="============================== ROUND"): - - return "\n ".join(out).split(marker)[-1] - - -def extract_markers(out, markers): - if isinstance(markers, str): - markers = (markers,) - res = [] - for line in out.split("\n"): - if any(marker in line for marker in markers): - res.append(line) - return res - def swat_version(): """The package version from BUILD_INFO.txt, written by scripts/package-svcomp.sh.""" try: @@ -221,6 +208,8 @@ def log_output(output: List[str]): cmd = generate_command(sys.argv[1:], property_file, property_path, input_paths) # Exactly one witness per run: drop whatever an earlier run left behind shutil.rmtree(WITNESS_DIR, ignore_errors=True) + if os.path.exists(os.path.join(LOG_DIR, 'stats.json')): + os.remove(os.path.join(LOG_DIR, 'stats.json')) local_target_dir = copy_target(target_dir) compile_target(sys.argv[1:], local_target_dir) @@ -228,22 +217,32 @@ def log_output(output: List[str]): log_dir = f'{path}/logs' cmd[-1] = os.path.join(SCRIPT_DIR, local_target_dir, 'build') #f"{local_target_dir}/build" + # The CPU time limit covers the whole run, so the explorer's budget starts with what copying + # and compiling the task took (the memory and CPU limits are the explorer's defaults). + t = os.times() + cmd[cmd.index('--classpath'):cmd.index('--classpath')] = [ + '--cpu-used', f'{t.user + t.system + t.children_user + t.children_system:.1f}'] #delete the dir if it exists if os.path.exists(log_dir): os.system(f'rm -rf {log_dir}') os.makedirs(log_dir) with pushd(path): - execution_status, output = run_command_with_timeout(cmd) - log_output(output) - print(f"Execution status: {execution_status}") - if execution_status == ExecutionStatus.TIMEOUT or execution_status == ExecutionStatus.ERROR: - print(f'Error: {output}') - exit(1) - - - out = extract_last_round(output) - verdict = extract_markers(out, f'[VERDICT {property_file}]') - if "== ERROR" in str(verdict): + # The explorer's output goes straight to ours: on long runs it is hundreds of MB, and + # collecting and re-logging it here took tens of seconds of CPU time and as much memory, + # after the explorer had already stopped in time for the limits. + logger.info(f'[TARGET EXECUTION]: Running command: {cmd}') + sys.stdout.flush() + sys.stderr.flush() + returncode = subprocess.run(cmd, stderr=subprocess.STDOUT).returncode + print(f"Execution status: {returncode}") + + # The explorer records its verdict in the testcase statistics + try: + with open(os.path.join(LOG_DIR, 'stats.json')) as f: + verdict = json.load(f)['verdict_raw'] + except (OSError, ValueError, KeyError): + verdict = None + if verdict == Verdict.VIOLATION.value: # The explorer wrote the witness from the trace of the violating execution print_witness() print(f'Verdict:\n {verdict}') \ No newline at end of file diff --git a/symbolic-explorer/SymbolicExplorer.py b/symbolic-explorer/SymbolicExplorer.py index be21affd..1640832a 100755 --- a/symbolic-explorer/SymbolicExplorer.py +++ b/symbolic-explorer/SymbolicExplorer.py @@ -139,6 +139,15 @@ def init_args(parser: argparse.ArgumentParser): parser.add_argument("--solver-rlimit", type=int, default=0, help="Z3 resource limit (rlimit) for the first query of each branch in sv-comp mode; a branch that runs out is retried later with a growing limit. 0 (default): unlimited") parser.add_argument("--parallel-sa", action="store_true", help="Run the static pre-analysis in the background and use it once it is ready (sv-comp mode). By default exploration waits for it: running it in parallel makes runs nondeterministic", default=False) + # SV-COMP resource budget, shared by the explorer and all executor JVMs (sv-comp mode). The + # defaults are the competition's limits (bench-defs swat.xml); a lower cgroup memory limit wins. + parser.add_argument("--memory-limit", type=int, default=15000, + help="Memory limit of the whole run in MB (10^6 bytes); the executor heap is derived from it") + parser.add_argument("--cpu-limit", type=float, default=900, + help="CPU time limit of the whole run in seconds; exploration stops with UNKNOWN shortly before it") + parser.add_argument("--cpu-used", type=float, default=0.0, + help="CPU seconds the run used before the explorer started (e.g. compiling the task)") + # SV-COMP witness arguments: the explorer writes the violation witness from the trace of the # violating execution (needs the task's input files and property file as given to the verifier) parser.add_argument("--witness-dir", default="witness", @@ -166,6 +175,11 @@ def init_args(parser: argparse.ArgumentParser): # Initialize loggers log.initialize_loggers(args.logdir) + if args.mode == 'sv-comp': + # The module loggers' DEBUG/INFO lines on stdout (several per branch and query) were 500 MB + # in 900 s on MinePump: CPU time, and memory in a BenchExec container. The explorer's own + # log (explorer.log) is not affected. + logging.getLogger().setLevel(logging.WARNING) # Set configuration flags in Database from data.Database import Database diff --git a/symbolic-explorer/driver/SVCompDriver.py b/symbolic-explorer/driver/SVCompDriver.py index a6c70f11..f4923779 100644 --- a/symbolic-explorer/driver/SVCompDriver.py +++ b/symbolic-explorer/driver/SVCompDriver.py @@ -1,4 +1,5 @@ import glob +import sys import signal import subprocess import os @@ -9,6 +10,7 @@ from solver.SolverHandler import SATResult +from solver.ConstraintCache import get_constraint_cache from data.Database import Database from data.BinaryExecutionTree.Node import Node @@ -21,6 +23,7 @@ from svcomp.SymbolicStorage import SymbolicStorage, DataTypes from svcomp.TestcaseStats import write_testcase_stats from svcomp.WitnessWriter import write_violation_witness +from svcomp.ResourceBudget import ResourceBudget, DEFAULT_MEMORY_LIMIT_MB, DEFAULT_CPU_LIMIT_S, descendants from timing.TimingManager import TimingManager @@ -28,6 +31,7 @@ logger = log.get_logger() verdict_logger = log.get_verdict_logger() import platform +import threading # With --solver-rlimit, each time a branch's query runs out of its resource limit, the limit for @@ -35,6 +39,15 @@ RLIMIT_GROWTH = 5 MAX_RLIMIT = 2**32 - 1 +# Execution errors after which an exhausted search proves nothing: a run was cut short, so the +# rest of its path was never explored. +CUT_SHORT_ERRORS = ('resource_limit', 'out_of_memory') + +# Executor JVM options besides the heap size. Every round starts a new, short-lived JVM, in which +# the C2 compiler spends more CPU time than it ever saves (more than a third of an executor's CPU +# time on MinePump/TCAS); C1 alone compiles fast enough. +EXECUTOR_JVM_OPTIONS = ['-XX:TieredStopAtLevel=1'] + # The (unique) endpoint ID for the SV-COMP target. As each target is handled separately, this ID is always 0. ENDPOINT_ID = 0 @@ -43,6 +56,8 @@ class ExecutionStatus(Enum): SUCCESS = 1 ERROR = 2 TIMEOUT = 3 + # Stopped by the resource budget before it ended (see svcomp/ResourceBudget.py) + RESOURCE_LIMIT = 4 class Verdict(Enum): VIOLATION = "== ERROR" @@ -98,6 +113,14 @@ def __init__(self, args): self.solver_limit_hits = 0 # queries that ran out of their resource limit # How often each branch (by gid) ran out of its resource limit; each time multiplies it. self.rlimit_level: dict[int, int] = {} + self.budget = ResourceBudget(memory_limit_mb=getattr(args, 'memory_limit', None) or DEFAULT_MEMORY_LIMIT_MB, + cpu_limit_s=getattr(args, 'cpu_limit', None) or DEFAULT_CPU_LIMIT_S, + cpu_used_s=getattr(args, 'cpu_used', None) or 0.0) + self.budget.on_emergency = self.end_run_now + # Only one verdict is reported, by the main thread or, in an emergency, by the monitor + self.verdict_lock = threading.Lock() + self.verdict_reported = False + self.stopped_by: str | None = None # why the budget stopped the last executor run, if it did @contextmanager def pushd(self, dirname): @@ -109,13 +132,15 @@ def pushd(self, dirname): finally: os.chdir(original_dir) - def build_command(self,javaagent:str, config:str, Z3_DIR: str, port, cp, java_path:str = 'java', mem:int=32) -> List[str]: - """Builds the Java command list with given parameters.""" + def build_command(self,javaagent:str, config:str, Z3_DIR: str, port, cp, java_path:str = 'java', heap_mib: int | None = None) -> List[str]: + """Builds the Java command list with given parameters. The heap size defaults to the + executor's share of the memory budget.""" classpath_separator = ";" if platform.system() == "Windows" else ":" cmd = [ java_path, - f'-Xmx{mem}g', + f'-Xmx{heap_mib or self.budget.heap_mib}m', + *EXECUTOR_JVM_OPTIONS, f'-Dconfig.path={config}', f'-Dexplorer.port={port}', f'-javaagent:{javaagent}', @@ -133,20 +158,27 @@ def build_command(self,javaagent:str, config:str, Z3_DIR: str, port, cp, java_p if var.dType == DataTypes.STRING and val is not None: val = escape_java_string(val) sym_vars.append(f'-Dswat.input.{var.dType.value}_{var.idx}={val}') - cmd[3:3] = sym_vars + at = cmd.index(f'-Dconfig.path={config}') + 1 + cmd[at:at] = sym_vars return cmd def run_command_with_timeout(self, cmd: List[str]) -> Tuple[ExecutionStatus, List[str]]: """ Executes the given command and returns output from both STDOUT and STDERR. - Todo: What about timeouts? + There is no timeout of its own: the resource budget stops the run when the memory or + CPU time of the whole verification run gets close to its limit (RESOURCE_LIMIT). """ logger.info(f'[SVCOMP] Executing target') output = [] with subprocess.Popen(cmd, stdout=subprocess.PIPE,stderr=subprocess.STDOUT, bufsize=1, universal_newlines=True) as proc: try: - stdout, _ = proc.communicate() #timeout=timeout + with self.budget.executor(proc) as stopped_by: + stdout, _ = proc.communicate() output = stdout.splitlines() + self.stopped_by = stopped_by() + if self.stopped_by: + logger.warning(f'[SVCOMP] Executor stopped by the resource budget ({self.stopped_by})') + return ExecutionStatus.RESOURCE_LIMIT, output return ExecutionStatus.SUCCESS, output except subprocess.TimeoutExpired: @@ -162,11 +194,21 @@ def run_command_with_timeout(self, cmd: List[str]) -> Tuple[ExecutionStatus, Lis def determine_next_step(self, status: ExecutionStatus, output: List[str] ) -> Action: - """Determines the next step based on the execution status.""" - if status == ExecutionStatus.SUCCESS: + """Determines the next step based on the execution status. - + A run the resource budget stopped (RESOURCE_LIMIT) still reports a violation it printed + before. Otherwise it ended without a trace, so its path stays unexplored: on a CPU stop the + verdict is UNKNOWN, on a memory stop exploration goes on with other branches. + """ + if status == ExecutionStatus.SUCCESS or status == ExecutionStatus.RESOURCE_LIMIT: + + out_of_memory_recorded = False for l in output: + if "java.lang.OutOfMemoryError" in l and not out_of_memory_recorded: + # The run was cut short by the JVM's memory limit, which the task does not have + logger.warning(f'[SVCOMP] Executor ran out of memory: {l.strip()}') + Database.instance().record_execution_error(ENDPOINT_ID, 'out_of_memory', l.strip()) + out_of_memory_recorded = True if "java.lang.AssertionError: [SWAT]" in l: # Internal assertion error in the DSE logger.critical(f'[SVCOMP] SWAT Assertion failed: {l}') @@ -190,6 +232,13 @@ def determine_next_step(self, status: ExecutionStatus, output: List[str] ) -> Ac self.state.verdict = Verdict.UNKNOWN return Action.REPORTVERDICT + if status == ExecutionStatus.RESOURCE_LIMIT: + Database.instance().record_execution_error(ENDPOINT_ID, 'resource_limit', + f'round {self.round_idx}: executor stopped ({self.stopped_by})') + if self.stopped_by == 'cpu' or len(Database.instance().get_tree(ENDPOINT_ID).symbolic_vars) == 0: + self.state.verdict = Verdict.UNKNOWN + return Action.REPORTVERDICT + return Action.SYMBOLICNEXT if len(Database.instance().get_tree(ENDPOINT_ID).symbolic_vars) == 0: # No symbolic vars found, safe to assume no violation @@ -251,8 +300,9 @@ def run_testcase(self, java_path, agentpath: str, configpath: str, z3path, port, executor_duration = time.perf_counter() - executor_start TimingManager.instance().record_executor_time(executor_duration) - # check if the target explored the branch as expected - if self.state.branch_to_explore: + # check if the target explored the branch as expected (a run stopped by the resource + # budget sent no trace to check) + if self.state.branch_to_explore and status != ExecutionStatus.RESOURCE_LIMIT: self.check_divergence(Database.instance().get_trace(-1)) # most recent trace self.state.branch_to_explore = None self.state.solution_is_guess = False @@ -277,6 +327,10 @@ def run_testcase(self, java_path, agentpath: str, configpath: str, z3path, port, # Explorer time is computed as residual in TimingManager.get_aggregates() # (total_time - executor_time - solver_time - witness_times) next_step = self.retrieve_solution() + if self.budget.cpu_exceeded.is_set() and next_step != Action.REPORTVERDICT: + logger.warning('[SVCOMP] CPU time deadline reached, stopping exploration') + self.state.verdict = Verdict.UNKNOWN + return self.state.verdict # If we are using SA and come to a SAFE verdict, then continue without SA # this should prevent false SAFE classifications from incomplete SA @@ -367,9 +421,16 @@ def solve_candidates(self, candidates: list[Node]) -> tuple[Node | None, dict, b limited = [] alone = retry and len(pending) == 1 for branch in pending: + if self.budget.cpu_exceeded.is_set(): + return None, {}, True rlimit = None if alone else self.solver_rlimit_for(branch) - sat, sol = StrategyService.solve_branch(branch, solver_timeout_ms=None, solver_rlimit=rlimit) + with self.budget.solver(get_constraint_cache().shared_context) as interrupted_by: + sat, sol = StrategyService.solve_branch(branch, solver_timeout_ms=None, solver_rlimit=rlimit) self.nr_solver_calls += 1 + if interrupted_by(): + logger.info(f'[SYMBOLIC EXPLORATION] Query for branch {branch.id} stopped by the resource budget ({interrupted_by()})') + gave_up = True + continue if sat == SATResult.SAT: return branch, sol, gave_up if sat == SATResult.LIMIT: @@ -433,6 +494,7 @@ def run(self): # Start total timing TimingManager.instance().start_total_timer() + self.budget.start() try: verdict = self.run_testcase(java_path=self.args.java_path, agentpath=self.args.agent, configpath=self.args.config, z3path=self.args.z3dir, port=self.args.port, cp=self.args.classpath) @@ -460,6 +522,11 @@ def run(self): logger.warning(f'[SVCOMP] The SA walk fell out of step with the execution (see [DFS] SA desync)') verdict = Verdict.UNKNOWN + if (verdict == Verdict.SAFE) and any(e.get('kind') in CUT_SHORT_ERRORS + for e in Database.instance().get_tree(ENDPOINT_ID).execution_errors): + logger.warning(f'[SVCOMP] An execution was cut short by a memory limit, its path is unexplored') + verdict = Verdict.UNKNOWN + if (verdict == Verdict.SAFE) and Database.instance().get_tree(ENDPOINT_ID).uncaught_exceptions > 0: logger.warning(f'[SVCOMP] Found uncaught exceptions during symbolic execution') verdict = Verdict.UNKNOWN @@ -470,7 +537,12 @@ def run(self): if verdict == Verdict.NO_SYMBOLIC_VARS: verdict = Verdict.SAFE - verdict_logger.info(f'[VERDICT {self.verification_category.value}] {verdict.value}') + with self.verdict_lock: + if self.verdict_reported: + return + self.verdict_reported = True + verdict_logger.info(f'[VERDICT {self.verification_category.value}] {verdict.value}') + self.budget.stop() logger.info(f'[EXPLORER] Symbolic execution terminated after {self.round_idx} iterations') # The witness is written after the verdict and never affects it @@ -487,12 +559,39 @@ def run(self): # Write the consolidated per-testcase statistics (verdict, soundness flags, missing # invocations and the context-loss subset) so the analysis can rely on structured data. - stats_file = os.path.join(log_dir, 'stats.json') - write_testcase_stats(Path(stats_file), verdict, self.verification_category, Database.instance().get_tree(ENDPOINT_ID), - self.round_idx, self.nr_solver_calls, self.sa.stats(self.sa_adopted_round), - self.solver_limit_hits) + self.write_stats(verdict) self.kill_current_process() + + def write_stats(self, verdict): + stats_file = os.path.join(self.args.logdir if hasattr(self.args, 'logdir') else 'logs', 'stats.json') + write_testcase_stats(Path(stats_file), verdict, self.verification_category, Database.instance().get_tree(ENDPOINT_ID), + self.round_idx, self.nr_solver_calls, self.sa.stats(self.sa_adopted_round), + self.solver_limit_hits, resources=self.budget.stats()) + + def end_run_now(self, why: str): + """Called by the resource budget's monitor when the run must end at once: reports UNKNOWN + (unless the main thread reported a verdict already) and exits. The main thread is stuck in + work the budget cannot stop, or the explorer alone is too large.""" + with self.verdict_lock: + if self.verdict_reported: + return + self.verdict_reported = True + logger.warning(f'[SVCOMP] Ending the run with UNKNOWN: {why}') + verdict_logger.info(f'[VERDICT {self.verification_category.value}] {Verdict.UNKNOWN.value}') + try: + self.write_stats(Verdict.UNKNOWN) + except Exception as e: # the tree may be changing under us + logger.error(f'[SVCOMP] Could not write the statistics: {e}') + for handler in logger.handlers + verdict_logger.handlers: + handler.flush() + sys.stdout.flush() + for pid in descendants(os.getpid()): # the executor or the pre-analysis, if one is running + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + os._exit(0) def write_witness(self): diff --git a/symbolic-explorer/strategy/DFS.py b/symbolic-explorer/strategy/DFS.py index b8d52078..db9c128e 100755 --- a/symbolic-explorer/strategy/DFS.py +++ b/symbolic-explorer/strategy/DFS.py @@ -3,8 +3,12 @@ from data.StaticAnalysisGraph.SAGraph import SANode, CallStack, is_interesting, \ apply_stack_delta, check_branch, resync +import logging + import log logger = log.get_logger() +# Log level of the per-node trace of the walk (enable with logger.setLevel(TRACE)) +TRACE = logging.DEBUG - 5 # The executor's marker for an exception the SA graph does not model # (SymbolicInstructionVisitor.UNMODELLED_EXCEPTION). @@ -81,7 +85,10 @@ def dfs(visited: set[Node], tree, node: Node | Leaf | None, solved_branches: set skip_is_interesting = mask_sa_node or (sa_node is None) or is_interesting(sa_node.get_fallthrough_child(), sa_stack) branch_is_interesting = mask_sa_node or (sa_node is None) or is_interesting(sa_node.get_branched_child(), sa_stack) - logger.info(f"[DFS] @{node.id}/{("CALLBACK" if node.callback else "CLINIT") if mask_sa_node else sa_node and sa_node.id} ({"branched" if node.branched else ""}{"skipped" if node.skipped else ""}): skip_is_interesting={skip_is_interesting}, branch_is_interesting={branch_is_interesting}") + # One line per branch node and round: at INFO this was most of the log (270 MB in 900 s on + # MinePump) and much of the explorer's own CPU time, so it is below DEBUG. + if logger.isEnabledFor(TRACE): + logger.log(TRACE, f"[DFS] @{node.id}/{("CALLBACK" if node.callback else "CLINIT") if mask_sa_node else sa_node and sa_node.id} ({"branched" if node.branched else ""}{"skipped" if node.skipped else ""}): skip_is_interesting={skip_is_interesting}, branch_is_interesting={branch_is_interesting}") # Add the node itself, if eligible if (node.skipped is None and skip_is_interesting) or (node.branched is None and branch_is_interesting): diff --git a/symbolic-explorer/svcomp/ResourceBudget.py b/symbolic-explorer/svcomp/ResourceBudget.py new file mode 100644 index 00000000..8fb9acc2 --- /dev/null +++ b/symbolic-explorer/svcomp/ResourceBudget.py @@ -0,0 +1,344 @@ +""" +The resource budget of an SV-COMP run: the memory and CPU time that the explorer and all executor +JVMs share. SV-COMP (bench-defs swat.xml) limits the whole run to 15 GB of memory and 15 min of CPU +time, summed over all processes, on 4 cores. + +A monitor thread polls the resident memory and CPU time of the explorer and its descendants. Close to +a limit it stops the work in progress itself, so that the run still ends with a verdict instead of +being killed from outside: +- memory: the executor run in progress is killed (it ends without a trace, so the path it was to + explore stays unexplored) and exploration goes on with the other branches; a Z3 query is + interrupted (the branch counts as undecided). Either rules out a SAFE verdict. +- CPU time: exploration stops and the run ends with UNKNOWN, unless a violation was found already. +""" +import ctypes +import os +import signal +import threading +import time +from contextlib import contextmanager + +import log + +logger = log.get_logger() + +MB = 1000 * 1000 +MIB = 1024 * 1024 + +# BenchExec counts in SI units: memlimit "15 GB" is 15e9 bytes. +DEFAULT_MEMORY_LIMIT_MB = 15000 +DEFAULT_CPU_LIMIT_S = 900 + +# Kept free below the memory limit: page cache and the files the run writes (which count against +# the limit in a BenchExec container), the wrapper scripts, and growth between two polls (the +# executor JVM was seen to grow by 2 GB/s). +HEADROOM_MB = 2000 +# If the explorer alone is above the limit minus this, nothing can be stopped to make room. +HARD_HEADROOM_MB = 1000 +# How long an interrupted Z3 query may stay over the memory threshold. Z3 does not react to an +# interrupt while it bit-blasts, during which it can allocate gigabytes in seconds. (Z3's own +# memory_max_size is no way out: Z3 then aborts the process in some places.) +INTERRUPT_GRACE_S = 1.0 +# The explorer's share (Python, the execution tree, Z3). The executor JVM gets half of the rest as +# heap; the other half is for its native memory, mostly JavaSMT's Z3 context, which grows with the +# trace about as fast as the heap. +EXPLORER_SHARE_MB = 1500 +MIN_HEAP_MB = 256 + +# Stop this much before the CPU limit: polling lag (up to 4 cores per poll interval), killing the +# executor and writing the verdict, the witness and the statistics. +CPU_MARGIN_S = 20 +POLL_S = 0.2 +# Finding all descendants means reading every process's parent, which is slow on a busy machine; +# in between, only the processes found last time and the executor are measured. +FULL_SCAN_S = 5 +# How long the main thread may take to wind down after the CPU deadline before the monitor ends +# the run itself. +GRACE_S = 10 + +CLK_TCK = os.sysconf('SC_CLK_TCK') +PAGE_SIZE = os.sysconf('SC_PAGE_SIZE') + + +def cgroup_memory_limit(proc_root: str = '/proc', cgroup_root: str = '/sys/fs/cgroup') -> int | None: + """The tightest memory limit (bytes) of this process's cgroup and its ancestors, or None. + + BenchExec (and a systemd scope with MemoryMax) enforce the limit through the cgroup, so it is + the limit that actually applies when it is lower than the configured one. + """ + return limiting_cgroup(proc_root, cgroup_root)[0] + + +def limiting_cgroup(proc_root: str = '/proc', cgroup_root: str = '/sys/fs/cgroup') -> tuple[int | None, str | None]: + """(limit, directory) of the cgroup with the tightest memory limit, see cgroup_memory_limit().""" + try: + lines = open(f'{proc_root}/self/cgroup').read().splitlines() + except OSError: + return None, None + limits = [] + for line in lines: + parts = line.split(':', 2) + if len(parts) != 3: + continue + _, controllers, path = parts + if controllers == '': # cgroup v2 + base, name = cgroup_root, 'memory.max' + elif 'memory' in controllers.split(','): # cgroup v1 + base, name = f'{cgroup_root}/memory', 'memory.limit_in_bytes' + else: + continue + path = path.rstrip('/') + while True: + try: + value = open(f'{base}{path}/{name}').read().strip() + if value.isdigit() and int(value) < 2**60: # v1 reports "no limit" as a huge number + limits.append((int(value), f'{base}{path}')) + except OSError: + pass + if not path: + break + path = path.rsplit('/', 1)[0] + return min(limits) if limits else (None, None) + + +def cgroup_cpu_time(directory: str | None) -> float | None: + """CPU seconds of all processes in a cgroup (v2 cpu.stat), or None.""" + if directory is None: + return None + try: + with open(f'{directory}/cpu.stat') as f: + for line in f: + if line.startswith('usage_usec '): + return int(line.split()[1]) / 1e6 + except (OSError, ValueError): + pass + return None + + +def _stat(pid: int, proc_root: str = '/proc') -> tuple[int, float, int] | None: + """(ppid, CPU seconds, RSS bytes) of a process, or None if it is gone.""" + try: + with open(f'{proc_root}/{pid}/stat') as f: + fields = f.read().rsplit(')', 1)[1].split() + except (OSError, IndexError): + return None + return int(fields[1]), (int(fields[11]) + int(fields[12])) / CLK_TCK, int(fields[21]) * PAGE_SIZE + + +def descendants(pid: int, proc_root: str = '/proc') -> dict[int, tuple[float, int]]: + """{pid: (CPU seconds, RSS bytes)} of the live descendants of a process.""" + stats = {} + for entry in os.listdir(proc_root): + if entry.isdigit(): + s = _stat(int(entry), proc_root) + if s is not None: + stats[int(entry)] = s + children: dict[int, list[int]] = {} + for p, (ppid, _, _) in stats.items(): + children.setdefault(ppid, []).append(p) + result, todo = {}, list(children.get(pid, [])) + while todo: + p = todo.pop() + result[p] = stats[p][1:] + todo.extend(children.get(p, [])) + return result + + +def _malloc_trim(): + """Returns freed heap memory to the system (glibc keeps it otherwise, e.g. after a Z3 query).""" + try: + ctypes.CDLL('libc.so.6').malloc_trim(0) + except (OSError, AttributeError): + pass + + +class ResourceBudget: + + def __init__(self, memory_limit_mb: int = DEFAULT_MEMORY_LIMIT_MB, cpu_limit_s: float = DEFAULT_CPU_LIMIT_S, + cpu_used_s: float = 0.0, cgroup_limit: int | None = None): + """cpu_used_s: CPU time the run used before the explorer started (e.g. compiling the task), + which counts against the same limit. cgroup_limit: the cgroup's memory limit; read from the + system unless given.""" + self.configured_memory_limit = memory_limit_mb * MB + if cgroup_limit is None: + cgroup_limit, self._cgroup_dir = limiting_cgroup() + else: + self._cgroup_dir = None + self.cgroup_memory_limit = cgroup_limit + self.memory_limit = min(self.configured_memory_limit, self.cgroup_memory_limit or self.configured_memory_limit) + self.memory_threshold = self.memory_limit - HEADROOM_MB * MB + self.hard_memory_threshold = self.memory_limit - HARD_HEADROOM_MB * MB + self.heap_mib = max(MIN_HEAP_MB * MB, (self.memory_threshold - EXPLORER_SHARE_MB * MB) // 2) // MIB + self.cpu_limit = cpu_limit_s + self.cpu_used_before = cpu_used_s + self.cpu_deadline = cpu_limit_s - CPU_MARGIN_S + + self.cpu_exceeded = threading.Event() + self.cpu_exceeded_at: float | None = None + self.peak_rss = 0 + self.cpu_used = cpu_used_s + self.memory_hits = 0 # how often work was stopped to stay within the memory limit + self.solver_interruptions = 0 + + self._lock = threading.Lock() + self._executor = None # the running executor process (subprocess.Popen) + self._executor_stopped: str | None = None # why the monitor killed it: 'memory' or 'cpu' + self._solver_context = None # the Z3 context of a running query + self._solver_interrupted: str | None = None + self._solver_interrupted_at = 0.0 + self._tracked: set[int] = set() # descendants found by the last full scan + self._last_scan = 0.0 + self._thread: threading.Thread | None = None + self._stop = threading.Event() + # Called (from the monitor thread) when the run must end now: the explorer alone is over + # the memory limit, or the main thread did not wind down after the CPU deadline. + self.on_emergency = None + + def describe(self) -> str: + cg = f', cgroup limit {self.cgroup_memory_limit // MB} MB' if self.cgroup_memory_limit else '' + return (f'memory limit {self.memory_limit // MB} MB{cg} (executor heap {self.heap_mib} MiB, ' + f'stop at {self.memory_threshold // MB} MB), CPU limit {self.cpu_limit:.0f} s ' + f'(stop at {self.cpu_deadline:.0f} s, {self.cpu_used_before:.1f} s used before)') + + # ---- measuring ---- + + def usage(self) -> tuple[int, float]: + """(RSS bytes, CPU seconds) of the run so far: the explorer and all its descendants, plus the + CPU time used before the explorer started. Where the memory limit comes from a cgroup (as + under BenchExec), the CPU time is at least that of the cgroup, which also counts the wrapper + scripts that started the explorer and keep running next to it.""" + own = _stat(os.getpid()) + rss = own[2] if own else 0 + t = os.times() + cpu = self.cpu_used_before + t.user + t.system + t.children_user + t.children_system + if time.monotonic() - self._last_scan >= FULL_SCAN_S: + self._tracked = set(descendants(os.getpid())) + self._last_scan = time.monotonic() + proc = self._executor + for pid in self._tracked | ({proc.pid} if proc is not None else set()): + st = _stat(pid) + if st is not None: + cpu += st[1] + rss += st[2] + return rss, max(cpu, cgroup_cpu_time(self._cgroup_dir) or 0.0) + + # ---- the work the monitor may stop ---- + + @contextmanager + def executor(self, proc): + """Marks proc as the running executor. Yields a callable that tells, once it ended, whether + the monitor stopped it, and why ('memory' or 'cpu'), or None.""" + with self._lock: + self._executor, self._executor_stopped = proc, None + try: + yield lambda: self._executor_stopped + finally: + with self._lock: + self._executor = None + + @contextmanager + def solver(self, context): + """Marks a Z3 query on context as running. Yields a callable that tells, once it returned, + whether the monitor interrupted it, and why, or None.""" + with self._lock: + self._solver_context, self._solver_interrupted = context, None + try: + yield lambda: self._solver_interrupted + finally: + with self._lock: + self._solver_context = None + if self._solver_interrupted: + _malloc_trim() + + def _stop_work(self, reason: str) -> bool | None: + """Stops the executor run or the Z3 query in progress: True if it was running, None if it + is being stopped already, False if there is none.""" + with self._lock: + proc = self._executor + if proc is not None and proc.poll() is None: + if self._executor_stopped: + return None + self._executor_stopped = reason + # SIGKILL: on SIGTERM the executor's shutdown hook would send the trace cut short + # as if the target had ended there. + for pid in [*descendants(proc.pid), proc.pid]: + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + return True + if self._solver_context is not None: + if self._solver_interrupted: + self._solver_context.interrupt() # for a check started since + return None + self._solver_interrupted = reason + self._solver_interrupted_at = time.monotonic() + self.solver_interruptions += 1 + self._solver_context.interrupt() + return True + return False + + # ---- the monitor ---- + + def start(self): + logger.info(f'[BUDGET] {self.describe()}') + self._thread = threading.Thread(target=self._monitor, name='resource-budget', daemon=True) + self._thread.start() + + def stop(self): + self._stop.set() + + def _monitor(self): + while not self._stop.wait(POLL_S): + try: + self.check() + except Exception as e: # the monitor must not die + logger.error(f'[BUDGET] Monitor error: {e}') + + def check(self): + """One poll: measure, and act if a limit is near.""" + rss, cpu = self.usage() + self.peak_rss = max(self.peak_rss, rss) + self.cpu_used = cpu + if cpu >= self.cpu_deadline: + if not self.cpu_exceeded.is_set(): + logger.warning(f'[BUDGET] CPU time {cpu:.1f} s reached the deadline of {self.cpu_deadline:.0f} s, stopping') + self.cpu_exceeded_at = time.monotonic() + self.cpu_exceeded.set() + # Also stops what was started after the deadline, until the main thread has wound down + self._stop_work('cpu') + if time.monotonic() - self.cpu_exceeded_at > GRACE_S: + self._emergency(f'the explorer did not stop within {GRACE_S} s of the CPU deadline') + if rss >= self.memory_threshold: + stopped = self._stop_work('memory') + if stopped: + self.memory_hits += 1 + logger.warning(f'[BUDGET] Memory {rss // MB} MB reached {self.memory_threshold // MB} MB, ' + f'stopped the work in progress') + elif stopped is None and self._solver_context is not None: + # An interrupted query that keeps its memory, or keeps growing + if rss >= self.hard_memory_threshold or time.monotonic() - self._solver_interrupted_at > INTERRUPT_GRACE_S: + self._emergency(f'a Z3 query uses {rss // MB} MB of the {self.memory_limit // MB} MB limit and does not stop') + elif stopped is False: + _malloc_trim() + if self.usage()[0] >= self.hard_memory_threshold: + self._emergency(f'the explorer uses {rss // MB} MB of the {self.memory_limit // MB} MB limit') + + def _emergency(self, why: str): + logger.error(f'[BUDGET] Ending the run: {why}') + if self.on_emergency is not None: + self.on_emergency(why) + + def stats(self) -> dict: + return { + 'memory_limit_mb': self.memory_limit // MB, + 'cgroup_memory_limit_mb': self.cgroup_memory_limit // MB if self.cgroup_memory_limit else None, + 'executor_heap_mib': self.heap_mib, + 'peak_rss_mb': self.peak_rss // MB, + 'memory_hits': self.memory_hits, + 'solver_interruptions': self.solver_interruptions, + 'cpu_limit_s': self.cpu_limit, + 'cpu_deadline_s': self.cpu_deadline, + 'cpu_used_s': round(self.cpu_used, 1), + 'cpu_deadline_reached': self.cpu_exceeded.is_set(), + } diff --git a/symbolic-explorer/svcomp/SVCompHandler.py b/symbolic-explorer/svcomp/SVCompHandler.py deleted file mode 100755 index a14c9ae4..00000000 --- a/symbolic-explorer/svcomp/SVCompHandler.py +++ /dev/null @@ -1,246 +0,0 @@ -import glob -import re -import signal -import subprocess -import os, sys -from contextlib import contextmanager - - -from solver.SolverHandler import SATResult - -from data.Database import Database - -from strategy.StrategyService import StrategyService - -from enum import Enum -from svcomp.SymbolicStorage import SymbolicStorage -import logging - -import log -logger = log.get_logger() -verdict_logger = log.get_verdict_logger() - - -class ExecutionStatus(Enum): - SUCCESS = 1 - ERROR = 2 - TIMEOUT = 3 - CRASH = 4 - VIOLATION = 5 - -class Verdict(Enum): - VIOLATION = "== ERROR" - SAFE = "== OK" - UNKNOWN = "== DONT-KNOW" - NO_SYMBOLIC_VARS = "== NON-SYMBOLIC" - -class Action(Enum): - RANDOMNEXT = 1 - SYMBOLICNEXT = 2 - REPORTVERDICT = 3 - -class INPUTTYPE(Enum): - RANDOM = 1 - SYMBOLIC = 2 - MAGIC = 3 - - - -class State: - def __init__(self): - self.verdict = Verdict.UNKNOWN - - -class SVCompHandler: - def __init__(self): - self.state = State() - self.symbolicStorage = SymbolicStorage() - self.shutdown_flag = False - - @contextmanager - def pushd(self, dirname): - """Context manager to temporarily change the working directory.""" - original_dir = os.getcwd() - os.chdir(dirname) - try: - yield - finally: - os.chdir(original_dir) - - def build_command(self, javaagent:str, config:str, Z3_DIR: str, mem:int=32) -> [str]: - """Builds the Java command list with given parameters.""" - cmd = [ - 'java', - f'-Xmx{mem}g', - f'-Dconfig.path={config}', - f'-javaagent:{javaagent}', - f"-Djava.library.path={Z3_DIR}", - '-cp', - f'build:{Z3_DIR}', - '-ea', - 'Main' - ] - if self.symbolicStorage.input_type == INPUTTYPE.SYMBOLIC: - - sym_vars = [] - for var in self.symbolicStorage.vars.values(): - val = var.newValue if var.newValue is not None else var.value - sym_vars.append(f'-Dswat.input.{var.dType.value}_{var.idx}={val}') - cmd[3:3] = sym_vars - return cmd - - - - def run_command_with_timeout(self, cmd: [str], timeout: int = 60) -> (ExecutionStatus, dict): - """Executes the given command and returns the status and message.""" - - logger.info(f'[SVCOMP] Java Output Begin') - try: - stdout = [] - with subprocess.Popen(cmd, stdout=subprocess.PIPE,stderr=subprocess.STDOUT, bufsize=1, universal_newlines=True) as proc: - for line in proc.stdout: - logger.info(f'[SWAT] --> {line.strip()}') - stdout.append(line) - logger.info(f'[SVCOMP] Java Output End') - for l in stdout: - if '*** java.lang.instrument ASSERTION FAILED ***' in l: - return ExecutionStatus.ERROR, stdout - if proc.returncode == 0: - return ExecutionStatus.SUCCESS, stdout - else: - for l in stdout: - if "java.lang.AssertionError" in l: - return ExecutionStatus.VIOLATION, stdout - return ExecutionStatus.ERROR, stdout - except subprocess.TimeoutExpired: - return ExecutionStatus.TIMEOUT, stdout - except Exception as e: - logger.critical(f'[SVCOMP] Exception: {e}') - return ExecutionStatus.CRASH, str(e) - - - - - def determine_next_step(self, status: ExecutionStatus, stdout: [str] ) -> Action: - """Determines the next step based on the execution status.""" - if status == ExecutionStatus.SUCCESS: - for line in stdout: - if re.match(r'.*\[DSE\] Initializing symbolic tracking for the following value .*', line): - return Action.SYMBOLICNEXT - - self.state.verdict = Verdict.NO_SYMBOLIC_VARS - - - return Action.REPORTVERDICT - - elif status == ExecutionStatus.VIOLATION: - - self.state.verdict = Verdict.VIOLATION - - return Action.REPORTVERDICT - - elif status == ExecutionStatus.TIMEOUT or status == ExecutionStatus.CRASH or status == ExecutionStatus.ERROR: - self.state.verdict = Verdict.UNKNOWN - return Action.REPORTVERDICT - - else: - self.state.verdict = Verdict.UNKNOWN - return Action.REPORTVERDICT - - def run_testcase(self, base_dir: str, classpath: [str], agentpath: str, configpath: str, z3path) -> Verdict: - """Runs the testcase using the constructed Java command.""" - db = Database.instance() - - next_step = Action.RANDOMNEXT - idx = 0 - while True: - - logger.info(f'[SVCOMP] ============================= ROUND ({idx}) =============================') - idx +=1 - command = self.build_command(javaagent=agentpath, config=configpath, Z3_DIR=z3path) - logger.info(f'[SVCOMP] Execution command: {command}') - - status, output = self.run_command_with_timeout(cmd=command, timeout=60) - logger.info(f'[STATUS] {status}') - next_step = self.determine_next_step(status, output) - - - if next_step == Action.REPORTVERDICT: - logger.info(f'[SVCOMP] Next step: REPORTVERDICT') - return self.state.verdict - - - elif next_step == Action.SYMBOLICNEXT: - logger.info(f'[SVCOMP] Next step: SYMBOLIC EXPLORATION') - - next_step = self.retrieve_solution() - if next_step == Action.REPORTVERDICT: - return self.state.verdict - elif next_step == Action.SYMBOLICNEXT: - self.symbolicStorage.input_type = INPUTTYPE.SYMBOLIC - - - - - def retrieve_solution(self): - possible_branches = StrategyService.select_branch(endpoint_id=0) - logger.info(f'[SYMBOLIC EXPLORATION] Found {len(possible_branches)} possible branches') - symbolic_vars = None - sat = None - branch_found = False - for branch in possible_branches: - if not StrategyService.is_symbolic_branch(branch): - continue - branch_found = True - sat, sol = StrategyService.solve_branch(branch, solver_timeout_ms=None) - - if sat == SATResult.SAT: - symbolic_vars = branch.inputs - break - - if not branch_found or sat == SATResult.UNSAT: - self.state.verdict = Verdict.SAFE - logger.info(f'[SYMBOLIC EXPLORATION] No symbolic branch found or UNSAT') - return Action.REPORTVERDICT - - if sat == SATResult.UNKNOWN: - logger.info(f'[SYMBOLIC EXPLORATION] SAT result is UNKNOWN') - self.state.verdict = Verdict.UNKNOWN - return Action.REPORTVERDICT - - sol_viz = [f'{key}: {val["plain_value"]}' for key, val in sol.items()] - logger.info(f'[SYMBOLIC EXPLORATION] Found new solution: {sol_viz}') - self.symbolicStorage.register_vars(symbolic_vars) - self.symbolicStorage.store_solution(sol) - return Action.SYMBOLICNEXT - - def check_missing_invocations(self): - for l in glob.glob("logs/invocation-*.log"): - with open(l, "r") as f: - if f.read().strip() != "": - return True - return False - - def run(self, basedir, classpath: [str], agentpath: str, configpath:str, z3dir:str): - Database.instance().add_endpoint(0) - logger.info(f'[SVCOMP] Beginning testcase analysis') - verdict = self.run_testcase(basedir, classpath, agentpath, configpath, z3dir) - - if (verdict == Verdict.SAFE) and self.check_missing_invocations(): - verdict = Verdict.UNKNOWN - - if verdict == Verdict.NO_SYMBOLIC_VARS: - verdict = Verdict.SAFE - - verdict_logger.info(f'[VERDICT ASSERT] {verdict.value}') - - - self.kill_current_process() - - - def kill_current_process(self): - pid = os.getpid() - os.kill(pid, signal.SIGTERM) # Send termination signal - # os.kill(pid, signal.SIGKILL) # Use this for a more forceful kill if needed - - diff --git a/symbolic-explorer/svcomp/TestcaseStats.py b/symbolic-explorer/svcomp/TestcaseStats.py index 6f129eaa..4b1fb936 100644 --- a/symbolic-explorer/svcomp/TestcaseStats.py +++ b/symbolic-explorer/svcomp/TestcaseStats.py @@ -25,7 +25,7 @@ def _signature(inv: dict) -> str: def build_testcase_stats(verdict, category, tree, symbolic_exec_iterations: int, nr_solver_calls: int, static_analysis: dict, - solver_limit_hits: int = 0) -> dict: + solver_limit_hits: int = 0, resources: dict | None = None) -> dict: """ Assemble the consolidated per-testcase statistics dictionary. @@ -34,6 +34,7 @@ def build_testcase_stats(verdict, category, tree, symbolic_exec_iterations: int, category: The VerificationCategory enum for the property. tree: The per-endpoint Tree holding accumulated metadata for the testcase. static_analysis: The pre-analysis outcome, see StaticPreAnalysis.stats(). + resources: The resource budget and its use, see ResourceBudget.stats(). Returns: A JSON-serializable dict describing the testcase outcome. @@ -73,13 +74,15 @@ def build_testcase_stats(verdict, category, tree, symbolic_exec_iterations: int, 'expected_divergences': tree.expected_divergences, }, 'static_analysis': static_analysis, + 'resources': resources, } def write_testcase_stats(filepath: Path, verdict, category, tree, symbolic_exec_iterations: int, nr_solver_calls: int, static_analysis: dict, - solver_limit_hits: int = 0): + solver_limit_hits: int = 0, resources: dict | None = None): """Write the consolidated per-testcase statistics to ``filepath`` as JSON.""" - data = build_testcase_stats(verdict, category, tree, symbolic_exec_iterations, nr_solver_calls, static_analysis, solver_limit_hits) + data = build_testcase_stats(verdict, category, tree, symbolic_exec_iterations, nr_solver_calls, static_analysis, solver_limit_hits, + resources) filepath.parent.mkdir(parents=True, exist_ok=True) with open(filepath, 'w') as f: diff --git a/symbolic-explorer/tests/test_resource_budget.py b/symbolic-explorer/tests/test_resource_budget.py new file mode 100644 index 00000000..0b3aa96e --- /dev/null +++ b/symbolic-explorer/tests/test_resource_budget.py @@ -0,0 +1,241 @@ +""" +Unit tests for the SV-COMP resource budget (svcomp/ResourceBudget.py) and how SVCompDriver acts on +it: the derived limits, the monitor stopping the executor or a Z3 query, and the verdicts after a run +the budget stopped. + +It imports the driver, which needs the explorer's dependencies (z3, fastapi, ...) that the system +Python lacks, so run it with the harness venv, from symbolic-explorer/: + ../targets/sv-comp/scripts/.venv/bin/python3 -m unittest tests.test_resource_budget -v +""" + +import os +import subprocess +import sys +import tempfile +import time +import unittest +from types import SimpleNamespace +from unittest import mock + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from driver import SVCompDriver as svcomp +from svcomp import ResourceBudget as rb +from svcomp.ResourceBudget import ResourceBudget, MB, MIB + + +class CgroupLimitTest(unittest.TestCase): + + def fs(self, cgroup_line, limits): + root = tempfile.mkdtemp() + os.makedirs(f'{root}/proc/self') + with open(f'{root}/proc/self/cgroup', 'w') as f: + f.write(cgroup_line + '\n') + for path, value in limits.items(): + os.makedirs(f'{root}/cg{path}', exist_ok=True) + with open(f'{root}/cg{path}/{"memory.limit_in_bytes" if path.startswith("/memory") else "memory.max"}', 'w') as f: + f.write(value + '\n') + return dict(proc_root=f'{root}/proc', cgroup_root=f'{root}/cg') + + def test_v2_takes_the_tightest_ancestor(self): + roots = self.fs('0::/bench/run/tool', {'/bench': '20000000000', '/bench/run': '15000000000', '/bench/run/tool': 'max'}) + self.assertEqual(rb.cgroup_memory_limit(**roots), 15_000_000_000) + + def test_v2_without_limit(self): + roots = self.fs('0::/user.slice/session.scope', {'/user.slice': 'max', '/user.slice/session.scope': 'max'}) + self.assertIsNone(rb.cgroup_memory_limit(**roots)) + + def test_v1_ignores_the_no_limit_value(self): + roots = self.fs('5:memory:/run', {'/memory/run': str(2**63 - 4096), '/memory': '8000000000'}) + self.assertEqual(rb.cgroup_memory_limit(**roots), 8_000_000_000) + + + def test_cpu_time_of_the_limiting_cgroup(self): + roots = self.fs('0::/bench/run', {'/bench': 'max', '/bench/run': '15000000000'}) + limit, directory = rb.limiting_cgroup(**roots) + self.assertEqual(limit, 15_000_000_000) + with open(f'{directory}/cpu.stat', 'w') as f: + f.write('usage_usec 12500000\nuser_usec 10000000\n') + self.assertEqual(rb.cgroup_cpu_time(directory), 12.5) + self.assertIsNone(rb.cgroup_cpu_time(None)) + + +class LimitsTest(unittest.TestCase): + + def test_competition_defaults(self): + b = ResourceBudget(cgroup_limit=0) + self.assertEqual(b.memory_limit, 15_000 * MB) + self.assertEqual(b.memory_threshold, 13_000 * MB) + # half of what is left after the explorer's share + self.assertEqual(b.heap_mib, (13_000 - 1_500) * MB // 2 // MIB) + self.assertEqual(b.cpu_deadline, 880) + + def test_lower_cgroup_limit_wins(self): + b = ResourceBudget(cgroup_limit=8_000 * MB) + self.assertEqual(b.memory_limit, 8_000 * MB) + self.assertLess(b.heap_mib * MIB + 1_500 * MB, b.memory_threshold) + + def test_cpu_used_before_counts(self): + b = ResourceBudget(cgroup_limit=0, cpu_used_s=12.5) + self.assertGreaterEqual(b.usage()[1], 12.5) + + def test_usage_includes_children(self): + b = ResourceBudget(cgroup_limit=0) + own = b.usage()[0] + with subprocess.Popen(['sleep', '5']) as p: + time.sleep(0.2) # until it is running + b._last_scan = 0.0 # due for a full scan + self.assertGreater(b.usage()[0], own) + p.kill() + + def test_usage_includes_the_executor_between_scans(self): + b = ResourceBudget(cgroup_limit=0) + own = b.usage()[0] + with subprocess.Popen(['sleep', '5']) as p, b.executor(p): + time.sleep(0.2) + self.assertGreater(b.usage()[0], own) + p.kill() + + +class MonitorTest(unittest.TestCase): + + def budget(self, rss, cpu): + b = ResourceBudget(cgroup_limit=0) + b.usage = lambda: (rss, cpu) + b.on_emergency = mock.Mock() + return b + + def test_cpu_deadline_kills_the_executor(self): + b = self.budget(100 * MB, 885) + with subprocess.Popen(['sleep', '30']) as p, b.executor(p) as stopped_by: + b.check() + self.assertEqual(p.wait(timeout=5), -9) # SIGKILL, so no shutdown hook sends a trace + self.assertEqual(stopped_by(), 'cpu') + self.assertTrue(b.cpu_exceeded.is_set()) + b.on_emergency.assert_not_called() + + def test_below_the_limits_nothing_happens(self): + b = self.budget(12_000 * MB, 800) + with subprocess.Popen(['sleep', '30']) as p, b.executor(p) as stopped_by: + b.check() + self.assertIsNone(p.poll()) + p.kill() + self.assertIsNone(stopped_by()) + self.assertFalse(b.cpu_exceeded.is_set()) + + def test_memory_interrupts_a_query(self): + b = self.budget(13_100 * MB, 10) + ctx = mock.Mock() + with b.solver(ctx) as interrupted_by: + b.check() + b.check() # still over: interrupted again, but counted once + self.assertEqual(ctx.interrupt.call_count, 2) + self.assertEqual(b.memory_hits, 1) + self.assertEqual(interrupted_by(), 'memory') + self.assertFalse(b.cpu_exceeded.is_set()) + + def test_query_that_does_not_stop_ends_the_run(self): + b = self.budget(13_100 * MB, 10) + with b.solver(mock.Mock()): + b.check() + b.check() + b.on_emergency.assert_not_called() + b._solver_interrupted_at -= rb.INTERRUPT_GRACE_S + 0.1 + b.check() + b.on_emergency.assert_called_once() + + def test_explorer_alone_over_the_limit_ends_the_run(self): + b = self.budget(14_100 * MB, 10) + b.check() + b.on_emergency.assert_called_once() + + def test_explorer_slightly_over_the_threshold_is_tolerated(self): + b = self.budget(13_100 * MB, 10) + b.check() + b.on_emergency.assert_not_called() + + def test_main_thread_stuck_after_the_deadline_ends_the_run(self): + b = self.budget(100 * MB, 890) + b.check() + b.on_emergency.assert_not_called() + b.cpu_exceeded_at -= rb.GRACE_S + 1 + b.check() + b.on_emergency.assert_called_once() + + +class DriverTest(unittest.TestCase): + + def setUp(self): + self.tree = SimpleNamespace(execution_errors=[], symbolic_vars={'I_0'}) + def record_execution_error(endpoint_id, kind, message): + self.tree.execution_errors.append({'kind': kind, 'message': message}) + db = SimpleNamespace(record_execution_error=record_execution_error, get_tree=lambda e: self.tree, + record_uncaught_exception=lambda e: None) + patcher = mock.patch.object(svcomp.Database, 'instance', return_value=db) + patcher.start() + self.addCleanup(patcher.stop) + d = svcomp.SVCompDriver.__new__(svcomp.SVCompDriver) + d.state = svcomp.State() + d.verification_category = svcomp.VerificationCategory.VALID_ASSERT_PRP + d.round_idx = 3 + d.stopped_by = None + d.budget = ResourceBudget(cgroup_limit=0) + self.driver = d + + def kinds(self): + return [e['kind'] for e in self.tree.execution_errors] + + def test_memory_stop_goes_on_exploring(self): + self.driver.stopped_by = 'memory' + step = self.driver.determine_next_step(svcomp.ExecutionStatus.RESOURCE_LIMIT, ['some output']) + self.assertEqual(step, svcomp.Action.SYMBOLICNEXT) + self.assertEqual(self.kinds(), ['resource_limit']) # rules out SAFE, see CUT_SHORT_ERRORS + + def test_memory_stop_of_the_first_run_is_unknown(self): + self.tree.symbolic_vars = set() # no trace yet: not "no symbolic inputs" + self.driver.stopped_by = 'memory' + step = self.driver.determine_next_step(svcomp.ExecutionStatus.RESOURCE_LIMIT, []) + self.assertEqual(step, svcomp.Action.REPORTVERDICT) + self.assertEqual(self.driver.state.verdict, svcomp.Verdict.UNKNOWN) + + def test_cpu_stop_is_unknown(self): + self.driver.stopped_by = 'cpu' + step = self.driver.determine_next_step(svcomp.ExecutionStatus.RESOURCE_LIMIT, []) + self.assertEqual(step, svcomp.Action.REPORTVERDICT) + self.assertEqual(self.driver.state.verdict, svcomp.Verdict.UNKNOWN) + + def test_violation_printed_before_the_stop_is_reported(self): + self.driver.stopped_by = 'cpu' + step = self.driver.determine_next_step(svcomp.ExecutionStatus.RESOURCE_LIMIT, + ['Exception in thread "main" java.lang.AssertionError']) + self.assertEqual(step, svcomp.Action.REPORTVERDICT) + self.assertEqual(self.driver.state.verdict, svcomp.Verdict.VIOLATION) + + def test_out_of_memory_error_is_recorded(self): + self.driver.verification_category = svcomp.VerificationCategory.NO_RUNTIME_EXCEPTION_PRP + out = ['java.lang.OutOfMemoryError: Requested array size exceeds VM limit', + 'Exception in thread "main" java.lang.OutOfMemoryError: Requested array size exceeds VM limit'] + step = self.driver.determine_next_step(svcomp.ExecutionStatus.SUCCESS, out) + self.assertEqual(step, svcomp.Action.SYMBOLICNEXT) + self.assertEqual(self.kinds(), ['out_of_memory']) + + def test_heap_and_jvm_options_in_the_command(self): + self.driver.symbolicStorage = SimpleNamespace(input_type=None) + cmd = self.driver.build_command('agent.jar', 'sv-comp.cfg', 'z3', 8000, ['cp'], java_path='java') + self.assertIn(f'-Xmx{self.driver.budget.heap_mib}m', cmd) + self.assertIn('-XX:TieredStopAtLevel=1', cmd) + + def test_no_query_after_the_cpu_deadline(self): + self.driver.args = SimpleNamespace(solver_rlimit=0) + self.driver.nr_solver_calls = self.driver.solver_limit_hits = 0 + self.driver.rlimit_level = {} + self.driver.budget.cpu_exceeded.set() + with mock.patch.object(svcomp.StrategyService, 'solve_branch') as solve: + found, _, gave_up = self.driver.solve_candidates([SimpleNamespace(gid=1, id=1)]) + solve.assert_not_called() + self.assertIsNone(found) + self.assertTrue(gave_up) # so the verdict is not SAFE + + +if __name__ == '__main__': + unittest.main() diff --git a/symbolic-explorer/tests/test_solver_rlimit.py b/symbolic-explorer/tests/test_solver_rlimit.py index 288d0e7f..d8f4ad08 100644 --- a/symbolic-explorer/tests/test_solver_rlimit.py +++ b/symbolic-explorer/tests/test_solver_rlimit.py @@ -15,6 +15,7 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from driver import SVCompDriver as svcomp +from svcomp.ResourceBudget import ResourceBudget from solver.SolverHandler import SATResult BASE = 1000 @@ -46,6 +47,7 @@ def driver(self, rlimit=BASE): d.nr_solver_calls = 0 d.solver_limit_hits = 0 d.rlimit_level = {} + d.budget = ResourceBudget(cgroup_limit=0) return d def solve(self, d, candidates, solver): diff --git a/targets/sv-comp/scripts/commands/test.py b/targets/sv-comp/scripts/commands/test.py index 05ad29a8..434ff9f4 100644 --- a/targets/sv-comp/scripts/commands/test.py +++ b/targets/sv-comp/scripts/commands/test.py @@ -138,7 +138,7 @@ def run_tests(ctx, mode, workers, benchmark_dir, config_file: str | None, catego # One timestamp ties this run's per-testcase logs to its results dir. run_timestamp = new_run_timestamp() - ver_tasks_with_commands = generate_commands(ver_tasks, config_file, run_timestamp=run_timestamp, no_sa=no_sa, sa_retry_without=sa_retry_without, parallel_sa=parallel_sa, sa_rank_distance=sa_rank_distance, sa_no_callback_targets=sa_no_callback_targets, solver_rlimit=solver_rlimit) + ver_tasks_with_commands = generate_commands(ver_tasks, config_file, run_timestamp=run_timestamp, no_sa=no_sa, sa_retry_without=sa_retry_without, parallel_sa=parallel_sa, sa_rank_distance=sa_rank_distance, sa_no_callback_targets=sa_no_callback_targets, solver_rlimit=solver_rlimit, cpu_limit_s=testcase_timeout_s) click.echo(f"Generated {len(ver_tasks_with_commands)} commands") # Check port availability diff --git a/targets/sv-comp/scripts/lib/command_gen.py b/targets/sv-comp/scripts/lib/command_gen.py index 913d0fc1..312bea67 100644 --- a/targets/sv-comp/scripts/lib/command_gen.py +++ b/targets/sv-comp/scripts/lib/command_gen.py @@ -35,7 +35,7 @@ def is_port_available(port: int) -> bool: return False -def generate_command(ver_task: VerificationTask, logging_dir: Path, port: int=8087, config_file:str = 'swat.cfg', no_sa: bool = False, sa_retry_without: bool = False, parallel_sa: bool = False, sa_rank_distance: bool = False, sa_no_callback_targets: bool = False, solver_rlimit: int = 0) -> list[str]: +def generate_command(ver_task: VerificationTask, logging_dir: Path, port: int=8087, config_file:str = 'swat.cfg', no_sa: bool = False, sa_retry_without: bool = False, parallel_sa: bool = False, sa_rank_distance: bool = False, sa_no_callback_targets: bool = False, solver_rlimit: int = 0, cpu_limit_s: Optional[int] = None) -> list[str]: test_case_dir = ver_task['file_path'].parent @@ -58,7 +58,8 @@ def generate_command(ver_task: VerificationTask, logging_dir: Path, port: int=80 (["--parallel-sa"] if parallel_sa and not no_sa else []) + \ (["--sa-rank-distance"] if sa_rank_distance and not no_sa else []) + \ (["--sa-no-callback-targets"] if sa_no_callback_targets and not no_sa else []) + \ - (["--solver-rlimit", str(solver_rlimit)] if solver_rlimit else []) + (["--solver-rlimit", str(solver_rlimit)] if solver_rlimit else []) + \ + (["--cpu-limit", str(cpu_limit_s)] if cpu_limit_s else []) cp: list[str] = [] inputs: list[str] = [] @@ -101,7 +102,7 @@ def write_run_info(run_dir: Path, **settings) -> Path: return path -def generate_commands(ver_tasks: list[VerificationTask], config_file: str = 'swat.cfg', run_timestamp: Optional[str] = None, no_sa: bool = False, sa_retry_without: bool = False, parallel_sa: bool = False, sa_rank_distance: bool = False, sa_no_callback_targets: bool = False, solver_rlimit: int = 0) -> list[VerificationTask]: +def generate_commands(ver_tasks: list[VerificationTask], config_file: str = 'swat.cfg', run_timestamp: Optional[str] = None, no_sa: bool = False, sa_retry_without: bool = False, parallel_sa: bool = False, sa_rank_distance: bool = False, sa_no_callback_targets: bool = False, solver_rlimit: int = 0, cpu_limit_s: Optional[int] = None) -> list[VerificationTask]: port = 9000 skipped_ports = [] @@ -146,7 +147,7 @@ def generate_commands(ver_tasks: list[VerificationTask], config_file: str = 'swa 'target_dir': target_dir, 'target': target, 'log_dir': logging_dir, - 'command': generate_command(ver_task, logging_dir, port=port, config_file=config_file, no_sa=no_sa, sa_retry_without=sa_retry_without, parallel_sa=parallel_sa, sa_rank_distance=sa_rank_distance, sa_no_callback_targets=sa_no_callback_targets, solver_rlimit=solver_rlimit) + 'command': generate_command(ver_task, logging_dir, port=port, config_file=config_file, no_sa=no_sa, sa_retry_without=sa_retry_without, parallel_sa=parallel_sa, sa_rank_distance=sa_rank_distance, sa_no_callback_targets=sa_no_callback_targets, solver_rlimit=solver_rlimit, cpu_limit_s=cpu_limit_s) } ver_task['command'] = command port += 1