Python Asyncio 之网络编程方法详解

import socket


def request(host: str) -> None:
    url: str = f"http://{host}"
    sock: socket.SocketType = socket.socket()
    sock.connect((host, 80))
    sock.send(f"GET {url} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode("ascii"))

    response_bytes: bytes = b""
    chunk: bytes = sock.recv(4096)
    while chunk:
        response_bytes += chunk
        chunk = sock.recv(4096)
    print("\n".join([i for i in response_bytes.decode().split("\r\n")]))


if __name__ == "__main__":
    request("so1n.me")

import asyncio
import socket


async def request(host: str) -> None:
    url: str = f"http://{host}"
    loop = asyncio.get_event_loop()
    sock: socket.SocketType = socket.socket()
    await loop.sock_connect(sock, (host, 80))
    await loop.sock_sendall(sock, f"GET {url} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode("ascii"))

    response_bytes: bytes = b""
    chunk: bytes = await loop.sock_recv(sock, 4096)
    while chunk:
        response_bytes += chunk
        chunk = await loop.sock_recv(sock, 4096)
    print("\n".join([i for i in response_bytes.decode().split("\r\n")]))


if __name__ == "__main__":
    asyncio.run(request("so1n.me"))

# 位于:Lib/asyncio/selector_events.py
async def sock_connect(self, sock, address):
    # 检查ssl sock以及检查是否为阻塞的sock
    base_events._check_ssl_socket(sock)
    if self._debug and sock.gettimeout() != 0:
        raise ValueError("the socket must be non-blocking")

    if sock.family == socket.AF_INET or (
            base_events._HAS_IPv6 and sock.family == socket.AF_INET6):
        # 通过dns将域名转为ip地址
        resolved = await self._ensure_resolved(
            address, family=sock.family, type=sock.type, proto=sock.proto,
            loop=self,
        )
        _, _, _, _, address = resolved[0]

    # 创建一个future，这个future会等待soc连连接成功才返回数据。
    fut = self.create_future()
    self._sock_connect(fut, sock, address)
    # 通过future等待soc床创建成功
    return await fut

# 位于:Lib/asyncio/selector_events.py
    async def _ensure_resolved(self, address, *, family=0, type=socket.SOCK_STREAM, proto=0, flags=0, loop):
        host, port = address[:2]
    	# 判断是否已经解析，已经解析了就直接使用
        info = _ipaddr_info(host, port, family, type, proto, *address[2:])
        if info is not None:
            # "host" is already a resolved IP.
            return [info]
        else:
    		# 没有解析则调用socket.getaddrinfo进行解析
            return await loop.getaddrinfo(host, port, family=family, type=type, proto=proto, flags=flags)
    
# 位于:Lib/asyncio/base_events.py
    async def getaddrinfo(self, host, port, *, family=0, type=0, proto=0, flags=0):
        if self._debug:
            getaddr_func = self._getaddrinfo_debug
        else:
            getaddr_func = socket.getaddrinfo

        return await self.run_in_executor(
            None, getaddr_func, host, port, family, type, proto, flags)

# 位于:Lib/asyncio/selector_events.py
def _sock_connect(self, fut, sock, address):
    fd = sock.fileno()
    try:
        sock.connect(address)
    except (BlockingIOError, InterruptedError):
        # 检查对应的文件描述符对应的连接是否还在运行
        # 如果还在运行则抛错
        self._ensure_fd_no_transport(fd)
        # 注册文件描述符的读回调
        handle = self._add_writer(fd, self._sock_connect_cb, fut, sock, address)
        # 通过fut添加fut完成时移除监听的回调
        fut.add_done_callback(functools.partial(self._sock_write_done, fd, handle=handle))
    except (SystemExit, KeyboardInterrupt):
        raise
    except BaseException as exc:
        fut.set_exception(exc)
    else:
        fut.set_result(None)

def _sock_write_done(self, fd, fut, handle=None):
    # 移除事件的监听
    if handle is None or not handle.cancelled():
        self.remove_writer(fd)

def _sock_connect_cb(self, fut, sock, address):
    if fut.done():
        return
    # 当有写事件除非时代表着连接建立这个行为已经执行完毕了，但是需要检查建立结果
    try:
        err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
        if err != 0:
            raise OSError(err, f'Connect call failed {address}')
    except (BlockingIOError, InterruptedError):
        pass
    except (SystemExit, KeyboardInterrupt):
        raise
    except BaseException as exc:
        fut.set_exception(exc)
    else:
        fut.set_result(None)

import asyncio


class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        # transport创建成功后的回调，这里需要跟Protocol进行绑定
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        # socket收到数据的回调事件，这里先接收消息，再把内容返回给客户端最后再关闭
        message = data.decode()
        print('Data received: {!r}'.format(message))

        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()


async def main():
    server = await asyncio.get_running_loop().create_server(
        lambda: EchoServerProtocol(), '127.0.0.1', 8888
    )
    async with server:
        await server.serve_forever()


asyncio.run(main())

import asyncio


class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, on_con_lost, received_queue):
        # 这里的on_con_lost是一个future对象，用于通知连接已经断开
        self.on_con_lost = on_con_lost
        # 这里的received_queue是一个queue对象，用于接收消息
        self.received_queue = received_queue

    def connection_made(self, transport):
        # transport创建成功后的回调，这里需要跟Protocol进行绑定
        self.transport = transport

    def data_received(self, data):
        self.received_queue.put_nowait(data.decode())

    def connection_lost(self, exc):
        # socket连接断开的回调事件，这里需要通知on_con_lost
        self.on_con_lost.set_result(True)


async def main():
    loop = asyncio.get_running_loop()
    on_con_lost = loop.create_future()
    received_queue = asyncio.Queue()

    # 初始化连接并接收返回的transport和protocol
    transport, protocol = await loop.create_connection(
        lambda: EchoClientProtocol(on_con_lost, received_queue),
        '127.0.0.1', 8888)

    # 通过transport发送消息，写入消息是直接投递到系统的socket缓冲区，不会阻塞
    transport.write('Hello World!'.encode())
    # 等待消息的返回
    rece_msg = await received_queue.get()
    print("Data received:" + rece_msg)
    try:
        # 等待连接被远方断开
        await on_con_lost
    finally:
        # 确保连接被关闭
        transport.close()


asyncio.run(main())

# 位于:Lib/asyncio/selector_events.py
def _start_serving(self):
   if self._serving:
       return
   self._serving = True
   for sock in self._sockets:
       sock.listen(self._backlog)  # <--重点，后面有说
       self._loop._start_serving(
           self._protocol_factory, sock, self._ssl_context,
           self, self._backlog, self._ssl_handshake_timeout,
           self._ssl_shutdown_timeout)

# 位于:Lib/asyncio/selector_events.py
def _start_serving(self, protocol_factory, sock,
                   sslcontext=None, server=None, backlog=100,
                   ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT,
                   ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT):
    self._add_reader(sock.fileno(), self._accept_connection,
                     protocol_factory, sock, sslcontext, server, backlog,
                     ssl_handshake_timeout, ssl_shutdown_timeout)

# 位于:Lib/asyncio/selector_events.py
def _accept_connection(
        self, protocol_factory, sock,
        sslcontext=None, server=None, backlog=100,
        ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT,
        ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT):
    # 执行n次accept，直到没有新的请求建立或者执行完毕
    for _ in range(backlog):
        try:
            conn, addr = sock.accept()
            conn.setblocking(False)
        except (BlockingIOError, InterruptedError, ConnectionAbortedError):
            # socket accept的缓冲区为空，所以不用处理了
            return None
        except OSError as exc:
            # 如果有错误，则应该移除监听，再重新调用`_start_serving`进行监听
            if exc.errno in (errno.EMFILE, errno.ENFILE,
                             errno.ENOBUFS, errno.ENOMEM):
                # 兼容accept多次后Linux一直将文件描述符标记为准备就绪的bug，先移除监听再重新监听
                self.call_exception_handler({
                    'message': 'socket.accept() out of system resource',
                    'exception': exc,
                    'socket': trsock.TransportSocket(sock),
                })
                self._remove_reader(sock.fileno())
                self.call_later(constants.ACCEPT_RETRY_DELAY,
                                self._start_serving,
                                protocol_factory, sock, sslcontext, server,
                                backlog, ssl_handshake_timeout,
                                ssl_shutdown_timeout)
            else:
                raise  # The event loop will catch, log and ignore it.
        else:
            extra = {'peername': addr}
            # 每个acccept都可以认为是一个新的客户端建立的请求，为了提高并发性，需要创建一个新的协程去处理。 
            accept = self._accept_connection2(
                protocol_factory, conn, extra, sslcontext, server,
                ssl_handshake_timeout, ssl_shutdown_timeout)
            self.create_task(accept)

# 位于:Lib/asyncio/selector_events.py
#  Transport有多个实现，而_SelectorSocketTransport是专门适配UnixSelector事件循环的Transport的实现，它有一定的代表意义。
class _SelectorSocketTransport(_SelectorTransport):
    def __init__(self, loop, sock, protocol, waiter=None,
                 extra=None, server=None):
        self._read_ready_cb = None
        # 初始化参数以及通过`set_protocol`设置Protocol
        super().__init__(loop, sock, protocol, extra, server)
        self._eof = False
        self._paused = False
        self._empty_waiter = None

        # 设置TCP_NODELAY，禁用Nagle算法以提升提升性能
        base_events._set_nodelay(self._sock)
        # transport创建完毕
        self._loop.call_soon(self._protocol.connection_made, self)
        # 注册读事件的回调
        self._loop.call_soon(self._add_reader, self._sock_fd, self._read_ready)
        if waiter is not None:
            self._loop.call_soon(futures._set_result_unless_cancelled, waiter, None)

    def set_protocol(self, protocol):
        # 根据协议设置对应的已读回调
        if isinstance(protocol, protocols.BufferedProtocol):
            self._read_ready_cb = self._read_ready__get_buffer
        else:
            self._read_ready_cb = self._read_ready__data_received

        super().set_protocol(protocol)

# 位于:Lib/asyncio/selector_events.py
class _SelectorSocketTransport(_SelectorTransport):
    ...
    def _read_ready__data_received(self):
        if self._conn_lost:
            return
        try:
            # 通过sock获获取数据
            data = self._sock.recv(self.max_size)
    	# 省去一些异常处理
        except BaseException as exc:
            self._fatal_error(exc, 'Fatal read error on socket transport')
            return
    
        if not data:
            # 没有数据则代表收到eof
            self._read_ready__on_eof()
            return
    
        try:
            self._protocol.data_received(data)
    	# 省去一些异常处理
        except BaseException as exc:
            self._fatal_error(
                exc, 'Fatal error: protocol.data_received() call failed.')
    
    def _read_ready__on_eof(self):
        try:
            keep_open = self._protocol.eof_received()
    	# 省去一些异常处理
        except BaseException as exc:
            self._fatal_error(
                exc, 'Fatal error: protocol.eof_received() call failed.')
            return
    
        if keep_open:
            self._loop._remove_reader(self._sock_fd)
        else:
            self.close()

# 位于:Lib/asyncio/selector_events.py
class _SelectorTransport(transports._FlowControlMixin, transports.Transport):
    ...
    def abort(self):
        # 强制关闭连接
        self._force_close(None)

    def is_closing(self):
        return self._closing

    def close(self):
        if self._closing:
            return
        # close只移除读事件监听，buffer没有数据才会强制关闭
        self._closing = True
        self._loop._remove_reader(self._sock_fd)
        if not self._buffer:
            self._conn_lost += 1
            self._loop._remove_writer(self._sock_fd)
            self._loop.call_soon(self._call_connection_lost, None)

    def __del__(self, _warn=warnings.warn):
        if self._sock is not None:
            _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
            # 确保sock被回收，如果没有这个操作，可能会导致内存溢出
            self._sock.close()

    def _fatal_error(self, exc, message='Fatal error on transport'):
        # 省略的异常处理...
        self._force_close(exc)

    def _force_close(self, exc):
        if self._conn_lost:
            # 如果连接丢失就不管了
            return
        if self._buffer:
            # 移除写事件监听以及缓存的buffer
            self._buffer.clear()
            self._loop._remove_writer(self._sock_fd)
        if not self._closing:
            # 移除读事件监听
            self._closing = True
            self._loop._remove_reader(self._sock_fd)
        # 稍后才关闭连接
        self._loop.call_soon(self._call_connection_lost, exc)

    def _call_connection_lost(self, exc):
        try:
            if self._protocol_connected:
                # 如果当前还处于连接中，那么需要通过Protocol的`connection_lost`方法把连接丢失的原因告诉使用方。
                self._protocol.connection_lost(exc)
        finally:
            # 关闭sock等一切资源
            self._sock.close()
            self._sock = None
            self._protocol = None
            self._loop = None
            server = self._server
            if server is not None:
                server._detach()
                self._server = None

# 位于:Lib/asyncio/selector_events.py
class _SelectorSocketTransport(_SelectorTransport):
    ...

    def write(self, data):
        ...
        # 省略一些校验的
        if self._eof:
            # 已经eof了，不能写数据
            raise RuntimeError('Cannot call write() after write_eof()')
        if not self._buffer:
            # 如果没有buffer,就代表当前的发送缓冲区可能为空，先发发试一试
            try:
                # 先看看能扔多少数据去缓冲区
                n = self._sock.send(data)
            except (BlockingIOError, InterruptedError):
                # 缓冲区满了，无法发送...
                pass
            # 省去一些异常处理
            except BaseException as exc:
                self._fatal_error(exc, 'Fatal write error on socket transport')
                return
            else:
                # 去掉已经扔到缓冲区的数据
                data = data[n:]
                if not data:
                    return
            # 如果缓冲区满了，那么还有剩余的数据，需要等待socket可写时才调用回调写数据
            self._loop._add_writer(self._sock_fd, self._write_ready)

        # 把数据加入buffer中，buffer默认是bytesarray，效率很高
        self._buffer.extend(data)
        # 如果写的压力很高，就需要暂停一下 
        self._maybe_pause_protocol()

    def _write_ready(self):

        if self._conn_lost:
            return
        try:
            n = self._sock.send(self._buffer)
        # 省去一些异常处理
        except BaseException as exc:
            # 如果有异常，就移除监听，清空buffer
            self._loop._remove_writer(self._sock_fd)
            self._buffer.clear()
            # 抛出异常并通过`force`强制关闭socket
            self._fatal_error(exc, 'Fatal write error on socket transport')
        else:
            if n:
                # 从buffer清除掉已经发送的buffer
                del self._buffer[:n]
            self._maybe_resume_protocol()  # May append to buffer.
            if not self._buffer:
                # 发送完了需要进行善后处理，因为有可能socket在发送完消息后就要关闭
                self._loop._remove_writer(self._sock_fd)
                if self._closing:
                    self._call_connection_lost(None)
                elif self._eof:
                    self._sock.shutdown(socket.SHUT_WR)

# 位于:Lib/asyncio/selector_events.py
class _SelectorSocketTransport(_SelectorTransport):
    ...
    def write_eof(self):
        if self._closing or self._eof:
            return
        self._eof = True
        if not self._buffer:
            self._sock.shutdown(socket.SHUT_WR)

    def can_write_eof(self):
        return True

import asyncio

async def tcp_echo_client(message):
    reader, writer = await asyncio.open_connection(
        '127.0.0.1', 8888)

    print(f'Send: {message!r}')
    writer.write(message.encode())
    await writer.drain()

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()
    await writer.wait_closed()

asyncio.run(tcp_echo_client('Hello World!'))

import asyncio

async def handle_echo(reader, writer):
    addr = writer.get_extra_info('peername')

    data = await reader.read(100)
    message = data.decode()
    print(f"Received {message!r} from {addr!r}")

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()
    await writer.wait_closed()

async def main():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)

    addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

