Revert "bpo-35621: Support running subprocesses in asyncio when loop is executed in non-main thread (#13630)" (GH-13793)

https://bugs.python.org/issue35621

1 files changed, 28 insertions, 212 deletions

diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index b943845..28128d2 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -2,7 +2,6 @@
 
 import errno
 import io
-import itertools
 import os
 import selectors
 import signal
@@ -30,9 +29,7 @@ from .log import logger
 __all__ = (
     'SelectorEventLoop',
     'AbstractChildWatcher', 'SafeChildWatcher',
-    'FastChildWatcher',
-    'MultiLoopChildWatcher', 'ThreadedChildWatcher',
-    'DefaultEventLoopPolicy',
+    'FastChildWatcher', 'DefaultEventLoopPolicy',
 )
 
 
@@ -187,13 +184,6 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
                                          stdin, stdout, stderr, bufsize,
                                          extra=None, **kwargs):
         with events.get_child_watcher() as watcher:
-            if not watcher.is_active():
-                # Check early.
-                # Raising exception before process creation
-                # prevents subprocess execution if the watcher
-                # is not ready to handle it.
-                raise RuntimeError("asyncio.get_child_watcher() is not activated, "
-                                   "subprocess support is not installed.")
             waiter = self.create_future()
             transp = _UnixSubprocessTransport(self, protocol, args, shell,
                                               stdin, stdout, stderr, bufsize,
@@ -848,15 +838,6 @@ class AbstractChildWatcher:
         """
         raise NotImplementedError()
 
-    def is_active(self):
-        """Watcher status.
-
-        Return True if the watcher is installed and ready to handle process exit
-        notifications.
-
-        """
-        raise NotImplementedError()
-
     def __enter__(self):
         """Enter the watcher's context and allow starting new processes
 
@@ -868,20 +849,6 @@ class AbstractChildWatcher:
         raise NotImplementedError()
 
 
-def _compute_returncode(status):
-    if os.WIFSIGNALED(status):
-        # The child process died because of a signal.
-        return -os.WTERMSIG(status)
-    elif os.WIFEXITED(status):
-        # The child process exited (e.g sys.exit()).
-        return os.WEXITSTATUS(status)
-    else:
-        # The child exited, but we don't understand its status.
-        # This shouldn't happen, but if it does, let's just
-        # return that status; perhaps that helps debug it.
-        return status
-
-
 class BaseChildWatcher(AbstractChildWatcher):
 
     def __init__(self):
@@ -891,9 +858,6 @@ class BaseChildWatcher(AbstractChildWatcher):
     def close(self):
         self.attach_loop(None)
 
-    def is_active(self):
-        return self._loop is not None and self._loop.is_running()
-
     def _do_waitpid(self, expected_pid):
         raise NotImplementedError()
 
@@ -934,6 +898,19 @@ class BaseChildWatcher(AbstractChildWatcher):
                 'exception': exc,
             })
 
