Как объединить результаты всех процессов Pool?

Question

[Vladislav_Polikarpov]

Не могу объединить результаты всех процессов Pool. Словарь outcomes содержит лишь 288 элементов из 411 необходимых. Вероятно это 1 или 2 потока из 3-х. Ниже код.

from itertools import repeat
from multiprocessing import Pool, Manager


def reader(entry, outcomes): #other args are constant
   .............................................
   #Some condition:
   prediction = min(distances) + (mlc_data[min(distances)[1]],)
   outcomes[entry_pairs[' '.join(entry)]] = prediction
   return outcomes

manager = Manager()
outcomes = manager.dict()
with Pool(3) as p:
  output = p.starmap(reader, zip(input[0], repeat(outcomes)))
  p.close()
  p.join()

В тоже время словарь output содержит 411** 2 элементов, что указывает на то, что код плохо оптимизирован.

Comments

[mayton2019]

В тоже время словарь output содержит 411** 2 элементов, что указывает на то, что код плохо оптимизирован.

Какое-то странное утверждение. Обычно говорят что код плохо оптимизирован когда он должен
работать 2 минуты а работает 2 часа.

Когда код выдает не верные результаты - он просто некорректен. Но это не про оптимизацию.
Это про другое.

Answer 1

[Михаил Р.]

Используйте outcomes после выполнения Pool, попробуйте так:

from itertools import repeat
from multiprocessing import Pool, Manager

def reader(entry, outcomes):
    # ...
    
    prediction = min(distances) + (mlc_data[min(distances)[1]],)
    outcomes[entry_pairs[' '.join(entry)]] = prediction

manager = Manager()
outcomes = manager.dict()

with Pool(3) as p:
    # Обрабатываем каждый элемент из input[0] и сохраняем результаты в outcomes:
    p.starmap(reader, zip(input[0], repeat(outcomes)))
    p.close()
    p.join()

print(outcomes)  # должен содержать все элементы

Comments

[Vladislav_Polikarpov]

Спасибо! Работает. Можете объяснить как print влияет на наполнение словаря?

[Михаил Р.]

Vladislav_Polikarpov,

Спасибо! Работает. Можете объяснить как print влияет на наполнение словаря?

print сам по себе не влияет на наполнение словаря, он лишь печатает то, что находится внутри него.

В Вашем случае, outcomes является Manager().dict(), что означает, что доступ к этому словарю происходит в безопасном для процессов режиме. Когда разные процессы пытаются записать данные в outcomes, они делают это последовательно, чтобы избежать конфликтов. Это гарантирует, что Ваш словарь обновляется корректно, но это также может замедлить процессы, особенно если обновления частые.

Когда Вы вызываете print(outcomes) после выполнения Pool, вы просто выводите текущее состояние словаря после завершения всех процессов. На это момент все изменения в словаре уже были сделаны.

[Vladislav_Polikarpov]

Насчет Manager().dict() мне все понятно, равно как и про его влияние на скорость. Вопрос лишь в том, почему все стало ок. Потому, что убрали return и присвоение?

[Михаил Р.]

Vladislav_Polikarpov,
Да, ранее reader() возвращала outcomes, который затем собирался в список output через starmap. Но это было лишним, потому что outcomes уже являлся общим ресурсом (совместно используемым словарем) между всеми процессами