Правильное использование MPI с многопоточными функциями NumPy
Прочитав документацию mpi4py, я не могу найти никакой информации об использовании MPI с многопоточными приложениями NumPy. То есть большая часть документации mpi4py, связанной с NumPy, включает операции перемещения данных (например, Bcast, Scatter, Gather), но меня интересует распараллеливание более тяжелых вычислительных процедур, например, решение линейной системы с помощью np.linalg.solve, что я понимаю может использовать несколько ядер/потоков, когда NumPy связан с несколькими библиотеками. Чего не хватает в документации, так это указаний по количеству используемых процессов MPI, когда каждый процесс сам может использовать несколько потоков/ядер, как в примере ниже. Аналогично, если бы существовал способ определить точное количество потоков/ядер, используемых данной процедурой NumPy, то, вероятно, можно было бы использовать некоторую базовую арифметику, чтобы определить, сколько процессов MPI использовать.
Как я могу рассуждать о количестве процессов MPI, которые можно/должно использовать, если целью является распараллеливание программы NumPy, которая сама использует несколько потоков?
Следующий базовый пример на моем ноутбуке (с поддержкой 12 потоков) имеет среду выполнения, которая масштабируется по существу линейно в зависимости от количества запущенных процессов MPI, что наводит меня на мысль, что код сериализуется «за кулисами», скорее всего, с момента появления BLAS/LAPACK. процедура уже использует несколько ядер/потоков.
import time
from mpi4py import MPI
import numpy as np
import numpy.random as npr
world = MPI.COMM_WORLD
world_size: int = world.Get_size()
rank: int = world.Get_rank()
master: bool = (rank == 0)
n: int = 8_000 # corresponds to ~2.5s runtime on my laptop
A: np.ndarray = npr.randint(0, 10, (n, n))
b: np.ndarray = npr.randint(0, 10, (n,))
start: float = time.perf_counter()
_ = np.linalg.solve(A, b)
runtime: float = time.perf_counter() - start
if master:
print(f"{runtime=:.3f}s")
Я ожидал, что запуск с помощью mpiexec -n 2 будет таким же, как и mpiexec -n 1, но это не так, это занимает примерно вдвое больше времени. Обратите внимание: если я использую не np.linalg.solve, а функцию типа time.sleep, использование вдвое большего количества процессов не увеличит время выполнения.
Обновлено: Чтобы прояснить то, что здесь кажется недоразумением, этот пример призван заменить приложение, в котором мы надеемся выполнить отдельное, независимое решение линейной системы (по одному на каждый процесс). То есть данные (A, b) будут разными для каждого процесса.
1. Количество процессов MPI, умноженное на количество потоков, равно количеству ядер. 2. Ваш код выполняет одно и то же решение для каждого процесса. Это кажется бессмысленным.
Процессы независимы. Каждый процесс решает одну и ту же задачу (в данном случае, возможно, это одна и та же система nxn, хотя это зависит от того, как здесь параллельно работает генератор случайных чисел numpy). Но тот факт, что это занимает вдвое больше времени, мне кажется не имеет смысла, поскольку каждый процесс должен выполнить задачу примерно за одинаковое время. Поэтому я ожидаю, что прошедшее время будет постоянным, а не возрастающим линейно (и определенно не уменьшающимся).
@VictorEijkhout это педагогический пример. В реальном приложении, конечно, A и b различны для каждого процесса.
@jared, верно, я ожидал бы, что время выполнения будет постоянным, но на самом деле оно линейно зависит от количества процессов (т. е. количества независимых задач), что для меня не имеет смысла.
@GillesGouaillardet последнее: они решают одну и ту же систему. Это интересная идея о пропускной способности памяти, и, похоже, она согласуется с результатами, которые я получаю, когда таким же образом измеряю операцию, явно связанную с памятью, например сложение векторов.
Numpy может использовать все ядра «под капотом», поэтому, когда операция привязана к вычислениям, вы в конечном итоге делитесь ресурсами при увеличении количества задач MPI и, следовательно, увеличивается затраченное время, поскольку предстоит выполнить больше работы.
Ответы 1
Как я могу рассуждать о количестве процессов MPI, которые можно/должно использовать, если целью является распараллеливание программы NumPy, которая сама использует несколько потоков?
В общем, вы хотите, чтобы уровень параллелизма NumPy, умноженный на уровень параллелизма MPI, равнялся количеству логических ядер.
Разумное предположение для этого, если у вас есть два вложенных уровня параллелизма, состоит в том, что количество процессов MPI должно быть квадратным корнем из числа логических ядер, округленным до ближайшего целого числа, и что параллелизм NumPy должен быть числом логических ядер, разделенных на количество процессов MPI, округленных до ближайшего целого числа.
В вашем случае это приведет к 3 процессам MPI, каждый из которых имеет пул потоков NumPy, содержащий 4 потока.
Однако я бы подчеркнул, что это всего лишь отправная точка, и вам следует экспериментально определить, что лучше всего подходит для вашего приложения. Как минимум, вам также следует попробовать отсутствие параллелизма MPI, полный параллелизм NumPy, а также полный параллелизм MPI и отсутствие параллелизма NumPy.
Аналогично, если бы существовал способ определить точное количество потоков/ядер, используемых данной процедурой NumPy, то, вероятно, можно было бы использовать некоторую базовую арифметику, чтобы определить, сколько процессов MPI использовать.
В общем, если вы его не настроите, NumPy будет использовать все ядра для алгоритмов, которые можно распараллелить, и одно ядро для алгоритмов, которые не могут быть распараллелены.
Алгоритмы, которые могут использовать несколько ядер, обычно не документируются и могут меняться в разных версиях вашей библиотеки BLAS. По этой причине предлагаю проверить экспериментально.
Если вы хотите экспериментально проверить, использует ли конкретный алгоритм NumPy несколько ядер, вы можете измерить реальное время, используя time.perf_counter_ns(), и измерить тактовую частоту процессора, используя time.process_time_ns(). Изменение тактовой частоты процессора, разделенное на изменение реального времени, дает средний уровень параллелизма для процесса.
Пример:
import numpy as np
import time
def measure_func(func, repeat=10):
time.sleep(1)
real_start = time.perf_counter_ns()
cpu_start = time.process_time_ns()
for i in range(repeat):
func()
real_end = time.perf_counter_ns()
cpu_end = time.process_time_ns()
real_delta = real_end - real_start
cpu_delta = cpu_end - cpu_start
print(f"Time taken: {real_delta / 1e9:.3f}s")
print(f"CPU time taken: {cpu_delta / 1e9:.3f}s")
print(f"Average concurrency: {cpu_delta / real_delta:.3f}")
A = np.random.rand(1000, 1000)
b = np.random.rand(1000, 1000)
print("Matrix solve")
measure_func(lambda: np.linalg.solve(A, b))
print("Adding matrix")
measure_func(lambda: A + b, repeat=1000)
Выбор NumPy использовать все ядра или одно ядро обычно не является оптимальным. Я бы рекомендовал настроить. Вы можете сделать это либо через переменную окружения OMP_NUM_THREADS, либо с помощью библиотеки threadpoolctl. Я рекомендую библиотеку threadpoolctl. Это позволяет вам изменять размер пула потоков во время работы вашей программы и это более удобно. Попробуйте изменить приведенную выше программу так, чтобы np.linalg.solve() использовало только одно ядро.
Просто обратите внимание, что threadpoolctl, похоже, не работает должным образом на моем MacBook Pro M3 на момент написания этой статьи. Их конструкции ограничения потоков, похоже, не имеют никакого эффекта, но когда я проверил отдельный экземпляр Ubuntu/x86, это руководство оказалось полезным. Спасибо!
@NickRichardson Это похоже на ошибку. Вы не против открыть проблему в системе отслеживания проблем threadpoolctl, чтобы проинформировать их?
Вы уверены, что обе задачи MPI взаимодействуют для решения системы? Или они оба решают одну и ту же систему дважды? В последнем случае, поскольку ядра и пропускная способность памяти являются общими, неудивительно, что затраченное время увеличивается линейно с количеством задач.