【问题标题】:Multiprocessing Pool hangs if child process killed如果子进程被杀死,多处理池会挂起
【发布时间】:2020-08-12 23:37:34
【问题描述】:

我启动了一个工作进程池并提交了一堆任务。系统内存不足,oomkiller 杀死了一个工作进程。父进程只是挂在那里等待任务完成并且从未返回。

这是一个重现问题的可运行示例。我没有等待 oomkiller 杀死一个工作进程,而是找到所有工作进程的进程 ID 并告诉第一个任务杀死该进程。 (对ps 的调用不适用于所有操作系统。)

import os
import signal
from multiprocessing import Pool
from random import choice
from subprocess import run, PIPE
from time import sleep


def run_task(task):
    target_process_id, n = task
    print(f'Processing item {n} in process {os.getpid()}.')
    delay = n + 1
    sleep(delay)
    if n == 0:
        print(f'Item {n} killing process {target_process_id}.')
        os.kill(target_process_id, signal.SIGKILL)
    else:
        print(f'Item {n} finished.')
    return n, delay


def main():
    print('Starting.')
    pool = Pool()

    ps_output = run(['ps', '-opid', '--no-headers', '--ppid', str(os.getpid())],
                    stdout=PIPE, encoding='utf8')
    child_process_ids = [int(line) for line in ps_output.stdout.splitlines()]
    target_process_id = choice(child_process_ids[1:-1])

    tasks = ((target_process_id, i) for i in range(10))
    for n, delay in pool.imap_unordered(run_task, tasks):
        print(f'Received {delay} from item {n}.')

    print('Closing.')
    pool.close()
    pool.join()
    print('Done.')


if __name__ == '__main__':
    main()

当我在一个有 8 个 CPU 的系统上运行它时,我看到了这个输出:

Starting.
Processing item 0 in process 303.
Processing item 1 in process 304.
Processing item 2 in process 305.
Processing item 3 in process 306.
Processing item 4 in process 307.
Processing item 5 in process 308.
Processing item 6 in process 309.
Processing item 7 in process 310.
Item 0 killing process 308.
Processing item 8 in process 303.
Received 1 from item 0.
Processing item 9 in process 315.
Item 1 finished.
Received 2 from item 1.
Item 2 finished.
Received 3 from item 2.
Item 3 finished.
Received 4 from item 3.
Item 4 finished.
Received 5 from item 4.
Item 6 finished.
Received 7 from item 6.
Item 7 finished.
Received 8 from item 7.
Item 8 finished.
Received 9 from item 8.
Item 9 finished.
Received 10 from item 9.

您可以看到项目 5 永远不会返回,并且池只是永远等待。

当子进程被杀死时,如何让父进程注意到?

【问题讨论】:

    标签: python multiprocessing


    【解决方案1】:

    Python bug 9205 中描述了此问题,但他们决定在 concurrent.futures module 而不是 multiprocessing 模块中修复它。为了利用该修复,请切换到较新的进程池。

    import os
    import signal
    from concurrent.futures.process import ProcessPoolExecutor
    from random import choice
    from subprocess import run, PIPE
    from time import sleep
    
    
    def run_task(task):
        target_process_id, n = task
        print(f'Processing item {n} in process {os.getpid()}.')
        delay = n + 1
        sleep(delay)
        if n == 0:
            print(f'Item {n} killing process {target_process_id}.')
            os.kill(target_process_id, signal.SIGKILL)
        else:
            print(f'Item {n} finished.')
        return n, delay
    
    
    def main():
        print('Starting.')
        pool = ProcessPoolExecutor()
    
        pool.submit(lambda: None)  # Force the pool to launch some child processes.
        ps_output = run(['ps', '-opid', '--no-headers', '--ppid', str(os.getpid())],
                        stdout=PIPE, encoding='utf8')
        child_process_ids = [int(line) for line in ps_output.stdout.splitlines()]
        target_process_id = choice(child_process_ids[1:-1])
    
        tasks = ((target_process_id, i) for i in range(10))
        for n, delay in pool.map(run_task, tasks):
            print(f'Received {delay} from item {n}.')
    
        print('Closing.')
        pool.shutdown()
        print('Done.')
    
    
    if __name__ == '__main__':
        main()
    

    现在,当您运行它时,您会收到一条清晰的错误消息。

    Starting.
    Processing item 0 in process 549.
    Processing item 1 in process 550.
    Processing item 2 in process 552.
    Processing item 3 in process 551.
    Processing item 4 in process 553.
    Processing item 5 in process 554.
    Processing item 6 in process 555.
    Processing item 7 in process 556.
    Item 0 killing process 556.
    Processing item 8 in process 549.
    Received 1 from item 0.
    Traceback (most recent call last):
      File "/home/don/.config/JetBrains/PyCharm2020.1/scratches/scratch2.py", line 42, in <module>
        main()
      File "/home/don/.config/JetBrains/PyCharm2020.1/scratches/scratch2.py", line 33, in main
        for n, delay in pool.map(run_task, tasks):
      File "/usr/lib/python3.7/concurrent/futures/process.py", line 483, in _chain_from_iterable_of_lists
        for element in iterable:
      File "/usr/lib/python3.7/concurrent/futures/_base.py", line 598, in result_iterator
        yield fs.pop().result()
      File "/usr/lib/python3.7/concurrent/futures/_base.py", line 428, in result
        return self.__get_result()
      File "/usr/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
        raise self._exception
    concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.
    

    【讨论】:

    • 如果你想要一个更有弹性的池实现,你可以试试pebble。它能够处理工人终止并通知主循环而不是崩溃。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2012-03-12
    • 2014-05-18
    • 1970-01-01
    • 2016-06-10
    • 2010-12-08
    • 2010-12-02
    相关资源
    最近更新 更多