Запуск параллельных файлов Stata в python с использованием многопроцесса и подпроцесса

Question

У меня есть файл stata do pyexample3.do, который использует свой аргумент как регресс для запуска регрессии. F-статистика из регрессии сохраняется в текстовом файле. Код выглядит следующим образом:

clear all 
set more off   
local y `1'   
display `"first parameter: `y'"' 

sysuse auto 
regress price `y' 
local f=e(F) 
display "`f'" 
file open myhandle using test_result.txt, write append 
file write myhandle "`f'" _n 
file close myhandle 
exit, STATA clear 

Теперь я пытаюсь запустить Stata сделать файл параллельно питона и записать все F-статистики в одном текстовом файле. Мой процессор имеет 4 ядра.

import multiprocessing 
    import subprocess 

    def work(staname): 
     dofile = "pyexample3.do" 
     cmd = ["StataMP-64.exe","/e", "do", dofile,staname] 
     return subprocess.call(cmd, shell=False) 

    if __name__ == '__main__': 

     my_list =[ "mpg","rep78","headroom","trunk","weight","length","turn","displacement","gear_ratio" ] 

     my_list.sort() 

     print my_list 

     # Get the number of processors available 
     num_processes = multiprocessing.cpu_count() 

     threads = [] 

     len_stas = len(my_list) 

     print "+++ Number of stations to process: %s" % (len_stas) 

     # run until all the threads are done, and there is no data left 

     for list_item in my_list: 

      # if we aren't using all the processors AND there is still data left to 
      # compute, then spawn another thread 

      if(len(threads) < num_processes): 

       p = multiprocessing.Process(target=work,args=[list_item]) 

       p.start() 

       print p, p.is_alive() 

       threads.append(p) 

      else: 
       for thread in threads: 

       if not thread.is_alive(): 

        threads.remove(thread) 

Хотя файл сделать предполагается запустить в 9 раз есть 9 строк в my_list, он был запущен только в 4 раза. Так где же не так?

Answer 1

В вашем for list_item in my_list цикла, после того, как первый 4 процесса получает инициирован, он затем переходит в else:

for thread in threads: 
    if not thread.is_alive(): 
     threads.remove(thread) 

Как вы можете видеть, поскольку thread.is_alive() не будет блокировать этот цикл получить немедленно выполняться без какого-либо из тех, 4 заканчивают свою задачу. Поэтому в первую очередь выполняются только первые 4 процесса.

Вы могли бы просто использовать while цикл постоянно проверять состояние процесса с небольшим интервалом:

keep_checking = True 

while keep_checking: 
    for thread in threads: 
     if not thread.is_alive(): 
      threads.remove(thread) 
      keep_checking = False 

    time.sleep(0.5) # wait 0.5s