FastAPI Engine: Inside Uvicorn, Building a Fast ASGI Server in Pure Python

Leapcell: The Best of Serverless Web Hosting

Technical Analysis of High-Performance ASGI Server Implementation Based on TCP

Ⅰ. Core Architecture of the ASGI Protocol

ASGI (Asynchronous Server Gateway Interface) defines a communication specification between asynchronous web servers and application frameworks, consisting of three core components:

1. Scope: Contains metadata such as protocol type (http/websocket), network addresses, and request methods.
2. Receive Channel: Asynchronously receives request bodies and messages.
3. Send Channel: Asynchronously sends response headers, response bodies, and close signals.

Typical ASGI Application Structure:

async def my_asgi_app(scope, receive, send):  
    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, ASGI!'  
    })

Ⅱ. Design of the TCP Server Infrastructure

2.1 Selection of Asynchronous IO Model

Use Python’s built-in asyncio framework to implement an asynchronous TCP server, with core components including:

• asyncio.start_server(): Creates a TCP listening socket.
• StreamReader/StreamWriter: Handles asynchronous IO reading and writing.
• Custom protocol classes inheriting from asyncio.Protocol.

2.2 Connection Management Module

import asyncio  
from typing import Dict, List, Any  

class ASGIServerProtocol(asyncio.Protocol):  
    def __init__(self):  
        self.reader = None  
        self.writer = None  
        self.scope: Dict[str, Any] = {}  
        self.app = None  # ASGI application instance  
    def connection_made(self, transport: asyncio.Transport):  
        self.transport = transport  
        self.reader = asyncio.StreamReader()  
        self.writer = asyncio.StreamWriter(  
            transport, self, self.reader, loop=transport.get_loop()  
        )

Ⅲ. Implementation of the HTTP Protocol Parsing Engine

3.1 Request Line Parsing

async def parse_request_line(self):  
    line = await self.reader.readline()  
    if not line:  
        return None  
    parts = line.split()  
    if len(parts) != 3:  
        await self.send_error_response(400, b"Bad Request")  
        return None  
    method, path, version = parts  
    return {  
        'method': method.decode(),  
        'path': path.decode(),  
        'version': version.decode()  
    }

3.2 Header Parsing Optimization

Pre-allocate buffers to reduce memory copying:

HEADERS_BUFFER_SIZE = 4096  

async def parse_headers(self):  
    headers = []  
    buffer = bytearray()  
    while True:  
        data = await self.reader.read(HEADERS_BUFFER_SIZE)  
        if not data:  
            break  
        buffer.extend(data)  
        while b'\r\n' in buffer:  
            line, buffer = buffer.split(b'\r\n', 1)  
            if not line:  # End of headers  
                return headers  
            key, value = line.split(b': ', 1)  
            headers.append((key.lower(), value))

3.3 Full Parsing Process

async def handle_connection(self):  
    request_line = await self.parse_request_line()  
    if not request_line:  
        return  
    headers = await self.parse_headers()  
    body = await self.reader.read()  

    self.scope = {  
        'type': 'http',  
        'method': request_line['method'],  
        'path': request_line['path'],  
        'headers': headers,  
        'query_string': b'',  # Simplified implementation, actual query parameter parsing needed  
        'server': ('127.0.0.1', 8000),  
        'client': ('127.0.0.1', 54321)  
    }  
    await self.invoke_asgi_app(body)

Ⅳ. Implementation of the ASGI Protocol Adapter

4.1 Channel Wrapper

