gh-159503: Avoid copying the unsent remainder of a bytes write in asyncio selector transports

Doc/library/asyncio-protocol.rst

@@ -333,6 +333,9 @@ Write-only Transports
    This method does not block; it buffers the data and arranges for it
    to be sent out asynchronously.
 
+   The transport keeps a reference to *data* rather than copying. *data*
+   must not be mutated while being sent.
+
 .. method:: WriteTransport.writelines(list_of_data)
 
    Write a list (or any iterable) of data bytes to the transport.

Lib/asyncio/selector_events.py

@@ -1095,7 +1095,9 @@ def write(self, data):
             # Not all was written; register write handler.
             self._add_writer(self._sock_fd, self._write_ready)
 
-        # Add it to the buffer.
+        # Add it to the buffer as a view, so partial sends don't copy it.
+        if not isinstance(data, memoryview):
+            data = memoryview(data)
         self._buffer.append(data)
         self._buffer_size += len(data)
         self._maybe_pause_protocol()

Lib/test/test_asyncio/test_selector_events.py

@@ -855,14 +855,23 @@ def test_write_partial_none(self):
         self.assertEqual(list_to_buffer([b'data']), transport._buffer)
 
     def test_write_tryagain(self):
-        self.sock.send.side_effect = BlockingIOError
-
         data = b'data'
-        transport = self.socket_transport()
-        transport.write(data)
-
-        self.loop.assert_writer(7, transport._write_ready)
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
+        for buf in data, bytearray(data), memoryview(data):
+            with self.subTest(type=type(buf).__name__):
+                self.sock.send.side_effect = BlockingIOError
+
+                transport = self.socket_transport()
+                transport.write(buf)
+
+                self.loop.assert_writer(7, transport._write_ready)
+                self.assertEqual(list_to_buffer([b'data']), transport._buffer)
+                # gh-159503: the data is buffered as a view, not a copy
+                self.assertIsInstance(transport._buffer[0], memoryview)
+                self.assertIs(transport._buffer[0].obj, memoryview(buf).obj)
+                if isinstance(buf, bytearray):
+                    # The buffered view pins the bytearray against resizing.
+                    with self.assertRaises(BufferError):
+                        buf.append(0)
 
     def test_write_sendmsg_no_data(self):
         self.sock.sendmsg = mock.Mock()

Misc/NEWS.d/next/Library/2026-10-16-01-56-15.gh-issue-159503.WgYiow.rst

@@ -0,0 +1,3 @@
+Fix quadratic copying in :mod:`asyncio` selector socket transports on partial
+sends. A :class:`bytearray` passed to :meth:`~asyncio.WriteTransport.write`
+now stays locked against resizing until it has been sent.