gh-159500: Speed up asyncio StreamReader.readline() and readuntil()

[cmaloney]

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

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 code

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 noreply@anthropic.com

• Issue: Improve performance of asyncio StreamReader .readline() and .readuntil() #159500

[maurycy]

@cmaloney Side note: while reading this one, I noticed two similar spots in io:

• cpython/Lib/_pyio.py

  Lines 2718 to 2723 in 21a36aa

  	else:
  	# non-universal
  	pos = line.find(self._readnl)
  	if pos >= 0:
  	endpos = pos + len(self._readnl)
  	break

• cpython/Modules/_io/iobase.c

  Lines 652 to 671 in 9028df3

  	if (PyBytes_GET_SIZE(readahead) > 0) {
  	Py_ssize_t n = 0;
  	const char *buf = PyBytes_AS_STRING(readahead);
  	if (limit >= 0) {
  	do {
  	if (n >= PyBytes_GET_SIZE(readahead) || n >= limit)
  	break;
  	if (buf[n++] == '\n')
  	break;
  	} while (1);
  	}
  	else {
  	do {
  	if (n >= PyBytes_GET_SIZE(readahead))
  	break;
  	if (buf[n++] == '\n')
  	break;
  	} while (1);
  	}
  	nreadahead = n;

Do you think they're worth it?