bpo-38323: Change MultiLoopChildWatcher to install handlers for all the event loops

Lib/asyncio/unix_events.py

@@ -1201,9 +1201,10 @@ def _do_waitpid_all(self):
 class MultiLoopChildWatcher(AbstractChildWatcher):
  """A watcher that doesn't require running loop in the main thread.
 
- This implementation registers a SIGCHLD signal handler on
- instantiation (which may conflict with other code that
- install own handler for this signal).
+ This implementation registers a SIGCHLD signal handler on every event loop
+ subprocesses are created from (which may conflict with other code that
+ install own handler for this signal). This implementation can only be used
+ in the main thread.
 
  The solution is safe but it has a significant overhead when
  handling a big number of processes (*O(n)* each time a
@@ -1212,29 +1213,28 @@ class MultiLoopChildWatcher(AbstractChildWatcher):
 
  # Implementation note:
  # The class keeps compatibility with AbstractChildWatcher ABC
- # To achieve this it has empty attach_loop() method
- # and doesn't accept explicit loop argument
- # for add_child_handler()/remove_child_handler()
- # but retrieves the current loop by get_running_loop()
+ # It retrieves the current loop by get_running_loop() and installs
+ # handler on it using add_child_handler()/remove_child_handler()
+
 
  def __init__(self):
  self._callbacks = {}
- self._saved_sighandler = None
+ self._handler_added_loops = []
 
  def is_active(self):
- return self._saved_sighandler is not None
+ return self._handler_added_loops is not []
 
  def close(self):
  self._callbacks.clear()
- if self._saved_sighandler is None:
+ if self._handler_added_loops is []:
  return
 
- handler = signal.getsignal(signal.SIGCHLD)
- if handler != self._sig_chld:
-  logger.warning("SIGCHLD handler was changed by outside code")
- else:
- signal.signal(signal.SIGCHLD, self._saved_sighandler)
- self._saved_sighandler = None
+ # Remove handler from every event loop we registered the handler
+ # on previously
+ if self._handler_added_loops is not []:
+  for loop in self._handler_added_loops:
+  loop.remove_signal_handler(signal.SIGCHLD)
+  self._handler_added_loops.remove(loop)
 
  def __enter__(self):
  return self
@@ -1243,35 +1243,47 @@ def __exit__(self, exc_type, exc_val, exc_tb):
  pass
 
  def add_child_handler(self, pid, callback, *args):
+ # Retreives the current running loop and installs the handler
+ # on it. Can only be called from the main thread.
  loop = events.get_running_loop()
+ for handler_added_loop in self._handler_added_loops:
+ if loop is handler_added_loop:
+ self.attach_loop(loop)
+ break
+
  self._callbacks[pid] = (loop, callback, args)
 
  # Prevent a race condition in case the child is already terminated.
  self._do_waitpid(pid)
 
  def remove_child_handler(self, pid):
  try:
- del self._callbacks[pid]
+ loop, callback, args = self._callbacks.pop(pid)
+ loop.remove_signal_handler(signal.SIGCHLD)
  return True
  except KeyError:
  return False
 
  def attach_loop(self, loop):
- # Don't save the loop but initialize itself if called first time
- # The reason to do it here is that attach_loop() is called from
- # unix policy only for the main thread.
- # Main thread is required for subscription on SIGCHLD signal
- if self._saved_sighandler is not None:
- return
+ # Install the handler on the loop and save it in a list.
+ # The reason to do it here is to avoid a race condition by
+ # indirectly calling set_wakeup_fd. Previously attach_loop
+ # installed the handler globally. See bpo-38323 for more
+ # details.
+ # Main thread is required for installing SIGCHLD handler
+ # on the event loop therefore attach_loop is only callable
+ # from the main thread.
 
- self._saved_sighandler = signal.signal(signal.SIGCHLD, self._sig_chld)
- if self._saved_sighandler is None:
- logger.warning("Previous SIGCHLD handler was set by non-Python code, "
- "restore to default handler on watcher close.")
- self._saved_sighandler = signal.SIG_DFL
 
- # Set SA_RESTART to limit EINTR occurrences.
- signal.siginterrupt(signal.SIGCHLD, False)
+ assert loop is None or isinstance(loop, events.AbstractEventLoop)
+
+ if loop is not None:
+ loop.add_signal_handler(signal.SIGCHLD, self._sig_chld)
+
+ # Prevent a race condition in case a child terminated
+ # during the switch.
+ self._do_waitpid_all()
+ self._handler_added_loops.append(loop)
 
  def _do_waitpid_all(self):
  for pid in list(self._callbacks):
@@ -1314,7 +1326,7 @@ def _do_waitpid(self, expected_pid):
  expected_pid, returncode)
  loop.call_soon_threadsafe(callback, pid, returncode, *args)
 
- def _sig_chld(self, signum, frame):
+ def _sig_chld(self):
  try:
  self._do_waitpid_all()
  except (SystemExit, KeyboardInterrupt):

Misc/NEWS.d/next/Library/2021-06-07-13-39-16.bpo-38323.FGViHn.rst

@@ -0,0 +1,20 @@
+Fix a race condition in ``MultiLoopChildWatcher``. The patch does the
+following -
+
+1) Creates a list that will keep track of the event loops where the handler
+has been installed by ``MultiLoopChildWatcher``.
+
+2) In ``attach_loop`` and ``add_child_handler``, the handler is installed for the
+loop using ``add_signal_handler`` and the event loop instance is also added to
+the list.
+
+3) In ``remove_child_handler``, the ``remove_signal_handler`` is called on the
+loop the provided pid is attached to.
+
+4) In ``close``, the ``remove_signal_handler`` is called on every loop and the
+loop is removed from the list.
+
+The patch changes the behavior of ``MultiLoopChildWatcher``. ``MultiLoopChildWatcher``
+can now only be used in the main thread.
+
+Patch by Shreyan Avigyan