Распределённое программирование
Распределённое программирование — процесс создания компьютерных программ, способных выполняться на нескольких или многих компьютерах сразу. То же: distributed programming.
Поддержка из коробки распределённого программирования — один из базовых принципов Эрланга, его серьёзное конкурентное преимущество перед многими другими языками программирования. Вместо того, чтобы спаунить акторы на текущей ноде (данном компьютере), мы спауним акторы на другой ноде другого компьютера. Получив пид созданного актора, далее мы с ним работаем как с обычным пидом обычного актора: посылаем сообщения, принимаем их, создаём линки и т.д.
Преимущества распределённого программирования:
- Производительность. Мы можем ускорить выполнение заданий, распределив их на разных машинах и задействовав таким образом объединённые мощности процессоров.
- Надёжность. Мы можем обеспечить отказоустойчивость Системы за счёт выполнения её на нескольких или многих машинах. Если одна из них выйдет из строя, другие это компенсируют.
- Масштабируемость. Даже самого мощного компьютера рано или поздно может не хватить для выполнения задач. Добавление новой машины в Эрланге не требует кардинальных изменений архитектуры Системы.
- Прагматичное распределение. Если пользователи нашей Системы разбросаны по всему миру, то не очень выгодно держать один сервер в одной точке планеты. Возможно, на каком-то этапе нам захочется иметь вычислительные мощности поближе к пользователям. И тогда мы создаём ряд нод, оптимизируя их распределение.
- Просто интересно. Распределённое программирование — весьма увлекательное занятие, когда можно сосредоточиться на решении творческих задач. Эрланг позволяет это.
Две основные модели распределения:
- Работа Системы (подсистемы) в надёжном окружении. Взаимосвязанные ноды работают в одной и той же локальной вычислительной сети (LAN). Если и есть доступ в интернет, то только через межсетевой экран. Технически можно распределить такую систему и по интернету, но это опасно. Зато этот вариант самый лёгкий и комфортный.
- В Системе есть ненадёжные элементы. Как правило это означает передачу данных через интернет. В этом случае правильно осуществлять распределение через сокеты. Этот вариант надёжен, но зато не так гибок и универсален. Тем не менее в Эрланге есть удобные вещи для работы с сокетами. В первую очередь — комфортная передача любых термов.
Джо Армстронг предлагает придерживаться следующего порядка при написании распределённого приложения:
- Написать и опробовать программу на одной ноде. Здесь надо сосредоточиться на ключевой логике и реализовать её. Конечно, все те места, где напрашивается конкурентность, мы сразу делаем конкурентными, то есть спауним столько акторов, сколько нужно.
- Совершенствуем свою программу и запускаем её на двух разных нодах, находящихся на одном компьютере. На этом этапе мы сосредотачиваемся на том, чтобы сделать программу распределённой.
- Тестируем свою программу на двух разных нодах, запущенных на двух разных компьютерах. Тут мы решаем разные вопросы, связанные со связью: правильно ли настроен межсетевой экран и др.
Эрланг помогает легко переходить от этапа к этапу, поэтому на каждом из них можно (и нужно) сконцентрироваться на решении текущих задач.
Строго говоря, распределённое программирование не означает, что мы распределяем в пространстве задания, которые можно было бы выполнить на одном компьютере. Часто органичным является сразу делать распределённую программу, например, когда на удалённом компьютере надо обрабатывать возникающие там файлы и результат этой обработки пересылать на другую ноду.
Пример распределённой программы
Придумаем несложную задачу, которую можно будет легко распараллелить. Допустим, у нас есть три файла, в каждом из которых содержится текст одного романа Ф.М. Достоевского. Наша задача — выявить, в каком из них большее число раз встречается слово “Петербург”.
1. Последовательная программа
Напишем сначала чисто последовательную программу.
-module(dostoev).
-export([main/0]).
main() ->
Files = ["prest.txt", "idiot.txt", "karamaz.txt"],
L = [ {analiz(X),X} || X <- Files],
lists:max(L).
analiz(F) ->
{ok,Bin} = file:read_file(F),
Text = unicode:characters_to_list(Bin),
{match, L} = re:run(Text, "Петербург", [unicode, global]),
length(L).
Функция analiz/1 возвращает число — сколько раз в заданном файле она нашла слово “Петербург”. Функция main строит список кортежей: {число слов, название файла}. Далее, составив список таких кортежей, находит максимальный (у кого первый элемент кортежа больше) и возвращает этот элемент.
1> dostoev:main().
{91,"idiot.txt"}
Мы выяснили, что из данных трёх романов чаще всего Петербург встречается в “Идиоте” — 91 раз.
2. Конкурентная программа
Теперь сделаем то же самое, но на акторах.
Конечно, архитектура разных конкурентных программ может быть очень разной, иногда весьма замысловатой. Но в данном случае у нас задача простая. Изначально есть некая общая ситуация, которую мы обрабатываем. У нас это просто определение совокупности трёх файлов, которые надо обработать (в одинаковом ключе).
Далее мы определяем несколько заданий, спауним эти задания, создавая акторы. Эти акторы работают независимо друг от друга, они даже не обмениваются сообщениями между собой (в данном случае нам это не нужно). В конце своей работы каждый актор посылает итоговое сообщение актору-родителю. Актор-родитель собирает эти сообщения в единую кучу.
После этого снова наступает общая ситуация. В данном случае нам предстоит сравнить сообщения от акторов и выяснить наконец, в каком файле чаще всего встречается слово “Петербург”. Конечно, мы это можем сделать только когда получили результаты от всех акторов. Иначе говоря, когда снова возникла общая ситуация.
Нам надо хорошо продумать переход от этапа к этапу. Переход I → II довольно понятен и очевиден. Где раньше мы вызывали функцию analiz/1, теперь будем спаунить задание. Задание будет включать модифицированную функцию analiz и передаваемый параметр — имя файла.
Переход II → III более замысловат. Тут возникает проблема полноты. Нам надо сформировать список L, в котором будут все результаты, полученные от акторов. В последовательном варианте нашей программы эта проблема решалась элементарно, результаты сразу выстраиваются в список L:
L = [ {analiz(X),X} || X <- Files],
Сколько бы раз мы ни запускали нашу программу, столько раз мы получим один и тот же результат, а именно:
[{49,"prest.txt"},{91,"idiot.txt"},{37,"karamaz.txt"}]
Сообщения от акторов могут приходить в разной последовательности, на то оно и конкурентное программирование.
Значит, волей-неволей при получении каждого сообщения нам придётся проверять полноту данных. И как только все данные будут собраны, мы перейдём к этапу III. Такого рода проблема полноты весьма характерна для конкурентного программирования.
Тут два ключевых момента.
- Обработчик сообщений так или иначе должен знать (помнить) текущее состояние: все или не все акторы отчитались о своей работе?
- Но что значит все или не все? Должен быть какой-то чёткий образ желаемого будущего (техзадание).
Обработчик, таким образом, делает вывод о полноте, основываясь на трёх моментах:
В одних случаях (как, например, в нашем) техзадание может состоять из простейшего терма — числа 3 (мы ждём трёх сообщений). В других случаях имеет смысл делать более сложное техзадание. Например, мы бы могли при первой итерации в качестве аргумента задать следующий терм-карту:
#{"prest.txt" => false, "idiot.txt" => false, "karamaz.txt" => false}
По ходу того, как приходят от акторов новые сообщения, соответствующие значения меняются с false на true. И когда везде получается true, тогда переходим к следующему этапу.
Как хранить текущее состояние, чтобы его сравнивать с техзаданием? Тут, как и в других подобных случаях, возможны два варианта:
- Хранить в терме, который будет передаваться, меняясь, к очередному вызову функции, которая обрабатывает сообщения.
- Хранить это состояние в словаре актора или ETS.
Воспользуемся первым способом (как базовым, да и более чистым). Просто от итерации к итерации будет передавать список L, куда будем добавлять сообщения, полученные от акторов. Как только в списке будет 3 элемента (что соответствует техзаданию), произойдёт переход к этапу III.
Есть ещё один момент. Надо быть всегда готовым, что могут прийти какие-то посторонние сообщения. Поэтому мы будем помечать пересылаемые сообщения атомом-тегом dostoev.
Контуры программы уже понятны, пора её написать.
-module(dostoev).
-export([main/0,analiz/2]).
main() ->
Files = ["prest.txt", "idiot.txt", "karamaz.txt"],
[spawn(dostoev,analiz,[self(),X]) || X <- Files],
TZ = length(Files), % техзадание
L = loop(TZ,[]),
lists:max(L).
analiz(Parent,F) ->
{ok,Bin} = file:read_file(F),
Text = unicode:characters_to_list(Bin),
{match, L} = re:run(Text, "Петербург", [unicode, global]),
Parent ! {dostoev,{length(L),F}}.
loop(TZ,L) ->
receive
{dostoev,Mess} -> L2 = [Mess | L],
case length(L2) of
TZ -> L2;
_ -> loop(TZ,L2)
end;
_ -> []
end.
Проверим её в работе:
1> dostoev:main().
{91,"idiot.txt"}
Результат тот же, что и у последовательной программы. Можно запустить много раз — результат не изменится.
Мы немного изменили функцию analiz. Теперь она стала принимать два аргумента. Parent нам нужна, чтобы знать, куда посылать сообщение. Функция теперь заканчивается как раз тем, что посылает результат родительскому актору.
Нам пришлось добавить функцию loop/2, чтобы она в итерациях обрабатывала приходящие сообщения. Проблему полноты мы решаем с помощью трёх переменных: TZ (техзадание), L (список сообщений на начало итерации) и Mess — новое сообщение. Если полнота не достигнута, функция loop вызывает саму себя (хвостовая рекурсия). Когда техзадание выполнено, выражение receive ... end возвращает L2 — нужный нам список. Этот же терм возвращает и функция loop/2.
3. Распределённая программа
Двигаемся дальше. Настала пора модифицировать нашу программу и запустить её на трёх разных нодах, но находящихся на одном компьютере.
Нам надо запустить три актора на трёх разных нодах. Конечно, в реальных условиях количество заданий может быть значительно больше количества доступных нод (или наоборот). Значит, нам надо найти какой-то универсальный алгоритм, как распределить акторы по доступным нодам.
1-й вариант такого универсального алгоритма — распределять акторы по случайным нодам. То есть мы перебираем задания и для каждого определяем случайную ноду. Плюс — не надо заморачиваться с равномерным распределением заданий по нодам. Минус — одной ноде может случайно перепасть слишком много заданий, поэтому работа и всей программы может замедлиться раза в 2-3.
2-й вариант — равномерное распределение заданий по нодам. Придётся заморочиться, однако нужный нам результат мы получим быстрее. На этом варианте мы и остановимся.
Предположим, у нас есть три ноды, которые обозначим буквами a, b, c, и три задания, которые обозначим цифрами 1, 2, 3. Тогда ноды и задания легко разбить по парам: a1, b2, c3. Если количество заданий увеличится до семи, тогда эти пары должны выглядеть так: a1, b2, c3, a4, b5, c6, a7. Если наоборот будет семь нод и три задания, тогда не все ноды будут задействованы: a1, b2, c3.
Придётся нам для удобства написать функцию “мозаичного зипования”:
mosaic_zip(L1,L2) ->
N = round(1+length(L2)/length(L1)),
L3 = lists:flatten(lists:duplicate(N,L1)),
lists:zip(L3,L2,trim).
Проверим её работу (1-й аргумент список нод, 2-й заданий):
1> spis:mosaic_zip([a,b,c],[1,2,3]).
[{a,1},{b,2},{c,3}]
2> spis:mosaic_zip([a,b,c],[1,2,3,4,5,6,7]).
[{a,1},{b,2},{c,3},{a,4},{b,5},{c,6},{a,7}]
3> spis:mosaic_zip([a,b,c,d,e,f,g],[1,2,3]).
[{a,1},{b,2},{c,3}]
Ноды будем запускать в трёх разных tty на одном компьютере, делая такие команды:
$ erl -sname node1
$ erl -sname node2
$ erl -sname node3
Когда нода запущена (мы использовали ключ -sname или -name), оболочка нам это показывает:
(node1@sis)1> node().
node1@sis
node1@sis это и есть имя нашей ноды (sis это имя хоста). Получить это имя можно также с помощью бифа node/0, что мы и проделали. Имена других доступных нод получаются с помощью бифа nodes/0:
(node1@sis)2> nodes().
[]
Хотя мы запустили три ноды, тут ни одна не отображается. Имя своей ноды nodes/0 не выводит, а другие ноды пока не доступны. Их надо просто пропинговать, чтобы в дальнейшем с ними можно было работать.
(node1@sis)3> net_adm:ping(node1@sis).
pong
(node1@sis)4> net_adm:ping(node2@sis).
pong
(node1@sis)5> net_adm:ping(node3@sis).
pong
(node1@sis)6> net_adm:ping(node4@sis).
pang
(node1@sis)7> nodes().
[node2@sis,node3@sis]
Заодно мы пропинговали свою ноду, а также несуществующую node4@sis. Получили атом pong (удача) и pang (неудача).
Нам понадобится список доступных нод. Раз мы будем запускать актор и на текущей ноде, её имя тоже не надо забыть добавить:
(node1@sis)8> Nodes = [node() | nodes()].
[node1@sis,node2@sis,node3@sis]
Спаунить задания будем с помощью бифа spawn/4:
(node1@sis)9> h(erlang, spawn, 4).
-spec spawn(Node, Module, Function, Args) -> pid()
when
Node :: node(),
Module :: module(),
Function :: atom(),
Args :: [term()].
Каждая запущенная нода имеет свой самостоятельный набор загруженных модулей. Когда мы спауним на другой ноде с помощью spawn/4, там будет запущена функция из тамошней ноды. То есть если у нас несколько работающих нод, мы при этом изменили один модуль и на одной ноде выполнили l(dostoev), только текущая нода будет использовать свежую версию. В оболочке вместо этого надо использовать команду nl(dostoev), и тогда свежий код будет везде. Поэтому не будет лишним добавить в код нашей программы следующую строку (после пинга нод):
c:nl(?MODULE),
Это можно рассматривать как команду синхронизации. Вряд ли стоит синхронизировать модули по несколько раз в секунду, это всё же затратная операция, но тем не менее в каких-то ключевых местах стоит позаботиться об этом.
С основными моментами разобрались, приступим к дальнейшему развитию модуля dostoev.
-module(dostoev).
-export([main/0,analiz/2]).
main() ->
Files = ["prest.txt", "idiot.txt", "karamaz.txt"],
Tasks = [ {dostoev,analiz,[self(),X]} || X <- Files],
net_adm:ping(node2@sis),
net_adm:ping(node3@sis),
c:nl(?MODULE),
Nodes = [node() | nodes()],
Plan = mosaic_zip(Nodes,Tasks),
[ spawn(Node,M,F,A) || {Node,{M,F,A}} <- Plan ],
TZ = length(Files), % техзадание
lists:max(loop(TZ,[])).
analiz(Parent,F) ->
{ok,Bin} = file:read_file(F),
Text = unicode:characters_to_list(Bin),
{match, L} = re:run(Text, "Петербург", [unicode, global]),
Parent ! {dostoev,{length(L),F}}.
loop(TZ,L) ->
receive
{dostoev,Mess} -> L2 = [Mess | L],
case length(L2) of
TZ -> L2;
_ -> loop(TZ,L2)
end;
_ -> []
end.
mosaic_zip(L1,L2) ->
N = round(1+length(L2)/length(L1)),
L3 = lists:flatten(lists:duplicate(N,L1)),
lists:zip(L3,L2,trim).
Переменная Plan содержит список всех заданий, которые в следующей строке будут спауниться. Каждое задание содержит в себе ноду, модуль, функцию, аргументы. Если угодно, можно посмотреть её содержание:
[{node1@sis,{dostoev,analiz,[<0.90.0>,"prest.txt"]}},
{node2@sis,{dostoev,analiz,[<0.90.0>,"idiot.txt"]}},
{node3@sis,{dostoev,analiz,[<0.90.0>,"karamaz.txt"]}}]
Проверим работу нашей распределённой программы:
(node1@sis)29> dostoev:main().
{91,"idiot.txt"}
Работает исправно. Для удовольствия можно добавить в функцию analiz/2 строку:
io:format(" ~p ~p~n", [node(), F]),
И тогда мы увидим, что акторы и правда выполняются на разных нодах:
(node1@sis)45> dostoev:main().
node1@sis "prest.txt"
node2@sis "idiot.txt"
node3@sis "karamaz.txt"
{91,"idiot.txt"}
(node1@sis)46> dostoev:main().
node1@sis "prest.txt"
node3@sis "karamaz.txt"
node2@sis "idiot.txt"
{91,"idiot.txt"}
Ура! У нас получилась распределённая программа.
4. Распределённая по разным хостам программа
Теперь мы запустим нашу распределённую программу на трёх разных компьютерах, находящихся в одной локальной сети.
Если в предыдущем разделе мы могли позволить запускать ноды с помощью ключа -sname, сейчас надо делать то же, но с ключом -name, при этом в названии ноды (которая имеет вид URI) должен быть чётко указан адрес хоста, на котором эту ноду будут искать другие ноды. Сейчас у нас это будет выглядеть так:
$ erl -name node1@192.168.3.16
$ erl -name node2@192.168.3.73
$ erl -name node3@192.168.3.74
Но перед этим надо не забыть сделать ещё одну вещь — убедиться, что на всех компьютерах нашей Системы имеется файл $HOME/.erlang.cookie с одинаковым содержимым. Иначе ноды не получат доступ друг к другу.
Другой вопрос, который сразу стоит продумать, — ресурсы. Для успешной работы нашей программы ей нужны ресурсы. В данном случае это файлы. В предыдущем разделе мы все три ноды запускали с одного компьютера, находясь при этом в одном и том же каталоге, содержащем три необходимых текстовых файла. Теперь мы запускаем программу на разных компьютерах, и необходимые файлы есть только на одном из них.
Тут может быть два выхода из ситуации:
- Обеспечить все компьютеры одним и тем же набором ресурсных файлов. Плюс — не надо тратиться на передачу данных по сети. Это может быть полезно, если данных очень много, а эти файлы в силу каких-то причин и так содержатся на всех компьютерах. Однако, если этих файлов нет, их придётся туда перенести, и скорее всего, по сети, хотя бы раз. Минус — ресурсные файлы могут поменяться, и тогда придётся снова совершать действия по синхронизации их.
- Ресурсные файлы читать на одном компьютере, а потом передавать бинарники в задании. Плюс — можно не беспокоиться о том, имеют ли другие ноды доступ к ресурсным файлам. Минус — это может быть медленнее. Если мы не распределяем чтение файлов, приходится отдуваться одному компьютеру, и если данные огромны, это может быть заметно.
В распределённом программировании всегда надо понимать, какие именно операции имеет смысл распределять, а какие — нет. Доступ для нод к ресурсным файлам можно было бы обеспечить с помощью какой-нибудь распределённой файловой системы. Это тоже рабочий вариант. Но мы выберем второй вариант, как более элегантный и не требующий посторонних инструментов. Будем считать, что для нас чтение и передача бинарников по сети — дешёвая операция (если сравнивать с обработкой полученных данных).
-module(dostoev).
-export([main/0,analiz/3]).
main() ->
Files = ["prest.txt", "idiot.txt", "karamaz.txt"],
Tasks = [ {dostoev,analiz,[self(),X,file:read_file(X)]} || X <- Files],
net_adm:ping('node2@192.168.3.73'),
net_adm:ping('node3@192.168.3.74'),
c:nl(?MODULE),
Nodes = [node() | nodes()],
Plan = mosaic_zip(Nodes,Tasks),
[ spawn(Node,M,F,A) || {Node,{M,F,A}} <- Plan ],
TZ = length(Files), % техзадание
lists:max(loop(TZ,[])).
analiz(Parent,F,{ok,Bin}) ->
Text = unicode:characters_to_list(Bin),
{match, L} = re:run(Text, "Петербург", [unicode, global]),
Parent ! {dostoev,{length(L),F}}.
loop(TZ,L) ->
receive
{dostoev,Mess} -> L2 = [Mess | L],
case length(L2) of
TZ -> L2;
_ -> loop(TZ,L2)
end;
_ -> []
end.
mosaic_zip(L1,L2) ->
N = round(1+length(L2)/length(L1)),
L3 = lists:flatten(lists:duplicate(N,L1)),
lists:zip(L3,L2,trim).
Изменений не очень много. В атомы-имена нод мы добавили IP-адреса. Функции analiz добавили третий аргумент — бинарник.
Следует заметить, что строка c:nl(?MODULE), автоматически загружает beam-файлы на все ноды. То есть не надо вручную копировать эти файлы по разным хостам в их файловые системы.
(node1@192.168.3.16)5> dostoev:main().
{91,"idiot.txt"}
Результат идентичен прежним — наша программа работает!
Добавим-ка снова в analiz строчку, которая показывает ноду, на которой исполняется актор, а также имя файла-аргумента.
(node1@192.168.3.16)1> dostoev:main().
'node1@192.168.3.16' "prest.txt"
'node2@192.168.3.73' "idiot.txt"
'node3@192.168.3.74' "karamaz.txt"
{91,"idiot.txt"}
Наша программа устроена достаточно гибко: список доступных нод определяется через nodes(), распределение заданий происходит автоматически по доступным нодам. Попробуем выключить одну ноду…
(node1@192.168.3.16)2> dostoev:main().
'node1@192.168.3.16' "prest.txt"
'node1@192.168.3.16' "karamaz.txt"
'node2@192.168.3.73' "idiot.txt"
{91,"idiot.txt"}
Как и следовало ожидать, одной ноде было поручено выполнить два задания. А теперь оставим только одну ноду…
(node1@192.168.3.16)3> dostoev:main().
'node1@192.168.3.16' "prest.txt"
'node1@192.168.3.16' "idiot.txt"
'node1@192.168.3.16' "karamaz.txt"
{91,"idiot.txt"}
Теперь все задания выполняются на одной ноде (конечно, конкурентно). На результате это не сказалось — всё равно победил “Идиот”.
Copyright © 2025-2026 Алексей Карманов