1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
--- a/tools/daemon_pool.py
+++ b/tools/daemon_pool.py
@@ -22,10 +22,23 @@
subagent timeout wrappers. Do NOT use it for work that must complete
before exit (durable writes) — those belong on foreground threads with
explicit bounded joins.
+
+CPython compatibility
+---------------------
+Python 3.14 rewrote ``concurrent.futures.thread``:
+
+ - ``ThreadPoolExecutor`` no longer stores ``_initializer`` / ``_initargs``
+ - worker threads receive a context from ``_create_worker_context()``
+ - ``_worker(executor_reference, ctx, work_queue)`` — 3-arg signature
+
+Pre-3.14 used ``_worker(ref, work_queue, initializer, initargs)``.
+This class detects the active stdlib shape and mirrors the matching
+``_adjust_thread_count`` (daemon=True, no ``_threads_queues`` registration).
"""
from __future__ import annotations
+import inspect
import threading
import weakref
from concurrent.futures import ThreadPoolExecutor
@@ -33,12 +46,16 @@
__all__ = ["DaemonThreadPoolExecutor"]
+# 3.14+ worker: (executor_reference, ctx, work_queue)
+# 3.8–3.13 worker: (executor_reference, work_queue, initializer, initargs)
+_WORKER_PARAM_COUNT = len(inspect.signature(_worker).parameters)
+
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 with two changes:
# daemon=True and no _threads_queues registration.
if self._idle_semaphore.acquire(timeout=0):
return
@@ -49,16 +66,30 @@
num_threads = len(self._threads)
if num_threads < self._max_workers:
thread_name = "%s_%d" % (self._thread_name_prefix or self, num_threads)
- t = threading.Thread(
- name=thread_name,
- target=_worker,
- args=(
- weakref.ref(self, weakref_cb),
- self._work_queue,
- self._initializer,
- self._initargs,
- ),
- daemon=True,
- )
+ if _WORKER_PARAM_COUNT >= 4:
+ # CPython 3.8–3.13
+ t = threading.Thread(
+ name=thread_name,
+ target=_worker,
+ args=(
+ weakref.ref(self, weakref_cb),
+ self._work_queue,
+ self._initializer,
+ self._initargs,
+ ),
+ daemon=True,
+ )
+ else:
+ # CPython 3.14+: context-object worker
+ t = threading.Thread(
+ name=thread_name,
+ target=_worker,
+ args=(
+ weakref.ref(self, weakref_cb),
+ self._create_worker_context(),
+ self._work_queue,
+ ),
+ daemon=True,
+ )
t.start()
self._threads.add(t)
|