gh-157025: Fix transport cleanup when cancelling asyncio native sendfile

Ensure cancellation while native sendfile waits for the write buffer to drain resets the empty waiter, restores the prior reading state, and restores the selector transport registry entry.

Author: lpyu001

Lib/asyncio/proactor_events.py

@@ -766,8 +766,9 @@ async def _sock_sendfile_native(self, sock, file, offset, count):
     async def _sendfile_native(self, transp, file, offset, count):
         resume_reading = transp.is_reading()
         transp.pause_reading()
-        await transp._make_empty_waiter()
+        empty_waiter = transp._make_empty_waiter()
         try:
+            await empty_waiter
             return await self.sock_sendfile(transp._sock, file, offset, count,
                                             fallback=False)
         finally:

Lib/asyncio/selector_events.py

@@ -739,8 +739,9 @@ async def _sendfile_native(self, transp, file, offset, count):
         del self._transports[transp._sock_fd]
         resume_reading = transp.is_reading()
         transp.pause_reading()
-        await transp._make_empty_waiter()
+        empty_waiter = transp._make_empty_waiter()
         try:
+            await empty_waiter
             return await self.sock_sendfile(transp._sock, file, offset, count,
                                             fallback=False)
         finally:

Lib/test/test_asyncio/test_sendfile.py

@@ -376,6 +376,47 @@ def test_sendfile(self):
         self.assertEqual(srv_proto.data, self.DATA)
         self.assertEqual(self.file.tell(), len(self.DATA))
 
+    def test_sendfile_cancel_empty_waiter(self):
+        for reading in (True, False):
+            with self.subTest(reading=reading):
+                srv_proto, cli_proto = self.prepare_sendfile()
+                transport = cli_proto.transport
+                if not reading:
+                    transport.pause_reading()
+                waiter = self.loop.create_future()
+
+                def make_empty_waiter():
+                    transport._empty_waiter = waiter
+                    return waiter
+
+                with mock.patch.object(transport, '_make_empty_waiter',
+                                       side_effect=make_empty_waiter):
+                    task = self.loop.create_task(
+                        self.loop.sendfile(transport, self.file))
+                    test_utils.run_briefly(self.loop)
+                    self.assertIs(transport._empty_waiter, waiter)
+                    self.assertFalse(waiter.done())
+                    self.assertFalse(transport.is_reading())
+                    task.cancel()
+                    with self.assertRaises(asyncio.CancelledError):
+                        self.run_loop(task)
+
+                try:
+                    self.assertIsNone(transport._empty_waiter)
+                    self.assertEqual(transport.is_reading(), reading)
+                    if isinstance(self.loop, asyncio.SelectorEventLoop):
+                        self.assertIs(
+                            self.loop._transports[transport._sock_fd],
+                            transport)
+                finally:
+                    transport._reset_empty_waiter()
+
+                ret = self.run_loop(self.loop.sendfile(transport, self.file))
+                transport.close()
+                self.run_loop(srv_proto.done)
+                self.assertEqual(ret, len(self.DATA))
+                self.assertEqual(srv_proto.data, self.DATA)
+
     def test_sendfile_force_fallback(self):
         srv_proto, cli_proto = self.prepare_sendfile()
 

Misc/NEWS.d/next/Library/2026-09-06-15-12-18.gh-issue-157025.ZvFpFT.rst

@@ -0,0 +1,2 @@
+Fix transport cleanup when cancelling :meth:`asyncio.loop.sendfile` while
+waiting for the write buffer to drain in the native implementation.