Como executar mapas de multiprocessamento de pathos hierárquicos e aninhados?
Tendo construído uma parte significativa do meu código na serialização / decapagem de endro, também estou tentando usar o multiprocessamento do pathos para paralelizar meus cálculos. Pathos é uma extensão natural do endro.
Ao tentar executar aninhado
from pathos.multiprocessing import ProcessingPool
ProcessingPool().map(fn, args)
dentro de um outroProcessingPool().map
, então eu recebo:
AssertionError: daemonic processes are not allowed to have children
Por exemplo.:
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))
rendimentos
AssertionError: daemonic processes are not allowed to have children
Eu tentei usaramap(...).get()
sem sucesso. Isso está no pathos 0.2.0.
Qual é a melhor maneira de permitir a paralelização aninhada?
Atualizar
Eu tenho que ser honesto neste momento, e confessar que removi a afirmação"daemonic processes are not allowed to have children"
do pathos. Eu também construí algo que cascataKeyboardInterrupt
aos trabalhadores e trabalhadores desses ... Partes da solução abaixo:
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()
Parece funcionar no console e no notebook IPython (com botão de parada), mas não tenho certeza de que é 100% correto em todos os casos de canto.