sanic

app.run:

def run(
    self,
    host=None,
    port=None,
    debug=False,
    ssl=None,
    sock=None,
    workers=1,
    protocol=None,
    backlog=100,
    stop_event=None,
    register_sys_signals=True,
    access_log=True,
    **kwargs
):
    try:
        self.is_running = True
        if workers == 1:
            if (
                auto_reload
                and os.environ.get("SANIC_SERVER_RUNNING") != "true"
            ):
                reloader_helpers.watchdog(2)
            else:
                serve(**server_settings)
        else:
            serve_multiple(server_settings, workers)
    except BaseException:
        error_logger.exception(
            "Experienced exception while trying to serve"
        )
        raise
    finally:
        self.is_running = False
    logger.info("Server Stopped")

sanic的启动函数除了配置相关的处理，主要是调用了server.serve_multiple函数来启动多个进程来接收服务，serve_multiple的实现如下：

server.serve_mutiple

def serve_multiple(server_settings, workers):
   if server_settings.get("sock") is None:
        sock = socket()
        sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
        sock.bind((server_settings["host"], server_settings["port"]))
        sock.set_inheritable(True)
        server_settings["sock"] = sock
        server_settings["host"] = None
        server_settings["port"] = None

    def sig_handler(signal, frame):
        logger.info("Received signal %s. Shutting down.", Signals(signal).name)
        for process in processes:
            os.kill(process.pid, SIGTERM)

    signal_func(SIGINT, lambda s, f: sig_handler(s, f))
    signal_func(SIGTERM, lambda s, f: sig_handler(s, f))

    processes = []

    for _ in range(workers):
        process = Process(target=serve, kwargs=server_settings)
        process.daemon = True
        process.start()
        processes.append(process)

    for process in processes:
        process.join()

    # the above processes will block this until they're stopped
    for process in processes:
        process.terminate()
    server_settings.get("sock").close()

serve_mutltiple主要是创建socket并绑定端口，然后根据参数中worker数量启动对应数量的进程，每个进程执行serve，serve的代码如下：
server.serve

def serve():
    if not run_async:
        # create new event_loop after fork
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    connections = connections if connections is not None else set()
    server = partial(
        protocol,
        loop=loop,
        connections=connections,
        signal=signal,
        request_handler=request_handler,
        error_handler=error_handler,
        request_timeout=request_timeout,
        response_timeout=response_timeout,
        keep_alive_timeout=keep_alive_timeout,
        request_max_size=request_max_size,
        request_class=request_class,
        access_log=access_log,
        keep_alive=keep_alive,
        is_request_stream=is_request_stream,
        router=router,
        websocket_max_size=websocket_max_size,
        websocket_max_queue=websocket_max_queue,
        websocket_read_limit=websocket_read_limit,
        websocket_write_limit=websocket_write_limit,
        state=state,
        debug=debug,
    )

    server_coroutine = loop.create_server(
        server,
        host,
        port,
        ssl=ssl,
        reuse_port=reuse_port,
        sock=sock,
        backlog=backlog,
    )

    try:
        http_server = loop.run_until_complete(server_coroutine)
    except BaseException:
        logger.exception("Unable to start server")
        return

    trigger_events(after_start, loop)

    pid = os.getpid()
    try:
        logger.info("Starting worker [%s]", pid)
        loop.run_forever()
    finally:
        logger.info("Stopping worker [%s]", pid)

        # Run the on_stop function if provided
        trigger_events(before_stop, loop)

        # Wait for event loop to finish and all connections to drain
        http_server.close()

server.serve主要通过asyncio库loop.create_server实现，这时一个针对四层的实现，具体http处理的实现需要用户实现，这个方法需要传入一个处理连接的类，这个类需要实现几个方法：connection_made,connection_lost, pause_writing, resume_writing, data_receved，sanic的http处理定义在server.HttpProtocol中，connection_made将传入的transport对象保存到对象的成员中，方便后续处理，同时启动request_timeout任务，即定时判断请求是否超时了。

   def connection_made(self, transport):
        self.connections.add(self)
        self._request_timeout_handler = self.loop.call_later(
            self.request_timeout, self.request_timeout_callback
        )
        self.transport = transport
        self._last_request_time = current_time

