Is there any operation like socket.recv_into in python 3 asyncio?

Question

The Socket module has a method socket.recv_into, so it can use a custom bytebuffer(for example bytearray) for a null copy. But it may BaseEventLoopnot have such a method. Is there a way to use a method like socket.recv_intoin asyncio?

Answer 1

Low-level socket operations specific to BaseEventLooprequire an object socket.socketto be transmitted, for example. BaseEventLoop.sock_recv(sock, nbytes). So, given what you have socket.socket, you can call sock.recv_into(). A good idea is another question.

Answer 2

, .recv_into(), , asyncio .recv_into() out-the-box.

: C, , , Python, .

Answer 3

: Python 3.7.0, -, , AbstractEventLoop.sock_recv_into().

: ...

asyncio sock_recv_into() :

byte_count = await loop.sock_recv_into(sock, buff)

- , Python, bytearray memoryview bytearray. bytearray .

The working demo code for asynchronous sockets necessarily includes a bunch of forests to configure both sides of the connections and start the event loop. The point here is to use asyncio sock_recv_into () in the sock_read_exactly () routine below.

#!/usr/bin/env python3

"""Demo the asyncio module sock_recv_into() facility."""

import sys
assert sys.version_info[:2] >= (3, 7), (
        'asyncio sock_recv_into() new in Python 3.7')

import socket
import asyncio

def local_echo_server(port=0):
    """Trivial treaded echo server with sleep delay."""
    import threading
    import time
    import random
    ssock = socket.socket()
    ssock.bind(('127.0.0.1', port))
    _, port = ssock.getsockname()
    ssock.listen(5)
    def echo(csock):
        while True:
            data = csock.recv(8192)
            if not data:
                break
            time.sleep(random.random())
            csock.sendall(data)
        csock.shutdown(1)
    def serve():
        while True:
            csock, client_addr = ssock.accept()
            tclient = threading.Thread(target=echo, args=(csock,), daemon=True)
            tclient.start()
    tserve = threading.Thread(target=serve, daemon=True)
    tserve.start()
    return port


N_COROS = 100
nrunning = 0

async def sock_read_exactly(sock, size, loop=None):
    "Read and return size bytes from sock in event-loop loop."
    if loop is None: loop = asyncio.get_event_loop()
    bytebuff = bytearray(size)
    sofar = 0
    while sofar < size:
        memview = memoryview(bytebuff)[sofar:]
        nread = await loop.sock_recv_into(sock, memview)
        print('socket', sock.getsockname(), 'read %d bytes' % nread)
        if not nread:
            raise RuntimeError('Unexpected socket shutdown.')
        sofar += nread
    return bytebuff

async def echo_client(port):
    "Send random data to echo server and test that we get back the same."
    from os import urandom
    global nrunning
    loop = asyncio.get_event_loop()
    sock = socket.socket()
    sock.setblocking(False)
    await loop.sock_connect(sock, ('127.0.0.1', port))
    for size in [1, 64, 1024, 55555]:
        sending = urandom(size)
        await loop.sock_sendall(sock, sending)
        received = await sock_read_exactly(sock, size)
        assert received == sending
    nrunning -= 1
    if not nrunning:
        loop.stop()


if __name__ == '__main__':
    port = local_echo_server()
    print('port is', port)
    loop = asyncio.get_event_loop()
    for _ in range(N_COROS):
        loop.create_task(echo_client(port))
        nrunning += 1
    print('Start loop.')
    loop.run_forever()