Improve performance of asyncio StreamReader .readline() and .readuntil()

Feature or enhancement

asyncio stream readuntil and readline currently execute a lot more instructions to compute their results than necessary. Oftentimes data already exists in the buffer which has the separated. As an example 64kb likely conains all the newline separated HTTP/1.1 headers.

cpython/Lib/asyncio/streams.py

Lines 628 to 697 in fdec4df

    
           # Consume whole buffer except last bytes, which length is 
        
           # one less than max_seplen. Let's check corner cases with 
        
           # separator[-1]='SEPARATOR': 
        
           # * we have received almost complete separator (without last 
        
           #   byte). i.e buffer='some textSEPARATO'. In this case we 
        
           #   can safely consume max_seplen - 1 bytes. 
        
           # * last byte of buffer is first byte of separator, i.e. 
        
           #   buffer='abcdefghijklmnopqrS'. We may safely consume 
        
           #   everything except that last byte, but this require to 
        
           #   analyze bytes of buffer that match partial separator. 
        
           #   This is slow and/or require FSM. For this case our 
        
           #   implementation is not optimal, since require rescanning 
        
           #   of data that is known to not belong to separator. In 
        
           #   real world, separator will not be so long to notice 
        
           #   performance problems. Even when reading MIME-encoded 
        
           #   messages :) 
        
           # `offset` is the number of bytes from the beginning of the buffer 
        
           # where there is no occurrence of any `separator`. 
        
           offset = 0 
        
           # Loop until we find a `separator` in the buffer, exceed the buffer size, 
        
           # 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) 
        
               # Complete message (with full separator) may be present in buffer 
        
               # even when EOF flag is set. This may happen when the last chunk 
        
               # adds data which makes separator be found. That's why we check for 
        
               # EOF *after* inspecting the buffer. 
        
               if self._eof: 
        
                   chunk = self._buffer.take_bytes() 
        
                   raise exceptions.IncompleteReadError(chunk, None) 
        
               # _wait_for_data() will resume reading if stream was paused. 
        
               await self._wait_for_data('readuntil') 
        
           if match_start > self._limit: 
        
               raise exceptions.LimitOverrunError( 
        
                   'Separator is found, but chunk is longer than limit', match_start) 
        
           chunk = self._buffer.take_bytes(match_end) 
        
           self._maybe_resume_transport() 
        
           return chunk

Reduce overheads by adding a fast-path for "separator is already in buffer" inside readuntil. Use the helper in readline to avoid requiring the overhead of another awaited coroutine.

Also fix an issue where readuntil when passed a tuple of separators was searching for every separator all the way to the end of the buffer even when a closer match has already been found. This made separators which never occur force extra full-length scans per separator. Bound by the best match so far instead.

In a small benchmark readline and readuntil get over 1.3x faster.

Has this already been discussed elsewhere?

This is a minor feature, which does not need previous discussion elsewhere

Links to previous discussion of this feature:

No response

Linked PRs

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

Navigation Menu

Repository navigation

Improve performance of asyncio StreamReader `.readline()` and `.readuntil()` #159500

Feature or enhancement

Has this already been discussed elsewhere?

Links to previous discussion of this feature:

Linked PRs

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

	# Consume whole buffer except last bytes, which length is
	# one less than max_seplen. Let's check corner cases with
	# separator[-1]='SEPARATOR':
	# * we have received almost complete separator (without last
	# byte). i.e buffer='some textSEPARATO'. In this case we
	# can safely consume max_seplen - 1 bytes.
	# * last byte of buffer is first byte of separator, i.e.
	# buffer='abcdefghijklmnopqrS'. We may safely consume
	# everything except that last byte, but this require to
	# analyze bytes of buffer that match partial separator.
	# This is slow and/or require FSM. For this case our
	# implementation is not optimal, since require rescanning
	# of data that is known to not belong to separator. In
	# real world, separator will not be so long to notice
	# performance problems. Even when reading MIME-encoded
	# messages :)

	# `offset` is the number of bytes from the beginning of the buffer
	# where there is no occurrence of any `separator`.
	offset = 0

	# Loop until we find a `separator` in the buffer, exceed the buffer size,
	# 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)

	# Complete message (with full separator) may be present in buffer
	# even when EOF flag is set. This may happen when the last chunk
	# adds data which makes separator be found. That's why we check for
	# EOF after inspecting the buffer.
	if self._eof:
	chunk = self._buffer.take_bytes()
	raise exceptions.IncompleteReadError(chunk, None)

	# _wait_for_data() will resume reading if stream was paused.
	await self._wait_for_data('readuntil')

	if match_start > self._limit:
	raise exceptions.LimitOverrunError(
	'Separator is found, but chunk is longer than limit', match_start)

	chunk = self._buffer.take_bytes(match_end)
	self._maybe_resume_transport()
	return chunk

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

Description

Feature or enhancement

Has this already been discussed elsewhere?

Links to previous discussion of this feature:

Linked PRs

Activity

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions

Improve performance of asyncio StreamReader `.readline()` and `.readuntil()` #159500