Как завершить длительные вычисления (задача, связанная с процессором) в Python с помощью asyncio и concurrent.futures.ProcessPoolExecutor?

Как мне завершить такие длительные вычисления, связанные с процессором, внутри метода?

Подход, который вы пробовали, не работает, потому что фьючерсы, возвращенные ProcessPoolExecutor не подлежат отмене. Хотя run_in_executor пытается распространить отмену он просто когда-то игнорируется Future.cancel задача начинает выполняться.

Для этого нет фундаментальной причины. В отличие от потоков, процессы могут быть безопасно завершены, поэтому ProcessPoolExecutor.submit вполне может вернуть будущее, cancel которое завершило соответствующий процесс. Сопрограммы asyncio определили семантику отмены и будут автоматически использовать ее. К сожалению, ProcessPoolExecutor.submit возвращает обычный concurrent.futures.Future, предполагающий наименьший общий знаменатель и рассматривает бегущее будущее как неприкосновенное.

В результате, чтобы отменить задачи, выполняемые в подпроцессах, нужно полностью обойти ProcessPoolExecutor и управлять собственными процессами. Проблема в том, как это сделать, не выполняя повторно половину multiprocessing. Один из вариантов, предлагаемых стандартной библиотекой, - (ab) использовать multiprocessing.Pool для этой цели, потому что он поддерживает надежное завершение рабочих процессов. CancellablePool может работать следующим образом:

• Вместо того, чтобы порождать фиксированное количество процессов, создайте фиксированное количество пулов с 1 рабочим.
• Назначьте задачи пулам из сопрограммы asyncio. Если сопрограмма отменяется в ожидании завершения задачи в другом процессе, завершить однопроцессный пул и создать новый.
• Поскольку все координируется из одного потока asyncio, не беспокойтесь об условиях гонки, таких как случайное завершение процесса, который уже начал выполнение другой задачи. (Этого нужно было бы предотвратить, если бы можно было поддерживать отмену в ProcessPoolExecutor.)

Вот пример реализации этой идеи:

import asyncio
import multiprocessing

class CancellablePool:
    def __init__(self, max_workers=3):
        self._free = {self._new_pool() for _ in range(max_workers)}
        self._working = set()
        self._change = asyncio.Event()

    def _new_pool(self):
        return multiprocessing.Pool(1)

    async def apply(self, fn, *args):
        """
        Like multiprocessing.Pool.apply_async, but:
         * is an asyncio coroutine
         * terminates the process if cancelled
        """
        while not self._free:
            await self._change.wait()
            self._change.clear()
        pool = usable_pool = self._free.pop()
        self._working.add(pool)

        loop = asyncio.get_event_loop()
        fut = loop.create_future()
        def _on_done(obj):
            loop.call_soon_threadsafe(fut.set_result, obj)
        def _on_err(err):
            loop.call_soon_threadsafe(fut.set_exception, err)
        pool.apply_async(fn, args, callback=_on_done, error_callback=_on_err)

        try:
            return await fut
        except asyncio.CancelledError:
            pool.terminate()
            usable_pool = self._new_pool()
        finally:
            self._working.remove(pool)
            self._free.add(usable_pool)
            self._change.set()

    def shutdown(self):
        for p in self._working | self._free:
            p.terminate()
        self._free.clear()

Минималистичный тестовый пример, показывающий отмену:

def really_long_process():
    print("I am a really long computation.....")
    large_val = 9729379273492397293479237492734 ** 344323
    print("I finally computed this large value: {}".format(large_val))

async def main():
    loop = asyncio.get_event_loop()
    pool = CancellablePool()

    tasks = [loop.create_task(pool.apply(really_long_process))
             for _ in range(5)]
    for t in tasks:
        try:
            await asyncio.wait_for(t, 1)
        except asyncio.TimeoutError:
            print('task timed out and cancelled')
    pool.shutdown()

asyncio.get_event_loop().run_until_complete(main())

Обратите внимание на то, что использование ЦП никогда не превышает 3 ядер и как оно начинает снижаться ближе к концу теста, указывая на то, что процессы завершаются, как и ожидалось.

Чтобы применить его к коду из вопроса, сделайте self._lmz_executor экземпляром CancellablePool и измените self._loop.run_in_executor(...) на self._loop.create_task(self._lmz_executor.apply(...)).