gh-130895: fix multiprocessing.Process join/wait/poll races

Lib/multiprocessing/popen_fork.py

@@ -1,6 +1,7 @@
 import atexit
 import os
 import signal
+import threading
 
 from . import util
 
@@ -17,23 +18,74 @@ def __init__(self, process_obj):
  util._flush_std_streams()
  self.returncode = None
  self.finalizer = None
+ self._init_locking()
  self._launch(process_obj)
 
+ def _init_locking(self):
+ self._exit_condition = threading.Condition()
+ self._exit_blockers = 0
+
  def duplicate_for_child(self, fd):
  return fd
 
  def poll(self, flag=os.WNOHANG):
- if self.returncode is None:
- try:
- pid, sts = os.waitpid(self.pid, flag)
- except OSError:
- # Child process not yet created. See #1731717
- # e.errno == errno.ECHILD == 10
- return None
+ with self._exit_condition:
+ if self.returncode is not None:
+ return self.returncode
+ elif flag & os.WNOHANG == os.WNOHANG:
+ return self._nonblocking_poll(flag)
+ else:
+ self._exit_blockers += 1
+
+ # We have released the lock, so may be racing with blocking &
+ # non-blocking calls at this point...
+ pid = None
+ try:
+ pid, sts = os.waitpid(self.pid, flag)
+ except OSError:
+ # Child process doesn't exist because it hasn't started yet (see
+ # bpo-1731717) or has already been awaited on a racing thread (see
+ # gh-130895)
+ pass
+
+ with self._exit_condition:
+ self._exit_blockers -= 1
  if pid == self.pid:
- self.returncode = os.waitstatus_to_exitcode(sts)
+ self._set_returncode(sts)
+ elif self._exit_blockers == 0:
+ self._exit_condition.notify_all()
+
+ # Wait until we get a definitive result, or we know there are no
+ # racing calls that might be about to set it
+ while self.returncode is None and self._exit_blockers > 0:
+ self._exit_condition.wait()
+
+ return self.returncode
+
+ def _nonblocking_poll(self, flag):
+ assert self._exit_condition._is_owned()
+ assert self.returncode is None
+ assert flag & os.WNOHANG == os.WNOHANG
+ try:
+ pid, sts = os.waitpid(self.pid, flag)
+ if pid == self.pid:
+ self._set_returncode(sts)
+ except OSError:
+ # See comments in the poll(...) except clause above
+ pass
+
+ # We may be racing with a blocking wait call, in which case (if we lose
+ # the race) it is arbitrary whether this returns None or the exit code
+ # (if there is one): calling code must always be prepared to handle a
+ # situation where this method returns None but the process has ended.
  return self.returncode
 
+ def _set_returncode(self, sts):
+ assert self._exit_condition._is_owned()
+ assert self.returncode is None
+ self.returncode = os.waitstatus_to_exitcode(sts)
+ self._exit_condition.notify_all()
+
  def wait(self, timeout=None):
  if self.returncode is None:
  if timeout is not None:

Lib/multiprocessing/popen_forkserver.py

@@ -1,5 +1,6 @@
 import io
 import os
+import threading
 
 from .context import reduction, set_spawning_popen
 if not reduction.HAVE_SEND_HANDLE:
@@ -34,6 +35,9 @@ def __init__(self, process_obj):
  self._fds = []
  super().__init__(process_obj)
 
+ def _init_locking(self):
+ self._lock = threading.Lock()
+
  def duplicate_for_child(self, fd):
  self._fds.append(fd)
  return len(self._fds) - 1
@@ -63,12 +67,15 @@ def poll(self, flag=os.WNOHANG):
  from multiprocessing.connection import wait
  timeout = 0 if flag == os.WNOHANG else None
  if not wait([self.sentinel], timeout):
- return None
- try:
- self.returncode = forkserver.read_signed(self.sentinel)
- except (OSError, EOFError):
- # This should not happen usually, but perhaps the forkserver
- # process itself got killed
- self.returncode = 255
+ return self.returncode
+
+ with self._lock:
+ if self.returncode is None:
+ try:
+ self.returncode = forkserver.read_signed(self.sentinel)
+ except (OSError, EOFError):
+ # This should not happen usually, but perhaps the
+ # forkserver process itself got killed
+ self.returncode = 255
 
  return self.returncode

Lib/test/_test_multiprocessing.py

@@ -1055,6 +1055,45 @@ def test_forkserver_without_auth_fails(self):
  proc.start()
  proc.join()
 
+ @staticmethod
+ def _wait_for_barrier(barrier):
+ barrier.wait()
+
+ def _wait_on_proc(self, barrier, proc, errs):
+ barrier.wait()
+ proc.join()
+ if proc.is_alive():
+ errs.append("process alive after join")
+ if proc.exitcode != 0:
+ errs.append("process reported non-zero exit code")
+
+ @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
+ def test_racing_joins(self):
+ if self.TYPE == "threads":
+ self.skipTest(f"test not appropriate for {self.TYPE}")
+
+ N = 5
+ ITERATIONS = 10
+ for _ in range(ITERATIONS):
+ barrier = self.Barrier(N+1)
+ proc = self.Process(target=self._wait_for_barrier, args=(barrier,))
+
+ errs = []
+ threads = [threading.Thread(target=self._wait_on_proc,
+ args=(barrier, proc, errs))
+ for _ in range(N)]
+ for t in threads:
+ t.start()
+
+ proc.start()
+ for t in threads:
+ t.join()
+
+ # On failure(s), report the first since they are likely the same
+ # error reported from multiple threads
+ if errs:
+ raise AssertionError(errs[0])
+
 #
 #
 #

Lib/test/test_concurrent_futures/test_process_pool.py

@@ -349,15 +349,7 @@ def test_force_shutdown_workers_stops_pool(self, function_name):
  # error since the process would be alive immediately after the
  # test run.. and die a moment later.
  worker_process.join(support.SHORT_TIMEOUT)
-
- # Oddly enough, even though join completes, sometimes it takes a
- # moment for the process to actually be marked as dead.
- # ... that seems a bit buggy.
- # We need it dead before ending the test to ensure it doesn't
- # get marked as an ENV CHANGE due to living child process.
- for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
- if not worker_process.is_alive():
- break
+ self.assertFalse(worker_process.is_alive())
 
 
 create_executor_tests(globals(), ProcessPoolExecutorTest,

Misc/NEWS.d/next/Library/2025-03-19-16-09-19.gh-issue-130895.6ILa73.rst

@@ -0,0 +1 @@
+Fix race with ``poll``/``wait``/``join`` in :mod:`multiprocessing`.``Process``.