# 位于:Lib/asyncio/stream.py
async def open_connection(host=None, port=None, *, limit=_DEFAULT_LIMIT, **kwds):
    # 客户端建立连接的方法
    loop = events.get_running_loop()
    reader = StreamReader(limit=limit, loop=loop)
    protocol = StreamReaderProtocol(reader, loop=loop)
    transport, _ = await loop.create_connection(
        lambda: protocol, host, port, **kwds)
    writer = StreamWriter(transport, protocol, reader, loop)
    return reader, writer


async def start_server(client_connected_cb, host=None, port=None, *, limit=_DEFAULT_LIMIT, **kwds):
    # 服务端建立连接的方法
    loop = events.get_running_loop()

    def factory():
        reader = StreamReader(limit=limit, loop=loop)
        protocol = StreamReaderProtocol(reader, client_connected_cb, loop=loop)
        return protocol

    return await loop.create_server(factory, host, port, **kwds)

# 位于:Lib/asyncio/stream.py
class StreamWriter:
    # 初始化...
    def __init__(self, transport, protocol, reader, loop):
        self._transport = transport
        self._protocol = protocol
        self._reader = reader
        self._loop = loop

    # 省略了一堆代理transport的方法
    ...

    # 相比transport多出来的方法
    async def wait_closed(self):
        await self._protocol._get_close_waiter(self)

    async def drain(self):
        if self._reader is not None:
            exc = self._reader.exception()
            if exc is not None:
                raise exc
        if self._transport.is_closing():
            await sleep(0)
        await self._protocol._drain_helper()

