Запуск вложенных процессов при многопроцессорной обработке

Question

У меня есть основной файл, запускающий несколько процессов, и один из процессов снова запускает несколько процессов. У меня возникают проблемы с запуском вложенного набора процессов.

У меня есть следующий код в одном файле:

# parallel_test.py 
import Queue 
import multiprocessing 
import time 
import threading 


def worker(q): 
    while not q.empty(): 
     try: 
      row = q.get(False) 
      print row 

      time.sleep(1) 

     except Queue.Empty: 
      break 


def main(): 
    print 'creating queue' 
    q = multiprocessing.Queue() 

    print 'enqueuing' 
    for i in range(100): 
     q.put(i) 

    num_processes = 15 
    pool = [] 

    for i in range(num_processes): 
     print 'launching process {0}'.format(i) 
     p = multiprocessing.Process(target=worker, args=(q,)) 
     p.start() 
     pool.append(p) 

    for p in pool: 
     p.join() 

if __name__ == '__main__': 
    main() 

Запуск этого файла в одиночку python parallel_test.py отлично работает и печатает число, как ожидается. Но запуск его из другого файла в качестве другого процесса вызывает проблему. Мой главный файл:

# main_loop_test.py 
import parallel_test 
from multiprocessing import Pool 
import time 


def main(): 
    targets = [parallel_test.main] 

    running = True 

    while running: 
     try: 
      p = Pool(12) 

      for target in targets: 
       p.apply_async(target) 

      p.close() # For some reason you need to run close() before join() 
      p.join() # What for all the steps to be done 

      print 'All steps done' 

      time.sleep(2) 

     except KeyboardInterrupt as e: 
      print "<<<<<<<<<<<<<<<<<<CAUGHT KEYBOARD INTERRUPT FROM USER>>>>>>>>>>>>>>>>>>>" 
      running = False 


if __name__ == '__main__': 
    main() 

Это parallel_test.py кажется, чтобы попытаться запустить один процесс (который ничего не делает), а затем выходит из функции и main_loop_test.py отпечатки «Все шаги сделаны». Никакие номера не печатаются. Выход:

creating queue 
enqueuing 
launching process 0 
All steps done 
creating queue 
enqueuing 
launching process 0 
All steps done 

Что случилось? Я получаю ту же проблему, используя Pool вместо управления процессами самостоятельно в parallel_test.py. Тем не менее, заменяет многопроцессорную обработку потоками.

Answer 1

Вы не можете создать дочерний процесс из parallel_test, когда вы вызываете его как дочерний процесс из другой программы по той причине, что процесс создается как демонический процесс, и, как упоминается в ссылке https://docs.python.org/2/library/multiprocessing.html, демонический процесс не разрешено создавать дочерний процесс. Вы должны создать процесс как недемонный процесс, установив для свойства daemon этого процесса значение false, как показано ниже.

p = multiprocessing.Process(target=test.main) 
p.daemon = False 
p.start() 
p.join() 

Я не уверен, как установить свойство daemon при создании дочернего процесса через модуль пула. Вы можете попробовать передать это свойство через список инициализаторов.

Answer 2

Вы хотите использовать иерархический параллелизм через модуль multiprocessing, как этот пример показывает, что карта обработки блокировки выполняется внутри асинхронной карты обработки?

>>> def squared(x): 
... return x**2 
... 
>>> def triple(x): 
... return 3*x 
... 
>>> from pathos.multiprocessing import ProcessingPool as PPool 
>>> res = PPool().amap(triple, PPool().map(squared, xrange(10))) 
>>> res.get() 
[0, 3, 12, 27, 48, 75, 108, 147, 192, 243] 

Я использую pathos вилку multiprocessing, так как он обеспечивает немного легче использовать, чем у стандартной версии библиотеки.

Также смотрите другой пример здесь: https://stackoverflow.com/questions/28203774/how-to-do-hierarchical-parallelism-in-ipython-parallel

Вы можете получить pathos здесь: https://github.com/uqfoundation