在异步IO任务序列中使用超时

Question

希望在每个任务中嵌入一个超时，并为这些任务的序列设置一个总超时

[编辑]代码根据@Barak的输入进行简化

import asyncio
from random import uniform, seed
seed(1234)

async def fast(n):
    s = uniform(1.5, 1.9)
    print(f'fast work {n} starts and sleeps for {s} seconds')
    await asyncio.sleep(s)
    r = f'fast work {n} is completed!'
    print(r)
    return r

async def slow(n):
    print(f'slow operation {n} starts and sleeps for 3 seconds')
    await asyncio.sleep(3)
    r = f'slow operation {n} is finished'
    r = print(r)
    return r

async def maketasks():
        # collect the tasks
        tasks = []
        tasks.append([asyncio.create_task(fast(i)) for i in range(3)])
        await slow(4) # this should immediately start!
        tasks.append(asyncio.create_task(fast(4)))

        return asyncio.gather(*tasks) # returns awaitables

async def dotasks(tasks, timeout):
    try:
        await asyncio.wait_for(tasks, timeout=timeout)
    except asyncio.TimeoutError:
        e_msg = f'Timed out after waiting {timeout} seconds in wait_for_complete'
        print(e_msg)
        return tasks, e_msg
    return tasks

if __name__ == "__main__":

    result = asyncio.run(dotasks(maketasks(), timeout=2))
    print(result)

[预期产出]

1. 对于超时<；3，我希望结果为我提供由fast和slow函数生成的字符串列表，作为任务中的result（）.done（）。相反，我得到一个coroutine对象，如下所示

...
fast work 0 is completed!
Timed out after waiting 2 seconds in wait_for_complete
(<coroutine object maketasks at 0x05AA4568>, 'Timed out after waiting 2 seconds in wait_for_complete')

1. 对于超时>；3，程序失败，出现TypeError: unhashable type: 'list'错误，原因如下：

if __name__ == "__main__":

    result = asyncio.run(dotasks(maketasks(), timeout=4))
    print(result)

…尽管它打印出消息并完成所有任务，包括fast（4）

1. 感谢你知道我错在哪里

2. 根据dotasks()中提供的超时，我应该如何从slow和fast函数中获取字符串列表的结果

Answer 1

如果要创建“聚集”任务，则该任务将在其内部的所有任务完成后才能完成。我不确定像在for循环中那样迭代它意味着什么

您可能可以通过在完成等待收集后检查各个任务的状态来提取各个任务的结果：

async def wait_for_complete(*tasks, timeout=2):
    # This code assumes that the inputs are tasks, not coroutines!
    combined = asyncio.gather(*tasks)
    try:
       await asyncio.wait_for(combined, timeout=timeout)
       return [t.result() for t in tasks]
    except asyncio.TimeoutError:
        return [t.result() if t.done() else 'Timeout'
                for t in tasks]