Как вы можете кодировать вложенный параллелизм в Python?

Question

Мой код имеет следующую схему:

class A():
    def evaluate(self):
        b = B()
        for i in range(30):
            b.run()

class B():
    def run(self):
        pass

if __name__ == '__main__':
    a = A()
    for i in range(10):
        a.evaluate()

И я хочу иметь два уровня параллелизма, первый на метод evaluate, а второй на метод run(вложенный параллелизм).Вопрос в том, как внедрить этот параллелизм, используя класс Pool многопроцессорного модуля ?Должен ли я передавать явно количество ядер?Решение не должно создавать процессы больше, чем число multiprocessing.cpu_count().

примечание: предположим, что количество ядер превышает 10.

Редактировать: Я видел много комментариев, в которых говорится, что Python не имеет истинного параллелизма из-заGIL, это верно для многопоточности Python, но для многопроцессорности это не совсем правильно, посмотрите здесь , также я рассчитал это также, что сделал статья , и результаты показывают, что это можетидти быстрее, чем последовательное выполнение.

Answer 1

Ваш комментарий касается возможного решения.Чтобы иметь «вложенный» параллелизм, у вас может быть 2 отдельных пула.Это привело бы к «плоской» структуре программы вместо гнездовой программы.Кроме того, он отделяет A от B, A теперь ничего не знает о b, который он просто публикует в общей очереди.В приведенном ниже примере используется один процесс для иллюстрации подключения параллельных рабочих, осуществляющих связь через асинхронную очередь, но его можно легко заменить на пул:

import multiprocessing as mp


class A():
    def __init__(self, in_q, out_q):
      self.in_q = in_q
      self.out_q = out_q

    def evaluate(self):
        """
        Reads from input does work and process output
        """
        while True:
          job = self.in_q.get()
          for i in range(30):
            self.out_q.put(i)

class B():
    def __init__(self, in_q):
      self.in_q = in_q

    def run(self):
        """
        Loop over queue and process items, optionally configure
        with another queue to "sink" the processing pipeline
        """
        while True:
           job = self.in_q.get()

if __name__ == '__main__':
    # create the queues to wire up our concurrent worker pools
    A_q = mp.Queue()
    AB_q = mp.Queue()

    a = A(in_q=A_q, out_q=AB_q)
    b = B(in_q=AB_q)

    p = mp.Process(target=a.evaluate)
    p.start()

    p2 = mp.Process(target=b.run)
    p2.start()

    for i in range(10):
        A_q.put(i)

    p.join()
    p2.join()

Это распространенный шаблон в golang.