8000 bpo-39812: Remove daemon threads in concurrent.futures (GH-19149) · python/cpython@b61b818 · GitHub
[go: up one dir, main page]
Skip to content

Commit b61b818

Browse files
aerosaeros
authored
bpo-39812: Remove daemon threads in concurrent.futures (GH-19149)Remove daemon threads from :mod:`concurrent.futures` by adding an internal `threading._register_atexit()`, which calls registered functions prior to joining all non-daemon threads. This allows for compatibility with subinterpreters, which don't support daemon threads.
1 parent 5f9c131 commit b61b818

File tree

6 files changed

+99
-32
lines changed

6 files changed

+99
-32
lines changed

‎Doc/whatsnew/3.9.rst

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,6 +195,11 @@ which have not started running, instead of waiting for them to complete before
195195
shutting down the executor.
196196
(Contributed by Kyle Stanley in :issue:`39349`.)
197197

198+
Removed daemon threads from :class:`~concurrent.futures.ThreadPoolExecutor`
199+
and :class:`~concurrent.futures.ProcessPoolExecutor`. This improves
200+
compatibility with subinterpreters and predictability in their shutdown
201+
processes. (Contributed by Kyle Stanley in :issue:`39812`.)
202+
198203
curses
199204
------
200205

‎Lib/concurrent/futures/process.py

Lines changed: 6 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -59,19 +59,6 @@
5959
import sys
6060
import traceback
6161

62-
# Workers are created as daemon threads and processes. This is done to allow the
63-
# interpreter to exit when there are still idle processes in a
64-
# ProcessPoolExecutor's process pool (i.e. shutdown() was not called). However,
65-
# allowing workers to die with the interpreter has two undesirable properties:
66-
# - The workers would still be running during interpreter shutdown,
67-
# meaning that they would fail in unpredictable ways.
68-
# - The workers could be killed while evaluating a work item, which could
69-
# be bad if the callable being evaluated has external side-effects e.g.
70-
# writing to a file.
71-
#
72-
# To work around this problem, an exit handler is installed which tells the
73-
# workers to exit when their work queues are empty and then waits until the
74-
# threads/processes finish.
7562

7663
_threads_wakeups = weakref.WeakKeyDictionary()
7764
_global_shutdown = False
@@ -107,6 +94,12 @@ def _python_exit():
10794
for t, _ in items:
10895
t.join()
10996

97+
# Register for `_python_exit()` to be called just before joining all
98+
# non-daemon threads. This is used instead of `atexit.register()` for
99+
# compatibility with subinterpreters, which no longer support daemon threads.
100+
# See bpo-39812 for context.
101+
threading._register_atexit(_python_exit)
102+
110103
# Controls how many more calls than processes will be queued in the call queue.
111104
# A smaller number will mean that processes spend more time idle waiting for
112105
# work while a larger number will make Future.cancel() succeed less frequently
@@ -306,9 +299,7 @@ def weakref_cb(_, thread_wakeup=self.thread_wakeup):
306299
# {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
307300
self.pending_work_items = executor._pending_work_items
308301

309-
# Set this thread to be daemonized
310302
super().__init__()
311-
self.daemon = True
312303

313304
def run(self):
314305
# Main loop for the executor manager thread.
@@ -732,5 +723,3 @@ def shutdown(self, wait=True, *, cancel_futures=False):
732723
self._executor_manager_thread_wakeup = None
733724

734725
shutdown.__doc__ = _base.Executor.shutdown.__doc__
735-
736-
atexit.register(_python_exit)

‎Lib/concurrent/futures/thread.py

Lines changed: 5 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -13,19 +13,6 @@
1313
import weakref
1414
import os
1515

16-
# Workers are created as daemon threads. This is done to allow the interpreter
17-
# to exit when there are still idle threads in a ThreadPoolExecutor's thread
18-
# pool (i.e. shutdown() was not called). However, allowing workers to die with
19-
# the interpreter has two undesirable properties:
20-
# - The workers would still be running during interpreter shutdown,
21-
# meaning that they would fail in unpredictable ways.
22-
# - The workers could be killed while evaluating a work item, which could
23-
# be bad if the callable being evaluated has external side-effects e.g.
24-
# writing to a file.
25-
#
26-
# To work around this problem, an exit handler is installed which tells the
27-
# workers to exit when their work queues are empty and then waits until the
28-
# threads finish.
2916

3017
_threads_queues = weakref.WeakKeyDictionary()
3118
_shutdown = False
@@ -43,7 +30,11 @@ def _python_exit():
4330
for t, q in items:
4431
t.join()
4532

46-
atexit.register(_python_exit)
33+
# Register for `_python_exit()` to be called just before joining all
34+
# non-daemon threads. This is used instead of `atexit.register()` for
35+
# compatibility with subinterpreters, which no longer support daemon threads.
36+
# See bpo-39812 for context.
37+
threading._register_atexit(_python_exit)
4738

