bpo-22087: Restructure code to be more robust and safe

Author: 1st1

Lib/asyncio/events.py

@@ -628,20 +628,25 @@ class _Local(threading.local):
         _pid = None
         _set_called = False
 
+        def __init__(self):
+            super().__init__()
+            self._pid = os.getpid()
+
     def __init__(self):
         self._local = self._Local()
-        self._local._pid = os.getpid()
-
-    def get_event_loop(self):
-        """Get the event loop.
 
-        This may be None or an instance of EventLoop.
-        """
+    def _check_pid(self):
         if self._local._pid != os.getpid():
             # If we detect we're in a child process forked by multiprocessing,
             # we reset self._local so that we'll get a new event loop.
             self._local = self._Local()
 
+    def get_event_loop(self):
+        """Get the event loop.
+
+        This may be None or an instance of EventLoop.
+        """
+        self._check_pid()
         if (self._local._loop is None and
                 not self._local._set_called and
                 isinstance(threading.current_thread(), threading._MainThread)):
@@ -655,6 +660,7 @@ def get_event_loop(self):
 
     def set_event_loop(self, loop):
         """Set the event loop."""
+        self._check_pid()
         self._local._set_called = True
         assert loop is None or isinstance(loop, AbstractEventLoop)
         self._local._loop = loop

Lib/test/test_asyncio/test_unix_events.py

@@ -1874,8 +1874,11 @@ def test_get_child_watcher_with_mainloop_existing(self):
         loop.close()
 
     def test_get_child_watcher_thread(self):
+        success = False
 
         def f():
+            nonlocal success
+
             policy.set_event_loop(policy.new_event_loop())
 
             self.assertIsInstance(policy.get_event_loop(),
@@ -1886,12 +1889,14 @@ def f():
             self.assertIsNone(watcher._loop)
 
             policy.get_event_loop().close()
+            success = True
 
         policy = self.create_policy()
 
         th = threading.Thread(target=f)
         th.start()
         th.join()
+        self.assertTrue(success)
 
     def test_child_watcher_replace_mainloop_existing(self):
         policy = self.create_policy()