Python Multiprocessing 조기 종료

Question

스크립트가 실행되는 동안 어떤 시점에서 오류가 발생할 수 있습니다. 이 경우 모든 프로세스가 제대로 종료되어야하며 오류 메시지가 반환되어야하며 스크립트는 종료되어야합니다.

나는 지금 이러한 요구 사항을 충족시키지 못하고있다. 오류가 발생하면 report_error()으로 보내지고 스크립트는 터미널에서 멈추게되고 Activity Monitor는 여전히 실행중인 많은 Python 프로세스를 표시합니다.

환경

• 맥 OS X 10.8.5
• 파이썬

스크립트의 어느 지점에서 모든 프로세스를 종료하는 올바른 방법은 무엇인가 3.3.3? 실제로 풀을 중단 할 자식 작업자 프로세스를 종료하고 map 명령을 영원히 정지 할 sys.exit()를 사용하여 첫 번째

#!/usr/bin/env python3 
# -*- coding: utf-8 -*- 


import sys 
from multiprocessing import Pool 


# Global variables. 

input_files = [ 
    'test_data_0.csv', 
    'test_data_1.csv' 
] 


def report_error(error): 

    # Reports errors then exits script. 
    print("Error: {0}".format(error), file=sys.stderr) 
    sys.exit(1) 

    # What I really want is to report the error, properly terminate all processes, 
    # and then exit the script. 


def read_file(file): 

    try: 
     # Read file into list. 
    except Exception as error: 
     report_error(error) 


def check_file(file): 

    # Do some error checking on file. 
    if error: 
     report_error(error) 


def job(file): 

    # Executed on each item in input_files. 

    check_file(file) 
    read_file(file) 


def main(): 

    # Sets up a process pool. Defaults to number of cores. 
    # Each input gets passed to job and processed in a separate process. 
    p = Pool() 
    p.map(job, input_files) 

    # Closing and joining a pool is important to ensure all resources are freed properly. 
    p.close() 
    p.join() 


if __name__ == '__main__': 
    main() 

Answer 1

. 현재 multiprocessing은 작업자가 작업을 처리하는 동안 작업자 프로세스의 충돌에서 제대로 복구되지 않습니다.이 문제를 해결하는 패치가있는 버그 보고서가 있습니다 (해당 문제가 here 인 경우).

실제로 수행 할 수있는 몇 가지 방법이 있습니다.. 작업자 함수에서 반환되는 값에 대해 신경 쓰지 않는 것처럼 보이기 때문에 가장 쉬운 방법은 map 대신 imap_unordered을 사용하고, 실패한 경우 작업자로부터 예외를 발생시킨 다음 imap_unordered에 의해 반환 된 반복기를 반복하는 것입니다 : imap_unordered와

def report_error(error): 

    # Reports errors then exits script. 
    print("Error: {0}".format(error), file=sys.stderr) 
    raise error # Raise the exception 

... 

def main(): 
    p = Pool() 
    try: 
     list(p.imap_unordered(job, input_files)) 
    except Exception: 
     print("a worker failed, aborting...") 
     p.close() 
     p.terminate() 
    else: 
     p.close() 
     p.join() 

if __name__ == '__main__': 
    main() 

는, 결과는 즉시 아이를 전송으로 부모에게 반환됩니다. 따라서 하위 프로세스가 상위 프로세스로 예외를 보내면 상위 프로세스에서 즉시 예외가 다시 발생합니다. 그 예외를 잡아서 메시지를 출력 한 다음 풀을 종료합니다.