class ASGIChannelWrapper:  
    def __init__(self, writer: asyncio.StreamWriter):  
        self.writer = writer  
        self.response_started = False  
        self.response_headers: List[List[bytes]] = []  
        self.response_body = bytearray()  

    async def receive(self):  
        # ASGI receive channel (simplified implementation, actual chunked request handling needed)  
        return {'type': 'http.request', 'body': b''}  
    async def send(self, message: Dict[str, Any]):  
        if message['type'] == 'http.response.start':  
            self.send_headers(message)  
        elif message['type'] == 'http.response.body':  
            self.send_body(message)  
    def send_headers(self, message: Dict[str, Any]):  
        status = message['status']  
        headers = message['headers']  
        # Build HTTP response headers  
        response = [  
            f"HTTP/1.1 {status} OK\r\n".encode(),  
            b''.join([k + b': ' + v + b'\r\n' for k, v in headers]),  
            b'\r\n'  
        ]  
        self.writer.write(b''.join(response))  
        self.response_started = True  
    def send_body(self, message: Dict[str, Any]):  
        body = message.get('body', b'')  
        self.writer.write(body)  
        if not message.get('more_body', False):  
            self.writer.write_eof()  
            self.writer.close()

4.2 Application Invocation Chain

async def invoke_asgi_app(self, body: bytes):  
    channel = ASGIChannelWrapper(self.writer)  
    # Construct ASGI receive channel  
    receive = channel.receive  
    send = channel.send  

    # Invoke ASGI application  
    await self.app(self.scope, receive, send)

Ⅴ. High-Performance Optimization Strategies

5.1 Event Loop Optimization

# Use Windows best practices (ProactorEventLoop has better performance on Windows)  
if sys.platform == 'win32':  
    loop = asyncio.ProactorEventLoop()  
    asyncio.set_event_loop(loop)  
else:  
    loop = asyncio.new_event_loop()  
    asyncio.set_event_loop(loop)

5.2 Buffer Management

• Use bytearray for zero-copy data concatenation.
• Set a reasonable read buffer size (default 4096 bytes).
• Process large request bodies in chunks (chunked transfer support required).

5.3 Connection Reuse

# Handle HTTP/1.1 keep-alive connections  
if b'connection: keep-alive' in headers:  
    while True:  
        await self.handle_connection()  
        # Add connection timeout detection logic

5.4 Asynchronous IO Best Practices

• Use asyncio.wait_for() to set operation timeouts.
• Manage concurrent connections with a task pool.
• Reasonably use create_task() to create background tasks.

Ⅵ. Full Server Implementation

6.1 Main Entry Module

class UvicornServer:  
    def __init__(self, app):  
        self.app = app  
        self.loop = asyncio.get_event_loop()  
        self.server = None  

    async def start(self, host='0.0.0.0', port=8000):  
        protocol_factory = lambda: ASGIServerProtocol(self.app)  
        self.server = await asyncio.start_server(  
            protocol_factory, host, port, loop=self.loop  
        )  
        print(f"Server running on http://{host}:{port}")  
    async def shutdown(self):  
        if self.server:  
            self.server.close()  
            await self.server.wait_closed()  
        self.loop.stop()  
# Usage example  
if __name__ == "__main__":  
    async def test_app(scope, receive, send):  
        await send({  
            'type': 'http.response.start',  
            'status': 200,  
            'headers': [[b'content-type', b'text/plain']]  
        })  
        await send({  
            'type': 'http.response.body',  
            'body': b'Hello from custom ASGI server!'  
        })  
    server = UvicornServer(test_app)  
    try:  
        server.loop.run_until_complete(server.start())  
        server.loop.run_forever()  
    except KeyboardInterrupt:  
        server.loop.run_until_complete(server.shutdown())

6.2 Full Protocol Handling Class

