Python 3.6中await调用赋值变量的协程为何报错？如何解决？

问题原因 #

这个AssertionError("yield from wasn't used with future",)错误看起来是asyncio内部的断言失败，但本质根源是文件对象arc是线程不安全的：当你把arc.read提交到线程池后，主线程继续执行循环逻辑，此时线程池中的read操作可能还未完成，共享的文件对象会出现并发访问的竞争条件——这会打乱asyncio Future的内部状态，最终触发断言检查失败。

当你把await和run_in_executor写在一行时，主线程会等待read操作完全结束后再继续，文件指针的移动是串行的，不会有竞争；但拆分写法中，你先提交read任务就直接修改arc_offs进入下一次循环，线程池的read操作和主线程的逻辑出现重叠，导致文件操作的状态混乱，进而影响了Future的正常生命周期。

另外还有个潜在问题：你手动维护的arc_offs偏移量，和arc.read自动移动的文件指针可能在并发场景下不一致，导致读取的数据完全不符合预期。

解决方案 #

要让拆分写法正常工作，核心是避免线程池任务共享同一个线程不安全的文件对象，改用手动控制偏移量的独立读取方式，确保每个线程池任务的读取操作互不干扰。

方案1：使用文件描述符实现安全的异步读取

用os.open获取文件描述符，配合os.read指定读取长度，这样每个线程池任务的读取都是独立的，不会互相影响：

import os
import asyncio as aio

# ... 你的其他初始化代码（如io_exec、hasher等） ...

read_ftr: aio.Future = None
hash_ftr: aio.Future = None
chunk = None
arc_offs = 0
arc_size = os.path.getsize(arc_path)
max_chunk_size = 1024 * 1024  # 示例值，可根据实际调整

# 用文件描述符打开文件，避免线程不安全的文件对象共享
fd = os.open(arc_path, os.O_RDONLY)
try:
    while arc_offs < arc_size:
        chunk_size = min(arc_size - arc_offs, max_chunk_size)
        if read_ftr:
            chunk = await read_ftr
        # 提交独立的读取任务，基于文件描述符而非共享文件对象
        read_ftr = evt_loop.run_in_executor(io_exec, os.read, fd, chunk_size)
        arc_offs += chunk_size
        
        if not chunk:
            continue
        if hash_ftr:
            await hash_ftr
        hash_ftr = hasher.async_update(chunk)
        if hash_ftr:
            await hash_ftr
    await hasher.async_digest()
finally:
    os.close(fd)

方案2：保持文件对象但取消异步重叠

如果你坚持使用open的文件对象，需要确保每次提交新的read任务前，上一次的read已经完成——这种写法不会报错，但会失去异步重叠的性能优势：

import asyncio as aio

# ... 你的其他初始化代码 ...

read_ftr: aio.Future = None
hash_ftr: aio.Future = None
chunk = None
arc_offs = 0
arc_size = os.path.getsize(arc_path)
max_chunk_size = 1024 * 1024

with open(arc_path, "rb") as arc:
    while arc_offs < arc_size:
        chunk_size = min(arc_size - arc_offs, max_chunk_size)
        # 先提交读取任务并等待完成，再继续后续逻辑
        read_ftr = evt_loop.run_in_executor(io_exec, arc.read, chunk_size)
        chunk = await read_ftr
        arc_offs += chunk_size
        
        if not chunk:
            continue
        if hash_ftr:
            await hash_ftr
        hash_ftr = hasher.async_update(chunk)
        if hash_ftr:
            await hash_ftr
await hasher.async_digest()

补充说明 #

asyncio的run_in_executor返回的是完全合法的asyncio Future，拆分写法本身没有语法问题，问题出在线程不安全资源的并发访问上。错误信息看似和Future相关，实际上是底层IO竞争导致Future的内部状态异常，触发了asyncio的断言检查。

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