uvicorn源码分析

async def app(scope, receive, send):
    # 一个最简单的ASGI应用程序
    assert scope['type'] == 'http'
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            [b'content-type', b'text/plain'],
        ]
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!',
    })


if __name__ == "__main__":
    # uvicorn服务
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=5000, log_level="info")

def run(app, **kwargs):
    # 加载配置
    config = Config(app, **kwargs)
    # 加载server
    server = Server(config=config)

    if (config.reload or config.workers > 1) and not isinstance(app, str):
        # 只有命令行模式才可以使用reload
        logger = logging.getLogger("uvicorn.error")
        logger.warning(
            "You must pass the application as an import string to enable 'reload' or "
            "'workers'."
        )
        sys.exit(1)

    if config.should_reload:
        # 启动reload逻辑
        sock = config.bind_socket()
        supervisor = ChangeReload(config, target=server.run, sockets=[sock])
        supervisor.run()
    elif config.workers > 1:
        # 多进程方式启动
        sock = config.bind_socket()
        supervisor = Multiprocess(config, target=server.run, sockets=[sock])
        supervisor.run()
    else:
        # 最普通的方法启动
        server.run()

def run(self, sockets=None):
    self.config.setup_event_loop()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(self.serve(sockets=sockets))

async def main_loop(self):
    counter = 0
    should_exit = await self.on_tick(counter)
    while not should_exit:
        counter += 1
        counter = counter % 864000
        await asyncio.sleep(0.1)
        should_exit = await self.on_tick(counter)

async def shutdown(self, sockets=None):
    logger.info("Shutting down")

    # 关闭socket， 不让有新的连接建立
    for server in self.servers:
        server.close()
    for sock in sockets or []:
        sock.close()
    for server in self.servers:
        await server.wait_closed()

    # 关闭已经创建的连接， 并等待他们处理完毕 
    for connection in list(self.server_state.connections):
        connection.shutdown()
    await asyncio.sleep(0.1)

    # 等待连接关闭或者用户强制关闭
    if self.server_state.connections and not self.force_exit:
        msg = "Waiting for connections to close. (CTRL+C to force quit)"
        logger.info(msg)
        while self.server_state.connections and not self.force_exit:
            await asyncio.sleep(0.1)

    # 等待后台任务完成或者用户强制关闭
    if self.server_state.tasks and not self.force_exit:
        msg = "Waiting for background tasks to complete. (CTRL+C to force quit)"
        logger.info(msg)
        while self.server_state.tasks and not self.force_exit:
            await asyncio.sleep(0.1)

    # 通过lifespan告诉ASGI应用程序即将关闭
    if not self.force_exit:
        await self.lifespan.shutdown()

class Protocol(BaseProtocol):
    def connection_made(self, transport):
        """
        在建立连接时调用.

        参数是表示管道连接的transport, 
        此时得到的transport需要设置为该类的transport， 方便后续connection_lost控制关闭管道.
        """

    def data_received(self, data):
        """
        通过该方法可以获取到客户端传输过来的数据
        """

    def eof_received(self):
        """
        当另一端调用write_eof()或等效函数时调用.

        如果返回一个假值(包括None)，则传输将关闭自身。
        如果它返回true值，则关闭传输取决于协议.
        """

    def connection_lost(self, exc):
        """
        当连接丢失或关闭时调用, 根据exc判断是否要关闭trnasport.
        参数是一个异常对象或None(后者表示接收到常规EOF或中止或关闭连接).
        """

    def pause_writing(self):
        """
        当transport缓冲区超过高水位(high-water mark)时调用, 
        此时应该能控制外部不再写入数据（通常是一个asyncio.Future）,
        同时应该通过transport.pause_reading来停止获取数据， 之后TCP就会通过拥塞机制使得客户端减缓发送数据的速度。
        """

    def resume_writing(self):
        """
        当transport缓冲区排放低于低水位线(low-water mark)时调用. 
        此时要释放标志， 使得外部可以继续写入数据， 
        同时通过transport.resume_reading来恢复获取数据， 
        之后TCP就会通过拥塞机制知道服务端的处理能力上来了， 使客户端加快发送速度。
        """

class HttpToolsProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        # 添加实例本身到集合， 代表当前还有连接在处理
        self.connections.add(self)

        self.transport = transport
        # 初始化流控制
        self.flow = FlowControl(transport)
        # 简单的初始化实例trsnaport的相关编列
        self.server = get_local_addr(transport)
        self.client = get_remote_addr(transport)
        self.scheme = "https" if is_ssl(transport) else "http"

        if self.logger.level <= TRACE_LOG_LEVEL:
            prefix = "%s:%d - " % tuple(self.client) if self.client else ""
            self.logger.log(TRACE_LOG_LEVEL, "%sConnection made", prefix)

    def connection_lost(self, exc):
        # 从集合删除实例本身， 代表当前连接已经处理玩了， 不需要进入统计容器
        self.connections.discard(self)

        if self.logger.level <= TRACE_LOG_LEVEL:
            prefix = "%s:%d - " % tuple(self.client) if self.client else ""
            self.logger.log(TRACE_LOG_LEVEL, "%sConnection lost", prefix)

        # 设置cycle， 告诉他连接已经断开
        if self.cycle and not self.cycle.response_complete:
            self.cycle.disconnected = True
        if self.cycle is not None:
            self.cycle.message_event.set()
        if self.flow is not None:
            self.flow.resume_writing()

    def _unset_keepalive_if_required(self):
        """取消keep alive timeout的任务，
        一般来说， 在发送数据后服务端会等待客户端发送数据， 如果超过多少秒没有发送数据则可以判断该客户端已经断开了， 服务端可以主动关闭连接
        而uvicorn通过timeout_keep_alive_task来实现
        """
        if self.timeout_keep_alive_task is not None:
            self.timeout_keep_alive_task.cancel()
            self.timeout_keep_alive_task = None

    def data_received(self, data):
        self._unset_keepalive_if_required()

        try:
            # 接受字节数据， 并交由http解析器进行解析
            self.parser.feed_data(data)
        except httptools.HttpParserError as exc:
            # 解析失败， 应该不是http协议的数据， 断开连接
            msg = "Invalid HTTP request received."
            self.logger.warning(msg, exc_info=exc)
            self.transport.close()
        except httptools.HttpParserUpgrade:
            # 已经超过了解析器能解析的协议版本， 应该交由更新的协议解析器处理
            self.handle_upgrade()