class ASGIServerProtocol(asyncio.Protocol):  
    def __init__(self, app):  
        super().__init__()  
        self.app = app  
        self.reader = None  
        self.writer = None  
        self.transport = None  
        self.scope = {}  
        self.channel = None  


    def connection_made(self, transport: asyncio.Transport):  
        self.transport = transport  
        self.reader = asyncio.StreamReader(limit=10*1024*1024)  # 10MB request limit  
        self.writer = asyncio.StreamWriter(  
            transport, self, self.reader, transport.get_loop()  
        )  
        self.loop = transport.get_loop()  
    async def handle_request(self):  
        try:  
            request_line = await self.parse_request_line()  
            if not request_line:  
                return  
            headers = await self.parse_headers()  
            body = await self.reader.read()
            self.build_scope(request_line, headers)  
            await self.invoke_app(body)  
        except Exception as e:  
            await self.send_error_response(500, str(e).encode())  
        finally:  
            self.writer.close()  
    def build_scope(self, request_line, headers):  
        self.scope = {  
            'type': 'http',  
            'method': request_line['method'],  
            'path': request_line['path'],  
            'headers': [(k.lower(), v) for k, v in headers],  
            'query_string': b'',  
            'server': ('0.0.0.0', 8000),  
            'client': self.transport.get_extra_info('peername') or ('127.0.0.1', 0)  
        }  
    async def invoke_app(self, body):  
        self.channel = ASGIChannelWrapper(self.writer)  
        receive = self.channel.receive  
        send = self.channel.send  
        await self.app(self.scope, receive, send)  
    # Omit parsing and error handling methods (same as previous implementations)

Ⅶ. In-Depth Analysis of Performance Optimization

7.1 Asynchronous IO Event-Driven Model

• Handles tens of thousands of concurrent connections in a single thread.
• Efficient event notification mechanisms based on epoll/kqueue.
• Low context-switching overhead from non-blocking IO operations.

7.2 Protocol Parsing Optimization

• Use state machines to parse the HTTP protocol.
• Pre-parse common header fields (e.g., Connection, Content-Length).
• Directly process binary data to avoid encoding conversion overhead.

7.3 Memory Management Strategies

• Use bytearray for zero-copy data concatenation.
• Reuse connection-level buffers (object pool implementation required).
• Process large request bodies in chunks to avoid memory spikes.

7.4 Concurrency Model Selection

# Multi-process mode (Linux-only)  
if sys.platform != 'win32':  
    import multiprocessing  
    workers = multiprocessing.cpu_count() * 2 + 1  
    for _ in range(workers):  
        process = multiprocessing.Process(target=run_single_process)  
        process.start()

Ⅷ. Production Environment Enhancements

8.1 Security Enhancements

• Add HTTP request body size limits.
• Implement request path security validation.
• Add CORS header support.

8.2 Protocol Extensions

• Support HTTPS (requires adding SSLContext).
• WebSocket protocol support (requires implementing a WSGI compatibility layer).
• HTTP/2 protocol support (requires upgrading the IO engine).

8.3 Monitoring and Debugging

• Add request processing time statistics.
• Implement connection count/throughput monitoring.
• Log error requests.

Ⅹ. Summary and Extension Directions

This implementation builds a basic ASGI server framework using asyncio, achieving core HTTP protocol parsing and ASGI protocol adaptation. Further improvements are needed for production environments:

1. Protocol Completeness: Implement chunked transfer, HTTPS, HTTP/2, and other protocol support.
2. Performance Optimization: Introduce connection pooling, object reuse, JIT compilation, and other technologies.
3. Functionality Extensions: Support WebSocket, startup parameter configuration, hot reloading, etc.
4. Stability: Improve error handling, connection timeouts, and resource leak detection.

By deeply understanding the ASGI protocol specification and asynchronous IO model, you can build web servers that meet high-concurrency scenarios. In practice, choose appropriate optimization strategies based on specific business needs to find the best balance between functional completeness and performance.

Leapcell: The Best of Serverless Web Hosting

Leapcell is recommended as the best platform for deploying Python services:

🚀 Build with Your Favorite Language

Develop effortlessly in JavaScript, Python, Go, or Rust.

🌍 Deploy Unlimited Projects for Free

Only pay for what you use — no requests, no charges.

⚡ Pay-as-You-Go, No Hidden Costs

No idle fees, just seamless scalability.

📖 Explore Our Documentation

🔹 Follow us on Twitter: @LeapcellHQ