gh-79033: Try to fix asyncio.Server.wait_closed() again (GH-111336)

* Try to fix asyncio.Server.wait_closed() again

I identified the condition that `wait_closed()` is intended
to wait for: the server is closed *and* there are no more
active connections.

When this condition first becomes true, `_wakeup()` is called
(either from `close()` or from `_detach()`) and it sets `_waiters`
to `None`. So we just check for `self._waiters is None`; if it's
not `None`, we know we have to wait, and do so.

A problem was that the new test introduced in 3.12 explicitly
tested that `wait_closed()` returns immediately when the server
is *not* closed but there are currently no active connections.
This was a mistake (probably a misunderstanding of the intended
semantics). I've fixed the test, and added a separate test that
checks exactly for this scenario.

I also fixed an oddity where in `_wakeup()` the result of the
waiter was set to the waiter itself. This result is not used
anywhere and I changed this to `None`, to avoid a GC cycle.

* Update Lib/asyncio/base_events.py

---------

Co-authored-by: Carol Willing <carolcode@willingconsulting.com>

Lib/test/test_asyncio/test_server.py

@@ -122,29 +122,59 @@ class SelectorStartServerTests(BaseStartServer, unittest.TestCase):
 
 class TestServer2(unittest.IsolatedAsyncioTestCase):
 
-    async def test_wait_closed(self):
+    async def test_wait_closed_basic(self):
         async def serve(*args):
             pass
 
         srv = await asyncio.start_server(serve, socket_helper.HOSTv4, 0)
+        self.addCleanup(srv.close)
 
-        # active count = 0
+        # active count = 0, not closed: should block
         task1 = asyncio.create_task(srv.wait_closed())
         await asyncio.sleep(0)
-        self.assertTrue(task1.done())
+        self.assertFalse(task1.done())
 
-        # active count != 0
+        # active count != 0, not closed: should block
         srv._attach()
         task2 = asyncio.create_task(srv.wait_closed())
         await asyncio.sleep(0)
+        self.assertFalse(task1.done())
         self.assertFalse(task2.done())
 
         srv.close()
         await asyncio.sleep(0)
+        # active count != 0, closed: should block
+        task3 = asyncio.create_task(srv.wait_closed())
+        await asyncio.sleep(0)
+        self.assertFalse(task1.done())
         self.assertFalse(task2.done())
+        self.assertFalse(task3.done())
 
         srv._detach()
+        # active count == 0, closed: should unblock
+        await task1
         await task2
+        await task3
+        await srv.wait_closed()  # Return immediately
+
+    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
+
+        srv = await asyncio.start_server(serve, socket_helper.HOSTv4, 0)
+        self.addCleanup(srv.close)
+
+        task = asyncio.create_task(srv.wait_closed())
+        await asyncio.sleep(0)
+        self.assertFalse(task.done())
+        srv._attach()
+        loop = asyncio.get_running_loop()
+        loop.call_soon(srv.close)
+        loop.call_soon(srv._detach)
+        await srv.wait_closed()
+
+
 
 
 @unittest.skipUnless(hasattr(asyncio, 'ProactorEventLoop'), 'Windows only')