Python многопроцессорность и менеджер

Question

Я использую Python multiprocessing для создания параллельного приложения.Процессы должны поделиться некоторыми данными, для которых я использую Manager.Тем не менее, у меня есть некоторые общие функции, которые нужно вызывать процессам и которые должны обращаться к данным, хранящимся в объекте Manager.Мой вопрос заключается в том, могу ли я избежать необходимости передавать экземпляр Manager этим общим функциям в качестве аргумента и скорее использовать его как глобальный.Другими словами, рассмотрим следующий код:

import multiprocessing as mp

manager = mp.Manager()
global_dict = manager.dict(a=[0])

def add():
    global_dict['a'] += [global_dict['a'][-1]+1]

def foo_parallel(var):
    add()
    print var

num_processes = 5
p = []
for i in range(num_processes):
    p.append(mp.Process(target=foo_parallel,args=(global_dict,)))

[pi.start() for pi in p]
[pi.join() for pi in p]

Это нормально работает и возвращает p=[0,1,2,3,4,5] на моей машине.Тем не менее, это «хорошая форма»?Это хороший способ сделать это, точно так же, как определить add(var) и вместо этого вызвать add(var)? 1013 *

Answer 1

Ваш пример кода, кажется, имеет больше проблем, чем форма.Вы получаете желаемый результат только при удаче.Повторное выполнение даст разные результаты.Это потому, что += не является атомарной операцией.Несколько процессов могут читать одно и то же старое значение один за другим, прежде чем какой-либо из них обновит его, и они будут записывать одни и те же значения.Чтобы предотвратить такое поведение, вам придется дополнительно использовать Manager.Lock.

---

К исходному вопросу о «хорошей форме».

ИМО было бы чище, чтобыглавная функция дочернего процесса foo_parallel, явно передать global_dict в обобщенную функцию add(var).Это будет форма внедрения зависимости и имеет некоторые преимущества.В вашем примере неисчерпывающе:

• позволяет изолированное тестирование
• увеличивает возможность повторного использования кода

• упрощает отладку (обнаруживает недоступность управляемогообъект не должен быть отложен до тех пор, пока не будет вызван add ( fail fast )

• меньше стандартного кода (например, блоки try-excepts для ресурсов, которые требуются нескольким функциям)

В качестве дополнительного примечания. Использование списочных представлений только для побочных эффектов считается «запахом кода». Если в результате вам не нужен список, просто используйтедля петли.

код:

import os
from multiprocessing import Process, Manager


def add(l):
    l += [l[-1] + 1]
    return l


def foo_parallel(global_dict, lock):
    with lock:
        l = global_dict['a']
        global_dict['a'] = add(l)
        print(os.getpid(), global_dict)


if __name__ == '__main__':

    N_WORKERS = 5

    with Manager() as manager:

        lock = manager.Lock()
        global_dict = manager.dict(a=[0])

        pool = [Process(target=foo_parallel, args=(global_dict, lock))
                for _ in range(N_WORKERS)]

        for p in pool:
            p.start()

        for p in pool:
            p.join()

        print('result', global_dict)