gh-141863: Use bytearray.take_bytes in asyncio.streams

Uses gh-139871 to improve performance over 10% on asyncio_tcp
pyperformance benchmark. All the patterns optimize here are in the
`take_bytes` Python 3.15 What's New entry.

tb_base.json
============

Performance version: 1.13.0
Python version: 3.15.0a2+ (64-bit) revision 227b9d3
Report on Linux-6.17.8-arch1-1-x86_64-with-glibc2.42
Number of logical CPUs: 32
Start date: 2025-11-22 16:18:13.127794
End date: 2025-11-22 16:18:41.207577

tb_asyncio.json
===============

Performance version: 1.13.0
Python version: 3.15.0a2+ (64-bit) revision 6982581e422
Report on Linux-6.17.8-arch1-1-x86_64-with-glibc2.42
Number of logical CPUs: 32
Start date: 2025-11-22 16:13:46.913998
End date: 2025-11-22 16:14:14.140152

Mean +- std dev: 169 ms +- 2 ms -> 143 ms +- 6 ms: 1.18x faster
Significant (t=18.89)

Mean +- std dev: 546 ms +- 8 ms -> 533 ms +- 6 ms: 1.02x faster
Significant (t=6.02)

Author: cmaloney

Lib/asyncio/streams.py

@@ -667,8 +667,7 @@ async def readuntil(self, separator=b'\n'):
             # adds data which makes separator be found. That's why we check for
             # EOF *after* inspecting the buffer.
             if self._eof:
-                chunk = bytes(self._buffer)
-                self._buffer.clear()
+                chunk = self._buffer.take_bytes()
                 raise exceptions.IncompleteReadError(chunk, None)
 
             # _wait_for_data() will resume reading if stream was paused.
@@ -678,10 +677,9 @@ async def readuntil(self, separator=b'\n'):
             raise exceptions.LimitOverrunError(
                 'Separator is found, but chunk is longer than limit', match_start)
 
-        chunk = self._buffer[:match_end]
-        del self._buffer[:match_end]
+        chunk = self._buffer.take_bytes(match_end)
         self._maybe_resume_transport()
-        return bytes(chunk)
+        return chunk
 
     async def read(self, n=-1):
         """Read up to `n` bytes from the stream.
@@ -716,20 +714,16 @@ async def read(self, n=-1):
             # collect everything in self._buffer, but that would
             # deadlock if the subprocess sends more than self.limit
             # bytes.  So just call self.read(self._limit) until EOF.
-            blocks = []
-            while True:
-                block = await self.read(self._limit)
-                if not block:
-                    break
-                blocks.append(block)
-            return b''.join(blocks)
+            joined = bytearray()
+            while block := await self.read(self._limit):
+                joined += block
+            return joined.take_bytes()
 
         if not self._buffer and not self._eof:
             await self._wait_for_data('read')
 
         # This will work right even if buffer is less than n bytes
-        data = bytes(memoryview(self._buffer)[:n])
-        del self._buffer[:n]
+        data = self._buffer.take_bytes(min(len(self._buffer), n))
 
         self._maybe_resume_transport()
         return data
@@ -760,18 +754,12 @@ async def readexactly(self, n):
 
         while len(self._buffer) < n:
             if self._eof:
-                incomplete = bytes(self._buffer)
-                self._buffer.clear()
+                incomplete = self._buffer.take_bytes()
                 raise exceptions.IncompleteReadError(incomplete, n)
 
             await self._wait_for_data('readexactly')
 
-        if len(self._buffer) == n:
-            data = bytes(self._buffer)
-            self._buffer.clear()
-        else:
-            data = bytes(memoryview(self._buffer)[:n])
-            del self._buffer[:n]
+        data = self._buffer.take_bytes(n)
         self._maybe_resume_transport()
         return data