use a stack of self._fds_to_close to prevent double closes (#481)

* Add test for preexec_fn fd double close issue
* use a stack of self._fds_to_close to prevent double closes

and make tests easier to write because the close order is deterministic
and in the order that opens happen in

this should also be a bit faster because list.append is faster
than set.add and we skip a call to os_close(-1) and catching an
OSError exception

* DRY os_dup call

Co-authored-by: Fantix King <fantix.king@gmail.com>

tests/test_process.py

@@ -7,6 +7,7 @@
 import subprocess
 import sys
 import tempfile
+import textwrap
 import time
 import unittest
 
@@ -796,7 +797,58 @@ async def test():
 
 
 class Test_UV_Process(_TestProcess, tb.UVTestCase):
-    pass
+    def test_process_double_close(self):
+        script = textwrap.dedent("""
+            import os
+            import sys
+            from unittest import mock
+
+            import asyncio
+
+            pipes = []
+            original_os_pipe = os.pipe
+            def log_pipes():
+                pipe = original_os_pipe()
+                pipes.append(pipe)
+                return pipe
+
+            dups = []
+            original_os_dup = os.dup
+            def log_dups(*args, **kwargs):
+                dup = original_os_dup(*args, **kwargs)
+                dups.append(dup)
+                return dup
+
+            with mock.patch(
+                "os.close", wraps=os.close
+            ) as os_close, mock.patch(
+                "os.pipe", new=log_pipes
+            ), mock.patch(
+                "os.dup", new=log_dups
+            ):
+                import uvloop
+
+
+            async def test():
+                proc = await asyncio.create_subprocess_exec(
+                    sys.executable, "-c", "pass"
+                )
+                await proc.communicate()
+
+            uvloop.install()
+            asyncio.run(test())
+
+            stdin, stdout, stderr = dups
+            (r, w), = pipes
+            assert os_close.mock_calls == [
+                mock.call(w),
+                mock.call(r),
+                mock.call(stderr),
+                mock.call(stdout),
+                mock.call(stdin),
+            ]
+        """)
+        subprocess.run([sys.executable, '-c', script], check=True)
 
 
 class Test_AIO_Process(_TestProcess, tb.AIOTestCase):

uvloop/handles/process.pxd

@@ -8,7 +8,7 @@ cdef class UVProcess(UVHandle):
         object _preexec_fn
         bint _restore_signals
 
-        set _fds_to_close
+        list _fds_to_close
 
         # Attributes used to compose uv_process_options_t:
         uv.uv_process_options_t options

uvloop/handles/process.pyx

@@ -7,7 +7,7 @@ cdef class UVProcess(UVHandle):
         self.uv_opt_args = NULL
         self._returncode = None
         self._pid = None
-        self._fds_to_close = set()
+        self._fds_to_close = list()
         self._preexec_fn = None
         self._restore_signals = True
         self.context = Context_CopyCurrent()
@@ -69,6 +69,11 @@ cdef class UVProcess(UVHandle):
                 'Racing with another loop to spawn a process.')
 
         self._errpipe_read, self._errpipe_write = os_pipe()
+        fds_to_close = self._fds_to_close
+        self._fds_to_close = None
+        fds_to_close.append(self._errpipe_read)
+        # add the write pipe last so we can close it early
+        fds_to_close.append(self._errpipe_write)
         try:
             os_set_inheritable(self._errpipe_write, True)
 
@@ -100,8 +105,8 @@ cdef class UVProcess(UVHandle):
 
             self._finish_init()
 
-            os_close(self._errpipe_write)
-            self._errpipe_write = -1
+            # close the write pipe early
+            os_close(fds_to_close.pop())
 
             if preexec_fn is not None:
                 errpipe_data = bytearray()
@@ -115,17 +120,8 @@ cdef class UVProcess(UVHandle):
                         break
 
         finally:
-            os_close(self._errpipe_read)
-            try:
-                os_close(self._errpipe_write)
-            except OSError:
-                # Might be already closed
-                pass
-
-            fds_to_close = self._fds_to_close
-            self._fds_to_close = None
-            for fd in fds_to_close:
-                os_close(fd)
+            while fds_to_close:
+                os_close(fds_to_close.pop())
 
             for fd in restore_inheritable:
                 os_set_inheritable(fd, False)
@@ -202,7 +198,7 @@ cdef class UVProcess(UVHandle):
         if self._fds_to_close is None:
             raise RuntimeError(
                 'UVProcess._close_after_spawn called after uv_spawn')
-        self._fds_to_close.add(fd)
+        self._fds_to_close.append(fd)
 
     def __dealloc__(self):
         if self.uv_opt_env is not NULL:
@@ -500,10 +496,7 @@ cdef class UVProcessTransport(UVProcess):
                     # shouldn't ever happen
                     raise RuntimeError('cannot apply subprocess.STDOUT')
 
-                newfd = os_dup(io[1])
-                os_set_inheritable(newfd, True)
-                self._close_after_spawn(newfd)
-                io[2] = newfd
+                io[2] = self._file_redirect_stdio(io[1])
             elif _stderr == subprocess_DEVNULL:
                 io[2] = self._file_devnull()
             else: