python - 如何使 ProcessPoolExecutor 中的任务表现得像守护进程？

Question

python 3.6.6

这是代码:

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


executor_processes = ProcessPoolExecutor(2)


def calculate():
    while True:
        print("while")
        time.sleep(1)


async def async_method():
    loop_ = asyncio.get_event_loop()
    loop_.run_in_executor(executor_processes, calculate)
    await asyncio.sleep(1)
    print("finish sleep")

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(async_method())
    print("main_thread is finished")

输出:

while
finish sleep
main_thread is finished
while
while
...

我希望子进程将被终止，就像 Process 用守护程序属性生成的情况一样:

import asyncio
import time
import multiprocessing


def calculate():
    while True:
        print("while")
        time.sleep(1)


async def async_method():
    proc = multiprocessing.Process(target=calculate)
    proc.daemon = True
    proc.start()
    await asyncio.sleep(1)
    print("finish sleep")

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(async_method())
    print("main_thread is finished")

输出:

while
finish sleep
main_thread is finished

问题:如何更改 loop_.run_in_executor(executor_processes, calculate) “类似守护进程”的行为？

Answer 1

您显示的代码显然只是一个小示例，用于演示您希望实现的目标。我们不知道您的实际任务/问题。但老实说，我不相信你在这里走的是正确的道路。
ProcessPoolExecutor是 concurrent.futures 的一部分标准库包。它返回一个 Future 调用时发送给调用者 submit() .那个Future是尚未完成的计算结果的代理。这是一个 promise ;尽管在这种情况下该术语在技术上并不完全正确。见 Wiki page为了区别。

这意味着，计算应在有限时间内完成并产生结果。这就是 ThreadPoolExecutor 的原因。和 ProcessPoolExecutor Python 中的实现不允许您生成守护进程。要求一个你实际上并不想要实现的结果的 promise 没有多大意义。

你怎么还能实现你的目标？

1 - 子类 ProcessPoolExecutor ? 可以拦截新进程的创建和启动，潜入p.daemon = True在 _adjust_process_count() .但是，由于 concurrent.futures设计时没有考虑到无限期运行的任务，这不会有太大帮助。不像 multiprocessing , concurrent.futures.process定义一个 exit handler不考虑守护进程。它只是试图 join()一切，这可能需要一些时间来无限循环。

2 - 定义您自己的退出处理程序! 两者都可以，multiprocessing和 concurrent.futures.process do:定义一个退出处理程序，在您的 Python 进程即将关闭时进行清理。 atexit可以提供帮助:

import atexit

executor_processes = ProcessPoolExecutor(2)

def calculate():
    while True:
        print("while")
        time.sleep(1)

def end_processes():
    [proc.terminate() for proc in multiprocessing.active_children()]

async def async_method():
    [...]

if __name__ == '__main__':
    atexit.register(end_processes)
    loop = asyncio.get_event_loop()
    [...]

注:这将终止在进程结束时处于事件状态的所有子进程。如果有您想要正常关闭的子进程，请保留一个句柄并在代码中的指令结束之前执行此操作。另请注意，进程可以拒绝兑现 terminate() . kill() 是你最后的手段。