Распределённые вычисления с использованием SharedArray

Введение

SharedArray – это специальный тип массива, который позволяет нескольким процессам на одном устройстве иметь прямой доступ к одним и тем же данным в памяти, без необходимости копировать их между процессами.

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

В данной работе представлен минимальный, но показательный пример: распределённое суммирование элементов матрицы случайных чисел двумя независимыми процессами. На этой модельной задаче иллюстрируются базовые операции:addprocs, @spawnat, wait, SharedArray. А также демонстрируется потенциальный выигрыш в производительности при переходе от последовательной схемы вычислений к параллельной.

Присоединение библиотек

Подключим необходимые библиотеки:

Distributed– для организации многопроцессности;
TickTock– для измерения времени выполнения.

for lib in ["Distributed", "TickTock"]
    try
        eval(Meta.parse("using $lib"))
    catch ex
        import Pkg
        Pkg.add(lib)
        eval(Meta.parse("using $lib"))
    end
end

Подготовка двух рабочих процессов

Запустим два дополнительных рабочих процесса командой addprocs(2) и убедимся, что они успешно зарегистрированы, выведя список активных воркеров через workers(). Получим вектор, идентификаторы которого будут использоваться далее для адресации заданий.

addprocs(2)
workers()

2-element Vector{Int64}:
 2
 3

Создание объектов SharedArray

Создадим две структуры данных в разделяемой памяти:

первая – SA, двумерный общий массив 6×2, заполненный случайными числами; он играет роль исходных данных.
вторая – SR, общий вектор из двух элементов, инициализированный нулями; он предназначен для накопления результатов, вычисленных разными процессами.

Использование SharedArray избавляет от необходимости копировать данные между процессами.

using SharedArrays
SA = convert(SharedArray{Float64, 2}, rand(6,2))

6×2 SharedMatrix{Float64}:
 0.783205  0.327205
 0.432006  0.914733
 0.390884  0.326474
 0.305342  0.319739
 0.242857  0.0980725
 0.59598   0.954856

SR = SharedArray{Float64,1}(2)

2-element SharedVector{Float64}:
 0.0
 0.0

Тесты распределённых вычислений

Проверим базовый механизм распределённых вычислений: с помощью макроса @spawnat отправим процессу 2 задание вычислить сумму элементов первого столбца SA и записать результат в SR[1].

@spawnat 2 SR[1] = sum(SA[:,1])

Future(2, 1, 18, ReentrantLock(), nothing)

display(SR)

2-element SharedVector{Float64}:
 0.0
 0.0

Аналогичное задание для второго столбца отправим процессу 3. Обратим внимание на важную деталь: @spawnat возвращает объект Future и не блокирует главный процесс, вычисления выполняются асинхронно. Поэтому немедленная проверка SR может показать ещё нулевые значения — это ожидаемое поведение, иллюстрирующее асинхронную природу распределённых заданий.

@spawnat 3 SR[2] = sum(SA[:,2])

Future(3, 1, 19, ReentrantLock(), nothing)

Сравнение производительности распределённого и однопроцессного вычисления

Теперь проведём более тяжёлый эксперимент с количеством итераций N = 10⁷. Сначала выполним вычисления последовательно в главном процессе, затем распределённо на рабочих процессах 2 и 3, дожидаясь их завершения через wait.([f1, f2]). Время в обоих случаях замерим с помощью tick() и tock(). Сравнение покажет, что распределённая версия быстрее: при таком объёме работы накладные расходы на параллельную организацию многократно перекрываются выигрышем от одновременного выполнения двух циклов на разных ядрах.

N = 10^7

1000000

# один процесс
SR[1:2] .= 0.0
display(SR)
tick()
SR[1] = begin x = 0.0; for i in 1:N x += sum(SA[:,1]); end; x; end;
SR[2] = begin x = 0.0; for i in 1:N x += sum(SA[:,2]); end; x; end;
display(SR)
tock()

2-element SharedVector{Float64}:
 0.0
 0.0

[ Info:  started timer at: 2026-10-15T05:17:04.778

2-element SharedVector{Float64}:
 2.750272925045633e6
 2.941078833171551e6

[ Info:          6.015979337s: 6 seconds, 15 milliseconds

# параллельные вычисления
SR[1:2] .= 0.0
display(SR)
tick()
f1 = @spawnat 2 SR[1] = begin x = 0.0; for i in 1:N x += sum(SA[:,1]); end; x; end;
f2 = @spawnat 3 SR[2] = begin x = 0.0; for i in 1:N x += sum(SA[:,2]); end; x; end;
@time wait.([f1, f2])
display(SR)
tock()

2-element SharedVector{Float64}:
 0.0
 0.0

[ Info:  started timer at: 2026-10-15T05:17:02.136

  1.802403 seconds (200.93 k allocations: 10.613 MiB, 10.99% compilation time)

2-element SharedVector{Float64}:
 2.750272925045633e6
 2.941078833171551e6

[ Info:          2.172321257s: 2 seconds, 172 milliseconds

Очистка ресурсов

Завершим рабочие процессы командой rmprocs(workers()) и убедимся, что в системе остался только главный процесс. Команда procs() вернёт вектор 1. Корректное освобождение ресурсов является важной частью работы с распределёнными системами, поскольку незавершённые рабочие процессы воркеры могут занимать память и процессорное время.

rmprocs(workers())
procs()

1-element Vector{Int64}:
 1

Заключение

Данный пример охватывает все основные этапы работы с распределёнными вычислениями в Engee: запуск рабочих процессов, создание разделяемых массивов, асинхронное распределение заданий и оценку ускорения. Связка модуля Distributed и типа SharedArray обеспечивает параллельную обработку данных без явного обмена сообщениями, а задания, возвращаемые @spawnat, требуют синхронизации через wait там, где важен порядок вычислений. Даже на тривиальной задаче суммирования достигнуто почти трёхкратное ускорение (≈2,17 с против ≈6,02 с).

Представленная схема лежит в основе широкого класса задач: параллельная агрегация больших массивов данных, независимые прогоны моделей в методах Монте-Карло, перебор гиперпараметров и обучение ансамблей в машинном обучении, многовариантный инженерный анализ, а также решение уравнений в частных производных и геномный анализ в научных вычислениях.

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

Распределённые вычисления