DocsPricingBlog
Contact Us
Log inSign up for FREE
Back to Tech Articles

FastAPI エンジン：Uvicorn がいかに高速性を実現するか：Python の ASGI サーバーの詳細な分析

Jun 09, 2025
# python

Ethan Miller

Product Engineer · Leapcell

This post is also available in English、한국어、Deutsch
Cover of "FastAPI エンジン：Uvicorn がいかに高速性を実現するか：Python の ASGI サーバーの詳細な分析"

FastAPI エンジン：Uvicorn がいかに高速性を実現するか：Python の ASGI サーバーの詳細な分析

TCP に基づく高性能 ASGI サーバー実装の技術分析

Ⅰ. ASGI プロトコルのコアアーキテクチャ

ASGI (Asynchronous Server Gateway Interface) は、非同期ウェブサーバーとアプリケーションフレームワーク間の通信仕様を定義し、次の 3 つのコアコンポーネントで構成されています。

  1. スコープ: プロトコルタイプ（http/websocket）、ネットワークアドレス、リクエストメソッドなどのメタデータが含まれます。
  2. 受信チャネル: リクエストボディとメッセージを非同期的に受信します。
  3. 送信チャネル: レスポンスヘッダー、レスポンスボディ、およびクローズシグナルを非同期的に送信します。

一般的な ASGI アプリケーション構造:

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!'
    })

Ⅱ. TCP サーバーインフラストラクチャの設計

2.1 非同期 IO モデルの選択

Python の組み込み asyncio フレームワークを使用して非同期 TCP サーバーを実装します。コアコンポーネントは次のとおりです。

  • asyncio.start_server(): TCP リスニングソケットを作成します。
  • StreamReader/StreamWriter: 非同期 IO の読み取りと書き込みを処理します。
  • asyncio.Protocol から継承するカスタムプロトコルクラス。

2.2 接続管理モジュール

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()
        )

Ⅲ. HTTP プロトコル解析エンジンの実装

3.1 リクエストラインの解析

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 ヘッダー解析の最適化

メモリコピーを削減するためにバッファを事前に割り当てます。

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 完全な解析プロセス

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)

Ⅳ. ASGI プロトコルアダプターの実装

4.1 チャネルラッパー

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 アプリケーションの呼び出しチェーン

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)

Ⅴ. 高性能最適化戦略

5.1 イベントループの最適化

# 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 バッファ管理

  • ゼロコピーデータ連結には bytearray を使用します。
  • 適切な読み取りバッファサイズ（デフォルトは 4096 バイト）を設定します。
  • 大きなリクエストボディをチャンクで処理します（チャンク転送のサポートが必要です）。

5.3 接続の再利用

# 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 非同期 IO のベストプラクティス

  • asyncio.wait_for() を使用して操作タイムアウトを設定します。
  • タスクプールを使用して同時接続を管理します。
  • バックグラウンドタスクを作成するには、 create_task() を適切に使用します。

Ⅵ. 完全なサーバー実装

6.1 メインエントリモジュール

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 完全なプロトコル処理クラス

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)

Ⅶ. パフォーマンス最適化の詳細な分析

7.1 非同期 IO イベント駆動型モデル

  • 単一のスレッドで数万の同時接続を処理します。
  • epoll/kqueue に基づく効率的なイベント通知メカニズム。
  • ノンブロッキング IO 操作による低いコンテキストスイッチングオーバーヘッド。

7.2 プロトコル解析の最適化

  • ステートマシンを使用して HTTP プロトコルを解析します。
  • 一般的なヘッダーフィールド（例：Connection、Content-Length）を事前に解析します。
  • エンコード変換のオーバーヘッドを回避するために、バイナリデータを直接処理します。

7.3 メモリ管理戦略

  • ゼロコピーデータ連結には bytearray を使用します。
  • 接続レベルのバッファを再利用します（オブジェクトプールの実装が必要です）。
  • メモリスパイクを回避するために、大きなリクエストボディをチャンクで処理します。

7.4 並行性モデルの選択

# 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()

Ⅷ. 本番環境の強化

8.1 セキュリティの強化

  • HTTP リクエストボディサイズの制限を追加します。
  • リクエストパスのセキュリティ検証を実装します。
  • CORS ヘッダーのサポートを追加します。

8.2 プロトコルの拡張

  • HTTPS をサポートします（SSLContext の追加が必要です）。
  • WebSocket プロトコルのサポート（WSGI 互換レイヤーの実装が必要です）。
  • HTTP/2 プロトコルのサポート（IO エンジンのアップグレードが必要です）。

8.3 監視とデバッグ

  • リクエスト処理時間の統計を追加します。
  • 接続数/スループットの監視を実装します。
  • エラーリクエストをログに記録します。

Ⅹ. 概要と拡張の方向性

この実装では、asyncio を使用して基本的な ASGI サーバーフレームワークを構築し、コア HTTP プロトコルの解析と ASGI プロトコルの適合を実現しています。本番環境では、さらなる改善が必要です。

  1. プロトコルの完全性: チャンク転送、HTTPS、HTTP/2、およびその他のプロトコルのサポートを実装します。
  2. パフォーマンスの最適化: 接続プーリング、オブジェクトの再利用、JIT コンパイル、およびその他のテクノロジーを導入します。
  3. 機能拡張: WebSocket、起動パラメータの設定、ホットリロードなどをサポートします。
  4. 安定性: エラー処理、接続タイムアウト、およびリソースリークの検出を改善します。

ASGI プロトコルの仕様と非同期 IO モデルを深く理解することで、高並行性シナリオに対応できるウェブサーバーを構築できます。実際には、特定のビジネスニーズに基づいて適切な最適化戦略を選択し、機能の完全性とパフォーマンスの最適なバランスを見つけてください。

Leapcell: 最高のサーバーレス Web ホスティング

Leapcell は、Python サービスをデプロイするための最適なプラットフォームとして推奨されます。

pic

🚀 お気に入りの言語で構築

JavaScript、Python、Go、または Rust で簡単に開発できます。

🌍 無制限のプロジェクトを無料でデプロイ

使用した分だけ料金が発生します - リクエストも料金もありません。

⚡ 従量課金制、隠れたコストなし

アイドル料金は発生せず、シームレスなスケーラビリティのみです。

pic

📖 ドキュメントを見る

🔹 Twitter でフォローしてください: @LeapcellHQ

Share this article

HackerNews

More Posts from Leapcell

cover

PythonとPostgreSQLの通信：PsycopgからORMまでの徹底分析

cover

Kubernetesから学ぶアーキテクチャデザイン

Popular Posts