class HttpToolsProtocol(asyncio.Protocol):
    def on_url(self, url):
        """这是收到一个请求后的第一次解析, 可以认为是该请求体的初始化， 此时会根据url和连接数据进行初始化， 并存放在实例的scope中"""
        method = self.parser.get_method()
        parsed_url = httptools.parse_url(url)
        raw_path = parsed_url.path
        path = raw_path.decode("ascii")
        if "%" in path:
            path = urllib.parse.unquote(path)
        self.url = url
        self.expect_100_continue = False
        self.headers = []
        self.scope = {
            "type": "http",
            "asgi": {"version": self.config.asgi_version, "spec_version": "2.1"},
            "http_version": "1.1",
            "server": self.server,
            "client": self.client,
            "scheme": self.scheme,
            "method": method.decode("ascii"),
            "root_path": self.root_path,
            "path": path,
            "raw_path": raw_path,
            "query_string": parsed_url.query if parsed_url.query else b"",
            "headers": self.headers,
        }

    def on_header(self, name: bytes, value: bytes):
        """解析器在解析header时， 是按照header一行一行进行解析的， 所以每即系一行header都会调用一次on_header, 并把他们存在实例的headers中"""
        name = name.lower()
        if name == b"expect" and value.lower() == b"100-continue":
            self.expect_100_continue = True
        self.headers.append((name, value))

    def on_headers_complete(self):
        """对于大部分前置web框架来说， 一般解析到header后就结束不再解析了， 会开始发送到正真处理的应用程序, uvicorn也是这样的"""
        http_version = self.parser.get_http_version()
        if http_version != "1.1":
            self.scope["http_version"] = http_version
        if self.parser.should_upgrade():
            # 如果发现当前http版本更加高级（比如websocket）, 则不再处理, 在另外一个逻辑会转到websocket处理
            return

        # Handle 503 responses when 'limit_concurrency' is exceeded.
        if self.limit_concurrency is not None and (
            len(self.connections) >= self.limit_concurrency
            or len(self.tasks) >= self.limit_concurrency
        ):
            # 当前并发数过高， 不再转发给后面的应用程序， 直接返回错误, 这里是一个具有ASGI标准函数签名的函数, 里面实现的功能是发送错误信息到socket
            app = service_unavailable
            message = "Exceeded concurrency limit."
            self.logger.warning(message)
        else:
            app = self.app

        # cycle相当于一个request的处理流程
        # 普通的HTTP请求只对应一个cycle就可以了， 这里是兼容Pipelined HTTP请求
        existing_cycle = self.cycle
        self.cycle = RequestResponseCycle(
            scope=self.scope,
            transport=self.transport,
            flow=self.flow,
            logger=self.logger,
            access_logger=self.access_logger,
            access_log=self.access_log,
            default_headers=self.default_headers,
            message_event=asyncio.Event(),
            expect_100_continue=self.expect_100_continue,
            keep_alive=http_version != "1.0",
            on_response=self.on_response_complete,
        )
        if existing_cycle is None or existing_cycle.response_complete:
            # 如果上个请求已经处理完了， 则开始处理这个请求(通过run_asgi来运行)
            task = self.loop.create_task(self.cycle.run_asgi(app))
            task.add_done_callback(self.tasks.discard)
            self.tasks.add(task)
        else:
            # 如果上个请求没有处理完， 就先暂停读取数据， 并把该cycle放到pipeline暂存
            self.flow.pause_reading()
            self.pipeline.insert(0, (self.cycle, app))

    def on_body(self, body: bytes):
        """读取到原生的body字节， 如果ASGI处理者还在运行， 且不是websocket, 则转给ASGI处理者
        注： 一个请求可能会触发多次on_body"""
        if self.parser.should_upgrade() or self.cycle.response_complete:
            return
        self.cycle.body += body
        if len(self.cycle.body) > HIGH_WATER_LIMIT:
            # 由于ASGI应用程序会根据调用者需要才来获取body(比如starlette的 await request.body())， 如果应用程序没有需要则会暂缓获取body数据
            self.flow.pause_reading()
        # 告诉ASGI应用程序， body已经获取结束（通常在cycle的more_body为False的时候, 才会检查message_event）
        self.cycle.message_event.set()

    def on_message_complete(self):
        if self.parser.should_upgrade() or self.cycle.response_complete:
            return
        # 表示body已经读取结束了 
        self.cycle.more_body = False
        self.cycle.message_event.set()

    def on_response_complete(self):
        """返回响应时的回调"""
        self.server_state.total_requests += 1

        if self.transport.is_closing():
            return

        # 设置一个keep_alive的机制， 服务端返回响应后会设置一个倒计时future, 该future只有在上面data_received收到请求的时候才会取消
        # 如果该future没有取消， 则会调用timeout_keep_alive_handler函数来关闭transport通道
        self._unset_keepalive_if_required()

        self.timeout_keep_alive_task = self.loop.call_later(
            self.timeout_keep_alive, self.timeout_keep_alive_handler
        )

        # 恢复读取数据
        self.flow.resume_reading()

        if self.pipeline:
            # 如果是pipeline请求， 则开始处理刚才暂存的cycle
            cycle, app = self.pipeline.pop()
            task = self.loop.create_task(cycle.run_asgi(app))
            task.add_done_callback(self.tasks.discard)
            self.tasks.add(task)