Почему я не вижу ускорения с помощью многопроцессорности в Python?

Я пытаюсь распараллелить неловко параллельный цикл (ранее заданный здесь) и поселился на эта реализация, которая соответствует моим параметрам:

    with Manager() as proxy_manager:
        shared_inputs = proxy_manager.list([datasets, train_size_common, feat_sel_size, train_perc,
                                            total_test_samples, num_classes, num_features, label_set,
                                            method_names, pos_class_index, out_results_dir, exhaustive_search])
        partial_func_holdout = partial(holdout_trial_compare_datasets, *shared_inputs)

        with Pool(processes=num_procs) as pool:
            cv_results = pool.map(partial_func_holdout, range(num_repetitions))

Причина, по которой мне нужно использовать прокси-объект (разделяемый между процессами), является первым элементом в общем списке прокси-сервера datasets, который представляет собой список крупных объектов (каждый около 200-300 МБ). Этот список datasets обычно имеет 5-25 элементов. Обычно мне нужно запустить эту программу в кластере HPC.

Вот вопрос, когда я запускаю эту программу с 32 процессами и 50 ГБ памяти (num_repetitions = 200, с наборами данных, являющимися списком из 10 объектов, каждый 250 МБ), я не вижу ускорения даже в 16 раз ( с 32 параллельными процессами). Я не понимаю, почему - какие-то подсказки? Любые очевидные ошибки или плохие выборы? Где я могу улучшить эту реализацию? Любые альтернативы?

Я уверен, что это обсуждалось ранее, и причины могут быть разными и очень конкретными для реализации, поэтому я прошу вас предоставить мне свои 2 цента. Спасибо.

Обновление. Я сделал некоторое профилирование с помощью cProfile, чтобы получить лучшую идею. Вот некоторая информация, отсортированная по кумулятивному времени.

In [19]: p.sort_stats('cumulative').print_stats(50)
Mon Oct 16 16:43:59 2017    profiling_log.txt

         555404 function calls (543552 primitive calls) in 662.201 seconds

   Ordered by: cumulative time
   List reduced from 4510 to 50 due to restriction <50>

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
    897/1    0.044    0.000  662.202  662.202 {built-in method builtins.exec}
        1    0.000    0.000  662.202  662.202 test_rhst.py:2(<module>)
        1    0.001    0.001  661.341  661.341 test_rhst.py:70(test_chance_classifier_binary)
        1    0.000    0.000  661.336  661.336 /Users/Reddy/dev/neuropredict/neuropredict/rhst.py:677(run)
        4    0.000    0.000  661.233  165.308 /Users/Reddy/anaconda/envs/py36/lib/python3.6/threading.py:533(wait)
        4    0.000    0.000  661.233  165.308 /Users/Reddy/anaconda/envs/py36/lib/python3.6/threading.py:263(wait)
       23  661.233   28.749  661.233   28.749 {method 'acquire' of '_thread.lock' objects}
        1    0.000    0.000  661.233  661.233 /Users/Reddy/anaconda/envs/py36/lib/python3.6/multiprocessing/pool.py:261(map)
        1    0.000    0.000  661.233  661.233 /Users/Reddy/anaconda/envs/py36/lib/python3.6/multiprocessing/pool.py:637(get)
        1    0.000    0.000  661.233  661.233 /Users/Reddy/anaconda/envs/py36/lib/python3.6/multiprocessing/pool.py:634(wait)
    866/8    0.004    0.000    0.868    0.108 <frozen importlib._bootstrap>:958(_find_and_load)
    866/8    0.003    0.000    0.867    0.108 <frozen importlib._bootstrap>:931(_find_and_load_unlocked)
    720/8    0.003    0.000    0.865    0.108 <frozen importlib._bootstrap>:641(_load_unlocked)
    596/8    0.002    0.000    0.865    0.108 <frozen importlib._bootstrap_external>:672(exec_module)
   1017/8    0.001    0.000    0.863    0.108 <frozen importlib._bootstrap>:197(_call_with_frames_removed)
   522/51    0.001    0.000    0.765    0.015 {built-in method builtins.__import__}

Данные профилирования теперь отсортированы по time

In [20]: p.sort_stats('time').print_stats(20)
Mon Oct 16 16:43:59 2017    profiling_log.txt

         555404 function calls (543552 primitive calls) in 662.201 seconds

   Ordered by: internal time
   List reduced from 4510 to 20 due to restriction <20>

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
       23  661.233   28.749  661.233   28.749 {method 'acquire' of '_thread.lock' objects}
   115/80    0.177    0.002    0.211    0.003 {built-in method _imp.create_dynamic}
      595    0.072    0.000    0.072    0.000 {built-in method marshal.loads}
        1    0.045    0.045    0.045    0.045 {method 'acquire' of '_multiprocessing.SemLock' objects}
    897/1    0.044    0.000  662.202  662.202 {built-in method builtins.exec}
        3    0.042    0.014    0.042    0.014 {method 'read' of '_io.BufferedReader' objects}
