Twisted框架高负载下deferred threads HTTP服务器异常问题

使用Twisted HTTP框架时遇到高负载下回调失效问题：为提升并发，采用threads.deferToThread搭配回调机制，低负载（1-2个客户端）时运行正常，但高负载时回调经常完全不执行，且无数据写入transport。

流程说明：每个请求的客户端数据包在延迟线程中处理，响应数据包写入队列，由on_request_done回调出队并返回响应。

简化代码如下：

class HttpPlayer(Player):
    def __init__(self, address: str, port: int) -> None:
        super().__init__(address, port)
        self.queue = Queue()
        self.token = ""

    def enqueue(self, data: bytes):
        self.queue.put(data)

    def dequeue(self, max: int = 4096) -> bytes:
        data = b""

        while not self.queue.empty():
            data += self.queue.get()

        return data

class HttpBanchoProtocol(Resource):
    isLeaf = True

    def __init__(self) -> None:
        self.player: HttpPlayer | None = None
        self.children = {}

    def handle_login_request(self, request: Request) -> bytes:
        username, password, client = (
            request.content.read().decode().splitlines()
        )

        deferred = threads.deferToThread(
            self.player.login_received,
            username,
            password,
            client
        )

        deferred.addCallbacks(
            lambda _: self.on_request_done(request),
            lambda f: self.on_request_error(request, f)
        )

        return NOT_DONE_YET

    def handle_request(self, request: Request) -> bytes:
        deferred = threads.deferToThread(
            self.process_packets,
            request.content.read()
        )

        deferred.addCallbacks(
            lambda _: self.on_request_done(request),
            lambda f: self.on_request_error(request, f)
        )

        return NOT_DONE_YET

    def login_received(self, username: str, password: str, client: str) -> None:
        # 登录处理逻辑
        ...

    def process_packets(self, request: bytes):
        # 数据包处理逻辑
        ...

    def on_request_done(self, request: Request) -> None:
        if request._disconnected:
            self.player.logger.warning('Client disconnected before response')
            return

        if request.finished:
            self.player.logger.warning('Request finished before response')
            return

        request.write(self.player.dequeue())
        request.finish()

    def on_request_error(
        self,
        request: Request,
        failiure: Failure
    ) -> None:
        request.setResponseCode(500)
        self.player.send_error()
        self.player.logger.error(
            f'Failed to process request: {failiure.getErrorMessage()}',
            exc_info=failiure.value
        )
        self.on_login_done(request)

    def render_POST(self, request: Request) -> bytes:
        request.setResponseCode(200)

        if not (token := request.getHeader('token')):
            return self.handle_login_request(request)

        if not (player := app.session.players.by_token(token)):
            request.setResponseCode(401)
            return b""

        self.player = player
        return self.handle_request(request)

核心问题点 #

1. 实例变量线程安全问题：HttpBanchoProtocol作为单实例Resource，高负载下多个请求会同时修改self.player，导致回调时使用的玩家实例与当前请求不匹配，队列数据混乱甚至出现空指针异常。
2. 队列操作竞态阻塞：dequeue方法中while not self.queue.empty()的判断与get()操作存在竞态，多线程下可能导致get()阻塞，回调无法完成。
3. 私有属性依赖：直接访问request._disconnected（私有属性）判断连接状态，Twisted不保证该属性的稳定性，可能导致判断失效。
4. 未捕获回调异常：回调中出现的异常会被Twisted静默吞掉，表现为回调未执行。
5. 线程池容量不足：默认线程池大小（10）在高负载下会被占满，新任务排队导致回调延迟或无响应。

具体修复步骤 #

1. 取消共享玩家实例，通过闭包传递请求绑定的玩家 #

删除self.player实例变量，在回调中通过闭包显式传递当前请求对应的玩家，避免多线程冲突：

def handle_login_request(self, request: Request) -> bytes:
    username, password, client = request.content.read().decode().splitlines()
    # 此处需确保创建/获取当前请求对应的player实例
    player = HttpPlayer(...)
    
    deferred = threads.deferToThread(
        player.login_received,
        username,
        password,
        client
    )
    # 显式传递player和request到回调，避免闭包延迟绑定问题
    deferred.addCallbacks(
        lambda _, req=request, p=player: self.on_request_done(req, p),
        lambda f, req=request, p=player: self.on_request_error(req, f, p)
    )
    return NOT_DONE_YET

def handle_request(self, request: Request) -> bytes:
    token = request.getHeader('token')
    player = app.session.players.by_token(token)
    if not player:
        request.setResponseCode(401)
        return b""
    
    data = request.content.read()
    deferred = threads.deferToThread(
        self.process_packets,
        data,
        player
    )
    deferred.addCallbacks(
        lambda _, req=request, p=player: self.on_request_done(req, p),
        lambda f, req=request, p=player: self.on_request_error(req, f, p)
    )
    return NOT_DONE_YET

2. 修复队列阻塞问题 #

将dequeue改为非阻塞读取，避免线程挂起：

# 修改HttpPlayer的dequeue方法
def dequeue(self) -> bytes:
    data = b""
    while True:
        try:
            chunk = self.queue.get_nowait()
            data += chunk
        except queue.Empty:
            break
    return data

3. 正确判断连接状态 #

替换私有属性request._disconnected为官方推荐的request.transport.disconnecting：

def on_request_done(self, request: Request, player: HttpPlayer) -> None:
    try:
        if request.transport.disconnecting:
            player.logger.warning('Client disconnected before response')
            return
        if request.finished:
            player.logger.warning('Request finished before response')
            return
        
        request.write(player.dequeue())
        request.finish()
    except Exception as e:
        player.logger.error(f"Failed to send response: {str(e)}", exc_info=True)
        # 确保请求被终止，避免挂起
        if not request.finished:
            request.setResponseCode(500)
            request.finish()

4. 增大线程池容量 #

在应用启动时调整Twisted线程池大小，适配高负载场景：

from twisted.internet import reactor

# 建议根据服务器CPU核心数设置，比如50
reactor.suggestThreadPoolSize(50)

5. 全局捕获未处理异常 #

添加全局错误回调，避免异常被静默吞掉：

from twisted.python import log

def unhandled_deferred_error(failure):
    log.err(failure, "Unhandled deferred error")

# 在所有deferred创建后添加
deferred.addErrback(unhandled_deferred_error)

内容的提问来源于stack exchange，提问作者Lekuru