From 9f0111ed872091df81ce7239f95581ee96296f51 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Mon, 22 Jan 2024 11:30:24 +0100
Subject: [PATCH 01/13] Don't use internals for wait_closed() tests

---
 Lib/test/test_asyncio/test_server.py | 27 +++++++++++++++++++--------
 1 file changed, 19 insertions(+), 8 deletions(-)

diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py
index 918faac909b9bf..0a49c72f95505f 100644
--- a/Lib/test/test_asyncio/test_server.py
+++ b/Lib/test/test_asyncio/test_server.py
@@ -125,8 +125,12 @@ async def main(srv):
 class TestServer2(unittest.IsolatedAsyncioTestCase):
 
     async def test_wait_closed_basic(self):
-        async def serve(*args):
-            pass
+        async def serve(rd, wr):
+            try:
+                await rd.read()
+            finally:
+                wr.close()
+                await wr.wait_closed()
 
         srv = await asyncio.start_server(serve, socket_helper.HOSTv4, 0)
         self.addCleanup(srv.close)
@@ -137,7 +141,8 @@ async def serve(*args):
         self.assertFalse(task1.done())
 
         # active count != 0, not closed: should block
-        srv._attach()
+        addr = srv.sockets[0].getsockname()
+        (rd, wr) = await asyncio.open_connection(addr[0], addr[1])
         task2 = asyncio.create_task(srv.wait_closed())
         await asyncio.sleep(0)
         self.assertFalse(task1.done())
@@ -152,7 +157,8 @@ async def serve(*args):
         self.assertFalse(task2.done())
         self.assertFalse(task3.done())
 
-        srv._detach()
+        wr.close()
+        await wr.wait_closed()
         # active count == 0, closed: should unblock
         await task1
         await task2
@@ -161,8 +167,12 @@ async def serve(*args):
 
     async def test_wait_closed_race(self):
         # Test a regression in 3.12.0, should be fixed in 3.12.1
-        async def serve(*args):
-            pass
+        async def serve(rd, wr):
+            try:
+                await rd.read()
+            finally:
+                wr.close()
+                await wr.wait_closed()
 
         srv = await asyncio.start_server(serve, socket_helper.HOSTv4, 0)
         self.addCleanup(srv.close)
@@ -170,10 +180,11 @@ async def serve(*args):
         task = asyncio.create_task(srv.wait_closed())
         await asyncio.sleep(0)
         self.assertFalse(task.done())
-        srv._attach()
+        addr = srv.sockets[0].getsockname()
+        (rd, wr) = await asyncio.open_connection(addr[0], addr[1])
         loop = asyncio.get_running_loop()
         loop.call_soon(srv.close)
-        loop.call_soon(srv._detach)
+        loop.call_soon(wr.close)
         await srv.wait_closed()
 
 

From c25eabbca6b3bac6ac9b30c25e20040b199d1733 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Mon, 22 Jan 2024 13:14:22 +0100
Subject: [PATCH 02/13] gh-113538: Allow client connections to be closed

Give applications the option of more forcefully terminating client
connections for asyncio servers. Useful when terminating a service and
there is limited time to wait for clients to finish up their work.
---
 Doc/library/asyncio-eventloop.rst             | 17 ++++++
 Doc/whatsnew/3.13.rst                         |  4 ++
 Lib/asyncio/base_events.py                    | 24 +++++---
 Lib/asyncio/events.py                         |  8 +++
 Lib/asyncio/proactor_events.py                |  4 +-
 Lib/asyncio/selector_events.py                |  4 +-
 Lib/test/test_asyncio/test_server.py          | 58 +++++++++++++++++++
 ...-01-22-15-50-58.gh-issue-113538.v2wrwg.rst |  3 +
 8 files changed, 110 insertions(+), 12 deletions(-)
 create mode 100644 Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst

diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst
index 828e506a72c937..2f02a143cabda2 100644
--- a/Doc/library/asyncio-eventloop.rst
+++ b/Doc/library/asyncio-eventloop.rst
@@ -1641,6 +1641,23 @@ Do not instantiate the :class:`Server` class directly.
       coroutine to wait until the server is closed (and no more
       connections are active).
 
