From 8d52be40b5af5c560284e92e043a28de314824fc Mon Sep 17 00:00:00 2001 From: Aleksei Menshutin Date: Thu, 15 Oct 2026 03:21:11 +0300 Subject: [PATCH] Terminate native class cell process groups on timeout and interruption --- usvm-ts-calls/scripts/run-class-experiment.py | 40 +++++++++-- .../scripts/test_run_class_experiment.py | 69 +++++++++++++++++++ 2 files changed, 105 insertions(+), 4 deletions(-) diff --git a/usvm-ts-calls/scripts/run-class-experiment.py b/usvm-ts-calls/scripts/run-class-experiment.py index 645f4b4ffe..05beab37ce 100644 --- a/usvm-ts-calls/scripts/run-class-experiment.py +++ b/usvm-ts-calls/scripts/run-class-experiment.py @@ -4,7 +4,10 @@ import csv import hashlib import json +import os +import signal import subprocess +import time from pathlib import Path PROFILES = ('EMPTY_STOP', 'EMPTY_FRESH', 'FROZEN_STOP', 'FROZEN_FRESH') @@ -41,13 +44,42 @@ def freeze(path, value): def invoke(calls, arguments, log, timeout): log.parent.mkdir(parents=True, exist_ok=True) + started = time.monotonic() + isolated_group = os.name == 'posix' with log.open('w', encoding='utf-8') as output: + process = subprocess.Popen([str(calls), *map(str, arguments)], stdout=output, + stderr=subprocess.STDOUT, start_new_session=isolated_group) + previous_termination_handler = signal.getsignal(signal.SIGTERM) + signal.signal(signal.SIGTERM, lambda signum, frame: sys_exit_for_signal(signum)) try: - result = subprocess.run([str(calls), *map(str, arguments)], stdout=output, - stderr=subprocess.STDOUT, timeout=timeout, check=False) - return {'exitCode': result.returncode, 'log': str(log)} + outcome = {'exitCode': process.wait(timeout=timeout), 'log': str(log)} except subprocess.TimeoutExpired: - return {'exitCode': None, 'timeoutSeconds': timeout, 'log': str(log)} + terminate_owned_process(process, isolated_group=isolated_group) + outcome = {'exitCode': None, 'timeoutSeconds': timeout, 'log': str(log), + 'processGroupKilled': isolated_group} + except BaseException: + terminate_owned_process(process, isolated_group=isolated_group) + raise + finally: + signal.signal(signal.SIGTERM, previous_termination_handler) + + return {**outcome, 'pid': process.pid, 'processGroupId': process.pid if isolated_group else None, + 'elapsedMillis': round((time.monotonic() - started) * 1000), 'logSha256': sha(log)} + + +def sys_exit_for_signal(signum): + raise SystemExit(128 + signum) + + +def terminate_owned_process(process, isolated_group): + try: + if isolated_group: + os.killpg(process.pid, signal.SIGKILL) + else: + process.kill() + except ProcessLookupError: + pass + process.wait() def select_classes(ledger, eligible): diff --git a/usvm-ts-calls/scripts/test_run_class_experiment.py b/usvm-ts-calls/scripts/test_run_class_experiment.py index 994bb4929d..9572ec2b75 100644 --- a/usvm-ts-calls/scripts/test_run_class_experiment.py +++ b/usvm-ts-calls/scripts/test_run_class_experiment.py @@ -1,6 +1,11 @@ import importlib.util import json +import os +import signal +import subprocess +import sys import tempfile +import time import unittest from pathlib import Path from unittest.mock import patch @@ -11,6 +16,70 @@ class CampaignIntegrityTest(unittest.TestCase): + def write_process_tree(self, directory): + heartbeat = directory / 'heartbeat' + child = directory / 'grandchild.py' + child.write_text('import sys,time\nfrom pathlib import Path\n' + 'path=Path(sys.argv[1])\nwhile True:\n' + ' with path.open("a") as stream: stream.write("alive\\n")\n' + ' time.sleep(0.02)\n') + parent = directory / 'parent.py' + parent.write_text('import subprocess,sys,time\n' + 'subprocess.Popen([sys.executable,sys.argv[1],sys.argv[2]])\n' + 'time.sleep(60)\n') + return parent, child, heartbeat + + def assert_heartbeat_stopped(self, heartbeat): + self.assertTrue(heartbeat.exists()) + after_cleanup = heartbeat.read_bytes() + + time.sleep(0.15) + + self.assertEqual(after_cleanup, heartbeat.read_bytes()) + + @unittest.skipUnless(os.name == 'posix', 'Owned process groups are a POSIX contract') + def test_cell_timeout_stops_grandchildren_and_records_log_identity(self): + with tempfile.TemporaryDirectory() as temporary: + directory = Path(temporary) + parent, child, heartbeat = self.write_process_tree(directory) + + outcome = campaign.invoke(Path(sys.executable), [parent, child, heartbeat], + directory / 'cell.log', timeout=0.5) + + self.assertIsNone(outcome['exitCode']) + self.assertTrue(outcome['processGroupKilled']) + self.assertEqual(outcome['pid'], outcome['processGroupId']) + self.assertEqual(campaign.sha(directory / 'cell.log'), outcome['logSha256']) + self.assert_heartbeat_stopped(heartbeat) + + @unittest.skipUnless(os.name == 'posix', 'Owned process groups are a POSIX contract') + def test_terminating_campaign_stops_active_cell_process_group(self): + with tempfile.TemporaryDirectory() as temporary: + directory = Path(temporary) + parent, child, heartbeat = self.write_process_tree(directory) + wrapper = directory / 'campaign.py' + wrapper.write_text('import importlib.util,sys\nfrom pathlib import Path\n' + 'spec=importlib.util.spec_from_file_location("campaign",sys.argv[1])\n' + 'module=importlib.util.module_from_spec(spec)\nspec.loader.exec_module(module)\n' + 'module.invoke(Path(sys.executable),sys.argv[2:5],Path(sys.argv[5]),timeout=60)\n') + process = subprocess.Popen([sys.executable, str(wrapper), str(Path(campaign.__file__).resolve()), + str(parent), str(child), str(heartbeat), str(directory / 'cell.log')]) + try: + deadline = time.monotonic() + 5 + while not heartbeat.exists() and time.monotonic() < deadline: + time.sleep(0.02) + self.assertTrue(heartbeat.exists()) + + process.send_signal(signal.SIGTERM) + code = process.wait(timeout=5) + + self.assertEqual(128 + signal.SIGTERM, code) + self.assert_heartbeat_stopped(heartbeat) + finally: + if process.poll() is None: + process.kill() + process.wait() + def make_phase_fixture(self, output): prepared = output / 'prepared' directory = prepared / 'class'