gh-150621: avoid quadratic bytes slicing in `asyncio.protocols._feed_data_to_buffered_proto` (#150622)

https://github.com/python/cpython/commit/2f4cca5467cfdf9d754a5d737b9e63ea0c1dc66f
commit: 2f4cca5467cfdf9d754a5d737b9e63ea0c1dc66f
branch: main
author: Timofei <[email protected]>
committer: kumaraditya303 <[email protected]>
date: 2026-07-25T13:14:49Z
summary:

gh-150621: avoid quadratic bytes slicing in `asyncio.protocols._feed_data_to_buffered_proto` (#150622)

files:
M Lib/asyncio/protocols.py
M Lib/test/test_asyncio/test_protocols.py

diff --git a/Lib/asyncio/protocols.py b/Lib/asyncio/protocols.py
index 09987b164c66e5..c67a26c9e2986c 100644
--- a/Lib/asyncio/protocols.py
+++ b/Lib/asyncio/protocols.py
@@ -199,6 +199,7 @@ def process_exited(self):
 
 def _feed_data_to_buffered_proto(proto, data):
     data_len = len(data)
+    start = 0
     while data_len:
         buf = proto.get_buffer(data_len)
         buf_len = len(buf)
@@ -206,11 +207,11 @@ def _feed_data_to_buffered_proto(proto, data):
             raise RuntimeError('get_buffer() returned an empty buffer')
 
         if buf_len >= data_len:
-            buf[:data_len] = data
+            buf[:data_len] = data[start:] if start else data
             proto.buffer_updated(data_len)
             return
         else:
-            buf[:buf_len] = data[:buf_len]
+            buf[:buf_len] = data[start:start + buf_len]
             proto.buffer_updated(buf_len)
-            data = data[buf_len:]
-            data_len = len(data)
+            start += buf_len
+            data_len -= buf_len
diff --git a/Lib/test/test_asyncio/test_protocols.py b/Lib/test/test_asyncio/test_protocols.py
index 643199962b8af0..38f1e3fba90cfd 100644
--- a/Lib/test/test_asyncio/test_protocols.py
+++ b/Lib/test/test_asyncio/test_protocols.py
@@ -2,6 +2,7 @@
 from unittest import mock
 
 import asyncio
+from asyncio import protocols
 
 
 def tearDownModule():
@@ -63,5 +64,32 @@ def test_subprocess_protocol(self):
         self.assertNotHasAttr(sp, '__dict__')
 
 
-if __name__ == '__main__':
+class FeedDataToBufferedProtoTests(unittest.TestCase):
+    def _make_proto(self, bufsize):
+        received = bytearray()
+        buf = bytearray(bufsize)
+
+        class P(asyncio.BufferedProtocol):
+            def get_buffer(self, sizehint):
+                return buf
+
+            def buffer_updated(self, nbytes):
+                received.extend(buf[:nbytes])
+
+        return P(), received
+
+    def test_large_multi_iteration(self):
+        proto, received = self._make_proto(64)
+        data = bytes(range(256)) * 16
+        protocols._feed_data_to_buffered_proto(proto, data)
+        self.assertEqual(bytes(received), data)
+
+    def test_memoryview_input(self):
+        proto, received = self._make_proto(64)
+        payload = b"y" * 200
+        protocols._feed_data_to_buffered_proto(proto, memoryview(payload))
+        self.assertEqual(bytes(received), payload)
+
+
+if __name__ == "__main__":
     unittest.main()

_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]