중첩 된 계층 적 경로 다중 처리 맵을 실행하는 방법은 무엇입니까?

Question

딜 직렬화/산세에 대한 내 코드의 상당 부분을 구축 했으므로 pathos 다중 처리를 사용하여 계산을 병렬화하려고합니다. Pathos 그것은 딜의 자연스러운 연장입니다.

는 내가받을 타 ProcessingPool().map 내부

from pathos.multiprocessing import ProcessingPool 
ProcessingPool().map(fn, args) 

중첩 실행할 때 :

AssertionError: daemonic processes are not allowed to have children 

예컨대 :

from pathos.multiprocessing import ProcessingPool 

def triple(x): 
    return 3*x 

def refork(x): 
    from pathos.multiprocessing import ProcessingPool 
    return ProcessingPool().map(triple, xrange(5)) 

ProcessingPool().map(refork, xrange(3)) 

수익률

AssertionError: daemonic processes are not allowed to have children 

amap(...).get()을 사용해 보았는데 성공하지 못했습니다. 이것은 pathos 0.2.0에 있습니다.

중첩 병렬 처리를 허용하는 가장 좋은 방법은 무엇입니까?

업데이트 나는이 시점에서 솔직히, 나는 비애의 주장 "daemonic processes are not allowed to have children"를 제거했는지 고백해야

. 나는 또한 ... 사람들의 노동자와 노동자 뭔가 폭포 KeyboardInterrupt을 구축 아래 솔루션의 부품 :

def run_parallel(exec_func, exec_args, num_workers_i) 
    pool = ProcessingPool(num_workers_i) 
    pool.restart(force=True) 
    pid_is = pool.map(get_pid_i, xrange(num_workers_i)) 
    try: 
     results = pool.amap(
      exec_func, 
      exec_args, 
     ) 
     counter_i = 0 
     while not results.ready(): 
      sleep(2) 
      if counter_i % 60 == 0: 
       print('Waiting for children running in pool.amap() with PIDs: {}'.format(pid_is)) 
      counter_i += 1 
     results = results.get() 
     pool.close() 
     pool.join() 
    except KeyboardInterrupt: 
     print('Ctrl+C received, attempting to terminate pool...') 
     hard_kill_pool(pid_is, pool) # sending Ctrl+C 
     raise 
    except: 
     print('Attempting to close parallel after exception: {}'.format(sys.exc_info()[0])) 
     cls.hard_kill_pool(pid_is, pool) # sending Ctrl+C 
     raise 


def hard_kill_pool(pid_is, pool): 
    for pid_i in pid_is: 
     os.kill(pid_i, signal.SIGINT) # sending Ctrl+C 
    pool.terminate() 

콘솔과 (정지 버튼) IPython 노트북에서 작동하는 것 같다,하지만 확실하지가 100 % 맞습니다 모든 구석에서.

Answer 1

정확히 동일한 문제가 발생했습니다. 필자의 경우, 내부 연산은 병렬 처리가 필요한 연산이므로 ProcessingPool 중 ThreadingPool을 수행했습니다. 여기에 귀하의 예가 나와 있습니다 :

from pathos.multiprocessing import ProcessingPool, ThreadingPool 

def triple(x): 
    return 3*x 

def refork(x): 
    from pathos.multiprocessing import ProcessingPool 
    return ProcessingPool().map(triple, xrange(5)) 

ThreadingPool().map(refork, xrange(3)) 

다른 외부 스레드 풀과 함께 다른 레이어를 가질 수도 있습니다. 사례에 따라 이러한 풀의 순서를 바꿀 수 있습니다. 그러나 프로세스 프로세스를 가질 수는 없습니다. 실제로 필요한 경우 https://stackoverflow.com/a/8963618/6522112을 참조하십시오. 나는 나 자신을 아직 시도하지 않았으므로 이것에 대해 자세히 설명 할 수는 없다.