pythongh-157025: Fix transport cleanup when cancelling asyncio native sendfile (python#157027)

Author: lpyu001

Lib/asyncio/proactor_events.py

@@ -766,8 +766,8 @@ 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()
         try:
+            await transp._make_empty_waiter()
             return await self.sock_sendfile(transp._sock, file, offset, count,
                                             fallback=False)
         finally:

Lib/asyncio/selector_events.py

@@ -739,8 +739,8 @@ 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()
         try:
+            await transp._make_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.