[3.10] GH-100573: Fix server hang caused by os.stat() on named pipe (Windows) (GH-100959) (#101020)

(cherry picked from commit 1bc7a73)

Author: miss-islington

Lib/asyncio/windows_events.py

@@ -366,6 +366,10 @@ def loop_accept_pipe(f=None):
                     return
 
                 f = self._proactor.accept_pipe(pipe)
+            except BrokenPipeError:
+                if pipe and pipe.fileno() != -1:
+                    pipe.close()
+                self.call_soon(loop_accept_pipe)
             except OSError as exc:
                 if pipe and pipe.fileno() != -1:
                     self.call_exception_handler({
@@ -377,6 +381,7 @@ def loop_accept_pipe(f=None):
                 elif self._debug:
                     logger.warning("Accept pipe failed on pipe %r",
                                    pipe, exc_info=True)
+                self.call_soon(loop_accept_pipe)
             except exceptions.CancelledError:
                 if pipe:
                     pipe.close()

Lib/test/test_asyncio/test_windows_events.py

@@ -250,6 +250,46 @@ def test_address_argument_type_error(self):
             proactor.sendto(sock, b'abc', addr=bad_address)
         sock.close()
 
+    def test_client_pipe_stat(self):
+        res = self.loop.run_until_complete(self._test_client_pipe_stat())
+        self.assertEqual(res, 'done')
+
+    async def _test_client_pipe_stat(self):
+        # Regression test for https://github.com/python/cpython/issues/100573
+        ADDRESS = r'\\.\pipe\test_client_pipe_stat-%s' % os.getpid()
+
+        async def probe():
+            # See https://github.com/python/cpython/pull/100959#discussion_r1068533658
+            h = _overlapped.ConnectPipe(ADDRESS)
+            try:
+                _winapi.CloseHandle(_overlapped.ConnectPipe(ADDRESS))
+            except OSError as e:
+                if e.winerror != _overlapped.ERROR_PIPE_BUSY:
+                    raise
+            finally:
+                _winapi.CloseHandle(h)
+
+        with self.assertRaises(FileNotFoundError):
+            await probe()
+
+        [server] = await self.loop.start_serving_pipe(asyncio.Protocol, ADDRESS)
+        self.assertIsInstance(server, windows_events.PipeServer)
+
+        errors = []
+        self.loop.set_exception_handler(lambda _, data: errors.append(data))
+
+        for i in range(5):
+            await self.loop.create_task(probe())
+
+        self.assertEqual(len(errors), 0, errors)
+
+        server.close()
+
+        with self.assertRaises(FileNotFoundError):
+            await probe()
+
+        return "done"
+
 
 class WinPolicyTests(test_utils.TestCase):
 

Misc/NEWS.d/next/Library/2023-01-12-01-18-13.gh-issue-100573.KDskqo.rst

@@ -0,0 +1 @@
+Fix a Windows :mod:`asyncio` bug with named pipes where a client doing ``os.stat()`` on the pipe would cause an error in the server that disabled serving future requests.