$ 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

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

HOST = "127.0.0.1"
PORT = 1234

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()

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")