+   .. method:: close_clients()
+
+      Close all existing incoming client connections.
+
+      Calls :meth:`Transport.close` on all associated transports.
+
+      .. versionadded:: 3.13
+
+   .. method:: abort_clients()
+
+      Close all existing incoming client connections immediately,
+      without waiting for pending operations to complete.
+
+      Calls :meth:`Transport.abort` on all associated transports.
+
+      .. versionadded:: 3.13
+
    .. method:: get_loop()
 
       Return the event loop associated with the server object.
diff --git a/Doc/whatsnew/3.13.rst b/Doc/whatsnew/3.13.rst
index 40f0cd37fe9318..8086d4f59e7120 100644
--- a/Doc/whatsnew/3.13.rst
+++ b/Doc/whatsnew/3.13.rst
@@ -168,6 +168,10 @@ asyncio
   the Unix socket when the server is closed.
   (Contributed by Pierre Ossman in :gh:`111246`.)
 
+* Add :meth:`asyncio.Server.close_clients` and
+  :meth:`asyncio.Server.abort_clients` methods which allows to more
+  forcefully close an asyncio server.
+
 copy
 ----
 
diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index a8870b636d1df5..6bd404e3fb7cdb 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -277,7 +277,7 @@ def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,
                  ssl_handshake_timeout, ssl_shutdown_timeout=None):
         self._loop = loop
         self._sockets = sockets
-        self._active_count = 0
+        self._clients = set()
         self._waiters = []
         self._protocol_factory = protocol_factory
         self._backlog = backlog
@@ -290,14 +290,14 @@ def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,
     def __repr__(self):
         return f'<{self.__class__.__name__} sockets={self.sockets!r}>'
 
-    def _attach(self):
+    def _attach(self, transport):
         assert self._sockets is not None
-        self._active_count += 1
+        self._clients.add(transport)
 
-    def _detach(self):
-        assert self._active_count > 0
-        self._active_count -= 1
-        if self._active_count == 0 and self._sockets is None:
+    def _detach(self, transport):
+        assert transport in self._clients
+        self._clients.remove(transport)
+        if len(self._clients) == 0 and self._sockets is None:
             self._wakeup()
 
     def _wakeup(self):
@@ -346,9 +346,17 @@ def close(self):
             self._serving_forever_fut.cancel()
             self._serving_forever_fut = None
 
-        if self._active_count == 0:
+        if len(self._clients) == 0:
             self._wakeup()
 
+    def close_clients(self):
+        for transport in self._clients.copy():
+            transport.close()
+
+    def abort_clients(self):
+        for transport in self._clients.copy():
+            transport.abort()
+
     async def start_serving(self):
         self._start_serving()
         # Skip one loop iteration so that all 'loop.add_reader'
diff --git a/Lib/asyncio/events.py b/Lib/asyncio/events.py
index 072a99fee123c3..73339b151b64eb 100644
--- a/Lib/asyncio/events.py
+++ b/Lib/asyncio/events.py
@@ -173,6 +173,14 @@ def close(self):
         """Stop serving.  This leaves existing connections open."""
         raise NotImplementedError
 
+    def close_clients(self):
+        """Close all active connections."""
+        raise NotImplementedError
+
+    def abort_clients(self):
+        """Close all active connections immediately."""
+        raise NotImplementedError
+
     def get_loop(self):
         """Get the event loop the Server object is attached to."""
         raise NotImplementedError
