improve freeze detection by removing global processing time assumption

Author: grungy-ado

tests/test_process_spawning.py

@@ -2,6 +2,7 @@
 import ctypes.util
 import logging
 from concurrent.futures import ThreadPoolExecutor
+from threading import Thread
 from unittest import TestCase
 
 import uvloop
@@ -44,12 +45,32 @@ 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=160)
+                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."""
 
@@ -71,25 +92,15 @@ def spawn_process():
                               ctypes.c_size_t, ctypes.c_void_p)
             fread.restype = ctypes.c_size_t
 
-            BUFFER_LENGTH = 1025
-            BufferType = ctypes.c_char * (BUFFER_LENGTH - 1)
-
-            for _ in range(10000):
-                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)
+            for iteration in range(10000):
+                t = Thread(target=run_echo,
+                           args=(popen, fread, pclose),
+                           daemon=True)
+                t.start()
+                t.join(timeout=20.0)
+                if t.is_alive():
+                    raise Exception('process freeze detected at {}'
+                                    .format(iteration))
 
             return True