Базовая многопроцессорность с циклом while

Вот подход, который я использовал пару раз с большим успехом:

1. Запустите многопроцессорный пул.

2. Используйте многопроцессорный SyncManager для создания нескольких очередей (одна для каждый тип данных должен обрабатываться по-разному).

3. Используйте apply_async для запуска функций, обрабатывающих данные. . Как и в случае с очередями, должна быть одна функция для каждого типа данных, которые нужно обрабатывать по-разному. Каждая запущенная функция получает в качестве входного аргумента очередь, соответствующую ее данным. Функции будут выполнять свою работу в бесконечном цикле, который начинается с получения данных из очереди.

4. Начать обработку. Во время обработки основной процесс сортирует данные и решает, какая функция должна их обрабатывать. После принятия решения данные помещаются в очередь, соответствующую этой функции.

5. После обработки всех данных основной процесс помещает в каждую очередь значение, называемое «таблеткой яда». Ядовитая пилюля — это значение, которое все рабочие процессы распознают как сигнал к выходу. Поскольку очереди работают по принципу «первым пришел – первым обслужен» (FIFO), то они гарантированно вытягивают ядовитую пилюлю как последний элемент в очередях.

6. Закройте и присоединитесь к многопроцессорному пулу.

Код

Ниже приведен пример этого алгоритма. Цель примера кода — использовать ранее описанный алгоритм для деления нечетных чисел на 2 и четных чисел на -2. Все результаты помещаются в общий список, доступный основному процессу.

import multiprocessing

POISON_PILL = "STOP"

def process_odds(in_queue, shared_list):

    while True:

        # block until something is placed on the queue
        new_value = in_queue.get() 

        # check to see if we just got the poison pill
        if new_value == POISON_PILL:
            break

        # we didn't, so do the processing and put the result in the
        # shared data structure
        shared_list.append(new_value/2)

    return

def process_evens(in_queue, shared_list):

    while True:    
        new_value = in_queue.get() 
        if new_value == POISON_PILL:
            break

        shared_list.append(new_value/-2)

    return

def main():

    # create a manager - it lets us share native Python object types like
    # lists and dictionaries without worrying about synchronization - 
    # the manager will take care of it
    manager = multiprocessing.Manager()

    # now using the manager, create our shared data structures
    odd_queue = manager.Queue()
    even_queue = manager.Queue()
    shared_list = manager.list()

    # lastly, create our pool of workers - this spawns the processes, 
    # but they don't start actually doing anything yet
    pool = multiprocessing.Pool()

    # now we'll assign two functions to the pool for them to run - 
    # one to handle even numbers, one to handle odd numbers
    odd_result = pool.apply_async(process_odds, (odd_queue, shared_list))
    even_result = pool.apply_async(process_evens, (even_queue, shared_list))
    # this code doesn't do anything with the odd_result and even_result
    # variables, but you have the flexibility to check exit codes
    # and other such things if you want - see docs for AsyncResult objects

    # now that the processes are running and waiting for their queues
    # to have something, lets give them some work to do by iterating
    # over our data, deciding who should process it, and putting it in
    # their queue
    for i in range(6):

        if (i % 2) == 0: # use mod operator to see if "i" is even
            even_queue.put(i)

        else:
            odd_queue.put(i)

    # now we've finished giving the processes their work, so send the 
    # poison pill to tell them to exit
    even_queue.put(POISON_PILL)
    odd_queue.put(POISON_PILL)

    # wait for them to exit
    pool.close()
    pool.join()

    # now we can check the results
    print(shared_list)

    # ...and exit!
    return


if __name__ == "__main__":
    main()

Выход

Этот код производит этот вывод:

[0.5, -0.0, 1.5, -1.0, 2.5, -2.0]

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

Обоснование

Я думаю, что это было бы хорошим решением вашей проблемы, потому что:

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

2. Он устраняет ваши опасения по поводу возможности обработки данных по-разному в зависимости от атрибутов данных. В своих комментариях вы выразили озабоченность по поводу возможности отправки данных в определенные процессы. При таком подходе вы можете выбирать, каким процессам отдавать данные, потому что вам нужно выбрать, в какую очередь их ставить. (Кстати, я думаю, вы имеете в виду pool.map, которая, как вы правильно считаете, не позволяет выполнять разные операции в одном задании. apply_async позволяет.)

3. Я нашел его очень расширяемым и гибким. Нужно добавить больше типов обработки данных? Просто напишите свою функцию-обработчик, добавьте еще одну очередь и добавьте логику в основную, чтобы направить данные в вашу новую функцию. Вы обнаруживаете, что одна очередь копируется и становится узким местом? Вы можете вызывать apply_async с одной и той же целевой функцией и ставить в очередь несколько раз, чтобы несколько рабочих процессов работали в одной очереди. Просто убедитесь, что вы дали в очередь достаточно ядовитых пилюль, чтобы все рабочие получили по одной.

Ограничения

Любые данные, которые вы хотите передать в очередь, должны быть доступны для выбора (сериализации) модулем pickle. Посмотрите здесь, чтобы узнать, можно и нельзя мариновать.

Вероятно, есть и другие ограничения, но я не могу придумать какие-либо другие в своей голове.