diff --git a/Lib/asyncio/proactor_events.py b/Lib/asyncio/proactor_events.py
index 1e2a730cf368a9..a646638adaef4b 100644
--- a/Lib/asyncio/proactor_events.py
+++ b/Lib/asyncio/proactor_events.py
@@ -63,7 +63,7 @@ def __init__(self, loop, sock, protocol, waiter=None,
         self._called_connection_lost = False
         self._eof_written = False
         if self._server is not None:
-            self._server._attach()
+            self._server._attach(self)
         self._loop.call_soon(self._protocol.connection_made, self)
         if waiter is not None:
             # only wake up the waiter when connection_made() has been called
@@ -167,7 +167,7 @@ def _call_connection_lost(self, exc):
             self._sock = None
             server = self._server
             if server is not None:
-                server._detach()
+                server._detach(self)
                 self._server = None
             self._called_connection_lost = True
 
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
index dcd5e0aa345029..4bc46153be8384 100644
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -787,7 +787,7 @@ def __init__(self, loop, sock, protocol, extra=None, server=None):
         self._paused = False  # Set when pause_reading() called
 
         if self._server is not None:
-            self._server._attach()
+            self._server._attach(self)
         loop._transports[self._sock_fd] = self
 
     def __repr__(self):
@@ -902,7 +902,7 @@ def _call_connection_lost(self, exc):
             self._loop = None
             server = self._server
             if server is not None:
-                server._detach()
+                server._detach(self)
                 self._server = None
 
     def get_write_buffer_size(self):
diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py
index 0a49c72f95505f..10fffa3a79ae5d 100644
--- a/Lib/test/test_asyncio/test_server.py
+++ b/Lib/test/test_asyncio/test_server.py
@@ -187,7 +187,65 @@ async def serve(rd, wr):
         loop.call_soon(wr.close)
         await srv.wait_closed()
 
+    async def test_close_clients(self):
+        async def serve(rd, wr):
+            try:
+                await rd.read()
+            finally:
+                wr.close()
+                await wr.wait_closed()
+
+        srv = await asyncio.start_server(serve, socket_helper.HOSTv4, 0)
+        self.addCleanup(srv.close)
+
+        addr = srv.sockets[0].getsockname()
+        (rd, wr) = await asyncio.open_connection(addr[0], addr[1])
+        self.addCleanup(wr.close)
+
+        task = asyncio.create_task(srv.wait_closed())
+        await asyncio.sleep(0)
+        self.assertFalse(task.done())
 
+        srv.close()
+        srv.close_clients()
+        await asyncio.sleep(0.1) # FIXME: flush call_soon()?
+        self.assertTrue(task.done())
+
+    async def test_abort_clients(self):
+        async def serve(rd, wr):
+            nonlocal s_rd, s_wr
+            s_rd = rd
+            s_wr = wr
+            await wr.wait_closed()
+
+        s_rd = s_wr = None
+        srv = await asyncio.start_server(serve, socket_helper.HOSTv4, 0)
+        self.addCleanup(srv.close)
+
+        addr = srv.sockets[0].getsockname()
+        (c_rd, c_wr) = await asyncio.open_connection(addr[0], addr[1])
+        self.addCleanup(c_wr.close)
+
+        # Make sure both sides are in a paused state
+        while (s_wr.transport.get_write_buffer_size() == 0 or
+               c_wr.transport.is_reading()):
+            while s_wr.transport.get_write_buffer_size() == 0:
+                s_wr.write(b'a' * 65536)
+                await asyncio.sleep(0)
+            await asyncio.sleep(0.1) # FIXME: More socket buffer space magically appears?
+
+        task = asyncio.create_task(srv.wait_closed())
+        await asyncio.sleep(0)
+        self.assertFalse(task.done())
+
+        # Sanity check
+        self.assertNotEqual(s_wr.transport.get_write_buffer_size(), 0)
+        self.assertFalse(c_wr.transport.is_reading())
+
+        srv.close()
+        srv.abort_clients()
+        await asyncio.sleep(0.1) # FIXME: flush call_soon()?
+        self.assertTrue(task.done())
 
 
 # Test the various corner cases of Unix server socket removal
diff --git a/Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst b/Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst
new file mode 100644
index 00000000000000..4af9bee0796afd
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst
@@ -0,0 +1,3 @@
+Add :meth:`asyncio.Server.close_clients` and
+:meth:`asyncio.Server.abort_clients` methods which allows to more forcefully
+close an asyncio server.

From b7fa1989d2466d56a1190b2412a9d8eb162bc6f8 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Mon, 22 Jan 2024 16:02:46 +0100
Subject: [PATCH 03/13] Fix method references in new documentation

---
 Doc/library/asyncio-eventloop.rst | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst
index 2f02a143cabda2..29ad90b6141b1b 100644
--- a/Doc/library/asyncio-eventloop.rst
+++ b/Doc/library/asyncio-eventloop.rst
@@ -1645,7 +1645,8 @@ Do not instantiate the :class:`Server` class directly.
 
       Close all existing incoming client connections.
 
-      Calls :meth:`Transport.close` on all associated transports.
+      Calls :meth:`~asyncio.BaseTransport.close` on all associated
+      transports.
 
       .. versionadded:: 3.13
 
@@ -1654,7 +1655,8 @@ Do not instantiate the :class:`Server` class directly.
       Close all existing incoming client connections immediately,
       without waiting for pending operations to complete.
 
-      Calls :meth:`Transport.abort` on all associated transports.
+      Calls :meth:`~asyncio.WriteTransport.abort` on all associated
+      transports.
 
       .. versionadded:: 3.13
 

From b3cd9c16363588aa381fa4b553c28cb0bbe9eb1c Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Tue, 23 Jan 2024 10:57:55 +0100
Subject: [PATCH 04/13] Add attribution

---
 Doc/whatsnew/3.13.rst | 1 +
 1 file changed, 1 insertion(+)

diff --git a/Doc/whatsnew/3.13.rst b/Doc/whatsnew/3.13.rst
index 8086d4f59e7120..d5141edff7dee9 100644
--- a/Doc/whatsnew/3.13.rst
+++ b/Doc/whatsnew/3.13.rst
@@ -171,6 +171,7 @@ asyncio
 * Add :meth:`asyncio.Server.close_clients` and
   :meth:`asyncio.Server.abort_clients` methods which allows to more
   forcefully close an asyncio server.
+  (Contributed by Pierre Ossman in :gh:`113538`.)
 
 copy
 ----

From c78a927c5c1de02f2bfd26cce7ec3301046bb097 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Fri, 2 Feb 2024 16:39:19 +0100
Subject: [PATCH 05/13] Only keep a weak client references

We want to be able to detect if the application fails to keep track of
the transports, so we cannot keep them alive by using a hard reference.
---
 Lib/asyncio/base_events.py | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index 6bd404e3fb7cdb..7e09bc2915233a 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -277,7 +277,8 @@ def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,
                  ssl_handshake_timeout, ssl_shutdown_timeout=None):
         self._loop = loop
         self._sockets = sockets
