aiohttp中aiojobs.spawn()创建任务无法被主循环all_tasks()获取且停服被取消

现有实现方案 #

当前服务关停流程中，计划通过asyncio.Task.all_tasks()区分两类任务：一类是关停时可直接取消的普通任务，一类是需要等待执行完成的阻塞任务，核心实现代码如下：

服务启动与关停入口 #

web.run_app(web_app, port=PORT, handle_signals=True)

# 应用停止后执行清理逻辑
loop.run_until_complete(wait_for_all_blocker_coroutines_to_finish())

阻塞任务等待逻辑 #

用于等待所有阻塞任务执行完成后再关停的函数实现：

async def wait_for_all_blocker_coroutines_to_finish() -> None:
    started_time = datetime.now()
    all_tasks = asyncio.Task.all_tasks() - {asyncio.current_task()}
    # 此处all_tasks不包含任何spawn()协程内创建的任务
    logging.debug(f"Total tasks unfinished: {len(all_tasks)}")
    loop = asyncio.get_event_loop()
    logging.debug(f"Checking in loop {loop._thread_id}")
    logging.debug(all_tasks)
    coroutines = list(filter(filter_tasks_with_meta, all_tasks))
    logging.debug(coroutines)
    total = len(coroutines)
    logging.debug(f"Waiting for all blocker coroutines to finish ({total} total)")
    await asyncio.gather(*coroutines, return_exceptions=True)
    duration = datetime.now() - started_time
    seconds = duration.total_seconds()
    logging.debug(f"Coroutines unblocked after {seconds} seconds")

阻塞任务标记逻辑 #

在spawn(coro)内部创建阻塞任务时，通过给task设置带元信息的name来标记需要等待的阻塞任务，方便后续过滤识别，实现代码如下：

class TaskMeta(TypedDict):
    is_meta: bool
    blocker: bool


def name_for_task_with_meta(task_meta: TaskMeta) -> str:
    return json.dumps(task_meta)


def create_app_blocking_task(coro) -> asyncio.Task:
    # 通过存入name参数的TaskMeta标记区分需要等待的阻塞任务，后续按名称过滤任务
    name = name_for_task_with_meta(TaskMeta(is_meta=True, blocker=True))
    loop = asyncio.get_running_loop()
    task = loop.create_task(coro, name=name)
    logging.debug(f"Creating blocking task with meta={name}, loop_id={loop._thread_id}")
    return task

job = create_app_blocking_task(coro)

异常表现 #

• 所有需要等待执行完成的目标阻塞任务，均在aiojobs.spawn()内部创建
• 在aiojobs.spawn()运行的协程内部调用asyncio.Task.all_tasks()时，可以正常获取到包含目标任务在内的完整任务列表
• 在aiojobs.spawn()外部的关停处理逻辑中调用asyncio.Task.all_tasks()时返回空集合，无法获取到实际正在运行的任务
• 最初推测是spawn()创建了独立的事件循环，导致其内部任务不属于主循环因此无法被获取，进一步排查后排除该原因，实际现象为：应用停止时这些spawn创建的任务会被直接自动取消

待解决问题 #

需要找到方案避免上述问题，正确实现这类阻塞任务的优雅关闭，保证标记为阻塞类型的任务执行完成后再走完服务关停流程。

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