DEV Community

Cover image for FastAPI Engine: Inside Uvicorn, Building a Fast ASGI Server in Pure Python
Leapcell
Leapcell

Posted on

2 1 1 1 1

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

#python #flask #fastapi #django

Image description

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!'  
    })
Enter fullscreen mode Exit fullscreen mode

Ⅱ. 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()  
        )
Enter fullscreen mode Exit fullscreen mode

Ⅲ. 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()  
    }
Enter fullscreen mode Exit fullscreen mode

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))
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

Ⅳ. 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()
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

Ⅴ. 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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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())
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

Ⅶ. 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()
Enter fullscreen mode Exit fullscreen mode

Ⅷ. 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:

Image description

🚀 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.

Image description

📖 Explore Our Documentation

🔹 Follow us on Twitter: @LeapcellHQ

profile
Redis
 Promoted

Redis image

Short-term memory for faster
AI agents 🤖💨

AI agents struggle with latency and context switching. Redis fixes it with a fast, in-memory layer for short-term context—plus native support for vectors and semi-structured data to keep real-time workflows on track.

Start building

Top comments (0)

Some comments may only be visible to logged-in visitors. Sign in to view all comments.

profile
Stellar Development Foundation
 Promoted

Dev Diairies image

User Feedback & The Pivot That Saved The Project

🔥 Check out Episode 3 of Dev Diairies, following a successful Hackathon project turned startup.

Watch full video 🎥

DEV Live Events

AWS Security LIVE!

Hosted by security experts, AWS Security LIVE! showcases AWS Partners tackling real-world security challenges. Join live and get your security questions answered.

Tune in to the full event

DEV is partnering to bring live events to the community. Join us or dismiss this billboard if you're not interested. ❤️