-        self._clients = set()
+        # Weak references so abandoned transports can be detected
+        self._clients = weakref.WeakSet()
         self._waiters = []
         self._protocol_factory = protocol_factory
         self._backlog = backlog
@@ -295,8 +296,10 @@ def _attach(self, transport):
         self._clients.add(transport)
 
     def _detach(self, transport):
-        assert transport in self._clients
-        self._clients.remove(transport)
+        # Note that 'transport' may already be missing from
+        # self._clients if it has been garbage collected
+        if transport in self._clients:
+            self._clients.remove(transport)
         if len(self._clients) == 0 and self._sockets is None:
             self._wakeup()
 

From 1ec06da9d744053b90c07f08d78a2158b48b450a Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Fri, 2 Feb 2024 16:40:50 +0100
Subject: [PATCH 06/13] Inform server of fallback transport cleanup

The application might be waiting for all transports to close, so we need
to properly inform the server that this transport is done.
---
 Lib/asyncio/selector_events.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
index 4bc46153be8384..772b87e947fcc8 100644
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -864,6 +864,8 @@ def __del__(self, _warn=warnings.warn):
         if self._sock is not None:
             _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
             self._sock.close()
+            if self._server is not None:
+                self._server._detach(self)
 
     def _fatal_error(self, exc, message='Fatal error on transport'):
         # Should be called from exception handler only.

From 2790ddf3348fa66361843219f8e47e92bf659e05 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Mon, 19 Feb 2024 13:26:59 +0100
Subject: [PATCH 07/13] Clarify recomended order of methods

---
 Doc/library/asyncio-eventloop.rst | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst
index 29ad90b6141b1b..0fd27509375815 100644
--- a/Doc/library/asyncio-eventloop.rst
+++ b/Doc/library/asyncio-eventloop.rst
@@ -1648,6 +1648,9 @@ Do not instantiate the :class:`Server` class directly.
       Calls :meth:`~asyncio.BaseTransport.close` on all associated
       transports.
 
+      :meth:`close` should be called before :meth:`close_clients` when
+      closing the server to avoid races with new clients connecting.
+
       .. versionadded:: 3.13
 
    .. method:: abort_clients()
@@ -1658,6 +1661,9 @@ Do not instantiate the :class:`Server` class directly.
       Calls :meth:`~asyncio.WriteTransport.abort` on all associated
       transports.
 
+      :meth:`close` should be called before :meth:`abort_clients` when
+      closing the server to avoid races with new clients connecting.
+
       .. versionadded:: 3.13
 
    .. method:: get_loop()

