_SelectorTransport.get_write_buffer_size() is O(n), causing quadratic behavior under write pressure

[Suor]

Bug report

What happened?

_SelectorTransport.get_write_buffer_size() recalculates the total buffer size on every call by iterating the entire _buffer deque:

def get_write_buffer_size(self):
    return sum(map(len, self._buffer))

This method is called from _maybe_pause_protocol() after every write() / writelines(), and from _maybe_resume_protocol() during _write_ready(). When the transport cannot drain fast enough and chunks accumulate in the buffer, this creates O(n²) behavior — each new write iterates over all previously buffered chunks.

In our production system (aiohttp WebSocket server sending many small binary messages), this caused get_write_buffer_size to consume ~90% CPU under load.

Other transports already use O(1) tracking

_SelectorTransport is the only transport with this problem. Even within the same module, _SelectorDatagramTransport already maintains a running self._buffer_size counter — incrementing on write, decrementing on drain — and returns it directly from get_write_buffer_size() in O(1).

The same O(1) pattern is used in:

• _ProactorBaseWritePipeTransport (self._buffer_size)
• SSLProtocol (self._write_buffer_size)

Suggested fix

Maintain a running total of buffered bytes in _SelectorTransport, consistent with _SelectorDatagramTransport in the same module.

CPython versions tested on

3.13 (but the master branch here has the same code for _SelectorTransport.get_write_buffer_size().

Operating system

Linux

Linked PRs

• gh-146507: Make _SelectorTransport.get_write_buffer_size() O(1) #146518

[Suor]

Here is a workaoround we are using for now. In case anyone interested.

NOTE: it will break if you pass a mutable something to .send_bytes() and when actually mutate it before it was sent. Don't do that ;)

Details

from collections import deque
from functools import cache

from asyncio.selector_events import _SelectorTransport


class _TrackedBuffer(deque):
    def __init__(self, iterable=()):
        deque.__init__(self, iterable)
        self.total_len = sum(len(item) for item in self)

    def append(self, x):
        deque.append(self, x)
        self.total_len += len(x)

    def appendleft(self, x):
        deque.appendleft(self, x)
        self.total_len += len(x)

    def extend(self, iterable):
        for item in iterable:
            deque.append(self, item)
            self.total_len += len(item)

    def popleft(self):
        item = deque.popleft(self)
        self.total_len -= len(item)
        return item

    def clear(self):
        deque.clear(self)
        self.total_len = 0


@cache
def _patch():
    _original_init = _SelectorTransport.__init__

    def _patched_init(self, *args, **kwargs):
        _original_init(self, *args, **kwargs)
        self._buffer = _TrackedBuffer(self._buffer)

    _SelectorTransport.__init__ = _patched_init
    _SelectorTransport.get_write_buffer_size = lambda self: self._buffer.total_len


_patch()