python - PIL 和使用 asyncio 的阻塞调用

Question

我有一个异步应用程序，它使用来自 aiohttp 的服务器和带有 asyncio.open_connection()

的异步套接字

我的代码包含一些来自 PIL 库的阻塞调用，例如

Image.save()
Image.resize()

1. 即使调用没有阻塞太长时间，如果我使用这些阻塞调用，我的 Web 服务器是否会卡住？更准确地说，事件循环是否有可能因为阻塞代码而错过事件？
2. 如果是，那么与 asyncio 集成的这些功能的替代品是什么？ PIL 没有异步版本。
3. 一般来说，asyncio 中的“阻塞代码”是什么？除了套接字、读取文件等明显的操作
   例如，os.path.join() 是否被认为是正确的？如何处理 numpy 数组？

Answer 1

can my web server freeze if I use these blocking calls? Moreprecisely, is it possible that the event loop will miss events becauseof blocking code?

服务器在执行图像函数时会精确卡住。您不会错过任何事件，但所有事件处理都将在图像函数执行时延迟。

卡住事件循环是一种糟糕的情况——你应该避免它。

If yes, what is the replacement for these functions, that integratewith asyncio? there is no asyncio version of PIL.

避免卡住事件循环的最简单和通用的方法 - 使用 asyncio.run_in_executor 在另一个线程或另一个进程中执行阻塞函数.那里的代码片段展示了如何做到这一点，并很好地解释了何时使用进程或线程:

def blocking_io():
    # File operations (such as logging) can block the
    # event loop: run them in a thread pool.
    with open('/dev/urandom', 'rb') as f:
        return f.read(100)

def cpu_bound():
    # CPU-bound operations will block the event loop:
    # in general it is preferable to run them in a
    # process pool.
    return sum(i * i for i in range(10 ** 7))

我只想补充一点，进程池可能并不总是适合每个 CPU 密集型操作的好解决方案。如果您的图像函数不会花费太多时间(或者特别是如果您的服务器没有多个处理器内核)，那么在一个线程中运行它们可能仍然更有效率。

In general, what is considered a 'blocking code' in asyncio? besidesthe obvious operations like socket, read a file, etc. For example, doesos.path.join() is considered ok? what about working on a numpy array?

粗略地说任何函数都是阻塞的:它会阻塞事件循环一段时间。但是像 os.path.join 这样的许多函数只需要很少的时间，所以它们不是问题，我们不称它们为“阻塞”。

当执行时间(和事件循环卡住)成为问题时，很难说出确切的限制，特别是考虑到不同硬件的时间会有所不同。我有偏见的建议 - 如果您的代码在将控制权返回给事件循环之前需要(或可能需要)> 50 毫秒，请考虑它阻塞并使用 run_in_executor。

更新:

Thanks, does it make sense to use one event loop (of the main thread),and using another thread that will add tasks using the same loop?

我不确定你在这里的意思，但我认为不是。我们需要另一个线程来运行一些作业，而不是在其中添加任务。

I need some way for the thread to inform the main thread after theimage processing is completed`

只需等待 run_in_executor 的结果或使用它启动任务。 run_in_executor - 是一个协程，它在后台线程中执行某些操作而不阻塞事件循环。

它看起来像这样:

thread_pool = ThreadPoolExecutor()


def process_image(img):
    # all stuff to process image here
    img.save()
    img.resize()


async def async_image_process(img):
    await loop.run_in_executor(
        thread_pool, 
        partial(process_image, img)
    )


async def handler(request):

    asyncio.create_task(
        async_image_process(img)
    )
    # we use a task to return the response immediately,
    # read https://stackoverflow.com/a/37345564/1113207

    return web.Response(text="Image processed without blocking other requests")