异步 python 函数调用，在 asyncio.gather 中不断重新调度函数，无需等待运行时间最长的任务

Question

所以目前我有如下所示的异步 python 代码（使用 asyncio）：

while datetime.now() < time_limit:
  last_start_run_time = datetime.now()
  result = await asyncio.gather(
    *(
      get_output(output_source)
    )
    for output_source in output_sources
  )
  for output in res:
    output_dict.update(my_output_dict)
  if (datetime.now() - last_start_run_time).seconds < upper_bound_wait:
    await asyncio.sleep(delay)

此代码的问题在于，它总是等待运行时间最长的 get_output 调用为所有输出源再次调用该函数。

我想知道如何重写此代码，使其在完成之前的运行后立即为每个 output_source 调用每个 get_output 调用（如果是在 upper_bound_wait 内），我还希望延迟是每个 get_output 函数调用，而不是在它完成所有函数调用之后。

如何使用 asyncio 来实现？

Answer 1

建议：将所有逻辑转移到协程中，并在一个简单的循环中创建任务。每个任务将自行决定何时延迟、何时重复以及何时退出。

async def get_output_repeatable(upper_bound_wait, output_source):
    while datetime.now() < time_limit:
        last_start_run_time = datetime.now()
        output_dict.update(await get_output(output_source))
        if (datetime.now() - last_start_run_time).seconds < upper_bound_wait:
            await asyncio.sleep(delay)

def run_them_all():            
  for output_source in output_sources:
      asyncio.create_task(get_output_repeatedly(upper_bound_wait, output_source))

Answer 2

您可以在循环中使用asyncio.wait 和FIRST_COMPLETED。您需要将原始来源存储在某处，例如在传递给asyncio.wait 和从asyncio.wait 返回的未来/任务对象上。这样，循环可以使用相同的源重新调用 get_output 以重新安排它。同样，为了使每个get_output-invocation 的延迟，您需要存储每个先前调用的开始时间，可能也在未来/任务上。例如（未经测试）：

async def delayed_run(aw):
    await asyncio.sleep(delay)
    return await aw

async def run_sources(output_sources, time_limit, upper_bound_wait):
    output_dict = {}

    pending = set()
    for output_source in output_sources:
        fut = asyncio.create_task(get_output(output_source))
        fut.my_start_time = time.time()
        fut.my_source = output_source
        pending.add(fut)

    while pending and time.time() < time_limit:
        done, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED,
            timeout=time_limit - time.time())

        for done_fut in done:
            output_dict.update(done_fut.result())
            new_coro = get_output(done_fut.my_source)
            if time.time() - done_fut.my_start_time < upper_bound_wait:
                new_coro = delayed_run(new_coro)
            new_fut = asyncio.create_task(new_coro)
            new_fut.my_start_time = time.time()
            new_fut.my_source = done_fut.my_source
            pending.add(new_fut)

    return output_dict