Clean up native class cell process groups on timeout

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):

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'