在不同的事件循环中竞速两个任务

Question

我正在使用 Docker SDK，我正在尝试将一个在几秒后超时的任务与另一个等待 Docker 容器完成的任务进行竞争。实际上，我想知道给定容器是否在我设置的超时时间内完成。

我有以下代码（改编自this post）：

container =  # ... create container with Docker SDK
timeout =  # ... some int
killed = None

# our tasks
async def __timeout():
  await asyncio.sleep(timeout)
  return True
async def __run():
  container.wait()
  return False

# loop and runner
wait_loop = asyncio.new_event_loop()
done, pending = wait_loop.run_until_complete(
  asyncio.wait({__run(), __timeout()}, return_when=asyncio.FIRST_COMPLETED)
)

# result extraction
for task in done:
  if killed is None:
    killed = task.result()
    # ... do something with result

# clean up
for task in pending:
  task.cancel()
  with contextlib.suppress(asyncio.CancelledError):
    wait_loop.run_until_complete(task)
wait_loop.close()

很遗憾，我不断收到以下错误：

  File "/usr/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
    return future.result()
  File "/usr/lib/python3.5/asyncio/futures.py", line 274, in result
    raise self._exception
  File "/usr/lib/python3.5/asyncio/tasks.py", line 241, in _step
    result = coro.throw(exc)
  File "/usr/lib/python3.5/asyncio/tasks.py", line 347, in wait
    return (yield from _wait(fs, timeout, return_when, loop))
  File "/usr/lib/python3.5/asyncio/tasks.py", line 430, in _wait
    yield from waiter
  File "/usr/lib/python3.5/asyncio/futures.py", line 361, in __iter__
    yield self  # This tells Task to wait for completion.
RuntimeError: Task <Task pending coro=<wait() running at /usr/lib/python3.5/asyncio/tasks.py:347> cb=[_run_until_complete_cb() at /usr/lib/python3.5/asyncio/base_events.py:164]> got Future <Future> pending> attached to a different loop

似乎我无法与等待任务竞争，因为它属于不同的循环。有什么办法可以绕过这个错误，以便我可以确定哪个任务先完成？

Answer 1

问题很简单，每个线程都有一个默认循环。由asyncio.set_event_loop(loop) 设置。然后你可以通过loop = asyncio.get_event_loop()得到这个循环。

所以问题是，大多数情况下，一些包默认使用asyncio.get_event_loop() 来获取当前运行循环。以aiohttp为例：

class aiohttp.ClientSession(*, connector=None, loop=None, cookies=None, headers=None, skip_auto_headers=None, auth=None, json_serialize=json.dumps, version=aiohttp.HttpVersion11, cookie_jar=None, read_timeout=None, conn_timeout=None, timeout=sentinel, raise_for_status=False, connector_owner=True, auto_decompress=True, requote_redirect_url=False, trust_env=False, trace_configs=None)

如您所见，它接受loop 参数来指定运行循环。但是您也可以将其留空以默认使用asyncio.get_event_loop()。

您的问题是您在新创建的循环中启动协程。但是您无法确认您所有的内部操作也都在使用这个新创建的操作。由于它们可能使用asyncio.get_event_loop()，它们将被附加到另一个循环中，这是当前线程中的默认循环。

---

在我看来，您实际上并不需要创建一个新的，而是让用户这样做。就像上面的例子，你接受一个参数loop，如果是None，使用默认的。

---

或者您需要仔细检查您的代码，以确保每个可能的协程都在使用您创建的循环。