Serialize process spawning across threads with a lock

Replace the RuntimeError-raising check for concurrent process spawning
with a threading.Lock that serializes spawns across different event
loops running in separate threads. The old approach would fail with
"Racing with another loop to spawn a process" when multiple threads
tried to spawn processes concurrently, since the global pthread_atfork
handlers can only be active for one loop at a time.

Now instead of failing, concurrent spawns wait for the lock, allowing
them to proceed sequentially. The lock is properly released in all
error paths via a finally block.

Fixes #508

Author: veeceey

uvloop/handles/process.pyx

@@ -68,21 +68,28 @@ cdef class UVProcess(UVHandle):
             self._abort_init()
             raise
 
-        if __forking or loop.active_process_handler is not None:
-            # Our pthread_atfork handlers won't work correctly when
-            # another loop is forking in another thread (even though
-            # GIL should help us to avoid that.)
+        # Acquire the global spawn lock to serialize process spawning
+        # across threads. Without this, concurrent spawns from different
+        # loops would race on the global pthread_atfork handlers.
+        __spawn_lock.acquire()
+        _spawn_lock_held = True
+
+        if loop.active_process_handler is not None:
+            __spawn_lock.release()
+            _spawn_lock_held = False
             self._abort_init()
             raise RuntimeError(
-                '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)
+                'Racing with the same loop to spawn a process.')
+
+        fds_to_close = None
         try:
+            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)
+
             os_set_inheritable(self._errpipe_write, True)
 
             self._preexec_fn = preexec_fn
@@ -103,6 +110,8 @@ cdef class UVProcess(UVHandle):
             __forking_loop = None
             system.resetForkHandler()
             loop.active_process_handler = None
+            __spawn_lock.release()
+            _spawn_lock_held = False
 
             PyOS_AfterFork_Parent()
 
@@ -128,8 +137,16 @@ cdef class UVProcess(UVHandle):
                         break
 
         finally:
-            while fds_to_close:
-                os_close(fds_to_close.pop())
+            if _spawn_lock_held:
+                __forking = 0
+                __forking_loop = None
+                system.resetForkHandler()
+                loop.active_process_handler = None
+                __spawn_lock.release()
+
+            if fds_to_close is not None:
+                while fds_to_close:
+                    os_close(fds_to_close.pop())
 
             for fd in restore_inheritable:
                 os_set_inheritable(fd, False)

uvloop/loop.pyx

@@ -41,6 +41,8 @@ from cpython cimport (
 from cpython.pycapsule cimport PyCapsule_New, PyCapsule_GetPointer
 
 from . import _noop
+import threading
+__spawn_lock = threading.Lock()
 
 
 include "includes/stdlib.pxi"