Fix - freeze when reading stdout from external process

tests/test_process_spawning.py

@@ -0,0 +1,109 @@
+import asyncio
+import ctypes.util
+import logging
+from concurrent.futures import ThreadPoolExecutor
+from threading import Thread
+from unittest import TestCase
+
+import uvloop
+
+
+class ProcessSpawningTestCollection(TestCase):
+
+    def test_spawning_external_process(self):
+        """Test spawning external process (using `popen` system call) that
+        cause loop freeze."""
+
+        async def run(loop):
+            event = asyncio.Event(loop=loop)
+
+            dummy_workers = [simulate_loop_activity(loop, event)
+                             for _ in range(5)]
+            spawn_worker = spawn_external_process(loop, event)
+            done, pending = await asyncio.wait([spawn_worker] + dummy_workers,
+                                               loop=loop)
+            exceptions = [result.exception()
+                          for result in done if result.exception()]
+            if exceptions:
+                raise exceptions[0]
+
+            return True
+
+        async def simulate_loop_activity(loop, done_event):
+            """Simulate loop activity by busy waiting for event."""
+            while True:
+                try:
+                    await asyncio.wait_for(done_event.wait(),
+                                           timeout=0.1, loop=loop)
+                except asyncio.TimeoutError:
+                    pass
+
+                if done_event.is_set():
+                    return None
+
+        async def spawn_external_process(loop, event):
+            executor = ThreadPoolExecutor()
+            try:
+                call = loop.run_in_executor(executor, spawn_process)
+                await asyncio.wait_for(call, loop=loop, timeout=3600)
+            finally:
+                event.set()
+                executor.shutdown(wait=False)
+            return True
+
+        BUFFER_LENGTH = 1025
+        BufferType = ctypes.c_char * (BUFFER_LENGTH - 1)
+
+        def run_echo(popen, fread, pclose):
+            fd = popen('echo test'.encode('ASCII'), 'r'.encode('ASCII'))
+            try:
+                while True:
+                    buffer = BufferType()
+                    data = ctypes.c_void_p(ctypes.addressof(buffer))
+
+                    # -> this call will freeze whole loop in case of bug
+                    read = fread(data, 1, BUFFER_LENGTH, fd)
+                    if not read:
+                        break
+            except Exception:
+                logging.getLogger().exception('read error')
+                raise
+            finally:
+                pclose(fd)
+
+        def spawn_process():
+            """Spawn external process via `popen` system call."""
+
+            stdio = ctypes.CDLL(ctypes.util.find_library('c'))
+
+            # popen system call
+            popen = stdio.popen
+            popen.argtypes = (ctypes.c_char_p, ctypes.c_char_p)
+            popen.restype = ctypes.c_void_p
+
+            # pclose system call
+            pclose = stdio.pclose
+            pclose.argtypes = (ctypes.c_void_p,)
+            pclose.restype = ctypes.c_int
+
+            # fread system call
+            fread = stdio.fread
+            fread.argtypes = (ctypes.c_void_p, ctypes.c_size_t,
+                              ctypes.c_size_t, ctypes.c_void_p)
+            fread.restype = ctypes.c_size_t
+
+            for iteration in range(1000):
+                t = Thread(target=run_echo,
+                           args=(popen, fread, pclose),
+                           daemon=True)
+                t.start()
+                t.join(timeout=10.0)
+                if t.is_alive():
+                    raise Exception('process freeze detected at {}'
+                                    .format(iteration))
+
+            return True
+
+        loop = uvloop.new_event_loop()
+        proc = loop.run_until_complete(run(loop))
+        self.assertTrue(proc)

uvloop/handles/process.pyx

@@ -28,6 +28,7 @@ cdef class UVProcess(UVHandle):
 
         global __forking
         global __forking_loop
+        global __forkHandler
 
         cdef int err
 
@@ -76,6 +77,7 @@ cdef class UVProcess(UVHandle):
             loop.active_process_handler = self
             __forking = 1
             __forking_loop = loop
+            __forkHandler = <OnForkHandler>&__get_fork_handler
 
             PyOS_BeforeFork()
 
@@ -85,6 +87,7 @@ cdef class UVProcess(UVHandle):
 
             __forking = 0
             __forking_loop = None
+            __forkHandler = NULL
             loop.active_process_handler = None
 
             PyOS_AfterFork_Parent()

uvloop/includes/fork_handler.h

@@ -0,0 +1,15 @@
+
+typedef void (*OnForkHandler)();
+
+OnForkHandler __forkHandler = NULL;
+
+/* Auxiliary function to call global fork handler if defined.
+
+Note: Fork handler needs to be in C (not cython) otherwise it would require
+GIL to be present, but some forks can exec non-python processes.
+*/
+void handleAtFork() {
+    if (__forkHandler != NULL) {
+        __forkHandler();
+    }
+}

uvloop/includes/system.pxd

@@ -59,12 +59,12 @@ cdef extern from "unistd.h" nogil:
     void _exit(int status)
 
 
-cdef extern from "pthread.h" nogil:
+cdef extern from "pthread.h":
 
     int pthread_atfork(
-        void (*prepare)() nogil,
-        void (*parent)() nogil,
-        void (*child)() nogil)
+        void (*prepare)(),
+        void (*parent)(),
+        void (*child)())
 
 
 cdef extern from "includes/compat.h" nogil:

uvloop/loop.pyx

@@ -3117,19 +3117,21 @@ cdef vint __atfork_installed = 0
 cdef vint __forking = 0
 cdef Loop __forking_loop = None
 
+cdef extern from "includes/fork_handler.h":
 
-cdef void __atfork_child() nogil:
-    # See CPython/posixmodule.c for details
+    ctypedef void (*OnForkHandler)()
+    cdef OnForkHandler __forkHandler
+    void handleAtFork()
+
+cdef void __get_fork_handler() nogil:
     global __forking
+    global __forking_loop
 
     with gil:
-        if (__forking and
-                __forking_loop is not None and
+        if (__forking and __forking_loop is not None and
                 __forking_loop.active_process_handler is not None):
-
             __forking_loop.active_process_handler._after_fork()
 
-
 cdef __install_atfork():
     global __atfork_installed
     if __atfork_installed:
@@ -3138,7 +3140,7 @@ cdef __install_atfork():
 
     cdef int err
 
-    err = system.pthread_atfork(NULL, NULL, &__atfork_child)
+    err = system.pthread_atfork(NULL, NULL, &handleAtFork)
     if err:
         __atfork_installed = 0
         raise convert_error(-err)