Asyncio BufferedProtocol with SSL is significantly slower than asyncio sockets with SSL

Bug report

Bug description:

With SSL enabled, asyncio.BufferedProtocol is significantly slower than using sockets:

$ python clients.py
Python: 3.13.0 (main, Oct 14 2024, 11:12:17) [Clang 15.0.0 (clang-1500.3.9.4)]
Running 100 trials with message size 100,000,000 bytes
Sockets: 10.36 seconds
Protocols: 17.50 seconds

Reproducible example:

shared.py:

MESSAGE_SIZE = 1_000_000 * 100
MESSAGE = b"a" * MESSAGE_SIZE 

HOST = "127.0.0.1"
PORT = 1234

server.py:

import socket
import ssl
from shared import MESSAGE, MESSAGE_SIZE, HOST, PORT


context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.verify_mode = ssl.CERT_NONE
context.load_cert_chain(certfile="cert.pem", keyfile="cert.pem")

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s = context.wrap_socket(s, server_side=True)

s.bind((HOST, PORT))
s.listen()

while True:
    conn, _= s.accept()
    bytes_read = 0

    mv = memoryview(bytearray(MESSAGE_SIZE))
    while bytes_read < MESSAGE_SIZE:
        read = conn.recv_into(mv[bytes_read:])
        if read == 0:
            raise OSError("Closed by peer")
        bytes_read += read
    conn.sendall(MESSAGE)
    conn.close()

clients.py:

import socket
import sys
import asyncio
import ssl
import timeit

from shared import MESSAGE, MESSAGE_SIZE, HOST, PORT


TRIALS=100

context = ssl.SSLContext()
context.verify_mode = ssl.CERT_NONE
context.check_hostname = False

class Protocol(asyncio.BufferedProtocol):
    def __init__(self):
        super().__init__()
        self._buffer = memoryview(bytearray(MESSAGE_SIZE))
        self._offset = 0
        self._done = None
        self._loop = asyncio.get_running_loop()

    def connection_made(self, transport):
        self.transport = transport
        self.transport.set_write_buffer_limits(MESSAGE_SIZE, MESSAGE_SIZE)

    async def write(self, message: bytes):
        self.transport.write(message)

    async def read(self):
        self._done = self._loop.create_future()
        await self._done

    def get_buffer(self, sizehint: int):
        return self._buffer[self._offset:]

    def buffer_updated(self, nbytes: int):
        if self._done and not self._done.done():
            self._offset += nbytes
            if self._offset == MESSAGE_SIZE:
                self._done.set_result(True)

    def data(self):
        return self._buffer

async def _async_socket_sendall_ssl(
    sock: ssl.SSLSocket, buf: bytes, loop: asyncio.AbstractEventLoop
) -> None:
    view = memoryview(buf)
    sent = 0

    def _is_ready(fut: asyncio.Future) -> None:
        if fut.done():
            return
        fut.set_result(None)

    while sent < len(buf):
        try:
            sent += sock.send(view[sent:])
        except (ssl.SSLWantReadError, ssl.SSLWantWriteError) as exc:
            fd = sock.fileno()
            # Check for closed socket.
            if fd == -1:
                raise ssl.SSLError("Underlying socket has been closed") from None
            if isinstance(exc, ssl.SSLWantReadError):
                fut = loop.create_future()
                loop.add_reader(fd, _is_ready, fut)
                try:
                    await fut
                finally:
                    loop.remove_reader(fd)
            if isinstance(exc, ssl.SSLWantWriteError):
                fut = loop.create_future()
                loop.add_writer(fd, _is_ready, fut)
                try:
                    await fut
                finally:
                    loop.remove_writer(fd)

async def _async_socket_receive_ssl(
    conn: ssl.SSLSocket, length: int, loop: asyncio.AbstractEventLoop
) -> memoryview:
    mv = memoryview(bytearray(length))
    total_read = 0

    def _is_ready(fut: asyncio.Future) -> None:
        if fut.done():
            return
        fut.set_result(None)

    while total_read < length:
        try:
            read = conn.recv_into(mv[total_read:])
            if read == 0:
                raise OSError("connection closed")
            total_read += read
        except (ssl.SSLWantReadError, ssl.SSLWantWriteError) as exc:
            fd = conn.fileno()
            # Check for closed socket.
            if fd == -1:
                raise ssl.SSLError("Underlying socket has been closed") from None
            if isinstance(exc, ssl.SSLWantReadError):
                fut = loop.create_future()
                loop.add_reader(fd, _is_ready, fut)
                try:
                    await fut
                finally:
                    loop.remove_reader(fd)
            if isinstance(exc, ssl.SSLWantWriteError):
                fut = loop.create_future()
                loop.add_writer(fd, _is_ready, fut)
                try:
                    await fut
                finally:
                    loop.remove_writer(fd)
    return mv


def socket_client():
    async def inner():
        loop = asyncio.get_running_loop()
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s = context.wrap_socket(s)
        s.connect((HOST, PORT))
        s.setblocking(False)
        await _async_socket_sendall_ssl(s, MESSAGE, loop)
        data = await _async_socket_receive_ssl(s, MESSAGE_SIZE, loop)
        assert len(data) == MESSAGE_SIZE and data[0]
        s.close()
    
    asyncio.run(inner())


def protocols_client():
    async def inner():
        loop = asyncio.get_running_loop()

        transport, protocol = await loop.create_connection(
            lambda: Protocol(),
            HOST, PORT, ssl=context)
        
        await asyncio.wait_for(protocol.write(MESSAGE), timeout=None)

        await asyncio.wait_for(protocol.read(), timeout=None)
        data = protocol.data()

        assert len(data) == MESSAGE_SIZE and data[0] == ord("a")
        transport.close()

    asyncio.run(inner())


def run_test(title, func):
    result = timeit.timeit(f"{func}()", setup=f"from __main__ import {func}", number=TRIALS)
    print(f"{title}: {result:.2f} seconds")


if __name__ == '__main__':
    print(f"Python: {sys.version}")
    print(f"Running {TRIALS} trials with message size {format(MESSAGE_SIZE, ',')} bytes")
    run_test("Sockets", "socket_client")
    run_test("Protocols", "protocols_client")

Profiling with cProfile + snakeviz shows that the protocol is calling ssl.write, while the socket is calling ssl.send, but that seems like an unlikely cause by itself.

Protocol:

Socket:

CPython versions tested on:

3.13

Operating systems tested on:

macOS, Linux

Start by running the server.py and clients.py reproducer on the listed Python versions and operating systems, then compare the socket and BufferedProtocol SSL profiles. Trace the asyncio BufferedProtocol and SSL transport path responsible for the difference. Done means identifying the cause and adding a verified fix or regression coverage for the reported performance gap.