asyncio.Event无法在信号处理器中设置，如何优雅终止asyncio服务器？

问题代码 #

import signal
import asyncio
from typing import Optional
from types import FrameType


async def main() -> None:
    server = (
        await asyncio.start_server(
            lambda reader, writer: None,
            '127.0.0.1',
            7070
        )
    )
    tasks = (
        asyncio.create_task(server.serve_forever()),
        asyncio.create_task(event.wait())
    )

    async with server:
        await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    # Some shutdown logic goes here.


def handle_signal(signal: int, frame: Optional[FrameType]) -> None:
    event.set()


if __name__ == '__main__':
    event = asyncio.Event()
    signal.signal(signal.SIGINT, handle_signal)
    asyncio.run(main())

问题描述 #

我希望在用户发送SIGINT（即按下CTRL+C）时优雅终止asyncio服务器。为此，我使用asyncio.Event：在主协程中通过asyncio.wait等待服务器停止或接收到SIGINT信号，且已设置对应信号处理器。但该方案在Alpine Linux上测试无效，请求解释具体原因及解决办法。

原因分析 #

Alpine Linux默认采用musl libc，而非常见的glibc，二者在信号处理的实现上存在关键差异：

1. asyncio.Event.set()并非信号安全函数，在musl的信号处理上下文直接调用该方法，无法正确触发asyncio事件循环的状态更新，导致event.wait()协程无法被唤醒。
2. musl对信号的分发机制与glibc不同，原生signal模块注册的处理器无法与asyncio事件循环正确交互，信号触发后事件循环无法及时感知Event的状态变化。

解决办法 #

方案一：使用asyncio原生信号处理API（推荐） #

asyncio提供的loop.add_signal_handler()是专门为异步场景设计的信号处理接口，能保证在事件循环上下文安全执行回调，完美适配musl环境。修改后的代码如下：

import asyncio


async def main() -> None:
    server = await asyncio.start_server(
        lambda reader, writer: None,
        '127.0.0.1',
        7070
    )
    shutdown_event = asyncio.Event()

    def trigger_shutdown():
        shutdown_event.set()

    # 获取当前运行的事件循环
    loop = asyncio.get_running_loop()
    # 注册SIGINT信号处理器
    loop.add_signal_handler(asyncio.signal.SIGINT, trigger_shutdown)

    tasks = (
        asyncio.create_task(server.serve_forever()),
        asyncio.create_task(shutdown_event.wait())
    )

    async with server:
        await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    # 执行自定义关闭逻辑
    print("服务器已优雅终止")


if __name__ == '__main__':
    asyncio.run(main())

方案二：用线程安全方法调用Event.set() #

如果必须使用原生signal模块，需通过loop.call_soon_threadsafe()确保在事件循环线程中安全调用Event.set()，避免信号上下文的安全问题：

import signal
import asyncio
from typing import Optional
from types import FrameType

loop = None
event = None


async def main() -> None:
    global loop, event
    loop = asyncio.get_running_loop()
    event = asyncio.Event()

    server = await asyncio.start_server(
        lambda reader, writer: None,
        '127.0.0.1',
        7070
    )
    tasks = (
        asyncio.create_task(server.serve_forever()),
        asyncio.create_task(event.wait())
    )

    async with server:
        await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    # 执行关闭逻辑
    print("服务器已优雅终止")


def handle_signal(signal: int, frame: Optional[FrameType]) -> None:
    if loop and event:
        loop.call_soon_threadsafe(event.set)


if __name__ == '__main__':
    signal.signal(signal.SIGINT, handle_signal)
    asyncio.run(main())

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