Как сделать так, чтобы задачи в 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")

Вывод:

в то время как
завершение сна
main_thread завершено
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")

Вывод:

пока
закончить сон
main_thread завершен

Вопрос: как изменить loop_.run_in_executor(executor_processes, calculate)поведение "демоноподобного"?

Answer 1

Код, который вы показываете, явно является небольшим примером, демонстрирующим то, чего вы надеетесь достичь.Мы не знаем вашу реальную задачу / проблему.Но, честно говоря, я не уверен, что вы на правильном пути.

ProcessPoolExecutor является частью стандартного пакета библиотеки concurrent.futures.При вызове submit() он возвращает Future вызывающей стороне.Это Future является прокси для результата вычисления, которое еще не завершено.Это обещание;хотя этот термин технически не совсем корректен в этом контексте.См. Wiki-страницу для различия.

Это означает, что вычисление ожидается , чтобы завершиться за конечное время и дать результат.
Вот почему реализации ThreadPoolExecutor и ProcessPoolExecutor в Python не позволяют создавать демонов.Требование обещания результата, которого вы на самом деле не хотите выполнить, не имеет особого смысла.

Как вы все еще можете достичь своей цели?

1 - Подкласс ProcessPoolExecutor?
Вы можете перехватить создание и запуск новых процессов, чтобы проникнуть в p.daemon = True в _adjust_process_count().Однако, поскольку concurrent.futures не предназначен для бесконечно запущенных задач, это не сильно поможет.В отличие от multiprocessing, concurrent.futures.process определяет обработчик выхода , который не учитывает демонические процессы.Он просто пытается 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() ваш последний курорт.