Skip to content

Commit 99d382c

Browse files
authored
gh-159500: Speed up asyncio StreamReader.readline() and readuntil() (#159505)
1 parent b84e0b7 commit 99d382c

3 files changed

Lines changed: 49 additions & 36 deletions

File tree

‎Lib/asyncio/streams.py‎

Lines changed: 37 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -553,6 +553,16 @@ async def _wait_for_data(self, func_name):
553553
finally:
554554
self._waiter = None
555555

556+
def _readuntil_fast(self, separator):
557+
# Fast linear search inside already buffered data.
558+
if type(separator) is bytes and separator and self._exception is None:
559+
isep = self._buffer.find(separator)
560+
if 0 <= isep <= self._limit:
561+
chunk = self._buffer.take_bytes(isep + len(separator))
562+
self._maybe_resume_transport()
563+
return chunk
564+
return None
565+
556566
async def readline(self):
557567
r"""Read chunk of data from the stream until newline (b'\n') is found.
558568
@@ -570,19 +580,18 @@ async def readline(self):
570580
needed.
571581
"""
572582
sep = b'\n'
573-
seplen = len(sep)
583+
if (line := self._readuntil_fast(sep)) is not None:
584+
return line
574585
try:
575-
line = await self.readuntil(sep)
586+
return await self.readuntil(sep)
576587
except exceptions.IncompleteReadError as e:
577588
return e.partial
578589
except exceptions.LimitOverrunError as e:
579-
if self._buffer.startswith(sep, e.consumed):
580-
del self._buffer[:e.consumed + seplen]
581-
else:
582-
self._buffer.clear()
590+
# e.consumed is the newline's index, or len(self._buffer) if
591+
# there is none: drop the line, or everything.
592+
del self._buffer[:e.consumed + 1]
583593
self._maybe_resume_transport()
584594
raise ValueError(e.args[0])
585-
return line
586595

587596
async def readuntil(self, separator=b'\n'):
588597
"""Read data from the stream until ``separator`` is found.
@@ -610,17 +619,18 @@ async def readuntil(self, separator=b'\n'):
610619
the shortest possible separator is considered to be the one that
611620
matched.
612621
"""
622+
if (chunk := self._readuntil_fast(separator)) is not None:
623+
return chunk
613624
if isinstance(separator, tuple):
614625
# Makes sure shortest matches wins
615626
separator = sorted(separator, key=len)
616627
else:
617628
separator = [separator]
618629
if not separator:
619630
raise ValueError('Separator should contain at least one element')
620-
min_seplen = len(separator[0])
621-
max_seplen = len(separator[-1])
622-
if min_seplen == 0:
631+
if len(separator[0]) == 0:
623632
raise ValueError('Separator should be at least one-byte string')
633+
max_seplen = len(separator[-1])
624634

625635
if self._exception is not None:
626636
raise self._exception
@@ -650,32 +660,23 @@ async def readuntil(self, separator=b'\n'):
650660
# or an EOF has happened.
651661
while True:
652662
buflen = len(self._buffer)
653-
654-
# Check if we now have enough data in the buffer for shortest
655-
# separator to fit.
656-
if buflen - offset >= min_seplen:
657-
match_start = None
658-
match_end = None
659-
for sep in separator:
660-
isep = self._buffer.find(sep, offset)
661-
662-
if isep != -1:
663-
# `separator` is in the buffer. `match_start` and
664-
# `match_end` will be used later to retrieve the
665-
# data.
666-
end = isep + len(sep)
667-
if match_end is None or end < match_end:
668-
match_end = end
669-
match_start = isep
670-
if match_end is not None:
671-
break
672-
673-
# see upper comment for explanation.
674-
offset = max(0, buflen + 1 - max_seplen)
675-
if offset > self._limit:
676-
raise exceptions.LimitOverrunError(
677-
'Separator is not found, and chunk exceed the limit',
678-
offset)
663+
match_end = buflen + 1
664+
for sep in separator:
665+
# Only a match that ends before the best one so far wins,
666+
# so don't search past it.
667+
isep = self._buffer.find(sep, offset, match_end - 1)
668+
if isep != -1:
669+
match_start = isep
670+
match_end = isep + len(sep)
671+
if match_end <= buflen:
672+
break
673+
674+
# see upper comment for explanation.
675+
offset = max(0, buflen + 1 - max_seplen)
676+
if offset > self._limit:
677+
raise exceptions.LimitOverrunError(
678+
'Separator is not found, and chunk exceed the limit',
679+
offset)
679680

680681
# Complete message (with full separator) may be present in buffer
681682
# even when EOF flag is set. This may happen when the last chunk

‎Lib/test/test_asyncio/test_streams.py‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -592,6 +592,14 @@ def test_exception(self):
592592
stream.set_exception(exc)
593593
self.assertIs(stream.exception(), exc)
594594

595+
# Raised even when the data is already buffered.
596+
stream.feed_data(b'line\n')
597+
self.assertRaises(
598+
ValueError, self.loop.run_until_complete, stream.readline())
599+
self.assertRaises(
600+
ValueError, self.loop.run_until_complete, stream.readuntil())
601+
self.assertEqual(b'line\n', stream._buffer)
602+
595603
def test_exception_waiter(self):
596604
stream = asyncio.StreamReader(loop=self.loop)
597605

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
Speed up :meth:`asyncio.StreamReader.readline` and
2+
:meth:`asyncio.StreamReader.readuntil` when the data is already buffered,
3+
and stop :meth:`~asyncio.StreamReader.readuntil` with a tuple of separators from
4+
scanning the whole buffer on every call.

0 commit comments

Comments
 (0)