다중 처리 스크립트 종료

Question

대상 기능에서 오류가 발생하면 다중 처리 스크립트를 종료하려고 시도하지만 종료하는 대신 상위 프로세스가 중지됩니다. 이것은 완벽하게 작동

#!/usr/bin/python3.5 

import time, multiprocessing as mp 

def myWait(wait, resultQueue): 
    startedAt = time.strftime("%H:%M:%S", time.localtime()) 
    time.sleep(wait) 
    endedAt = time.strftime("%H:%M:%S", time.localtime()) 
    name = mp.current_process().name 
    resultQueue.put((name, wait, startedAt, endedAt)) 

# queue initialisation 
resultQueue = mp.Queue() 

# process creation arg: (process number, sleep time, queue) 
proc = [ 
    mp.Process(target=myWait, name = ' _One_', args=(2, resultQueue,)), 
    mp.Process(target=myWait, name = ' _Two_', args=(2, resultQueue,)) 
    ] 

# starting processes 
for p in proc: 
    p.start() 

for p in proc: 
    p.join() 

# print results 
results = {} 
for p in proc: 
    name, wait, startedAt, endedAt = resultQueue.get() 
    print('Process %s started at %s wait %s ended at %s' % (name, startedAt, wait, endedAt)) 

, 내가 부모 스크립트가 htop 두 자식 프로세스를 산란을 볼 수 있지만 내가 부모 스크립트를 강제로 할 때 종료 :

이

내가 문제를 복제하는 데 사용하는 테스트 스크립트입니다 myWait 대상 함수에서 오류가 발생하면 상위 프로세스가 중지되고 하위 프로세스가 생성되지 않습니다. 나는 그것을 죽일 ctrl-c해야합니다.

def myWait(wait, resultQueue): 
    try: 
     # do something wrong 
    except: 
     raise SystemExit 

나는 아무 소용 기능을 (예를 들어 exit(), sys.exit(), os._exit() ...) 종료하기 위해 모든 방법을 시도했다.

Answer 1

첫째, 코드에는 대기열 내용이 플러시되기 전에 프로세스에 참여하려고 시도하는 중입니다 (있는 경우). 교착 상태가 발생할 수 있습니다. 이 예외가 myWait 함수에서와 데이터가이 없다고 발생하면 결코 을 발생하지 않습니다 일부 데이터를 수신 할 때까지 https://docs.python.org/3/library/multiprocessing.html#multiprocessing-programming

둘째, resultQueue.get()에 대한 호출이 차단됩니다 : 여기에 '큐를 사용하여 가입 프로세스를'섹션을 참조하십시오 그 전에 대기열에 푸시되었습니다. 그래서 그것을 블로킹 (non-blocking)으로 만들고 마침내 무언가를 받거나 무언가 잘못 될 때까지 루프의 모든 데이터를 확인합니다.

#!/usr/bin/python3.5 

import multiprocessing as mp 
import queue 
import time 

def myWait(wait, resultQueue): 
    raise Exception("error!") 

# queue initialisation 
resultQueue = mp.Queue() 

# process creation arg: (process number, sleep time, queue) 
proc = [ 
    mp.Process(target=myWait, name = ' _One_', args=(2, resultQueue,)), 
    mp.Process(target=myWait, name = ' _Two_', args=(2, resultQueue,)) 
    ] 

# starting processes 
for p in proc: 
    p.start() 

# print results 
results = {} 
for p in proc: 
    while True: 
     if not p.is_alive(): 
      break 

     try: 
      name, wait, startedAt, endedAt = resultQueue.get(block=False) 
      print('Process %s started at %s wait %s ended at %s' 
        % (name, startedAt, wait, endedAt)) 
      break 
     except queue.Empty: 
      pass 

for p in proc: 
    p.join() 

기능 myWait는 예외가 발생합니다하지만 두 프로세스가 여전히 가입하고 프로그램이 잘 종료됩니다 :

여기 당신에게 아이디어를 제공하기 위해 quick'n'dirty 수정합니다.

Answer 2

귀하의 프로세스를 관리하려면 multiprocessing.Pool을 사용해야합니다. 그런 다음 Pool.imap_unordered을 사용하여 결과가 완료된 순서대로 반복됩니다. 첫 번째 예외가 발생하자마자 풀과 하위 프로세스를 중지 할 수 있습니다 (이 작업은 with Pool() as pool 블록을 종료하면 자동으로 수행됩니다). 그것은 단지 아이의 수와 같은 병렬 작업의 많은을 실행으로 예를 들어

from multiprocessing import Pool 
import time 

def my_wait(args): 
    name, wait = args 
    if wait == 2: 
     raise ValueError("error!") 
    else: 
     startedAt = time.strftime("%H:%M:%S", time.localtime()) 
     time.sleep(wait) 
     endedAt = time.strftime("%H:%M:%S", time.localtime()) 
     return name, wait, startedAt, endedAt 

if __name__ == "__main__": 
    try: 
     with Pool() as pool: 
      args = [["_One_", 2], ["_Two_", 3]] 
      for name, wait, startedAt, endedAt in pool.imap_unordered(my_wait, args):  
       print('Task %s started at %s wait %s ended at %s' % (name, 
        startedAt, wait, endedAt)) 
    except ValueError as e: 
     print(e) 

이 방법은, 긴 시간 초과, 낮은 부하 작업에 적합하지 않습니다 처리가 관리하고있는 (그러나 이것은 뭔가 당신입니다 설정할 수 있습니다). 다른 기능을 실행해야하는 경우에는 좋지 않습니다.