Merge remote-tracking branch 'upstream/main' into fix-structseq-validate-desc

Author: XuehaiPan

Lib/asyncio/windows_events.py

@@ -760,6 +760,46 @@ def _get_accept_socket(self, family):
         s.settimeout(0)
         return s
 
+    def _process_completion_status(self, status):
+        """Process a single status from the completion port.
+
+        A caller that waits on the completion port itself can pass each
+        status it receives here.
+        """
+        err, transferred, key, address = status
+        try:
+            f, ov, obj, callback = self._cache.pop(address)
+        except KeyError:
+            if self._loop.get_debug():
+                self._loop.call_exception_handler({
+                    'message': ('GetQueuedCompletionStatus() returned an '
+                                'unexpected event'),
+                    'status': ('err=%s transferred=%s key=%#x address=%#x'
+                               % (err, transferred, key, address)),
+                })
+
+            # key is either zero, or it is used to return a pipe
+            # handle which should be closed to avoid a leak.
+            if key not in (0, _overlapped.INVALID_HANDLE_VALUE):
+                _winapi.CloseHandle(key)
+            return
+
+        if obj in self._stopped_serving:
+            f.cancel()
+        # Don't call the callback if _register() already read the result or
+        # if the overlapped has been cancelled
+        elif not f.done():
+            try:
+                value = callback(transferred, key, ov)
+            except OSError as e:
+                f.set_exception(e)
+                self._results.append(f)
+            else:
+                f.set_result(value)
+                self._results.append(f)
+            finally:
+                f = None
+
     def _poll(self, timeout=None):
         if timeout is None:
             ms = INFINITE
@@ -778,39 +818,8 @@ def _poll(self, timeout=None):
                 break
             ms = 0
 
-            err, transferred, key, address = status
-            try:
-                f, ov, obj, callback = self._cache.pop(address)
-            except KeyError:
-                if self._loop.get_debug():
-                    self._loop.call_exception_handler({
-                        'message': ('GetQueuedCompletionStatus() returned an '
-                                    'unexpected event'),
-                        'status': ('err=%s transferred=%s key=%#x address=%#x'
-                                   % (err, transferred, key, address)),
-                    })
-
-                # key is either zero, or it is used to return a pipe
-                # handle which should be closed to avoid a leak.
-                if key not in (0, _overlapped.INVALID_HANDLE_VALUE):
-                    _winapi.CloseHandle(key)
-                continue
-
-            if obj in self._stopped_serving:
-                f.cancel()
-            # Don't call the callback if _register() already read the result or
-            # if the overlapped has been cancelled
-            elif not f.done():
-                try:
-                    value = callback(transferred, key, ov)
-                except OSError as e:
-                    f.set_exception(e)
-                    self._results.append(f)
-                else:
-                    f.set_result(value)
-                    self._results.append(f)
-                finally:
-                    f = None
+            # gh-154971: split out so custom event loops can call it directly
+            self._process_completion_status(status)
 
         # Remove unregistered futures
         for ov in self._unregistered:

Lib/test/test_asyncio/test_windows_events.py

@@ -327,6 +327,28 @@ def threadMain():
         stop.set()
         thr.join()
 
+    def test_custom_poll_integration(self):
+        # gh-154971: a caller can wait on the completion port and process statuses itself
+        proactor = self.loop._proactor
+
+        a, b = socket.socketpair()
+        self.addCleanup(a.close)
+        self.addCleanup(b.close)
+
+        fut = proactor.recv(a, 100)
+        self.assertFalse(fut.done())
+
+        b.send(b'data')
+
+        deadline = time.monotonic() + support.SHORT_TIMEOUT
+        while not fut.done() and time.monotonic() < deadline:
+            status = _overlapped.GetQueuedCompletionStatus(proactor._iocp, 100)
+            if status is not None:
+                proactor._process_completion_status(status)
+
+        self.assertTrue(fut.done())
+        self.assertEqual(fut.result(), b'data')
+
 
 class ProactorPipeObjectSupportTests(unittest.TestCase):
 

Lib/test/test_class.py

@@ -2,7 +2,7 @@
 
 import unittest
 from test import support
-from test.support import cpython_only, import_helper, script_helper
+from test.support import cpython_only, import_helper, isolation
 
 testmeths = [
 
@@ -1014,34 +1014,43 @@ class C:
         C.a = X()
 
     @support.nomemtest
+    @isolation.runInSubprocess()
     def test_detach_materialized_dict_no_memory(self):
-        code = """if 1:
-            import test.support
-            import _testcapi
-
-            class A:
-                def __init__(self):
-                    self.a = 1
-                    self.b = 2
+        import _testcapi
+
+        class A:
+            def __init__(self):
+                self.a = 1
+                self.b = 2
+
+        # The failing allocation should be the one which detaches the
+        # dictionary from the object, but other allocations can happen
+        # first, so try to fail every one of the first allocations.
+        raised = False
+        for n in range(20):
             a = A()
             d = a.__dict__
-
-            test.support.gc_collect()
-            with test.support.catch_unraisable_exception() as ex:
-                _testcapi.set_nomemory(0, 1)
-                del a
-                assert ex.unraisable.exc_type is MemoryError
             try:
-                d["a"]
-            except KeyError:
-                pass
-            else:
-                assert False, "KeyError not raised"
-        """
-        rc, out, err = script_helper.assert_python_ok("-c", code)
-        self.assertEqual(rc, 0)
-        self.assertFalse(out, msg=out.decode('utf-8'))
-        self.assertFalse(err, msg=err.decode('utf-8'))
+                with support.catch_unraisable_exception() as ex:
+                    _testcapi.set_nomemory(n, n + 1)
+                    try:
+                        del a
+                    finally:
+                        _testcapi.remove_mem_hooks()
+                    exc_type = ex.unraisable and ex.unraisable.exc_type
+            except MemoryError:
+                # The failing allocation was not in the deallocation code.
+                continue
+            if exc_type is not MemoryError:
+                continue
+            raised = True
+            if "a" not in d:
+                # The dictionary was cleared, as expected.
+                break
+        else:
+            if not raised:
+                self.fail("MemoryError was not raised during deallocation")
+            self.fail("the dictionary was not cleared")
 
 if __name__ == '__main__':
     unittest.main()