2037/1974    0.037    0.000    0.082    0.000 {built-in method builtins.__build_class__}
      286    0.022    0.000    0.061    0.000 /Users/Reddy/anaconda/envs/py36/lib/python3.6/site-packages/scipy/misc/doccer.py:12(docformat)
     2886    0.021    0.000    0.021    0.000 {built-in method posix.stat}
       79    0.016    0.000    0.016    0.000 {built-in method posix.read}
      597    0.013    0.000    0.021    0.000 <frozen importlib._bootstrap_external>:830(get_data)
      276    0.011    0.000    0.013    0.000 /Users/Reddy/anaconda/envs/py36/lib/python3.6/sre_compile.py:250(_optimize_charset)
      108    0.011    0.000    0.038    0.000 /Users/Reddy/anaconda/envs/py36/lib/python3.6/site-packages/scipy/stats/_distn_infrastructure.py:626(_construct_argparser)
     1225    0.011    0.000    0.050    0.000 <frozen importlib._bootstrap_external>:1233(find_spec)
     7179    0.009    0.000    0.009    0.000 {method 'splitlines' of 'str' objects}
       33    0.008    0.000    0.008    0.000 {built-in method posix.waitpid}
      283    0.008    0.000    0.015    0.000 /Users/Reddy/anaconda/envs/py36/lib/python3.6/site-packages/scipy/misc/doccer.py:128(indentcount_lines)
        3    0.008    0.003    0.008    0.003 {method 'poll' of 'select.poll' objects}
     7178    0.008    0.000    0.008    0.000 {method 'expandtabs' of 'str' objects}
      597    0.007    0.000    0.007    0.000 {method 'read' of '_io.FileIO' objects}

Дополнительная информация профилирования, отсортированная по percall info:

Обновление 2

Элементы в большом списке datasets, о которых я упоминал ранее, обычно не такие большие - они обычно 10-25 МБ каждый. Но в зависимости от используемой точности с плавающей запятой, количества образцов и функций, это может легко вырасти до 500 МБ-1 ГБ на элемент. поэтому я бы предпочел решение, которое может масштабироваться.

Обновление 3:

Код внутри holdout_trial_compare_datasets использует метод GridSearchCV scikit-learn, который внутренне использует библиотеку joblib, если мы установили n_jobs > 1 (или когда бы мы даже не установили его). Это может привести к некорректному взаимодействию между многопроцессорной обработкой и joblib. Поэтому попробуйте другую конфигурацию, где я вообще не устанавливаю n_jobs (что по умолчанию не должно быть parallelism в scikit-learn). Будет держать вас в курсе.

Ответ 1

Основываясь на обсуждении в комментариях, я сделал мини-эксперимент, сравнив три версии реализации:

v1: в основном так же, как и ваш подход, на самом деле, поскольку partial(f1, *shared_inputs) немедленно распаковывает proxy_manager.list, Manager.List не участвует здесь, данные передаются работнику с внутренней очередью Pool.
v2: v2 использовал Manager.List, рабочая функция получит объект ListProxy, он извлекает общие данные через внутреннее соединение с серверным процессом.
v3: дочерние данные процесса передаются от родителя, используйте системный вызов fork(2).

def f1(*args):
    for e in args[0]: pow(e, 2)

def f2(*args):
    for e in args[0][0]: pow(e, 2)

def f3(n):
    for i in datasets: pow(i, 2)

def v1(np):
    with mp.Manager() as proxy_manager:
        shared_inputs = proxy_manager.list([datasets,])
        pf = partial(f1, *shared_inputs)
        with mp.Pool(processes=np) as pool:
            r = pool.map(pf, range(16))

def v2(np):
    with mp.Manager() as proxy_manager:
        shared_inputs = proxy_manager.list([datasets,])
        pf = partial(f2, shared_inputs)
        with mp.Pool(processes=np) as pool:
            r = pool.map(pf, range(16))

def v3(np):
    with mp.Pool(processes=np) as pool:
        r = pool.map(f3, range(16))

datasets = [2.0 for _ in range(10 * 1000 * 1000)]
for f in (v1, v2, v3):
    print(f.__code__.co_name)
    for np in (2, 4, 8, 16):
        s = time()
        f(np)
        print("%s %.2fs" % (np, time()-s))

полученных на 16-ядерном E5-2682 VPC, очевидно, что v3 лучше масштабируется:

Ответ 2

{method 'acquire' of '_thread.lock' objects}

Глядя на ваш профайлер, я бы сказал, что накладные расходы на блокировку/разблокирование объектов перекрывают скорость получения многопоточности.

Рефакторинг, чтобы работа была обработана для работников, которым не нужно разговаривать друг с другом.

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

Вот почему очереди могут казаться намного быстрее: они связаны с типом работы, которая не требует объекта, который должен быть "управляемым" и поэтому заблокирован/разблокирован.

Только "управлять" вещами, которые абсолютно необходимо разделить между процессами. В вашем управляемом списке есть несколько очень сложных объектов...

Более быстрая парадигма:

allwork = manager.list([a, b,c])
theresult = manager.list()

а затем

while mywork:
    unitofwork = allwork.pop()
    theresult = myfunction(unitofwork)

Ответ 3

Если вам не нужен сложный общий объект, используйте только список самых простых объектов, которые можно себе представить.

Затем сообщите рабочим о приобретении сложных данных, которые они могут обрабатывать в своем маленьком мире.

Try:

allwork = manager.list([datasetid1, datasetid2 ,...])
theresult = manager.list()

while mywork:
    unitofworkid = allwork.pop()
    theresult = myfunction(unitofworkid)

def myfunction(unitofworkid):
    thework = acquiredataset(unitofworkid)
    result = holdout_trial_compare_datasets(thework, ...)

Я надеюсь, что это имеет смысл. Это не должно занимать слишком много времени для рефакторинга в этом направлении. И вы должны увидеть, что {метод "приобретать" объектов "_thread.lock" } падает при сканировании как камень.