Как объединить асинхронную обработку Python и многопроцессорность? ⇐ Python

[Anonymous]

У меня есть устройство, которому требуется многопроцессорная обработка для обработки десериализации и декодирования входящих данных с привязкой к ЦП; но остальная часть приложения представляет собой более медленный код с ограничением ввода-вывода, что отлично подходит для asyncio. Однако, похоже, нет хорошего способа объединить многопроцессорную обработку и асинхронную обработку.

Я пробовал https://github.com/dano/aioprocessing, который использует поточную обработку. исполнители для многопроцессорных операций. Однако эта библиотека изначально не поддерживает распространенные операции asyncio; например, отмена ожидания сопрограммы в очереди.get с помощью этой библиотеки приведет к взаимоблокировке.

Я также пытался использовать ProcessPoolExecutor< /code>, но передача объектов многопроцессорности этому исполнителю не работает, поскольку объекты очереди не передаются при создании процесса.

Код: Выделить всё

import multiprocessing
import asyncio
import atexit
from concurrent.futures import ProcessPoolExecutor

@atexit.register
def kill_children():
[p.kill() for p in multiprocessing.active_children()]

async def queue_get(queue: multiprocessing.Queue):
executor = ProcessPoolExecutor(max_workers=1)
loop = asyncio.get_running_loop()
return await loop.run_in_executor(executor, queue.get)

async def main():
queue = multiprocessing.Queue()
get_task = asyncio.create_task(queue_get(queue))

queue.put(None)

print(await get_task)

if __name__ == "__main__":
asyncio.run(main())

Выполнение этого кода приводит к такому исключению:

Код: Выделить всё

RuntimeError: Queue objects should only be shared between processes through inheritance

Есть ли какой-нибудь способ устранить разрыв между многопроцессорной обработкой и асинхронной обработкой?

Подробнее здесь: https://stackoverflow.com/questions/569 ... processing