如何運行嵌套的分層病態多處理映射？

Question

在蒔蘿序列化/酸洗中構建我的代碼的重要部分之後，我還試圖使用pathos多處理來並行化我的計算。悲傷是蒔蘿的自然延伸。

當試圖運行嵌套

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

的其他ProcessingPool().map裏面，然後我得到：

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()沒有成功。這是在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

我遇到完全相同的問題。在我的情況下，內部操作是需要並行性的操作，所以我做了ThreadingPool的ProcessingPool。這裏是你的例子：

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。我還沒有嘗試過，所以我不能詳細說明這一點。