From 6a56a80bce06e824dc08a9e33e2348a09dbdc535 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Mon, 19 Feb 2024 13:27:26 +0100
Subject: [PATCH 08/13] Fix typos

---
 Doc/whatsnew/3.13.rst                                           | 2 +-
 .../next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/Doc/whatsnew/3.13.rst b/Doc/whatsnew/3.13.rst
index d5141edff7dee9..d3b9b847b8c0d8 100644
--- a/Doc/whatsnew/3.13.rst
+++ b/Doc/whatsnew/3.13.rst
@@ -169,7 +169,7 @@ asyncio
   (Contributed by Pierre Ossman in :gh:`111246`.)
 
 * Add :meth:`asyncio.Server.close_clients` and
-  :meth:`asyncio.Server.abort_clients` methods which allows to more
+  :meth:`asyncio.Server.abort_clients` methods which allow to more
   forcefully close an asyncio server.
   (Contributed by Pierre Ossman in :gh:`113538`.)
 
diff --git a/Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst b/Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst
index 4af9bee0796afd..5c59af98e136bb 100644
--- a/Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst
+++ b/Misc/NEWS.d/next/Library/2024-01-22-15-50-58.gh-issue-113538.v2wrwg.rst
@@ -1,3 +1,3 @@
 Add :meth:`asyncio.Server.close_clients` and
-:meth:`asyncio.Server.abort_clients` methods which allows to more forcefully
+:meth:`asyncio.Server.abort_clients` methods which allow to more forcefully
 close an asyncio server.

From 69298883c83b4505a19aef8bef683f3ef6ec932f Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Mon, 19 Feb 2024 13:27:31 +0100
Subject: [PATCH 09/13] Use discard() instead of conditional remove()

---
 Lib/asyncio/base_events.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index 7e09bc2915233a..e09c263375f9eb 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -298,8 +298,7 @@ def _attach(self, transport):
     def _detach(self, transport):
         # Note that 'transport' may already be missing from
         # self._clients if it has been garbage collected
-        if transport in self._clients:
-            self._clients.remove(transport)
+        self._clients.discard(transport)
         if len(self._clients) == 0 and self._sockets is None:
             self._wakeup()
 

From 831619942e9619c270ddb65d7e55ed158e5b064e Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Tue, 27 Feb 2024 12:52:01 +0100
Subject: [PATCH 10/13] Improve some comments

One could be made clearar, and the other is probably superfluous.
---
 Lib/asyncio/base_events.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index e09c263375f9eb..8afe14a04a79c0 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -277,7 +277,8 @@ def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,
                  ssl_handshake_timeout, ssl_shutdown_timeout=None):
         self._loop = loop
         self._sockets = sockets
-        # Weak references so abandoned transports can be detected
+        # Weak references so we don't break Transport's ability to
+        # detect abandoned transports
         self._clients = weakref.WeakSet()
         self._waiters = []
         self._protocol_factory = protocol_factory
@@ -296,8 +297,6 @@ def _attach(self, transport):
         self._clients.add(transport)
 
     def _detach(self, transport):
-        # Note that 'transport' may already be missing from
-        # self._clients if it has been garbage collected
         self._clients.discard(transport)
         if len(self._clients) == 0 and self._sockets is None:
             self._wakeup()

From 3e1705b27c9b6202a1980403b42e63995e4cfda5 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Tue, 27 Feb 2024 14:08:51 +0100
Subject: [PATCH 11/13] Do multiple sleeps to flush out callbacks

---
 Lib/test/test_asyncio/test_server.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py
index 10fffa3a79ae5d..f1717919cbabb8 100644
--- a/Lib/test/test_asyncio/test_server.py
+++ b/Lib/test/test_asyncio/test_server.py
@@ -208,7 +208,8 @@ async def serve(rd, wr):
 
         srv.close()
         srv.close_clients()
-        await asyncio.sleep(0.1) # FIXME: flush call_soon()?
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
         self.assertTrue(task.done())
 
     async def test_abort_clients(self):
@@ -244,7 +245,8 @@ async def serve(rd, wr):
 
         srv.close()
         srv.abort_clients()
-        await asyncio.sleep(0.1) # FIXME: flush call_soon()?
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
         self.assertTrue(task.done())
 
 

