asyncio: Add support for running subprocesses on Windows with the IOCP event loop (Richard Oudkerk).

Lib/asyncio/unix_events.py

@@ -1,6 +1,5 @@
 """Selector eventloop for Unix with signal handling."""
 
-import collections
 import errno
 import fcntl
 import os
@@ -11,6 +10,7 @@ import subprocess
 import sys
 
 
+from . import base_subprocess
 from . import constants
 from . import events
 from . import protocols
@@ -406,159 +406,20 @@ class _UnixWritePipeTransport(transports.WriteTransport):
             self._loop = None
 
 
-class _UnixWriteSubprocessPipeProto(protocols.BaseProtocol):
-    pipe = None
+class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):
 
-    def __init__(self, proc, fd):
-        self.proc = proc
-        self.fd = fd
-        self.connected = False
-        self.disconnected = False
-        proc._pipes[fd] = self
-
-    def connection_made(self, transport):
-        self.connected = True
-        self.pipe = transport
-        self.proc._try_connected()
-
-    def connection_lost(self, exc):
-        self.disconnected = True
-        self.proc._pipe_connection_lost(self.fd, exc)
-
-
-class _UnixReadSubprocessPipeProto(_UnixWriteSubprocessPipeProto,
-                                   protocols.Protocol):
-
-    def data_received(self, data):
-        self.proc._pipe_data_received(self.fd, data)
-
-    def eof_received(self):
-        pass
-
-
-class _UnixSubprocessTransport(transports.SubprocessTransport):
-
-    def __init__(self, loop, protocol, args, shell,
-                 stdin, stdout, stderr, bufsize,
-                 extra=None, **kwargs):
-        super().__init__(extra)
-        self._protocol = protocol
-        self._loop = loop
-
-        self._pipes = {}
+    def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
         stdin_w = None
         if stdin == subprocess.PIPE:
-            self._pipes[STDIN] = None
             # Use a socket pair for stdin, since not all platforms
             # support selecting read events on the write end of a
             # socket (which we use in order to detect closing of the
             # other end).  Notably this is needed on AIX, and works
             # just fine on other platforms.
             stdin, stdin_w = self._loop._socketpair()
-        if stdout == subprocess.PIPE:
-            self._pipes[STDOUT] = None
-        if stderr == subprocess.PIPE:
-            self._pipes[STDERR] = None
-        self._pending_calls = collections.deque()
-        self._finished = False
-        self._returncode = None
-
         self._proc = subprocess.Popen(
             args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
             universal_newlines=False, bufsize=bufsize, **kwargs)
         if stdin_w is not None:
             stdin.close()
             self._proc.stdin = open(stdin_w.detach(), 'rb', buffering=bufsize)
-        self._extra['subprocess'] = self._proc
-
-    def close(self):
-        for proto in self._pipes.values():
-            proto.pipe.close()
-        if self._returncode is None:
-            self.terminate()
-
-    def get_pid(self):
-        return self._proc.pid
-
-    def get_returncode(self):
-        return self._returncode
-
-    def get_pipe_transport(self, fd):
-        if fd in self._pipes:
-            return self._pipes[fd].pipe
-        else:
-            return None
-
-    def send_signal(self, signal):
-        self._proc.send_signal(signal)
-
-    def terminate(self):
-        self._proc.terminate()
-
-    def kill(self):
-        self._proc.kill()
-
-    @tasks.coroutine
-    def _post_init(self):
-        proc = self._proc
-        loop = self._loop
-        if proc.stdin is not None:
-            transp, proto = yield from loop.connect_write_pipe(
-                lambda: _UnixWriteSubprocessPipeProto(self, STDIN),
-                proc.stdin)
-        if proc.stdout is not None:
-            transp, proto = yield from loop.connect_read_pipe(
-                lambda: _UnixReadSubprocessPipeProto(self, STDOUT),
-                proc.stdout)
-        if proc.stderr is not None:
-            transp, proto = yield from loop.connect_read_pipe(
-                lambda: _UnixReadSubprocessPipeProto(self, STDERR),
-                proc.stderr)
-        if not self._pipes:
-            self._try_connected()
-
-    def _call(self, cb, *data):
-        if self._pending_calls is not None:
-            self._pending_calls.append((cb, data))
-        else:
-            self._loop.call_soon(cb, *data)
-
-    def _try_connected(self):
-        assert self._pending_calls is not None
-        if all(p is not None and p.connected for p in self._pipes.values()):
-            self._loop.call_soon(self._protocol.connection_made, self)
-            for callback, data in self._pending_calls:
-                self._loop.call_soon(callback, *data)
-            self._pending_calls = None
-
-    def _pipe_connection_lost(self, fd, exc):
-        self._call(self._protocol.pipe_connection_lost, fd, exc)
-        self._try_finish()
-
-    def _pipe_data_received(self, fd, data):
-        self._call(self._protocol.pipe_data_received, fd, data)
-
-    def _process_exited(self, returncode):
-        assert returncode is not None, returncode
-        assert self._returncode is None, self._returncode
-        self._returncode = returncode
-        self._loop._subprocess_closed(self)
-        self._call(self._protocol.process_exited)
-        self._try_finish()
-
-    def _try_finish(self):
-        assert not self._finished
-        if self._returncode is None:
-            return
-        if all(p is not None and p.disconnected
-               for p in self._pipes.values()):
-            self._finished = True
-            self._loop.call_soon(self._call_connection_lost, None)
-
-    def _call_connection_lost(self, exc):
-        try:
-            self._protocol.connection_lost(exc)
-        finally:
-            self._proc = None
-            self._protocol = None
-            self._loop = None