[3.9] gh-116773: Fix overlapped memory corruption crash (GH-116774) (GH-117080)

Co-authored-by: Łukasz Langa <lukasz@langa.pl>

Lib/test/test_asyncio/test_windows_events.py

@@ -36,7 +36,23 @@ def data_received(self, data):
             self.trans.close()
 
 
-class ProactorLoopCtrlC(test_utils.TestCase):
+class WindowsEventsTestCase(test_utils.TestCase):
+    def _unraisablehook(self, unraisable):
+        # Storing unraisable.object can resurrect an object which is being
+        # finalized. Storing unraisable.exc_value creates a reference cycle.
+        self._unraisable = unraisable
+        print(unraisable)
+
+    def setUp(self):
+        self._prev_unraisablehook = sys.unraisablehook
+        self._unraisable = None
+        sys.unraisablehook = self._unraisablehook
+
+    def tearDown(self):
+        sys.unraisablehook = self._prev_unraisablehook
+        self.assertIsNone(self._unraisable)
+
+class ProactorLoopCtrlC(WindowsEventsTestCase):
 
     def test_ctrl_c(self):
 
@@ -58,7 +74,7 @@ def SIGINT_after_delay():
         thread.join()
 
 
-class ProactorMultithreading(test_utils.TestCase):
+class ProactorMultithreading(WindowsEventsTestCase):
     def test_run_from_nonmain_thread(self):
         finished = False
 
@@ -79,7 +95,7 @@ def func():
         self.assertTrue(finished)
 
 
-class ProactorTests(test_utils.TestCase):
+class ProactorTests(WindowsEventsTestCase):
 
     def setUp(self):
         super().setUp()
@@ -239,8 +255,32 @@ def test_read_self_pipe_restart(self):
         self.close_loop(self.loop)
         self.assertFalse(self.loop.call_exception_handler.called)
 
+    def test_loop_restart(self):
+        # We're fishing for the "RuntimeError: <_overlapped.Overlapped object at XXX>
+        # still has pending operation at deallocation, the process may crash" error
+        stop = threading.Event()
+        def threadMain():
+            while not stop.is_set():
+                self.loop.call_soon_threadsafe(lambda: None)
+                time.sleep(0.01)
+        thr = threading.Thread(target=threadMain)
 
-class WinPolicyTests(test_utils.TestCase):
+        # In 10 60-second runs of this test prior to the fix:
+        # time in seconds until failure: (none), 15.0, 6.4, (none), 7.6, 8.3, 1.7, 22.2, 23.5, 8.3
+        # 10 seconds had a 50% failure rate but longer would be more costly
+        end_time = time.time() + 10 # Run for 10 seconds
+        self.loop.call_soon(thr.start)
+        while not self._unraisable: # Stop if we got an unraisable exc
+            self.loop.stop()
+            self.loop.run_forever()
+            if time.time() >= end_time:
+                break
+
+        stop.set()
+        thr.join()
+
+
+class WinPolicyTests(WindowsEventsTestCase):
 
     def test_selector_win_policy(self):
         async def main():