Fix asyncio to support instantiation of new event loops in subprocesses

See PR #452 for context.

Author: 1st1

asyncio/events.py

@@ -11,6 +11,7 @@
 
 import functools
 import inspect
+import os
 import reprlib
 import socket
 import subprocess
@@ -611,6 +612,9 @@ def new_event_loop(self):
 # A TLS for the running event loop, used by _get_running_loop.
 class _RunningLoop(threading.local):
     _loop = None
+    _pid = None
+
+
 _running_loop = _RunningLoop()
 
 
@@ -620,7 +624,8 @@ def _get_running_loop():
     This is a low-level function intended to be used by event loops.
     This function is thread-specific.
     """
-    return _running_loop._loop
+    if _running_loop._pid == os.getpid():
+        return _running_loop._loop
 
 
 def _set_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
 
 

asyncio/test_utils.py

@@ -449,12 +449,15 @@ def new_test_loop(self, gen=None):
         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)
 

tests/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')
@@ -2103,6 +2113,18 @@ def connect(cmd=None, **kwds):
         with self.assertRaises(ValueError):
             self.loop.run_until_complete(connect(shell=False))
 
+    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 sys.platform == 'win32':