Как ждать и получить результат в шаблоне производитель-потребитель?

Question

У меня есть приложение, которое использует шаблон производитель-потребитель: несколько производителей, один потребитель.Подвох в том, что внутри каждого производителя, после отправки задачи, я хочу подождать и получить результат внутри того же процесса производителя :

from multiprocessing import Process
from queue import Queue


def producer(queue, work_item):
    queue.put(work_item)

    # ??? How to wait for the work_item to be done, and get back the result of processing work_item???

    # "result" is the result of producer processing this item. I want to print the result "INSIDE THE PRODUCER PROCESS"
    print("the result of processing {} is {}".format(work_item, result))


def consumer(queue)    :
    while True:
        item = queue.get()
        # process the item
        result = process(item)
        # ??? How to pass this result back to producer???
        # processing done
        queue.task_done()


if __name__== "__main__":
    queue=Queue()
    p1 = Process(target=producer, args=(queue,1))
    p2 = Process(target=producer, args=(queue,2))
    p3 = Process(target=producer, args=(queue,3))

    c1 = Process(target=consumer, args=(queue,))

    p1.start()
    p2.start()
    p3.start()
    c1.start()

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

Поскольку в моем реальном приложении:

1. производитель может быть обработчиком веб-запросов на веб-сервере Flask,
2. потребитель может быть классификатором изображения тензорного потока.

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

Так как же ждать и получить результат в шаблоне производитель-потребитель?

---

пс.Я уже попробовал сельдерей.Но здесь меня больше интересует реализация Core Python (или общая схема параллелизма), если таковая имеется.

Answer 1

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

from multiprocessing import Process, Event
from queue import Queue


def producer(queue, work_item, yoursignal):
    # do stuff
    if yoursignal.is_set():
        yoursignal.clear() # block others
        queue.put(block=True)
        yoursignal.wait() # wait for signal to become True
        print("the result of processing {} is {}".format(work_item, result))


def consumer(queue, yoursignal):
    while True:
        item = queue.get(block=True) # wait until something is available
        # process the item
        result = process(item)
        # ??? How to pass this result back to producer???
        # processing done
        queue.task_done()
        yoursignal.set()


if __name__== "__main__":
    queue = Queue(1) # only one slot in the queue
    signal = Event()
    p1 = Process(target=producer, args=(queue,1, signal))
    p2 = Process(target=producer, args=(queue,2, signal))
    p3 = Process(target=producer, args=(queue,3, signal))

    c1 = Process(target=consumer, args=(queue, signal))

    p1.start()
    p2.start()
    p3.start()
    c1.start()