Как остановить процесс с помощью цикла обработки событий и исполнителя - PullRequest
Новая
       8

Как остановить процесс с помощью цикла обработки событий и исполнителя

0 голосов
/ 29 августа 2018

Рассмотрим следующую программу. 

import asyncio
import multiprocessing
from multiprocessing import Queue
from concurrent.futures.thread import ThreadPoolExecutor
import sys


def main():
    executor = ThreadPoolExecutor()
    loop = asyncio.get_event_loop()
    # comment the following line and the shutdown will work smoothly
    asyncio.ensure_future(print_some(executor))

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("shutting down")
        executor.shutdown()
        loop.stop()
        loop.close()
        sys.exit()


async def print_some(executor):
    print("Waiting...Hit CTRL+C to abort")
    queue = Queue()
    loop = asyncio.get_event_loop()
    some = await loop.run_in_executor(executor, queue.get)
    print(some)


if __name__ == '__main__':
    main()

Все, что я хочу, это постепенное отключение, когда я нажимаю "CTRL + C". Тем не менее, поток исполнителя, кажется, предотвращает это (хотя я и называю shutdown)

1 Ответ

0 голосов
/ 29 августа 2018

Вам нужно отправить таблетку с ядом, чтобы рабочие перестали слушать вызов queue.get. Рабочие потоки в пуле ThreadPoolExecutor будут блокировать выход Python, если они активно работают. В исходном коде есть комментарий , который описывает причину такого поведения: 

# Workers are created as daemon threads. This is done to allow the interpreter
# to exit when there are still idle threads in a ThreadPoolExecutor's thread
# pool (i.e. shutdown() was not called). However, allowing workers to die with
# the interpreter has two undesirable properties:
#   - The workers would still be running during interpreter shutdown,
#     meaning that they would fail in unpredictable ways.
#   - The workers could be killed while evaluating a work item, which could
#     be bad if the callable being evaluated has external side-effects e.g.
#     writing to a file.
#
# To work around this problem, an exit handler is installed which tells the
# workers to exit when their work queues are empty and then waits until the
# threads finish.

Вот полный пример, который выходит чисто: 

import asyncio
import multiprocessing
from multiprocessing import Queue
from concurrent.futures.thread import ThreadPoolExecutor
import sys


def main():
    executor = ThreadPoolExecutor()
    loop = asyncio.get_event_loop()
    # comment the following line and the shutdown will work smoothly
    fut = asyncio.ensure_future(print_some(executor))

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("shutting down")
        queue.put(None)  # Poison pill
        loop.run_until_complete(fut)
        executor.shutdown()
        loop.stop()
        loop.close()


async def print_some(executor):
    print("Waiting...Hit CTRL+C to abort")
    loop = asyncio.get_event_loop()
    some = await loop.run_in_executor(executor, queue.get)
    print(some)


queue = None
if __name__ == '__main__':
    queue = Queue()
    main()

Вызов run_until_complete(fut) необходим, чтобы избежать предупреждения об ожидающей задаче, зависшей при выходе из цикла событий asyncio. Если вас это не волнует, вы можете пропустить этот звонок.

...