asyncio.protocols._feed_data_to_buffered_proto: avoid O(N^2) bytes slicing

[deadlovelll]

Bug report

Bug description:

asyncio.protocols._feed_data_to_buffered_proto feeds incoming data into a BufferedProtocol reusable buffer. The current multi-iteration path is O(N^2) because it slices data as bytes on every iteration:

# Lib/asyncio/protocols.py:200-216
def _feed_data_to_buffered_proto(proto, data):
    data_len = len(data)
    while data_len:
        buf = proto.get_buffer(data_len)
        buf_len = len(buf)
        if not buf_len:
            raise RuntimeError('get_buffer() returned an empty buffer')

        if buf_len >= data_len:
            buf[:data_len] = data
            proto.buffer_updated(data_len)
            return
        else:
            buf[:buf_len] = data[:buf_len]
            proto.buffer_updated(buf_len)
            data = data[buf_len:]
            data_len = len(data)

data = data[buf_len:] executes N / buf_len times, each time recopying the remaining tail.

Proposed fix

Track the offset into data manually instead of reslicing:

def _feed_data_to_buffered_proto(proto, data):
    data_len = len(data)
    start = 0
    while data_len:
        buf = proto.get_buffer(data_len)
        buf_len = len(buf)
        if not buf_len:
            raise RuntimeError('get_buffer() returned an empty buffer')

        if buf_len >= data_len:
            buf[:data_len] = data[start:start + data_len] if start else data
            proto.buffer_updated(data_len)
            return
        else:
            buf[:buf_len] = data[start:start + buf_len]
            proto.buffer_updated(buf_len)
            start += buf_len
            data_len -= buf_len

Benchmarks

Benchmarks made via pyperf

+-------------------------+---------+-------------------------+
| Benchmark               | old     | new                     |
+=========================+=========+=========================+
| data=   65536 buf= 4096 | 48.7 us | 16.8 us: 2.89x faster   |
+-------------------------+---------+-------------------------+
| data=  262144 buf= 4096 | 637 us  | 67.4 us: 9.45x faster   |
+-------------------------+---------+-------------------------+
| data= 1048576 buf= 4096 | 8.28 ms | 269 us: 30.77x faster   |
+-------------------------+---------+-------------------------+
| data= 4194304 buf= 4096 | 205 ms  | 1.08 ms: 189.36x faster |
+-------------------------+---------+-------------------------+
| data= 1048576 buf=65536 | 741 us  | 237 us: 3.13x faster    |
+-------------------------+---------+-------------------------+
| data=     100 buf= 4096 | 558 ns  | 571 ns: 1.02x slower    |
+-------------------------+---------+-------------------------+
| data=    1024 buf= 4096 | 636 ns  | 641 ns: 1.01x slower    |
+-------------------------+---------+-------------------------+
| data=    4096 buf= 4096 | 812 ns  | 823 ns: 1.01x slower    |
+-------------------------+---------+-------------------------+
| data=   32768 buf=65536 | 2.53 us | 2.55 us: 1.01x slower   |
+-------------------------+---------+-------------------------+
| Geometric mean          | (ref)   | 4.27x faster            |
+-------------------------+---------+-------------------------+

The scenarios when buf_len >= data_len are not regressed. The 1.01-1.02x differences can be treated as statistical noise

Full benchmark code I'll pin in the PR

CPython versions tested on:

CPython main branch

Operating systems tested on:

Windows, macOS

Linked PRs

• gh-150621: avoid quadratic bytes slicing in asyncio.protocols._feed_data_to_buffered_proto #150622

[picnixz]

Can you show us the benchmark script itself?

[deadlovelll]

Can you show us the benchmark script itself?

here it is:

import pyperf
from asyncio import BufferedProtocol


def feed_old(proto, data):
    data_len = len(data)
    while data_len:
        buf = proto.get_buffer(data_len)
        buf_len = len(buf)
        if not buf_len:
            raise RuntimeError('get_buffer() returned an empty buffer')

        if buf_len >= data_len:
            buf[:data_len] = data
            proto.buffer_updated(data_len)
            return
        else:
            buf[:buf_len] = data[:buf_len]
            proto.buffer_updated(buf_len)
            data = data[buf_len:]
            data_len = len(data)


def feed_new(proto, data):
    data_len = len(data)
    start = 0
    while data_len:
        buf = proto.get_buffer(data_len)
        buf_len = len(buf)
        if not buf_len:
            raise RuntimeError('get_buffer() returned an empty buffer')

        if buf_len >= data_len:
            buf[:data_len] = data[start:start + data_len] if start else data
            proto.buffer_updated(data_len)
            return
        else:
            buf[:buf_len] = data[start:start + buf_len]
            proto.buffer_updated(buf_len)
            start += buf_len
            data_len -= buf_len


class Proto(BufferedProtocol):
    def __init__(self, buf_size):
        self._buf = bytearray(buf_size)

    def get_buffer(self, sizehint):
        return self._buf

    def buffer_updated(self, nbytes):
        pass


def bench(loops, feed, data, proto):
    range_it = range(loops)
    t0 = pyperf.perf_counter()
    for _ in range_it:
        feed(proto, data)
    return pyperf.perf_counter() - t0


def add_cmdline_args(cmd, args):
    cmd.extend(("--variant", args.variant))


runner = pyperf.Runner(add_cmdline_args=add_cmdline_args)
runner.argparser.add_argument(
    "--variant", choices=["old", "new"], required=True,
    help="impl to bench",
)
args = runner.parse_args()
feed = feed_old if args.variant == "old" else feed_new

scenarios = [
    (64 * 1024,        4096),
    (256 * 1024,       4096),
    (1024 * 1024,      4096),
    (4 * 1024 * 1024,  4096),
    (1024 * 1024,      65536),

    (100,              4096),    
    (1024,             4096),    
    (4096,             4096),    
    (32 * 1024,        65536),
]

for ds, bs in scenarios:
    proto = Proto(bs)
    data = b"x" * ds
    runner.bench_time_func(f"data={ds:>8} buf={bs:>5}", bench, feed, data, proto)

To get the results of new version run: ./python.exe buffer_memory_view_bench.py --variant new -o new.json
To get the results of old one: ./python.exe buffer_memory_view_bench.py --variant old -o old.json
Get the compare table: ./python.exe -m pyperf compare_to old.json new.json --table

[picnixz]

I'm afraid that this could break objects that don't expose a memoryview-compatible interface. If we cannot take a mv on data, this will fail. Instead, I would suggest remembering start/stop indices and take slice of the data as is.

[picnixz]

Note: I'm not saying it's common, or even accepted, but changing this may not be desirable for downstream users.