如何使用asyncio和concurrent.futures.ProcessPoolExecutor终止Python中长

Question

类似的问题(但答案对我不起作用(：如何取消使用concurrent.futures.ProcessPoolExecutor运行的长时间运行的子进程？

与上面链接的问题和提供的解决方案不同，在我的情况下，计算本身相当长(CPU 受限(，无法循环运行以检查是否发生了某些事件。

以下代码的简化版本：

import asyncio
import concurrent.futures as futures
import time
class Simulator:
def __init__(self):
self._loop = None
self._lmz_executor = None
self._tasks = []
self._max_execution_time = time.monotonic() + 60
self._long_running_tasks = []
def initialise(self):
# Initialise the main asyncio loop
self._loop = asyncio.get_event_loop()
self._loop.set_default_executor(
futures.ThreadPoolExecutor(max_workers=3))
# Run separate processes of long computation task
self._lmz_executor = futures.ProcessPoolExecutor(max_workers=3)
def run(self):
self._tasks.extend(
[self.bot_reasoning_loop(bot_id) for bot_id in [1, 2, 3]]
)
try:
# Gather bot reasoner tasks
_reasoner_tasks = asyncio.gather(*self._tasks)
# Send the reasoner tasks to main monitor task
asyncio.gather(self.sample_main_loop(_reasoner_tasks))
self._loop.run_forever()
except KeyboardInterrupt:
pass
finally:
self._loop.close()
async def sample_main_loop(self, reasoner_tasks):
"""This is the main monitor task"""
await asyncio.wait_for(reasoner_tasks, None)
for task in self._long_running_tasks:
try:
await asyncio.wait_for(task, 10)
except asyncio.TimeoutError:
print("Oops. Some long operation timed out.")
task.cancel()  # Doesn't cancel and has no effect
task.set_result(None)  # Doesn't seem to have an effect
self._lmz_executor.shutdown()
self._loop.stop()
print('And now I am done. Yay!')
async def bot_reasoning_loop(self, bot):
import math
_exec_count = 0
_sleepy_time = 15
_max_runs = math.floor(self._max_execution_time / _sleepy_time)
self._long_running_tasks.append(
self._loop.run_in_executor(
self._lmz_executor, really_long_process, _sleepy_time))
while time.monotonic() < self._max_execution_time:
print("Bot#{}: thinking for {}s. Run {}/{}".format(
bot, _sleepy_time, _exec_count, _max_runs))
await asyncio.sleep(_sleepy_time)
_exec_count += 1
print("Bot#{} Finished Thinking".format(bot))
def really_long_process(sleepy_time):
print("I am a really long computation.....")
_large_val = 9729379273492397293479237492734 ** 344323
print("I finally computed this large value: {}".format(_large_val))
if __name__ == "__main__":
sim = Simulator()
sim.initialise()
sim.run()

这个想法是有一个主要的模拟循环来运行和监视三个机器人线程。然后，这些机器人线程中的每一个都执行一些推理，但也使用ProcessPoolExecutor启动一个非常长的后台进程，最终可能会运行更长的自己的阈值/最大执行时间来推理事物。

正如您在上面的代码中看到的，我试图在发生超时时.cancel()这些任务。虽然这并没有真正取消实际的计算，但实际计算在后台不断发生，并且asyncio循环直到所有长时间运行的计算完成后才会终止。

如何在方法中终止如此长时间运行的 CPU 密集型计算？

其他类似的SO问题，但不一定相关或有帮助：

1. asyncio：是否可以取消由执行者运行的未来？
2. 如果单个异步任务超过 Python 中的阈值时间，如何在多处理中终止该异步任务
3. Python 中带有工作线程池的异步多处理：超时后如何继续运行？

Answer 1

如何在方法中终止如此长时间运行的 CPU 密集型计算？

您尝试的方法不起作用，因为ProcessPoolExecutor返回的未来是不可取消的。尽管 asyncio 的run_in_executor试图传播取消，但一旦任务开始执行，Future.cancel就会忽略它。

这没有根本原因。与线程不同，进程可以安全地终止，因此ProcessPoolExecutor.submit完全有可能返回其cancel终止相应进程的未来。Asyncio 协程具有明确定义的取消语义，并且可以自动使用它。不幸的是，ProcessPoolExecutor.submit返回一个常规concurrent.futures.Future，它假设底层执行者的最低公分母，并将运行的未来视为不可触碰。

因此，要取消在子流程中执行的任务，必须完全规避ProcessPoolExecutor并管理自己的流程。挑战是如何做到这一点 不重新实现一半的multiprocessing.标准库提供的一个选项是(滥用(multiprocessing.Pool用于此目的，因为它支持工作进程的可靠关闭。CancellablePool可以按如下方式工作：

• 与其生成固定数量的进程，不如生成固定数量的 1 个工作线程池。
• 从异步协程将任务分配给池。如果在等待任务在其他进程中完成时取消协程，请终止单进程池并创建一个新池。
• 由于一切都是从单个 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())

请注意 CPU 使用率如何永远不会超过 3 个内核，以及它在测试结束时如何开始下降，这表明进程正在按预期终止。

要将其应用于问题中的代码，请将self._lmz_executor设置为CancellablePool的实例，并将self._loop.run_in_executor(...)更改为self._loop.create_task(self._lmz_executor.apply(...))。