[llvm] [lit] Migrate lit to ProcessPoolExecutor (PR #202681)

via llvm-commits llvm-commits at lists.llvm.org
Tue Jun 9 07:59:59 PDT 2026


https://github.com/prasoon054 created https://github.com/llvm/llvm-project/pull/202681

This PR is a foundational refactor for the lit single-process re-architecture.

It migrates test execution from `multiprocessing.Pool` to `concurrent.futures.ProcessPoolExecutor`. While the process model remains unchanged (this is purely correctness and API modernization with no behavior change on a passing suite), this migration establishes the `concurrent.futures` API foundation required to introduce a `ThreadPoolExecutor` backend in future PRs.

By collecting results with `as_completed` via an explicit `{future: test}` map, this refactor also fixes two latent bugs:
1. **Stale timeout bug**: The per-iteration timeout budget was previously computed once and reused. It is now correctly anchored to an absolute deadline.
2. **Submission-order coupling**: Results are now safely routed by future identity rather than submission index.


>From 6cc5b168a9e5454021448242d9f36a15fd211d26 Mon Sep 17 00:00:00 2001
From: Prasoon Kumar <prasoonkumar054 at gmail.com>
Date: Tue, 9 Jun 2026 20:13:17 +0530
Subject: [PATCH] [lit] Migrate lit to ProcessPoolExecutor

This patch replaces multiprocessing.Pool with concurrent.futures.ProcessPoolExecutor in lit's core test execution to address bugs related to stale per-iteration timeouts and submission index result coupling.

Signed-off-by: Prasoon Kumar <prasoonkumar054 at gmail.com>
---
 llvm/utils/lit/lit/main.py |   9 ++++
 llvm/utils/lit/lit/run.py  | 102 +++++++++++++++++++++----------------
 2 files changed, 66 insertions(+), 45 deletions(-)

diff --git a/llvm/utils/lit/lit/main.py b/llvm/utils/lit/lit/main.py
index a3bd153040a69..d9be04ca79f9e 100755
--- a/llvm/utils/lit/lit/main.py
+++ b/llvm/utils/lit/lit/main.py
@@ -272,12 +272,21 @@ def run_tests(tests, lit_config, opts, discovered_tests):
     try:
         execute_in_tmp_dir(run, lit_config)
     except KeyboardInterrupt:
+        # Clear concurrent.futures thread registry to prevent atexit deadlocks.
+        try:
+            import concurrent.futures.process
+            if hasattr(concurrent.futures.process, '_threads_wakeups'):
+                concurrent.futures.process._threads_wakeups.clear()
+        except ImportError:
+            pass
         interrupted = True
         error = "  interrupted by user"
     except lit.run.MaxFailuresError:
         error = "warning: reached maximum number of test failures"
     except lit.run.TimeoutError:
         error = "warning: reached timeout"
+    except lit.run.WorkerCrashError as e:
+        lit_config.error(f"a worker process crashed: {e}")
 
     display.clear(interrupted)
     if error:
diff --git a/llvm/utils/lit/lit/run.py b/llvm/utils/lit/lit/run.py
index 6c6d464a6881a..3ac8d14bd31e9 100644
--- a/llvm/utils/lit/lit/run.py
+++ b/llvm/utils/lit/lit/run.py
@@ -2,6 +2,9 @@
 import os
 import platform
 import time
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from concurrent.futures import TimeoutError as FuturesTimeoutError
+from concurrent.futures.process import BrokenProcessPool
 
 import lit.Test
 import lit.util
@@ -24,6 +27,10 @@ class TimeoutError(Exception):
     pass
 
 
+class WorkerCrashError(Exception):
+    """A worker process died abrupty (segfault, OOM-kill, abort) instead of returning a result."""
+    pass
+
 class Run:
     """A concrete, configured testing run."""
 
@@ -103,59 +110,64 @@ def _execute(self, deadline):
                 % (num_pools, self.workers, workers_per_pool_list)
             )
 
-        # Create multiple pools
-        pools = []
-        for pool_size in workers_per_pool_list:
-            pool = multiprocessing.Pool(
-                pool_size, lit.worker.initialize, (self.lit_config, semaphores)
+        executors = [
+            ProcessPoolExecutor(
+                max_workers=pool_size,
+                initializer=lit.worker.initialize,
+                initargs=(self.lit_config, semaphores),
             )
-            pools.append(pool)
-
-        # Distribute tests across pools
-        tests_per_pool = _ceilDiv(len(self.tests), num_pools)
-        async_results = []
-
-        for pool_idx, pool in enumerate(pools):
-            start_idx = pool_idx * tests_per_pool
-            end_idx = min(start_idx + tests_per_pool, len(self.tests))
-            for test in self.tests[start_idx:end_idx]:
-                ar = pool.apply_async(
-                    lit.worker.execute, args=[test], callback=self.progress_callback
-                )
-                async_results.append(ar)
+            for pool_size in workers_per_pool_list
+        ]
 
-        # Close all pools
-        for pool in pools:
-            pool.close()
+        future_to_test = {}
+        for i, test in enumerate(self.tests):
+            ex = executors[i % len(executors)]
+            future_to_test[ex.submit(lit.worker.execute, test)] = test
 
         try:
-            self._wait_for(async_results, deadline)
-        except:
-            # Terminate all pools on exception
-            for pool in pools:
-                pool.terminate()
-            raise
-        finally:
-            # Join all pools
-            for pool in pools:
-                pool.join()
-
-    def _wait_for(self, async_results, deadline):
-        timeout = deadline - time.time()
-        idx = 0
-        while len(async_results) > 0:
+            self._wait_for(future_to_test, deadline)
+        except BaseException:
             try:
-                ar = async_results.pop(0)
-                test = ar.get(timeout)
-            except multiprocessing.TimeoutError:
-                raise TimeoutError()
-            else:
-                self._update_test(self.tests[idx], test)
-                if test.isFailure():
+                tree_kill_ok, _ = lit.util.killProcessAndChildrenIsSupported()
+
+                # Prevent multiprocessing from joining queue feeder threads at exit
+                for ex in executors:
+                    if hasattr(ex, '_call_queue') and ex._call_queue is not None:
+                        ex._call_queue.cancel_join_thread()
+
+                for ex in executors:
+                    for pid, proc in list((ex._processes or {}).items()):
+                        if tree_kill_ok:
+                            lit.util.killProcessAndChildren(pid)
+                        else:
+                            proc.kill()
+
+                # Skip future.cancel() and ex.shutdown() to prevent 
+                # ProcessPoolExecutor deadlocks and InvalidStateErrors on Python 3.8.
+            except Exception:
+                pass
+            raise
+        else:
+            for ex in executors:
+                ex.shutdown(wait=True)
+
+
+    def _wait_for(self, future_to_test, deadline):
+        try:
+            for future in as_completed(future_to_test, timeout=deadline - time.time()):
+                remote_test = future.result()
+                local_test = future_to_test[future]
+                self._update_test(local_test, remote_test)
+                self.progress_callback(remote_test)
+                if remote_test.isFailure():
                     self.failures += 1
                     if self.failures == self.max_failures:
                         raise MaxFailuresError()
-            idx += 1
+        except FuturesTimeoutError:
+            raise TimeoutError()
+        except BrokenProcessPool as e:
+            raise WorkerCrashError(str(e))
+
 
     # Update local test object "in place" from remote test object.  This
     # ensures that the original test object which is used for printing test



More information about the llvm-commits mailing list