Remove daemon threads in concurrent.futures

python · pitrou · Mar 27, 2020 · Mar 25, 2020 · Mar 25, 2020 · Mar 25, 2020
commit 3f03bd7dda85f6a67d0a45f21f143aab8029d42a
diff --git a/Lib/concurrent/futures/process.py b/Lib/concurrent/futures/process.py
@@ -59,20 +59,6 @@
 import sys
 import traceback
 
-# Workers are created as daemon threads and processes. This is done to allow the
-# interpreter to exit when there are still idle processes in a
-# ProcessPoolExecutor's process pool (i.e. shutdown() was not called). However,
-# allowing workers to die with the interpreter has two undesirable properties:
-#   - The workers would still be running during interpreter shutdown,
-#     meaning that they would fail in unpredictable ways.
-#   - The workers could be killed while evaluating a work item, which could
-#     be bad if the callable being evaluated has external side-effects e.g.
-#     writing to a file.
-#
-# To work around this problem, an exit handler is installed which tells the
-# workers to exit when their work queues are empty and then waits until the
-# threads/processes finish.
-
 _threads_wakeups = weakref.WeakKeyDictionary()
 _global_shutdown = False
 
@@ -107,6 +93,12 @@ def _python_exit():
     for t, _ in items:
         t.join()
 
+# Register for `_python_exit()` to be called just before joining all
+# non-daemon threads. This is used instead of `atexit.register()` for
+# compatibility with subinterpreters, which no longer support daemon threads.
+# See bpo-39812 for context.
+threading._register_atexit(_python_exit)
+
 # Controls how many more calls than processes will be queued in the call queue.
 # A smaller number will mean that processes spend more time idle waiting for
 # work while a larger number will make Future.cancel() succeed less frequently
@@ -306,9 +298,7 @@ def weakref_cb(_, thread_wakeup=self.thread_wakeup):
         #     {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
         self.pending_work_items = executor._pending_work_items
 
-        # Set this thread to be daemonized
         super().__init__()
-        self.daemon = True
 
     def run(self):
         # Main loop for the executor manager thread.
@@ -732,5 +722,3 @@ def shutdown(self, wait=True, *, cancel_futures=False):
             self._executor_manager_thread_wakeup = None
 
     shutdown.__doc__ = _base.Executor.shutdown.__doc__
-
-atexit.register(_python_exit)
diff --git a/Lib/concurrent/futures/thread.py b/Lib/concurrent/futures/thread.py
@@ -13,20 +13,6 @@
 import weakref
 import os
 
-# Workers are created as daemon threads. This is done to allow the interpreter
-# to exit when there are still idle threads in a ThreadPoolExecutor's thread
-# pool (i.e. shutdown() was not called). However, allowing workers to die with
-# the interpreter has two undesirable properties:
-#   - The workers would still be running during interpreter shutdown,
-#     meaning that they would fail in unpredictable ways.
-#   - The workers could be killed while evaluating a work item, which could
-#     be bad if the callable being evaluated has external side-effects e.g.
-#     writing to a file.
-#
-# To work around this problem, an exit handler is installed which tells the
-# workers to exit when their work queues are empty and then waits until the
-# threads finish.
-
 _threads_queues = weakref.WeakKeyDictionary()
 _shutdown = False
 # Lock that ensures that new workers are not created while the interpreter is
@@ -43,7 +29,11 @@ def _python_exit():
     for t, q in items:
         t.join()
 
-atexit.register(_python_exit)
+# Register for `_python_exit()` to be called just before joining all
+# non-daemon threads. This is used instead of `atexit.register()` for
+# compatibility with subinterpreters, which no longer support daemon threads.
+# See bpo-39812 for context.
+threading._register_atexit(_python_exit)
 
 
 class _WorkItem(object):
@@ -197,7 +187,6 @@ def weakref_cb(_, q=self._work_queue):
                                        self._work_queue,
                                        self._initializer,
                                        self._initargs))
-            t.daemon = True
             t.start()
             self._threads.add(t)
             _threads_queues[t] = self._work_queue

diff --git a/Lib/threading.py b/Lib/threading.py
@@ -3,6 +3,7 @@
 import os as _os
 import sys as _sys
 import _thread
+import functools
 
 from time import monotonic as _time
 from _weakrefset import WeakSet
@@ -1346,6 +1347,26 @@ def enumerate():
     with _active_limbo_lock:
         return list(_active.values()) + list(_limbo.values())
 
+
+_threading_atexits = None
+
+def _register_atexit(func, *arg, **kwargs):
+    """CPython internal: register *func* to be called before joining threads.
+
+    The registered *func* is called with its arguments just before all
+    non-daemon threads are joined in `_shutdown()`. It provides a similar
+    purpose to `atexit.register()`, but its functions are called prior to
+    threading shutdown instead of interpreter shutdown.
+
+    For similarity to atexit, the registed functions are called in reverse.
+    """
+    global _threading_atexits
+    if _threading_atexits is None:
+        _threading_atexits = []
+    call = functools.partial(func, *arg, **kwargs)
+    _threading_atexits.append(call)
+
+
 from _thread import stack_size
 
 # Create the main thread object,
@@ -1376,6 +1397,12 @@ def _shutdown():
     tlock.release()
     _main_thread._stop()
 
+    # Call registered threading atexit functions before threads are joined
+    if _threading_atexits is not None:
+        # Order is reversed, similar to atexit.
+        for atexit_call in reversed(_threading_atexits):
+            atexit_call()
+
     # Join all non-deamon threads
     while True:
         with _shutdown_locks_lock: