Многопроцессорство в трубопроводе выполнено правильно

Ответ 1

Что было бы неправильно с использованием идеи 1, но с каждым рабочим процессом (f2), когда он был создан пользовательский объект с его идентификатором? Затем f3 просто прекратит работу этого работника, пока не останется рабочий процесс.

Кроме того, новый в Python 3.2 представляет собой пакет concurrent.futures в стандартной библиотеке, который должен делать то, что вы пытаетесь "правильно" (tm) - http://docs.python.org/dev/library/concurrent.futures.html

Возможно, можно найти backport of concurrent.futures для серии Python 2.x.

Ответ 2

С MPipe модуль, просто выполните следующее:

from mpipe import OrderedStage, Pipeline

def f1(value):
    return value * 2

def f2(value):
    print(value)

s1 = OrderedStage(f1, size=2)
s2 = OrderedStage(f2)
p = Pipeline(s1.link(s2))

for task in 1, 2, 3, 4, 5, None:
    p.put(task)

В приведенном выше примере выполняется 4 процесса:

• два для первого этапа (функция f1)
• один для второго этапа (функция f2)
• и еще один для основной программы, которая передает конвейер.

кулинарная книга MPipe предлагает некоторое объяснение того, как процессы закрываются внутри, используя None в качестве последней задачи.

Чтобы запустить код, установите MPipe:

virtualenv venv
venv/bin/pip install mpipe
venv/bin/python prog.py

Вывод:

2
4
6
8
10

Ответ 3

Для идеи 1, как насчет:

import multiprocessing as mp

sentinel=None

def f2(inq,outq):
    while True:
        val=inq.get()
        if val is sentinel:
            break
        outq.put(val*2)

def f3(outq):
    while True:
        val=outq.get()
        if val is sentinel:
            break
        print(val)

def f1():
    num_workers=2
    inq=mp.Queue()
    outq=mp.Queue()
    for i in range(5):
        inq.put(i)
    for i in range(num_workers):        
        inq.put(sentinel)
    workers=[mp.Process(target=f2,args=(inq,outq)) for i in range(2)]
    printer=mp.Process(target=f3,args=(outq,))
    for w in workers:
        w.start()
    printer.start()
    for w in workers:
        w.join()
    outq.put(sentinel)
    printer.join()

if __name__=='__main__':
    f1()

Единственное отличие от описания Идеи 1 состоит в том, что f2 выходит из while-loop, когда получает дозорный (таким образом, завершая себя). f1 блокируется до тех пор, пока работники не будут выполнены (используя w.join()), а затем отправит f3 дозорный сигнал (сигнализируя, что он вырвался из его while-loop).

Ответ 4

Самый простой способ сделать это с помощью семафоров.

F1

F1 заполняет "очередь" данными, которые вы хотите обработать. Завершите конец этого нажатия, вы поместите n 'Stop' в свою очередь. n = 2 для вашего примера, но обычно количество задействованных работников. Код будет выглядеть так:

for n in no_of_processes:
    tasks.put('Stop')

F2

F2 вытаскивает из предоставленной очереди команду get. Элемент берется из очереди и удаляется в очереди. Теперь вы можете поместить pop в цикл, обращая внимание на сигнал останова:

for elem in iter(tasks.get, 'STOP'):
   do something

F3

Это немного сложно. Вы можете сгенерировать семафор в F2, который действует как сигнал к F3. Но вы не знаете, когда поступит этот сигнал, и вы можете потерять данные. Однако F3 вытаскивает данные так же, как F2, и вы можете вставить это в try... except -statement. queue.get вызывает queue.Empty, когда в очереди нет элементов. Таким образом, ваше нажатие F3 будет выглядеть так:

while control:
    try:
        results.get()
    except queue.Empty:
        control = False

С tasks и results очереди. Поэтому вам не нужно ничего, что еще не включено в Python.

Ответ 5

Я использую concurent.futures и три пула, которые соединяются вместе через future.add_done_callback. Затем я жду завершения всего процесса, вызвав shutdown в каждом пуле.

from concurrent.futures import ProcessPoolExecutor
import time
import random


def worker1(arg):
    time.sleep(random.random())
    return arg


def pipe12(future):
    pool2.submit(worker2, future.result()).add_done_callback(pipe23)


def worker2(arg):
    time.sleep(random.random())
    return arg


def pipe23(future):
    pool3.submit(worker3, future.result()).add_done_callback(spout)


def worker3(arg):
    time.sleep(random.random())
    return arg


def spout(future):
    print(future.result())


if __name__ == "__main__":
    __spec__ = None  # Fix multiprocessing in Spyder IPython
    pool1 = ProcessPoolExecutor(2)
    pool2 = ProcessPoolExecutor(2)
    pool3 = ProcessPoolExecutor(2)
    for i in range(10):
        pool1.submit(worker1, i).add_done_callback(pipe12)
    pool1.shutdown()
    pool2.shutdown()
    pool3.shutdown()