等待正在运行的容器的正确异步方式

Question

我在基于aiohttp 的应用程序中使用python docker SDK。我的目标是在请求到来时启动一个容器，等到它完成所有工作并返回响应。我想同时处理很多这样的请求。

我在分离模式下运行容器，所以问题不在于启动容器本身。问题是如何定义容器何时完成工作。 Container docker 包的实例有一个方法 wait() 但它是阻塞的，所以我不能“像那样”使用它。所以我想出了别的办法：我启动一个容器，创建新的 asyncio 任务，并且（在那个任务中）我正在检查容器是否改变了它的状态。它看起来或多或少是这样的：

import docker
import asyncio
import time

def wait_for_finish(event, container):
    try:
        container.reload()
        timeout = time.time() + 600
        while time.time() < timeout:
            if container.status == 'running' or container.status == 'created':
                await asyncio.sleep(0.5)
                try:
                    container.reload()
                except requests.exceptions.HTTPError:
                    break
            else:
                break
    finally:
        event.set()

docker_client = docker.from_env()
container = docker_client.containers.run(**kwargs)
event = asyncio.Event()
asyncio.create_task(
            wait_for_finish(event, container)
        )
await event.wait()
event.clear()

这很简单，但效果很好。但我的问题是：这种等待容器完成的“虚拟”方式（状态不同于running 或created）是一个好方法吗？说实话我不知道喜欢这个解决方案，因为几个原因，我认为这种“等待”有问题，但我不知道究竟是什么......

我知道诸如aiodocker 或run_in_executor/threading 之类的解决方案，甚至用subprocess 替换docker sdk，但现在我想使用 - 如果可能的话 - 组合asyncio 和 docker sdk 没有例如阅读

---

编辑

我最终得到了这样的解决方案：我通过 aiohttp 直接向 Docker API 发出异步请求，这允许我异步使用提到的 wait() 方法：

await docker_socket_session.post(f'http://localhost/containers/{container_id}/wait', timeout=timeout)

它运行良好，至少目前我没有看到任何有意义的缺点。
感谢您的帮助

Answer 1

您可以将两个现有函数链接在一起以避免轮询循环并通常简化此代码。

docker-py Container 对象有一个 wait method ，它会阻塞直到容器停止，然后返回。这将替换轮询循环，并且在容器退出之前不会返回；但是，这是一个阻塞调用。

asyncio.to_thread 接受一个函数，在一个单独的线程中运行它，并返回一个等待函数，该函数将在函数完成时触发。这将让您将阻塞 I/O 函数桥接到 asyncio，而不会实际阻塞主执行。 （文档指出这是 Python 3.9 中的新功能。）

您应该能够将这些组合在一起：

import docker
import asyncio

docker_client = docker.from_env()
container = docker_client.containers.run(**kwargs)
awaitable = asyncio.to_thread(container.wait)
result = await awaitable

（请记住，容器可以指定自己的用户 ID 并挂载任意主机内容；如果容器以 root 身份运行并挂载主机 / 目录，则这种方法可能会成为整个系统的 root。考虑使用消息队列，例如RabbitMQ 或 Celery 之类的框架来调度作业，无需添加 Docker 依赖项、限制并发性、在作业失败时增加一些弹性，并且没有接管主机的可能性。）