[llvm] Revert "[lit] Migrate lit to ProcessPoolExecutor (#202681)" (PR #206138)
via llvm-commits
llvm-commits at lists.llvm.org
Fri Jun 26 11:09:24 PDT 2026
https://github.com/prasoon054 created https://github.com/llvm/llvm-project/pull/206138
This reverts the commit 1e2d1bbc12f6.
ProcessPoolExecutor.shutdown(wait=True) hangs on macOS 14 with Python 3.9:
join_executor_internals() calls call_queue.join_thread() before p.join(),
but macOS requires the inverse order. The feeder thread cannot drain
until worker processes are joined, so join_thread() blocks forever. This
is fixed upstream in CPython >= 3.12 but affects all earlier versions on
macOS. Reverting to unblock the aarch64-darwin buildbot while a proper fix
is worked out.
The original changes and context can be found in https://github.com/llvm/llvm-project/pull/202681
>From ba03ce9b2e6a142663fcf8ba8d8a3f6f75e5685a Mon Sep 17 00:00:00 2001
From: Prasoon Kumar <prasoonkumar054 at gmail.com>
Date: Fri, 26 Jun 2026 23:21:26 +0530
Subject: [PATCH] Revert "[lit] Migrate lit to ProcessPoolExecutor (#202681)"
This reverts commit 1e2d1bbc12f6a5f5931c77d39894ee1b8679f5f8.
---
llvm/utils/lit/lit/main.py | 2 -
llvm/utils/lit/lit/run.py | 125 +++++++++++++------------------------
2 files changed, 45 insertions(+), 82 deletions(-)
diff --git a/llvm/utils/lit/lit/main.py b/llvm/utils/lit/lit/main.py
index 54ae2fceb7775..a3bd153040a69 100755
--- a/llvm/utils/lit/lit/main.py
+++ b/llvm/utils/lit/lit/main.py
@@ -278,8 +278,6 @@ def run_tests(tests, lit_config, opts, discovered_tests):
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 15dd8d5f94a64..6c6d464a6881a 100644
--- a/llvm/utils/lit/lit/run.py
+++ b/llvm/utils/lit/lit/run.py
@@ -1,12 +1,7 @@
import multiprocessing
import os
import platform
-import sys
import time
-from concurrent.futures import ProcessPoolExecutor, as_completed
-from concurrent.futures import TimeoutError as FuturesTimeoutError
-from concurrent.futures.process import BrokenProcessPool
-import concurrent.futures.process
import lit.Test
import lit.util
@@ -29,11 +24,6 @@ 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."""
@@ -81,46 +71,6 @@ def execute(self):
if test.result is None:
test.setResult(skipped)
- def _abort_executors(self, executors, future_to_test):
- """SIGKILL all workers on abort (ctrl-C, --max-failures, --max-time,
- worker crash). Pre-3.14 ProcessPoolExecutor has no force-stop."""
- try:
- # We don't call ex.shutdown() here: it joins the management thread,
- # which is blocked reading the queue we just corrupted.
- # On 3.8 / 3.9, cancel() races with the call-queue feeder thread and can
- # deadlock or corrupt the queue (https://github.com/python/cpython/issues/94440).
- # Skipping it is safe because we SIGKILL workers below, so no pending future
- # will ever be dispatched. cancel() on 3.10+ is a clean hint.
- if sys.version_info >= (3, 10):
- for future in future_to_test:
- future.cancel()
- # Killing worker processes can corrupt the executor's queues, which makes it
- # unsafe for its atexit hooks to join their threads. Disable those hooks
- # before terminating workers (a second ctrl-C should not bypass this cleanup).
- # This applies to call-queue feeder threads and management threads.
- # Otherwise, a thread blocked on a partially written pipe may require multiple
- # ctrl-C to unblock.
- # See: https://github.com/python/cpython/issues/125886
- # These threads are daemonic on Python 3.8, so disabling them is harmless.
- for ex in executors:
- if hasattr(ex, "_call_queue") and ex._call_queue is not None:
- ex._call_queue.cancel_join_thread()
- if hasattr(concurrent.futures.process, "_threads_wakeups"):
- concurrent.futures.process._threads_wakeups.clear()
- tree_kill_ok, _ = lit.util.killProcessAndChildrenIsSupported()
- for ex in executors:
- for pid, proc in list((ex._processes or {}).items()):
- if tree_kill_ok:
- lit.util.killProcessAndChildren(pid)
- else:
- proc.kill()
- # TODO: Python>=3.14 adds ex.kill_workers(), which stops the workers cleanly
- # without corrupting the queues. However kill_workers() won't reap the
- # llc / FileCheck grandchildren the workers spawned.
- # https://github.com/python/cpython/issues/128041
- except Exception:
- pass
-
def _execute(self, deadline):
self._increase_process_limit()
@@ -153,44 +103,59 @@ def _execute(self, deadline):
% (num_pools, self.workers, workers_per_pool_list)
)
- executors = [
- ProcessPoolExecutor(
- max_workers=pool_size,
- initializer=lit.worker.initialize,
- initargs=(self.lit_config, semaphores),
+ # Create multiple pools
+ pools = []
+ for pool_size in workers_per_pool_list:
+ pool = multiprocessing.Pool(
+ pool_size, lit.worker.initialize, (self.lit_config, semaphores)
)
- for pool_size in workers_per_pool_list
- ]
+ 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)
- 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
+ # Close all pools
+ for pool in pools:
+ pool.close()
try:
- self._wait_for(future_to_test, deadline)
- except BaseException:
- self._abort_executors(executors, future_to_test)
+ self._wait_for(async_results, deadline)
+ except:
+ # Terminate all pools on exception
+ for pool in pools:
+ pool.terminate()
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():
+ 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:
+ 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():
self.failures += 1
if self.failures == self.max_failures:
raise MaxFailuresError()
- except FuturesTimeoutError:
- raise TimeoutError()
- except BrokenProcessPool as e:
- raise WorkerCrashError(str(e))
+ idx += 1
# 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