Detect crashed test runners and report and exit

Probably a better job could be done than this, but this should get us
a clean failure instead of a long hang.

Author: msullivan

edb/tools/test/mproc_fixes.py

@@ -28,6 +28,7 @@
 
 
 _orig_pool_worker_handler = None
+_orig_pool_join_exited_workers = None
 
 logger = logging.getLogger(__name__)
 
@@ -92,9 +93,17 @@ def multiprocessing_worker_handler(*args):
         worker_process.join(timeout=10)
 
 
+def join_exited_workers(pool):
+    # Our use case shouldn't have workers exiting really, so we skip
+    # doing the joins so that we can detect crashes ourselves in the
+    # test runner.x
+    pass
+
+
 def patch_multiprocessing(debug: bool):
     global _orig_pool_worker
     global _orig_pool_worker_handler
+    global _orig_pool_join_exited_workers
 
     if debug:
         multiprocessing.util.log_to_stderr(logging.DEBUG)
@@ -111,3 +120,9 @@ def patch_multiprocessing(debug: bool):
     # Allow workers some time to shut down gracefully.
     _orig_pool_worker_handler = multiprocessing.pool.Pool._handle_workers
     multiprocessing.pool.Pool._handle_workers = multiprocessing_worker_handler
+
+    _orig_pool_join_exited_workers = (
+        multiprocessing.pool.Pool._join_exited_workers)
+    multiprocessing.pool.Pool._join_exited_workers = join_exited_workers
+
+    multiprocessing.pool.Pool._crashed_workers = None

edb/tools/test/runner.py

@@ -166,15 +166,17 @@ def _run(self, test, result):
                 getattr(result, '_moduleSetUpFailed', False)):
             return
 
+        result.annotate_test(test, {
+            'py-hash-secret': py_hash_secret,
+            'py-random-seed': py_random_seed,
+            'runner-pid': os.getpid(),
+        })
+
         start = time.monotonic()
         test.run(result)
         elapsed = time.monotonic() - start
 
         result.record_test_stats(test, {'running-time': elapsed})
-        result.annotate_test(test, {
-            'py-hash-secret': py_hash_secret,
-            'py-random-seed': py_random_seed,
-        })
 
         result._testRunEntered = False
         return result
@@ -374,6 +376,28 @@ def run(self, result):
                     try:
                         ar.get(timeout=0.1)
                     except multiprocessing.TimeoutError:
+                        # multiprocessing doesn't handle processes
+                        # crashing very well, so we check ourselves
+                        # (having disabled its own child pruning in
+                        # mproc_fixes)
+                        #
+                        # TODO: Should we look into using
+                        # concurrent.futures.ProcessPoolExecutor
+                        # instead?
+                        for p in pool._pool:
+                            if p.exitcode:
+                                tmsg = ''
+                                if isinstance(result, ParallelTextTestResult):
+                                    test = result.current_pids.get(p.pid)
+                                    tmsg = f' while running {test}'
+                                print(
+                                    f"ERROR: Test worker {p.pid} crashed with "
+                                    f"exit code {p.exitcode}{tmsg}",
+                                    file=sys.stderr,
+                                )
+                                sys.stderr.flush()
+                                os._exit(1)
+
                         if self.stop_requested:
                             break
                         else:
@@ -744,6 +768,7 @@ def __init__(self, *, stream, verbosity, warnings, tests,
         self.warnings = []
         self.notImplemented = []
         self.currently_running = {}
+        self.current_pids = {}
         # An index of all seen warnings to keep track
         # of repeated warnings.
         self._warnings = {}
@@ -775,7 +800,14 @@ def report_still_running(self):
         for test, start in self.currently_running.items():
             running_for = now - start
             if running_for > 5.0:
-                still_running[test] = running_for
+                key = str(test)
+                if (
+                    test in self.test_annotations
+                    and (pid := self.test_annotations[test].get('runner-pid'))
+                ):
+                    key = f'{key} (pid={pid})'
+
+                still_running[key] = running_for
         if still_running:
             self.ren.report_still_running(still_running)
 
@@ -800,6 +832,11 @@ def startTest(self, test):
         self.currently_running[test] = time.monotonic()
         self.ren.report_start(
             test, currently_running=list(self.currently_running))
+        if (
+            test in self.test_annotations
+            and (pid := self.test_annotations[test].get('runner-pid'))
+        ):
+            self.current_pids[pid] = test
 
     def addSuccess(self, test):
         super().addSuccess(test)