sanic的data_received是借助于httptools库的HttpRequestParser来处理的，

def data_received(self, data):
    ...
    if self.parser is None:
        assert self.request is None
        self.headers = []
        self.parser = HttpRequestParser(self)
    ...        
    try:
        self.parser.feed_data(data)
    except HttpParserError:
        message = "Bad Request"
        if self._debug:
            message += "\n" + traceback.format_exc()
        self.write_error(InvalidUsage(message))

可以通过提供on_headers,on_body,on_message_complete等方法来干预不同阶段的请求处理, sanic中除了on_message_complete阶段，其他阶段用户均不能干涉，主要的做的是构建request对象，将必要的信息赋给这个对象。on_message_complete阶段创建一个任务调用request_handler来处理请求

def on_message_complete(self):
    # Entire request (headers and whole body) is received.
    # We can cancel and remove the request timeout handler now.
    if self._request_timeout_handler:
        self._request_timeout_handler.cancel()
        self._request_timeout_handler = None
    if self.is_request_stream and self._is_stream_handler:
        self._request_stream_task = self.loop.create_task(
            self.request.stream.put(None)
        )
        return
    self.request.body_finish()
    self.execute_request_handler()

def execute_request_handler(self):
    self._response_timeout_handler = self.loop.call_later(
        self.response_timeout, self.response_timeout_callback
    )
    self._last_request_time = current_time
    self._request_handler_task = self.loop.create_task(
        self.request_handler(
            self.request, self.write_response, self.stream_response
        )
    )

request_handler即app中定义的handle_request

async def handle_request(self, request, write_callback, stream_callback):
    ...
    response = None
    cancelled = False
    ...
    handler, args, kwargs, uri = self.router.get(request)
    try:
    ...
        response = handler(request, *args, **kwargs)
        if isawaitable(response):
            response = await response
    ...
    if isinstance(response, StreamingHTTPResponse):
        await stream_callback(response)
    else:
        write_callback(response)

这个方法主要先获取处理对应路由的方法及用户定义的方法，根据是否awaitable来获取response即用户定义路由处理方法的返回，返回对象是一个response.HTTPResponse／StreamingHTTPResponse对象，然后调用write_callback来将结果返回给client，write_callback是server.HttpProtocol的write_response方法，实现如下：

def write_response(self, response):
    ...
    try:
        keep_alive = self.keep_alive
        self.transport.write(
            response.output(
                self.request.version, keep_alive, self.keep_alive_timeout
            )
        )
    ...
    finally:
        if not keep_alive:
            self.transport.close()
            self.transport = None
        else:
            self._keep_alive_timeout_handler = self.loop.call_later(
                self.keep_alive_timeout, self.keep_alive_timeout_callback
            )
            self._last_response_time = current_time
            self.cleanup()

其主要通过transport.write写repsonse.output按照http协议格式输出的bytes response，response.output的定义如下

   def output(self, version="1.1", keep_alive=False, keep_alive_timeout=None):

        timeout_header = b""
        if keep_alive and keep_alive_timeout is not None:
            timeout_header = b"Keep-Alive: %d\r\n" % keep_alive_timeout

        body = b""
        if has_message_body(self.status):
            body = self.body
            self.headers["Content-Length"] = self.headers.get(
                "Content-Length", len(self.body)
            )

        self.headers["Content-Type"] = self.headers.get(
            "Content-Type", self.content_type
        )

        if self.status in (304, 412):
            self.headers = remove_entity_headers(self.headers)

        headers = self._parse_headers()

        if self.status is 200:
            status = b"OK"
        else:
            status = STATUS_CODES.get(self.status, b"UNKNOWN RESPONSE")

        return (
            b"HTTP/%b %d %b\r\n" b"Connection: %b\r\n" b"%b" b"%b\r\n" b"%b"
        ) % (
            version.encode(),
            self.status,
            status,
            b"keep-alive" if keep_alive else b"close",
            timeout_header,
            headers,
            body,
        )

不论是否有body返回header之后都会有一个空行，_parse_headers的返回是以\r\n结尾的，再接%b\r\n构成了一个空行