python map_async, откуда возникают накладные расходы?

Question

Я использую map_async по назначению - для отображения итератора над несколькими процессорными ядрами с помощью:

cores = mp.cpu_count() 
pool = mp.Pool() 

r = pool.map_async(func, offsets,callback=mycallback) 
r.wait() 

func возвращает Dict, поэтому обратный вызов «сливает» в dicts с помощью:

ddict = defaultdict(set) 
def mycallback(w): 
    for l in w: 
     for key, value in l.items(): 
      for v in value: 
       ddict[key].add(v) 

взаимозачетов повторяемый, который я тестировал с 1000 - 50 000 элементов.

Если я удаляю r.wait(), невозможно вернуть весь вывод из звонка map_async.

Использование r.wait(), я вижу времена обработки, которые уступают последовательной реализации и не масштабируются, т. Е. Параллельная реализация возрастает во времени экспоненциально, в то время как последовательная версия линейно возрастает.

Я знаю, что func достаточно дорог, как в серийном, так и параллельно он привязывает мои обрабатывающие ядра.

Где я ввел накладные расходы, используя map_async? Он не находится в функции обратного вызова, так как удаление и замена с помощью result.append не влияет на время.

РЕДАКТИР Комментариев:

1. Я двигаюсь большая dicts вокруг, где-то от 1000 - 100000 элементов. Значение - это наборы, которые обычно составляют 3-5 элементов. Таким образом, травление определенно может быть проблемой. Какие альтернативные структуры данных можно было бы предложить, не переходя к чему-то в общей памяти?

2. apply_async с аналогичным обратным вызовом, сохранить строку for l in w, возвращает примерно одинаковые результаты. Скорость немного лучше, чем map_async для некоторых наборов задач и немного хуже для других. Значительно хуже используется управляемый dict и присоединяемая очередь.

3. Некоторые испытания времени. Это использует 2 ядра. Поскольку я добавляю дополнительные ядра, я вижу экспоненциальное увеличение, поэтому могу только предположить, что это увеличение вызвано процессом нереста или травления для возврата данных.

принимает точку данных и ищет соседние страны. Это идентичная функция для всех случаев, за исключением необходимости прохождения смещений, указывающих параллельный код, который указывает данные для поиска. Это, по сути, функция поиска KDTree.

Распределенные

Гомогенно

1000 точек данных: Последовательный 0.098659992218 | apply_async 0.120759010315 | map_async 0.080078125

10 000 точек данных < ====== ТОЛЬКО УЛУЧШЕНИЕ ПАРАЛЛЕЛЬНЫМИ | Последовательный 0.507845163345 | apply_async 0.446543931961 | map_async 0,477811098099

Случайным Распределенные

10000 точек данных: Последовательный 0.584854841232 | apply_async 1.03224301338 | map_async 0.948460817337

50 000 точек данных: серийный номер 3.66075992584 | apply_async 4.95467185974 | map_async

Answer 1

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

def mycallback(w): 
    for l in w: 
     for key, value in l.items(): 
      ddict[key].update(value) 

Это должно помочь как с последовательным, так и с параллельным временем обработки.

К сожалению, я думаю, что @Dougal прав насчет травления/распиловки всех этих данных при передаче между потоками. Возможно, быстрее записать двоичные данные на диск и прочитать его снова, вместо того, чтобы передавать его в памяти из-за накладных расходов на травление. Вы можете использовать такой формат, как:

key value1 value2 value3 ... 
key2 valueA valueB valueC ... 
... 

который должен быть легким и для записи, и для чтения.