diff --git a/tests/tools/test_daemon_pool.py b/tests/tools/test_daemon_pool.py index 8112e78f2a884..428a813d4b469 100644 --- a/tests/tools/test_daemon_pool.py +++ b/tests/tools/test_daemon_pool.py @@ -45,6 +45,22 @@ def _init(tag): pool.shutdown(wait=True) +def test_worker_args_match_stdlib_worker_signature(): + """CPython changes _worker's signature between versions (3.14 moved + initializer/initargs into a worker context), so the mirrored spawn + path must adapt instead of passing a stale argument tuple.""" + import inspect + import weakref + from concurrent.futures.thread import _worker + + expected = len(inspect.signature(_worker).parameters) + pool = DaemonThreadPoolExecutor(max_workers=1) + try: + assert len(pool._worker_args(weakref.ref(pool))) == expected + finally: + pool.shutdown(wait=True) + + def test_idle_worker_reuse(): pool = DaemonThreadPoolExecutor(max_workers=4) try: diff --git a/tools/daemon_pool.py b/tools/daemon_pool.py index 2fb5a61d0a24d..50840aba1b0ab 100644 --- a/tools/daemon_pool.py +++ b/tools/daemon_pool.py @@ -38,7 +38,7 @@ class DaemonThreadPoolExecutor(ThreadPoolExecutor): """ThreadPoolExecutor variant whose workers do not block process exit.""" def _adjust_thread_count(self) -> None: - # Mirrors CPython's implementation (3.8–3.13) with two changes: + # Mirrors CPython's implementation (3.8–3.14) with two changes: # daemon=True and no _threads_queues registration. if self._idle_semaphore.acquire(timeout=0): return @@ -52,13 +52,22 @@ def weakref_cb(_, q=self._work_queue): t = threading.Thread( name=thread_name, target=_worker, - args=( - weakref.ref(self, weakref_cb), - self._work_queue, - self._initializer, - self._initargs, - ), + args=self._worker_args(weakref.ref(self, weakref_cb)), daemon=True, ) t.start() self._threads.add(t) + + def _worker_args(self, executor_ref: weakref.ref) -> tuple: + # CPython 3.14 replaced the (queue, initializer, initargs) worker + # arguments with a per-worker context object built by the executor, + # so reading self._initializer there raises AttributeError. + create_context = getattr(self, "_create_worker_context", None) + if create_context is not None: + return (executor_ref, create_context(), self._work_queue) + return ( + executor_ref, + self._work_queue, + self._initializer, + self._initargs, + )