Как породить параллельные дочерние процессы на многопроцессорной системе?

Я бы определенно использовал многопроцессорность вместо того, чтобы использовать собственное решение с помощью подпроцесса.

Я не думаю, что вам нужна очередь, если вы не собираетесь получать данные из приложений (что, если вам действительно нужны данные, я думаю, что в любом случае будет проще добавить их в базу данных)

но попробуйте это для размера:

поместите все содержимое вашего скрипта create_graphs.py в функцию под названием «create_graphs»

import threading
from create_graphs import create_graphs

num_processes = 64
my_list = [ 'XYZ', 'ABC', 'NYU' ]

threads = []

# run until all the threads are done, and there is no data left
while threads or 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) and my_list:
        t = threading.Thread(target=create_graphs, args=[ my_list.pop() ])
        t.setDaemon(True)
        t.start()
        threads.append(t)

    # in the case that we have the maximum number of threads check if any of them
    # are done. (also do this when we run out of data, until all the threads are done)
    else:
        for thread in threads:
            if not thread.isAlive():
                threads.remove(thread)

Я знаю, что это приведет к на 1 потоку меньше, чем процессоров, что, вероятно, хорошо , он оставляет процессор для управления потоками, дисковым вводом-выводом и другими вещами, происходящими на компьютере. Если вы решите, что хотите использовать последнее ядро, просто добавьте к нему одно

edit : Я думаю, что я мог неверно истолковать назначение my_list. Вам не нужен my_list , чтобы отслеживать потоки вообще (поскольку на все они ссылаются элементы в списке thread ). Но это прекрасный способ подачи входных данных процессов - или даже лучше: использовать функцию генератора;)

Назначение my_list и потоков

my_list хранит данные которые вам нужно обработать в вашей функции
потоки - это просто список текущих запущенных потоков

, цикл while выполняет две вещи: запускает новые потоки для обработки данных и проверяет, выполняются ли какие-либо потоки .

Итак, пока у вас есть либо (а) больше данных для обработки, либо (б) потоки, которые не завершены ... вы хотите, чтобы программа продолжала работу. После того, как оба списка станут пустыми, они будут вычислены как False , и цикл while завершится

Вот решение, которое я придумал, основываясь на комментариях Нади и Джима. Я не уверен, что это лучший способ, но он работает. Первоначальный вызываемый дочерний скрипт должен быть скриптом оболочки, потому что мне нужно использовать некоторые сторонние приложения, включая Matlab. Так что мне пришлось извлечь это из Python и закодировать на bash.

import sys
import os
import multiprocessing
import subprocess

def work(staname):
    print 'Processing station:',staname
    print 'Parent process:', os.getppid()
    print 'Process id:', os.getpid()
    cmd = [ "/bin/bash" "/path/to/executable/create_graphs.sh","--name=%s" % (staname) ]
    return subprocess.call(cmd, shell=False)

if __name__ == '__main__':

    my_list = [ 'XYZ', 'ABC', 'NYU' ]

    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)

Кажется ли это разумным решением? Я попытался использовать формат цикла while Джима, но мой скрипт ничего не вернул. Я не уверен, почему это так. Вот результат, когда я запускаю скрипт с циклом «while» Джима, заменяющим цикл «for»:

hostname{me}2% controller.py 
['ABC', 'NYU', 'XYZ']
Number of processes: 64
+++ Number of stations to process: 3
hostname{me}3%

Когда я запускаю его с циклом «for», я получаю кое-что более значимое:

hostname{me}6% controller.py 
['ABC', 'NYU', 'XYZ']
Number of processes: 64
+++ Number of stations to process: 3
Processing station: ABC
Parent process: 1056
Process id: 1068
Processing station: NYU
Parent process: 1056
Process id: 1069
Processing station: XYZ
Parent process: 1056
Process id: 1071
hostname{me}7%

Итак, это работает, и Я счастлив. Однако я до сих пор не понимаю, почему я не могу использовать цикл стиля «while» Джима вместо цикла «for», который я использую.