[llvm] [lit] Migrate lit to ProcessPoolExecutor (PR #202681)
via llvm-commits
llvm-commits at lists.llvm.org
Tue Jun 9 08:01:10 PDT 2026
https://github.com/prasoon054 updated https://github.com/llvm/llvm-project/pull/202681
>From b5c7be68e87ab0949a3f3dc67201652428c7b3b4 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 | 17 ++++--
llvm/utils/lit/lit/run.py | 103 +++++++++++++++++++++----------------
2 files changed, 70 insertions(+), 50 deletions(-)
diff --git a/llvm/utils/lit/lit/main.py b/llvm/utils/lit/lit/main.py
index a3bd153040a69..81b4e19e0dfc1 100755
--- a/llvm/utils/lit/lit/main.py
+++ b/llvm/utils/lit/lit/main.py
@@ -47,9 +47,7 @@ def main(builtin_params={}):
maxIndividualTestTime=opts.maxIndividualTestTime,
)
- discovered_tests = lit.discovery.find_tests_for_inputs(
- lit_config, opts.test_paths
- )
+ discovered_tests = lit.discovery.find_tests_for_inputs(lit_config, opts.test_paths)
if not discovered_tests:
sys.stderr.write("error: did not discover any tests for provided path(s)\n")
sys.exit(2)
@@ -69,7 +67,6 @@ def main(builtin_params={}):
print(" ".join(sorted(features)))
sys.exit(0)
-
determine_order(discovered_tests, opts.order)
selected_tests = [
@@ -272,12 +269,22 @@ 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:
@@ -366,7 +373,7 @@ def print_summary(total_tests, tests_by_code, quiet, elapsed):
max_label_len = max(len(label) for label, _ in groups)
max_count_len = max(len(str(count)) for _, count in groups)
- for (label, count) in groups:
+ for label, count in groups:
label = label.ljust(max_label_len)
count = str(count).rjust(max_count_len)
print(" %s: %s (%.2f%%)" % (label, count, float(count) / total_tests * 100))
diff --git a/llvm/utils/lit/lit/run.py b/llvm/utils/lit/lit/run.py
index 6c6d464a6881a..4ebdb0e5aad95 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
@@ -16,6 +19,7 @@
def _ceilDiv(a, b):
return (a + b - 1) // b
+
class MaxFailuresError(Exception):
pass
@@ -24,6 +28,12 @@ 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 +113,62 @@ 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