process: Refactor process transport to handle late stdio init

Author: 1st1

tests/test_process.py

@@ -464,6 +464,92 @@ def cancel_make_transport():
 
 
 class Test_UV_Process(_TestProcess, tb.UVTestCase):
+
+    def test_process_lated_stdio_init(self):
+
+        class TestProto:
+            def __init__(self):
+                self.lost = 0
+                self.stages = []
+
+            def connection_made(self, transport):
+                self.stages.append(('CM', transport))
+
+            def pipe_data_received(self, fd, data):
+                if fd == 1:
+                    self.stages.append(('STDOUT', data))
+
+            def pipe_connection_lost(self, fd, exc):
+                if fd == 1:
+                    self.stages.append(('STDOUT', 'LOST'))
+
+            def process_exited(self):
+                self.stages.append('PROC_EXIT')
+
+            def connection_lost(self, exc):
+                self.stages.append(('CL', self.lost, exc))
+                self.lost += 1
+
+        async def run(**kwargs):
+            return await self.loop.subprocess_shell(
+                lambda: TestProto(),
+                'echo 1',
+                **kwargs)
+
+        with self.subTest('paused, stdin pipe'):
+            transport, proto = self.loop.run_until_complete(
+                run(stdin=subprocess.PIPE,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    __uvloop_sleep_after_fork=True))
+            self.assertIsNot(transport, None)
+            self.assertEqual(transport.get_returncode(), 0)
+            self.assertEqual(
+                set(proto.stages),
+                {
+                    ('CM', transport),
+                    'PROC_EXIT',
+                    ('STDOUT', b'1\n'),
+                    ('STDOUT', 'LOST'),
+                    ('CL', 0, None)
+                })
+
+        with self.subTest('paused, no stdin'):
+            transport, proto = self.loop.run_until_complete(
+                run(stdin=None,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    __uvloop_sleep_after_fork=True))
+            self.assertIsNot(transport, None)
+            self.assertEqual(transport.get_returncode(), 0)
+            self.assertEqual(
+                set(proto.stages),
+                {
+                    ('CM', transport),
+                    'PROC_EXIT',
+                    ('STDOUT', b'1\n'),
+                    ('STDOUT', 'LOST'),
+                    ('CL', 0, None)
+                })
+
+        with self.subTest('no pause, no stdin'):
+            transport, proto = self.loop.run_until_complete(
+                run(stdin=None,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE))
+            self.loop.run_until_complete(transport._wait())
+            self.assertEqual(transport.get_returncode(), 0)
+            self.assertIsNot(transport, None)
+            self.assertEqual(
+                set(proto.stages),
+                {
+                    ('CM', transport),
+                    'PROC_EXIT',
+                    ('STDOUT', b'1\n'),
+                    ('STDOUT', 'LOST'),
+                    ('CL', 0, None)
+                })
+
     def test_process_streams_redirect(self):
         # This won't work for asyncio implementation of subprocess
 

uvloop/handles/pipe.pyx

@@ -111,6 +111,13 @@ cdef class UVWritePipeTransport(UVWriteTransport):
                                   object waiter):
         cdef UVWritePipeTransport handle
         handle = UVWritePipeTransport.__new__(UVWritePipeTransport)
+
+        # We listen for read events on write-end of the pipe. When
+        # the read-end is close, the uv_stream_t.read callback will
+        # receive an error -- we want to silence that error, and just
+        # close the transport.
+        handle._close_on_read_error()
+
         handle._init(loop, protocol, server, waiter)
         __pipe_init_uv_handle(<UVStream>handle, loop)
         return handle

uvloop/handles/process.pxd

@@ -17,7 +17,8 @@ cdef class UVProcess(UVHandle):
 
     cdef _init(self, Loop loop, list args, dict env, cwd,
                start_new_session,
-               stdin, stdout, stderr, pass_fds)
+               stdin, stdout, stderr, pass_fds,
+               debug_flags)
 
     cdef char** __to_cstring_array(self, list arr)
     cdef _init_args(self, list args)
@@ -35,24 +36,35 @@ cdef class UVProcess(UVHandle):
 cdef class UVProcessTransport(UVProcess):
     cdef:
         list _exit_waiters
+        list _init_futs
+        bint _stdio_ready
+        list _pending_calls
         object _protocol
+        bint _finished
 
         UVWritePipeTransport stdin
         UVReadPipeTransport stdout
         UVReadPipeTransport stderr
 
+        object stdin_proto
+        object stdout_proto
+        object stderr_proto
+
     cdef _file_redirect_stdio(self, int fd)
     cdef _file_devnull(self)
+    cdef _file_inpipe(self)
     cdef _file_outpipe(self)
 
     cdef _check_proc(self)
     cdef _pipe_connection_lost(self, int fd, exc)
     cdef _pipe_data_received(self, int fd, data)
 
     cdef _call_connection_made(self, waiter)
+    cdef _try_finish(self)
 
     @staticmethod
     cdef UVProcessTransport new(Loop loop, protocol, args, env, cwd,
                                 start_new_session,
                                 stdin, stdout, stderr, pass_fds,
-                                waiter)
+                                waiter,
+                                debug_flags)