Применить уменьшение на выходе генератора с многопроцессорностью

Question

У меня есть функция генератора (Python), которая работает так

def Mygenerator(x, y, z, ...):
    while True:
        # code that makes two matrices based on sequences of input arrays
        yield (matrix1, matrix2)

То, что я хочу сделать, это добавить выход из этого генератора. Эта строка делает работу:

M1, M2 = reduce(lambda x, y: x[0] + y[0], x[1] + y[1], Mygenerator(x, y, z, ...))

Я хотел бы распараллелить это, чтобы ускорить вычисления. Важно, чтобы выходы из Mygenerator уменьшались по мере его выдачи, поскольку list(Mygenerator(...)) займет слишком много памяти.

Answer 1

Чтобы ответить на мой собственный вопрос, я нашел решение, которое, кажется, работает так, как я надеялся:

Первый, Mygenerator больше не генератор, а функция. Кроме того, вместо того, чтобы проходить по сегментам x, y и z, теперь я передаю один сегмент в функцию за раз:

def Myfunction(x_segment, y_segment, z_segment):
        # code that makes two matrices based on input arrays
        return (matrix1, matrix2)

С помощью multiprocessing.Pool с imap (генератор) функция, кажется, работает:

pool = multiprocessing.Pool(ncpus)
results = pool.imap(Myfunction, 
                    ( (x[i], y[i], z[i]) for i in range(len(x)) )
M1, M2 = reduce(lambda r1, r2: (r1[0] + r2[0], r1[1] + r2[1]), 
                    (result for result in results))
pool.close()
pool.join()

где я изменил x а также y в лямбда-выражении r1 а также r2 чтобы избежать путаницы с другими переменными с таким же именем. При попытке использовать генератор с multiprocessing У меня проблемы с рассолом.

Единственное разочарование в этом решении заключается в том, что оно не сильно ускоряет вычисления. Я думаю, это связано с операциями над головой. При использовании 8 ядер скорость обработки была увеличена примерно на 10%. При уменьшении до 4 ядер скорость удвоилась. Кажется, это лучшее, что я могу сделать со своей конкретной задачей, если только нет другого способа сделать распараллеливание...

imap Функция была необходима для использования здесь, так как map будет хранить все возвращенные значения в памяти до reduce операция, и в этом случае это было бы невозможно.