From 6c078d67811889ae6416656ef74144a2e626a183 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Tue, 27 Feb 2024 15:43:00 +0100
Subject: [PATCH 12/13] More deterministic abort_clients() test

Try to get the streams and the kernel in to a more deterministic state
by specifying fixed buffering limits.
---
 Lib/test/test_asyncio/test_server.py | 30 +++++++++++++++++-----------
 1 file changed, 18 insertions(+), 12 deletions(-)

diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py
index f1717919cbabb8..15721eb7208de6 100644
--- a/Lib/test/test_asyncio/test_server.py
+++ b/Lib/test/test_asyncio/test_server.py
@@ -224,25 +224,31 @@ async def serve(rd, wr):
         self.addCleanup(srv.close)
 
         addr = srv.sockets[0].getsockname()
-        (c_rd, c_wr) = await asyncio.open_connection(addr[0], addr[1])
+        (c_rd, c_wr) = await asyncio.open_connection(addr[0], addr[1], limit=4096)
         self.addCleanup(c_wr.close)
 
-        # Make sure both sides are in a paused state
-        while (s_wr.transport.get_write_buffer_size() == 0 or
-               c_wr.transport.is_reading()):
-            while s_wr.transport.get_write_buffer_size() == 0:
-                s_wr.write(b'a' * 65536)
-                await asyncio.sleep(0)
-            await asyncio.sleep(0.1) # FIXME: More socket buffer space magically appears?
+        # Limit the send buffer so we can reliably overfill it
+        s_sock = s_wr.get_extra_info('socket')
+        s_sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)
+
+        # Get the reader in to a paused state by sending more than twice
+        # the configured limit
+        s_wr.write(b'a' * 4096)
+        s_wr.write(b'a' * 4096)
+        s_wr.write(b'a' * 4096)
+        while c_wr.transport.is_reading():
+            await asyncio.sleep(0)
+
+        # Get the writer in a waiting state by sending data until the
+        # kernel stops accepting more in to the send buffer
+        while s_wr.transport.get_write_buffer_size() == 0:
+            s_wr.write(b'a' * 4096)
+            await asyncio.sleep(0)
 
         task = asyncio.create_task(srv.wait_closed())
         await asyncio.sleep(0)
         self.assertFalse(task.done())
 
-        # Sanity check
-        self.assertNotEqual(s_wr.transport.get_write_buffer_size(), 0)
-        self.assertFalse(c_wr.transport.is_reading())
-
         srv.close()
         srv.abort_clients()
         await asyncio.sleep(0)

From 1158151d212afb03c0ab58f267b67178a2158454 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Tue, 5 Mar 2024 17:11:41 +0100
Subject: [PATCH 13/13] Even more deterministic

No possibly infinite loop. Instead ask the system how much buffer space
it has and fill that.
---
 Lib/test/test_asyncio/test_server.py | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py
index 15721eb7208de6..0c55661bfb88f4 100644
--- a/Lib/test/test_asyncio/test_server.py
+++ b/Lib/test/test_asyncio/test_server.py
@@ -227,9 +227,11 @@ async def serve(rd, wr):
         (c_rd, c_wr) = await asyncio.open_connection(addr[0], addr[1], limit=4096)
         self.addCleanup(c_wr.close)
 
-        # Limit the send buffer so we can reliably overfill it
+        # Limit the socket buffers so we can reliably overfill them
         s_sock = s_wr.get_extra_info('socket')
         s_sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)
+        c_sock = c_wr.get_extra_info('socket')
+        c_sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)
 
         # Get the reader in to a paused state by sending more than twice
         # the configured limit
@@ -240,10 +242,11 @@ async def serve(rd, wr):
             await asyncio.sleep(0)
 
         # Get the writer in a waiting state by sending data until the
-        # kernel stops accepting more in to the send buffer
-        while s_wr.transport.get_write_buffer_size() == 0:
-            s_wr.write(b'a' * 4096)
-            await asyncio.sleep(0)
+        # socket buffers are full on both server and client sockets and
+        # the kernel stops accepting more data
+        s_wr.write(b'a' * c_sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF))
+        s_wr.write(b'a' * s_sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF))
+        self.assertNotEqual(s_wr.transport.get_write_buffer_size(), 0)
 
         task = asyncio.create_task(srv.wait_closed())
         await asyncio.sleep(0)