From 29b1613b5205aeb6f55cb7f64bdb3c72488b11b2 Mon Sep 17 00:00:00 2001 From: Cody Maloney Date: Thu, 15 Oct 2026 16:53:45 -0700 Subject: [PATCH] gh-159500: Speed up asyncio StreamReader.readline() and readuntil() Reduce work in `readline` and `readuntil` when the separator is bytes and data is already buffered. When the separator is a tuple, stop each separator's search where the best match sofar ends instead of scanning the rest of the buffer for every separator on every call. Check in test_exception that readline() and readuntil() still raise a stored exception when the data is already buffered.
Performance: readline() 1.8x, readuntil() 1.4x faster on buffered data; tuple separators no longer scan the whole buffer pyperf, PGO+LTO, `--rigorous`, 37-byte lines. The tuple rows use `(b'\n', b'\r\n')`; their speedup grows with the amount buffered (37 KB and 64 KiB here). | Benchmark | base | exp | |---------------------------------|:-------:|:---------------------:| | readline, buffered | 318 ns | 174 ns: 1.82x faster | | readline, 64 KiB chunks | 319 ns | 178 ns: 1.80x faster | | readuntil, buffered | 258 ns | 182 ns: 1.42x faster | | readuntil, 64 KiB chunks | 263 ns | 187 ns: 1.40x faster | | readuntil, tuple, buffered | 11.2 us | 422 ns: 26.50x faster | | readuntil, tuple, 64 KiB chunks | 19.1 us | 430 ns: 44.50x faster | callgrind, instructions per call, buffered: readline() 7,105 -> 3,890, readuntil() 5,843 -> 4,037.
Benchmark ```python import asyncio import time import pyperf async def buffered(loops, name, args): data = (b'x' * 36 + b'\n') * 1000 r = asyncio.StreamReader(limit=1 << 30) rl = getattr(r, name) total = 0.0 for _ in range(loops): r.feed_data(data) t0 = time.perf_counter() for _ in range(1000): await rl(*args) total += time.perf_counter() - t0 return total async def chunked(loops, name, args): # 64 KiB chunks that split lines, fed as the reader waits. chunk = ((b'x' * 36 + b'\n') * 1772)[:65536] r = asyncio.StreamReader(limit=1 << 30) async def feed(): for _ in range(loops): r.feed_data(chunk) await asyncio.sleep(0) r.feed_eof() t0 = time.perf_counter() task = asyncio.create_task(feed()) rl = getattr(r, name) try: while await rl(*args): pass except asyncio.IncompleteReadError: # readuntil() at EOF pass await task return time.perf_counter() - t0 runner = pyperf.Runner() for label, name, args in (('readline', 'readline', ()), ('readuntil', 'readuntil', ()), ('readuntil, tuple', 'readuntil', ((b'\n', b'\r\n'),))): runner.bench_time_func(f'{label}, buffered', lambda n, name=name, args=args: asyncio.run(buffered(n, name, args)), inner_loops=1000) runner.bench_time_func(f'{label}, 64 KiB chunks', lambda n, name=name, args=args: asyncio.run(chunked(n, name, args)), inner_loops=1771) ```
Co-Authored-By: Claude Opus 5.5 --- Lib/asyncio/streams.py | 73 ++++++++++--------- Lib/test/test_asyncio/test_streams.py | 8 ++ ...-10-15-18-38-22.gh-issue-159500.OjAQ8W.rst | 4 + 3 files changed, 49 insertions(+), 36 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2026-10-15-18-38-22.gh-issue-159500.OjAQ8W.rst diff --git a/Lib/asyncio/streams.py b/Lib/asyncio/streams.py index 954e132617ce42b..d57460e4ca30f5e 100644 --- a/Lib/asyncio/streams.py +++ b/Lib/asyncio/streams.py @@ -553,6 +553,16 @@ async def _wait_for_data(self, func_name): finally: self._waiter = None + def _readuntil_fast(self, separator): + # Fast linear search inside already buffered data. + if type(separator) is bytes and separator and self._exception is None: + isep = self._buffer.find(separator) + if 0 <= isep <= self._limit: + chunk = self._buffer.take_bytes(isep + len(separator)) + self._maybe_resume_transport() + return chunk + return None + async def readline(self): r"""Read chunk of data from the stream until newline (b'\n') is found. @@ -570,19 +580,18 @@ async def readline(self): needed. """ sep = b'\n' - seplen = len(sep) + if (line := self._readuntil_fast(sep)) is not None: + return line try: - line = await self.readuntil(sep) + return await self.readuntil(sep) except exceptions.IncompleteReadError as e: return e.partial except exceptions.LimitOverrunError as e: - if self._buffer.startswith(sep, e.consumed): - del self._buffer[:e.consumed + seplen] - else: - self._buffer.clear() + # e.consumed is the newline's index, or len(self._buffer) if + # there is none: drop the line, or everything. + del self._buffer[:e.consumed + 1] self._maybe_resume_transport() raise ValueError(e.args[0]) - return line async def readuntil(self, separator=b'\n'): """Read data from the stream until ``separator`` is found. @@ -610,6 +619,8 @@ async def readuntil(self, separator=b'\n'): the shortest possible separator is considered to be the one that matched. """ + if (chunk := self._readuntil_fast(separator)) is not None: + return chunk if isinstance(separator, tuple): # Makes sure shortest matches wins separator = sorted(separator, key=len) @@ -617,10 +628,9 @@ async def readuntil(self, separator=b'\n'): separator = [separator] if not separator: raise ValueError('Separator should contain at least one element') - min_seplen = len(separator[0]) - max_seplen = len(separator[-1]) - if min_seplen == 0: + if len(separator[0]) == 0: raise ValueError('Separator should be at least one-byte string') + max_seplen = len(separator[-1]) if self._exception is not None: raise self._exception @@ -650,32 +660,23 @@ async def readuntil(self, separator=b'\n'): # or an EOF has happened. while True: buflen = len(self._buffer) - - # Check if we now have enough data in the buffer for shortest - # separator to fit. - if buflen - offset >= min_seplen: - match_start = None - match_end = None - for sep in separator: - isep = self._buffer.find(sep, offset) - - if isep != -1: - # `separator` is in the buffer. `match_start` and - # `match_end` will be used later to retrieve the - # data. - end = isep + len(sep) - if match_end is None or end < match_end: - match_end = end - match_start = isep - if match_end is not None: - break - - # see upper comment for explanation. - offset = max(0, buflen + 1 - max_seplen) - if offset > self._limit: - raise exceptions.LimitOverrunError( - 'Separator is not found, and chunk exceed the limit', - offset) + match_end = buflen + 1 + for sep in separator: + # Only a match that ends before the best one so far wins, + # so don't search past it. + isep = self._buffer.find(sep, offset, match_end - 1) + if isep != -1: + match_start = isep + match_end = isep + len(sep) + if match_end <= buflen: + break + + # see upper comment for explanation. + offset = max(0, buflen + 1 - max_seplen) + if offset > self._limit: + raise exceptions.LimitOverrunError( + 'Separator is not found, and chunk exceed the limit', + offset) # Complete message (with full separator) may be present in buffer # even when EOF flag is set. This may happen when the last chunk diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 172f183849c3057..a49236b9a9ada7d 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -591,6 +591,14 @@ def test_exception(self): stream.set_exception(exc) self.assertIs(stream.exception(), exc) + # Raised even when the data is already buffered. + stream.feed_data(b'line\n') + self.assertRaises( + ValueError, self.loop.run_until_complete, stream.readline()) + self.assertRaises( + ValueError, self.loop.run_until_complete, stream.readuntil()) + self.assertEqual(b'line\n', stream._buffer) + def test_exception_waiter(self): stream = asyncio.StreamReader(loop=self.loop) diff --git a/Misc/NEWS.d/next/Library/2026-10-15-18-38-22.gh-issue-159500.OjAQ8W.rst b/Misc/NEWS.d/next/Library/2026-10-15-18-38-22.gh-issue-159500.OjAQ8W.rst new file mode 100644 index 000000000000000..2cfaec1e5999a6f --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-10-15-18-38-22.gh-issue-159500.OjAQ8W.rst @@ -0,0 +1,4 @@ +Speed up :meth:`asyncio.StreamReader.readline` and +:meth:`asyncio.StreamReader.readuntil` when the data is already buffered, +and stop :meth:`~asyncio.StreamReader.readuntil` with a tuple of separators from +scanning the whole buffer on every call.