Skip to content

Commit 6f08a59

Browse files
gpsheadmiss-islington
gpshead
authored and
miss-islington
committed
pythongh-90622: Do not spawn ProcessPool workers on demand via fork method. (pythonGH-91598)
Do not spawn ProcessPool workers on demand when they spawn via fork. This avoids potential deadlocks in the child processes due to forking from a multithreaded process. (cherry picked from commit ebb37fc) Co-authored-by: Gregory P. Smith <greg@krypto.org>
1 parent 8037fb7 commit 6f08a59

3 files changed

Lines changed: 49 additions & 11 deletions

File tree

‎Lib/concurrent/futures/process.py‎

Lines changed: 34 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -652,6 +652,10 @@ def __init__(self, max_workers=None, mp_context=None,
652652
mp_context = mp.get_context()
653653
self._mp_context = mp_context
654654

655+
# https://github.com/python/cpython/issues/90622
656+
self._safe_to_dynamically_spawn_children = (
657+
self._mp_context.get_start_method(allow_none=False) != "fork")
658+
655659
if initializer is not None and not callable(initializer):
656660
raise TypeError("initializer must be a callable")
657661
self._initializer = initializer
@@ -714,6 +718,8 @@ def __init__(self, max_workers=None, mp_context=None,
714718
def _start_executor_manager_thread(self):
715719
if self._executor_manager_thread is None:
716720
# Start the processes so that their sentinels are known.
721+
if not self._safe_to_dynamically_spawn_children: # ie, using fork.
722+
self._launch_processes()
717723
self._executor_manager_thread = _ExecutorManagerThread(self)
718724
self._executor_manager_thread.start()
719725
_threads_wakeups[self._executor_manager_thread] = \
@@ -726,15 +732,32 @@ def _adjust_process_count(self):
726732

727733
process_count = len(self._processes)
728734
if process_count < self._max_workers:
729-
p = self._mp_context.Process(
730-
target=_process_worker,
731-
args=(self._call_queue,
732-
self._result_queue,
733-
self._initializer,
734-
self._initargs,
735-
self._max_tasks_per_child))
736-
p.start()
737-
self._processes[p.pid] = p
735+
# Assertion disabled as this codepath is also used to replace a
736+
# worker that unexpectedly dies, even when using the 'fork' start
737+
# method. That means there is still a potential deadlock bug. If a
738+
# 'fork' mp_context worker dies, we'll be forking a new one when
739+
# we know a thread is running (self._executor_manager_thread).
740+
#assert self._safe_to_dynamically_spawn_children or not self._executor_manager_thread, 'https://github.com/python/cpython/issues/90622'
741+
self._spawn_process()
742+
743+
def _launch_processes(self):
744+
# https://github.com/python/cpython/issues/90622
745+
assert not self._executor_manager_thread, (
746+
'Processes cannot be fork()ed after the thread has started, '
747+
'deadlock in the child processes could result.')
748+
for _ in range(len(self._processes), self._max_workers):
749+
self._spawn_process()
750+
751+
def _spawn_process(self):
752+
p = self._mp_context.Process(
753+
target=_process_worker,
754+
args=(self._call_queue,
755+
self._result_queue,
756+
self._initializer,
757+
self._initargs,
758+
self._max_tasks_per_child))
759+
p.start()
760+
self._processes[p.pid] = p
738761

739762
def submit(self, fn, /, *args, **kwargs):
740763
with self._shutdown_lock:
@@ -755,7 +778,8 @@ def submit(self, fn, /, *args, **kwargs):
755778
# Wake up queue management thread
756779
self._executor_manager_thread_wakeup.wakeup()
757780

758-
self._adjust_process_count()
781+
if self._safe_to_dynamically_spawn_children:
782+
self._adjust_process_count()
759783
self._start_executor_manager_thread()
760784
return f
761785
submit.__doc__ = _base.Executor.submit.__doc__

‎Lib/test/test_concurrent_futures.py‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -497,10 +497,16 @@ def acquire_lock(lock):
497497
lock.acquire()
498498

499499
mp_context = self.get_context()
500+
if mp_context.get_start_method(allow_none=False) == "fork":
501+
# fork pre-spawns, not on demand.
502+
expected_num_processes = self.worker_count
503+
else:
504+
expected_num_processes = 3
505+
500506
sem = mp_context.Semaphore(0)
501507
for _ in range(3):
502508
self.executor.submit(acquire_lock, sem)
503-
self.assertEqual(len(self.executor._processes), 3)
509+
self.assertEqual(len(self.executor._processes), expected_num_processes)
504510
for _ in range(3):
505511
sem.release()
506512
processes = self.executor._processes
@@ -1021,6 +1027,8 @@ def test_saturation(self):
10211027
def test_idle_process_reuse_one(self):
10221028
executor = self.executor
10231029
assert executor._max_workers >= 4
1030+
if self.get_context().get_start_method(allow_none=False) == "fork":
1031+
raise unittest.SkipTest("Incompatible with the fork start method.")
10241032
executor.submit(mul, 21, 2).result()
10251033
executor.submit(mul, 6, 7).result()
10261034
executor.submit(mul, 3, 14).result()
@@ -1029,6 +1037,8 @@ def test_idle_process_reuse_one(self):
10291037
def test_idle_process_reuse_multiple(self):
10301038
executor = self.executor
10311039
assert executor._max_workers <= 5
1040+
if self.get_context().get_start_method(allow_none=False) == "fork":
1041+
raise unittest.SkipTest("Incompatible with the fork start method.")
10321042
executor.submit(mul, 12, 7).result()
10331043
executor.submit(mul, 33, 25)
10341044
executor.submit(mul, 25, 26).result()

‎Misc/NEWS.d/next/Library/2022-04-15-22-07-36.gh-issue-90622.0C6l8h.rst‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
Worker processes for :class:`concurrent.futures.ProcessPoolExecutor` are no
2+
longer spawned on demand (a feature added in 3.9) when the multiprocessing
3+
context start method is ``"fork"`` as that can lead to deadlocks in the
4+
child processes due to a fork happening while threads are running.

0 commit comments

Comments
 (0)