Учитывая N генераторов, можно ли создать генератор, который запускает их в параллельных процессах и выдает zip этих генераторов?

Можно получить такой Unified Parallel Generator (UPG) (попытка придумать имя) с некоторыми усилиями, но, как уже упоминал @jasonharper, вам определенно нужно собрать подгенераторы внутри дочернего процесса. процессы, так как работающий генератор не может быть замаринован.

Приведенный ниже шаблон можно использовать повторно, только функция генератора gen() является пользовательской для этого примера. В проекте используется multiprocessing.SimpleQueue для возврата результатов генератора родительскому и multiprocessing.Barrier для синхронизации.

Вызов Barrier.wait() заблокирует вызывающую сторону (поток в любом процессе) до тех пор, пока указанное число parties не вызовет .wait(), после чего все потоки, ожидающие в данный момент Barrier, освобождаются одновременно. Использование Barrier здесь гарантирует, что дальнейшие результаты генератора начинают вычисляться только после родителя, получившего все результаты итерации, что может быть желательно для сохранения общего потребления памяти. в чеке.

Количество используемых параллельных рабочих процессов равно количеству аргументов-кортежей, которые вы предоставляете в gen_args_tuples-iterable, поэтому gen_args_tuples=zip(range(4)) будет использовать, например, четыре рабочих процесса. Дополнительные сведения см. в комментариях в коде.

import multiprocessing as mp

SENTINEL = 'SENTINEL'


def gen(a):
    """Your individual generator function."""
    lst = ['a', 'b', 'c']
    for ch in lst:
        for _ in range(int(10e6)):  # some dummy computation
            pass
        yield ch + str(a)


def _worker(i, barrier, queue, gen_func, gen_args):
    for x in gen_func(*gen_args):
        print(f"WORKER-{i} sending item.")
        queue.put((i, x))
        barrier.wait()
    queue.put(SENTINEL)


def parallel_gen(gen_func, gen_args_tuples):
    """Construct and yield from parallel generators
     build from `gen_func(gen_args)`.
     """
    gen_args_tuples = list(gen_args_tuples)  # ensure list
    n_gens = len(gen_args_tuples)
    sentinels = [SENTINEL] * n_gens
    queue = mp.SimpleQueue()
    barrier = mp.Barrier(n_gens + 1)  # `parties`: + 1 for parent

    processes = [
        mp.Process(target=_worker, args=(i, barrier, queue, gen_func, args))
        for i, args in enumerate(gen_args_tuples)
    ]

    for p in processes:
        p.start()

    while True:
        results = [queue.get() for _ in range(n_gens)]
        if results != sentinels:
            results.sort()
            yield tuple(r[1] for r in results)  # sort and drop ids
            barrier.wait()  # all workers are waiting
            # already, so this will unblock immediately
        else:
            break

    for p in processes:
        p.join()


if __name__ == '__main__':

    for res in parallel_gen(gen_func=gen, gen_args_tuples=zip(range(4))):
        print(res)

Вывод:

WORKER-1 sending item.
WORKER-0 sending item.
WORKER-3 sending item.
WORKER-2 sending item.
('a0', 'a1', 'a2', 'a3')
WORKER-1 sending item.
WORKER-2 sending item.
WORKER-3 sending item.
WORKER-0 sending item.
('b0', 'b1', 'b2', 'b3')
WORKER-2 sending item.
WORKER-3 sending item.
WORKER-1 sending item.
WORKER-0 sending item.
('c0', 'c1', 'c2', 'c3')

Process finished with exit code 0

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

from multiprocessing import Pool

pool = Pool(processes=4)

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

def parallel_generators(generators, pool):
results = ['placeholder']
while len(results) != 0:
    batch = pool.map_async(next, generators)  # defines the next round of values
    results = list(batch.get)  # actual calculation done here
    yield results
return 

Мы определяем условие результатов в цикле while таким образом, потому что объекты сопоставления с next и генераторами возвращают пустой список, когда генераторы перестают создавать значения. Итак, в этот момент мы просто отключаем параллельный генератор.

РЕДАКТИРОВАТЬ

Таким образом, очевидно, многопроцессорный пул и карта не работают с генераторами, из-за чего приведенный выше код не работает должным образом, поэтому не используйте до более позднего обновления.

Что касается ошибки рассола, кажется, что некоторые связанные функции не поддерживают рассол, который необходим в библиотеке многопроцессорной обработки для передачи объектов и функций, для обходного пути пафосная библиотека многопроцессорной обработки использует укроп, который устраняет необходимость рассола и является вариантом, который вы могли бы хотите попробовать, ищите в Stack Overflow свою ошибку, вы также можете найти более сложные решения с пользовательским кодом для выбора необходимых функций.