4839

4940
class _WorkItem(object):
@@ -197,7 +188,6 @@ def weakref_cb(_, q=self._work_queue):
197188
self._work_queue,
198189
self._initializer,
199190
self._initargs))
200-
t.daemon = True
201191
t.start()
202192
self._threads.add(t)
203193
_threads_queues[t] = self._work_queue

‎Lib/test/test_threading.py

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1397,5 +1397,55 @@ def test_interrupt_main_noerror(self):
13971397
signal.signal(signal.SIGINT, handler)
13981398

13991399

1400+
class AtexitTests(unittest.TestCase):
1401+
1402+
def test_atexit_output(self):
1403+
rc, out, err = assert_python_ok("-c", """if True:
1404+
import threading
1405+
1406+
def run_last():
1407+
print('parrot')
1408+
1409+
threading._register_atexit(run_last)
1410+
""")
1411+
1412+
self.assertFalse(err)
1413+
self.assertEqual(out.strip(), b'parrot')
1414+
1415+
def test_atexit_called_once(self):
1416+
rc, out, err = assert_python_ok("-c", """if True:
1417+
import threading
1418+
from unittest.mock import Mock
1419+
1420+
mock = Mock()
1421+
threading._register_atexit(mock)
1422+
mock.assert_not_called()
1423+
# force early shutdown to ensure it was called once
1424+
threading._shutdown()
1425+
mock.assert_called_once()
1426+
""")
1427+
1428+
self.assertFalse(err)
1429+
1430+
def test_atexit_after_shutdown(self):
1431+
# The only way to do this is by registering an atexit within
1432+
# an atexit, which is intended to raise an exception.
1433+
rc, out, err = assert_python_ok("-c", """if True:
1434+
import threading
1435+
1436+
def func():
1437+
pass
1438+
1439+
def run_last():
1440+
threading._register_atexit(func)
1441+
1442+
threading._register_atexit(run_last)
1443+
""")
1444+
1445+
self.assertTrue(err)
1446+
self.assertIn("RuntimeError: can't register atexit after shutdown",
1447+
err.decode())
1448+
1449+
14001450
if __name__ == "__main__":
14011451
unittest.main()

‎Lib/threading.py

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import os as _os
44
import sys as _sys
55
import _thread
6+
import functools
67

78
from time import monotonic as _time
89
from _weakrefset import WeakSet
@@ -1346,6 +1347,27 @@ def enumerate():
13461347
with _active_limbo_lock:
13471348
return list(_active.values()) + list(_limbo.values())
13481349

1350+
1351+
_threading_atexits = []
1352+
_SHUTTING_DOWN = False
1353+
1354+
def _register_atexit(func, *arg, **kwargs):
1355+
"""CPython internal: register *func* to be called before joining threads.
1356+
1357+
The registered *func* is called with its arguments just before all
1358+
non-daemon threads are joined in `_shutdown()`. It provides a similar
1359+
purpose to `atexit.register()`, but its functions are called prior to
1360+
threading shutdown instead of interpreter shutdown.
1361+
1362+
For similarity to atexit, the registered functions are called in reverse.
1363+
"""
1364+
if _SHUTTING_DOWN:
1365+
raise RuntimeError("can't register atexit after shutdown")
1366+
1367+
call = functools.partial(func, *arg, **kwargs)
1368+
_threading_atexits.append(call)
1369+
1370+
13491371
from _thread import stack_size
13501372

13511373
# Create the main thread object,
@@ -1367,6 +1389,8 @@ def _shutdown():
13671389
# _shutdown() was already called
13681390
return
13691391

1392+
global _SHUTTING_DOWN
1393+
_SHUTTING_DOWN = True
13701394
# Main thread
13711395
tlock = _main_thread._tstate_lock
13721396
# The main thread isn't finished yet, so its thread state lock can't have
@@ -1376,6 +1400,11 @@ def _shutdown():
13761400
tlock.release()
13771401
_main_thread._stop()
13781402

1403+
# Call registered threading atexit functions before threads are joined.
1404+
# Order is reversed, similar to atexit.
1405+
for atexit_call in reversed(_threading_atexits):
1406+
atexit_call()
1407+
13791408
# Join all non-deamon threads
13801409
while True:
13811410
with _shutdown_locks_lock:

‎Misc/NEWS.d/next/Library/2020-03-25-00-35-48.bpo-39812.rIKnms.rst

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
Removed daemon threads from :mod:`concurrent.futures` by adding
2+
an internal `threading._register_atexit()`, which calls registered functions
3+
prior to joining all non-daemon threads. This allows for compatibility
4+
with subinterpreters, which don't support daemon threads.

0 commit comments

Comments
 (0)
0