bpo-29703: asyncio: Fix creating new event loops in child processes. (#404) (#410)

Lib/asyncio/events.py

@@ -11,6 +11,7 @@ __all__ = ['AbstractEventLoopPolicy',
 
 import functools
 import inspect
+import os
 import reprlib
 import socket
 import subprocess
@@ -611,6 +612,9 @@ _lock = threading.Lock()
 # A TLS for the running event loop, used by _get_running_loop.
 class _RunningLoop(threading.local):
     _loop = None
+    _pid = None
+
+
 _running_loop = _RunningLoop()
 
 
@@ -620,6 +624,7 @@ def _get_running_loop():
     This is a low-level function intended to be used by event loops.
     This function is thread-specific.
     """
+    if _running_loop._pid == os.getpid():
         return _running_loop._loop
 
 
@@ -629,6 +634,7 @@ def _set_running_loop(loop):
     This is a low-level function intended to be used by event loops.
     This function is thread-specific.
     """
+    _running_loop._pid = os.getpid()
     _running_loop._loop = loop
 
 

Lib/asyncio/test_utils.py

@@ -449,12 +449,15 @@ class TestCase(unittest.TestCase):
         self.set_event_loop(loop)
         return loop
 
+    def unpatch_get_running_loop(self):
+        events._get_running_loop = self._get_running_loop
+
     def setUp(self):
         self._get_running_loop = events._get_running_loop
         events._get_running_loop = lambda: None
 
     def tearDown(self):
-        events._get_running_loop = self._get_running_loop
+        self.unpatch_get_running_loop()
 
         events.set_event_loop(None)
 

Lib/test/test_asyncio/test_events.py

@@ -1,6 +1,7 @@
 """Tests for events.py."""
 
 import collections.abc
+import concurrent.futures
 import functools
 import gc
 import io
@@ -57,6 +58,15 @@ def osx_tiger():
     return version < (10, 5)
 
 
+def _test_get_event_loop_new_process__sub_proc():
+    async def doit():
+        return 'hello'
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    return loop.run_until_complete(doit())
+
+
 ONLYCERT = data_file('ssl_cert.pem')
 ONLYKEY = data_file('ssl_key.pem')
 SIGNED_CERTFILE = data_file('keycert3.pem')
@@ -2181,6 +2191,18 @@ else:
             asyncio.set_child_watcher(None)
             super().tearDown()
 
+        def test_get_event_loop_new_process(self):
+            async def main():
+                pool = concurrent.futures.ProcessPoolExecutor()
+                return await self.loop.run_in_executor(
+                    pool, _test_get_event_loop_new_process__sub_proc)
+
+            self.unpatch_get_running_loop()
+
+            self.assertEqual(
+                self.loop.run_until_complete(main()),
+                'hello')
+
     if hasattr(selectors, 'KqueueSelector'):
         class KqueueEventLoopTests(UnixEventLoopTestsMixin,
                                    SubprocessTestsMixin,

Misc/NEWS

@@ -79,6 +79,9 @@ Extension Modules
 Library
 -------
 
+- bpo-29703: Fix asyncio to support instantiation of new event loops
+  in child processes.
+
 - bpo-29376: Fix assertion error in threading._DummyThread.is_alive().
 
 - bpo-28624: Add a test that checks that cwd parameter of Popen() accepts