write.write()
await write.drain()

class StreamReader:

    def _wakeup_waiter(self):
        # 告诉read系列的方法，已经有数据进来了，可以继续执行了
        waiter = self._waiter
        if waiter is not None:
            self._waiter = None
            if not waiter.cancelled():
                waiter.set_result(None)

    def feed_eof(self):
        # 投递了一个eof的信息
        self._eof = True
        self._wakeup_waiter()

    def at_eof(self):
        """给StreamProtocol判断是否需要投递`eof`"""
        return self._eof and not self._buffer

    def feed_data(self, data):
        assert not self._eof, 'feed_data after feed_eof'

        if not data:
            return

        # 把数据添加到buffer中
        self._buffer.extend(data)
        # 通过`wakeup_waiter`唤醒read系列方法
        self._wakeup_waiter()

        if (self._transport is not None and not self._paused and len(self._buffer) > 2 * self._limit):
            # 如果当前的buffer数据超过了限制，就暂停数据的读取
            try:
                self._transport.pause_reading()
            except NotImplementedError:
                # 如果当前transport不支持暂停，就直接忘掉它，后面需要调用者通过`set_transport`重新设置transport
                self._transport = None
            else:
                self._paused = True

class StreamReader:
    ... # 省略设置异常的方法

    def _maybe_resume_transport(self):
        # 由于读取了数据了，`buffer`当前的水位已经降低，需要尝试恢复从transport读取数据。
        if self._paused and len(self._buffer) <= self._limit:
            self._paused = False
            self._transport.resume_reading()

    async def _wait_for_data(self, func_name):
        if self._waiter is not None:
            raise RuntimeError(f'{func_name}() called while another coroutine is already waiting for incoming data')

        # 如果已经被暂停，那么应该恢复数据的读取
        if self._paused:
            self._paused = False
            self._transport.resume_reading()

        # 创建_waiter并等待被唤醒。
        self._waiter = self._loop.create_future()
        try:
            await self._waiter
        finally:
            self._waiter = None

    async def read(self, n=-1):
        # 有异常直接抛异常
        if self._exception is not None:
            raise self._exception

        if n == 0:
            # 如果为0,那么直接返回
            return b''

        # 省去小于0的情况，小于0时会一直获取数据，直到接收到EOF
        ...
        
        # 如果没有buffer，则需要等待
        if not self._buffer and not self._eof:
            await self._wait_for_data('read')

        # 从buffer获取数据（同时也清空了它们在buffer中的占用）
        data = bytes(self._buffer[:n])
        del self._buffer[:n]

        self._maybe_resume_transport()
        return data