为何独立任务中分步异步迭代与无任务时表现不同？

我在使用AsyncSSH生成的进程stdout流时遇到一个异常：通过异步任务调用add_next逐行读取流时，会出现预期外的StopAsyncIteration导致流被截断；但直接await add_next则运行正常。这个问题并非所有异步可迭代对象都会出现——比如create_subprocess_exec创建的流就可以正常工作。在慢网络环境连接远程机器时，日志显示进程终止后会立即触发StopAsyncIteration。我想知道为什么有无独立任务会导致这种表现差异。

问题复现代码（触发截断） #

async def add_next(generator, collector) -> bool:
    with suppress(StopAsyncIteration):
        collector.append(await anext(generator))
        return True
    return False

async def thrtl(generator):
    bucket = []
    while True:
        if not await asyncio.create_task(add_next(generator, bucket)):
            break
    yield bucket

正常运行代码（无截断） #

async def thrtl(generator):
    bucket = []
    while True:
        if not await add_next(generator, bucket):
            break
    yield bucket

完整参考脚本 #

import asyncio, asyncssh, logging
from contextlib import suppress

async def add_next(generator, collector) -> bool:
    with suppress(StopAsyncIteration):
        collector.append(await anext(generator))
        return True
    return False

async def thrtl(generator):
    bucket = []
    while True:
        # good:
        #if not await add_next(generator, bucket):
        # bad:
        if not await asyncio.create_task(add_next(generator, bucket)):
            break
    yield bucket

async def streamhandler(stream):
    result = []
    async for rawlines in thrtl(aiter(stream)):
        result += rawlines
    return result
    
async def main():
    ssh_connection = await asyncssh.connect("localhost")

    while True:
        ssh_process = await ssh_connection.create_process("df -P")
        stdout, stderr, completed = await asyncio.gather(
            streamhandler(ssh_process.stdout),
            streamhandler(ssh_process.stderr),
            asyncio.ensure_future(ssh_process.wait()),
        )
        print(stdout)
        print(stderr)
        await asyncio.sleep(2)

logging.basicConfig(level=logging.DEBUG)
asyncio.run(main())

核心原因：异步任务调度与AsyncSSH流的终止逻辑冲突 #

1. 直接await的同步读取逻辑
   当直接await add_next时，读取操作是严格按顺序执行的：每次读取一行后才进入下一次循环。AsyncSSH的流会在数据到达时触发anext返回，直到进程结束且所有数据都被读取完毕后，才会抛出StopAsyncIteration。读取操作和流的状态更新完全同步，不会提前终止。

2. 使用create_task的异步调度逻辑
   用asyncio.create_task创建独立任务时，add_next会被放入事件循环的任务队列等待调度，主循环会立即进入下一次迭代（await create_task只是等待任务完成，但任务本身的执行是异步的）。

关键问题出在AsyncSSH流的终止条件：远程进程终止后，AsyncSSH会立即标记流为结束状态，此时如果有未完成的读取任务，anext会直接抛出StopAsyncIteration——哪怕流中还有未被读取的缓存数据。

在慢网络环境下，进程终止的信号可能先于最后一批数据到达本地。当create_task的读取任务还在等待调度时，进程已经终止，流被标记为结束，此时执行anext就会直接抛出异常，导致后续数据被丢弃。

而create_subprocess_exec的流没有这个问题，是因为它的终止逻辑不同：本地子进程的stdout流会在所有数据写入完毕后才触发结束信号，不会出现“进程终止但数据未读完”的情况；或者它的流实现会在进程终止后继续返回缓存中的数据，直到全部读取完毕才抛出异常。

3. 任务并发带来的时序问题
   另外，asyncio.gather同时运行streamhandler、streamhandler和ssh_process.wait()，当ssh_process.wait()先完成时，AsyncSSH可能会主动关闭流。而使用create_task的读取方式因为调度延迟，可能还没来得及读取缓存中的数据，就被流的关闭触发了StopAsyncIteration。

解决方案 #

保持直接await add_next的同步读取方式，确保每次读取操作都能在流状态更新前完成，避免因为任务调度延迟导致的提前终止。如果需要并发读取，要确保流的终止逻辑和读取操作的时序兼容——比如等待所有数据读取完毕后再处理进程终止的信号。

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