+    def _compute_returncode(self, status):
+        if os.WIFSIGNALED(status):
+            # The child process died because of a signal.
+            return -os.WTERMSIG(status)
+        elif os.WIFEXITED(status):
+            # The child process exited (e.g sys.exit()).
+            return os.WEXITSTATUS(status)
+        else:
+            # The child exited, but we don't understand its status.
+            # This shouldn't happen, but if it does, let's just
+            # return that status; perhaps that helps debug it.
+            return status
+
 
 class SafeChildWatcher(BaseChildWatcher):
     """'Safe' child watcher implementation.
@@ -957,6 +934,11 @@ class SafeChildWatcher(BaseChildWatcher):
         pass
 
     def add_child_handler(self, pid, callback, *args):
+        if self._loop is None:
+            raise RuntimeError(
+                "Cannot add child handler, "
+                "the child watcher does not have a loop attached")
+
         self._callbacks[pid] = (callback, args)
 
         # Prevent a race condition in case the child is already terminated.
@@ -992,7 +974,7 @@ class SafeChildWatcher(BaseChildWatcher):
                 # The child process is still alive.
                 return
 
-            returncode = _compute_returncode(status)
+            returncode = self._compute_returncode(status)
             if self._loop.get_debug():
                 logger.debug('process %s exited with returncode %s',
                              expected_pid, returncode)
@@ -1053,6 +1035,11 @@ class FastChildWatcher(BaseChildWatcher):
     def add_child_handler(self, pid, callback, *args):
         assert self._forks, "Must use the context manager"
 
+        if self._loop is None:
+            raise RuntimeError(
+                "Cannot add child handler, "
+                "the child watcher does not have a loop attached")
+
         with self._lock:
             try:
                 returncode = self._zombies.pop(pid)
@@ -1085,7 +1072,7 @@ class FastChildWatcher(BaseChildWatcher):
                     # A child process is still alive.
                     return
 
-                returncode = _compute_returncode(status)
+                returncode = self._compute_returncode(status)
 
             with self._lock:
                 try:
@@ -1114,177 +1101,6 @@ class FastChildWatcher(BaseChildWatcher):
                 callback(pid, returncode, *args)
 
 
-class MultiLoopChildWatcher(AbstractChildWatcher):
-    # 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()
-
-    def __init__(self):
-        self._callbacks = {}
-        self._saved_sighandler = None
-
-    def is_active(self):
-        return self._saved_sighandler is not None
-
-    def close(self):
-        self._callbacks.clear()
-        if self._saved_sighandler is not None:
-            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
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        pass
-
-    def add_child_handler(self, pid, callback, *args):
-        loop = events.get_running_loop()
-        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]
-            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 None:
-            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)
-
-    def _do_waitpid_all(self):
-        for pid in list(self._callbacks):
-            self._do_waitpid(pid)
-
-    def _do_waitpid(self, expected_pid):
-        assert expected_pid > 0
-
-        try:
-            pid, status = os.waitpid(expected_pid, os.WNOHANG)
-        except ChildProcessError:
-            # The child process is already reaped
-            # (may happen if waitpid() is called elsewhere).
-            pid = expected_pid
-            returncode = 255
-            logger.warning(
-                "Unknown child process pid %d, will report returncode 255",
-                pid)
-            debug_log = False
-        else:
-            if pid == 0:
-                # The child process is still alive.
-                return
-
-            returncode = _compute_returncode(status)
-            debug_log = True
-        try:
-            loop, callback, args = self._callbacks.pop(pid)
-        except KeyError:  # pragma: no cover
-            # May happen if .remove_child_handler() is called
-            # after os.waitpid() returns.
-            logger.warning("Child watcher got an unexpected pid: %r",
-                           pid, exc_info=True)
-        else:
-            if loop.is_closed():
-                logger.warning("Loop %r that handles pid %r is closed", loop, pid)
-            else:
-                if debug_log and loop.get_debug():
-                    logger.debug('process %s exited with returncode %s',
-                                 expected_pid, returncode)
-                loop.call_soon_threadsafe(callback, pid, returncode, *args)
-
-    def _sig_chld(self, signum, frame):
-        try:
-            self._do_waitpid_all()
-        except (SystemExit, KeyboardInterrupt):
-            raise
-        except BaseException:
-            logger.warning('Unknown exception in SIGCHLD handler', exc_info=True)
-
-
-class ThreadedChildWatcher(AbstractChildWatcher):
-    # The watcher uses a thread per process
-    # for waiting for the process finish.
-    # It doesn't require subscription on POSIX signal
-
-    def __init__(self):
-        self._pid_counter = itertools.count(0)
-
-    def is_active(self):
-        return True
-
-    def close(self):
-        pass
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        pass
-
-    def add_child_handler(self, pid, callback, *args):
-        loop = events.get_running_loop()
-        thread = threading.Thread(target=self._do_waitpid,
-                                  name=f"waitpid-{next(self._pid_counter)}",
-                                  args=(loop, pid, callback, args),
-                                  daemon=True)
-        thread.start()
-
-    def remove_child_handler(self, pid):
-        # asyncio never calls remove_child_handler() !!!
-        # The method is no-op but is implemented because
-        # abstract base classe requires it
-        return True
-
-    def attach_loop(self, loop):
-        pass
-
-    def _do_waitpid(self, loop, expected_pid, callback, args):
-        assert expected_pid > 0
-
-        try:
-            pid, status = os.waitpid(expected_pid, 0)
-        except ChildProcessError:
-            # The child process is already reaped
-            # (may happen if waitpid() is called elsewhere).
-            pid = expected_pid
-            returncode = 255
-            logger.warning(
-                "Unknown child process pid %d, will report returncode 255",
-                pid)
-        else:
-            returncode = _compute_returncode(status)
-            if loop.get_debug():
-                logger.debug('process %s exited with returncode %s',
-                             expected_pid, returncode)
-
-        if loop.is_closed():
-            logger.warning("Loop %r that handles pid %r is closed", loop, pid)
-        else:
-            loop.call_soon_threadsafe(callback, pid, returncode, *args)
-
-
 class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
     """UNIX event loop policy with a watcher for child processes."""
     _loop_factory = _UnixSelectorEventLoop
@@ -1296,7 +1112,7 @@ class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
     def _init_watcher(self):
         with events._lock:
             if self._watcher is None:  # pragma: no branch
-                self._watcher = ThreadedChildWatcher()
+                self._watcher = SafeChildWatcher()
                 if isinstance(threading.current_thread(),
                               threading._MainThread):
                     self._watcher.attach_loop(self._local._loop)
@@ -1318,7 +1134,7 @@ class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
     def get_child_watcher(self):
         """Get the watcher for child processes.
 
-        If not yet set, a ThreadedChildWatcher object is automatically created.
+        If not yet set, a SafeChildWatcher object is automatically created.
         """
         if self._watcher is None:
             self._init_watcher()