Распределенные математические вычисления на языке Erlang

В статье иллюстрируется пример настройки окружения и серверов для возможности проведения распределенных вычислений на языке Erlang, а также их осуществление. В качестве примера предлагается перемножение матрицы на вектор с использованием пула pool(3) серверов Erlang. В данном случае пул состоит из 3 машин под управлением ОС FreeBSD.


Введение
Многие из вас, наверняка, уже использовали вызов spawn[_link] в своих приложениях, но немногие, как я заметил, знают о более лаконичном -- но также и более функциональном -- его аналоге pspawn[_link]. Попробуем заполнить этот пробел, решив простую математическую задачу.


Подготовка
Настройки серверов для хоста с двумя тюрьмами, расположенными на нем:

[root@conductor ~/crunching]# uname -prs
FreeBSD 7.1-RELEASE i386
[root@conductor ~/crunching]# grep '\.loc' /etc/hosts
192.168.32.40
192.168.32.201
192.168.32.202
[root@conductor ~/crunching]# jls
JID  IP Address      Hostname
 1  192.168.32.201  j1.loc
 2  192.168.32.202  j2.loc
[root@conductor ~/crunching]# ifconfig em0 | grep inet
inet 192.168.32.40 netmask 0xffffff00 broadcast 192.168.32.255
inet 192.168.32.201 netmask 0xffffff00 broadcast 192.168.32.201
inet 192.168.32.202 netmask 0xffffff00 broadcast 192.168.32.202
[root@conductor ~/crunching]# cat ~/.hosts.erlang
'j1.loc'
'j2.loc'
[root@conductor ~/crunching]# pkg_info -E erlang*
[root@conductor ~/crunching]#

Запуск мастер-ноды poolmaster@conductor Erlang на хосте, затем запуск пула ведомых нод (slaves) в тюрьмах. Возвращение списка адресов машин, составляющих пул:

[root@conductor ~/crunching]# erl -sname poolmaster -rsh ssh
Erlang R13B01 (erts-5.7.2) [source] [rq:1] [async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.7.2  (abort with ^G)
(poolmaster@conductor)1> c(pwnd).
{ok,pwnd}
(poolmaster@conductor)2> pool:start(crunchpool).
[crunchpool@j1,crunchpool@j2]
(poolmaster@conductor)3> pool:get_nodes().
[poolmaster@conductor,crunchpool@j1,crunchpool@j2]
(poolmaster@conductor)4>

Удаленное подключение к shell Erlang-ноды, находящейся на другой машине (в данном случае -- в тюрьме), и исполнение произвольного кода:

[root@conductor ~/crunching]# erl -sname client -remsh crunchpool@j1
Erlang R13B01 (erts-5.7.2) [source] [rq:1] [async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.7.2  (abort with ^G)
(crunchpool@j1)1> os:cmd("hostname").
"j1.loc\n"
(crunchpool@j1)2>

На всех узлах пула (в т. ч. на ведущем (master)) запускаются Erlang Port Mapper Daemon (epmd(1)) и beam(1):

[root@conductor ~] jexec 1 sockstat -4
USER    COMMAND   PID   FD    PROTO    LOCAL ADDRESS         FOREIGN ADDRESS
root    beam      2863  7     tcp4     192.168.32.201:63037  *:*
root    beam      2863  8     tcp4     192.168.32.201:50707  192.168.32.201:4369
root    beam      2863  10    tcp4     192.168.32.201:54136  192.168.32.40:61473
root    beam      2863  13    tcp4     192.168.32.201:63037  192.168.32.202:61991
root    epmd      1143  3     tcp4     192.168.32.201:4369   *:*
root    epmd      1143  4     tcp4     192.168.32.201:4369   192.168.32.201:50707
root    inetd     861   5     tcp4     192.168.32.201:22     *:*
root    sendmail  837   5     tcp4     192.168.32.201:25     *:*
root    syslogd   777   6     udp4     192.168.32.201:514    *:*
[root@conductor ~] jexec 2 sockstat -4
USER    COMMAND   PID   FD    PROTO    LOCAL ADDRESS         FOREIGN ADDRESS
root    beam      2880  7     tcp4     192.168.32.202:64315  *:*
root    beam      2880  8     tcp4     192.168.32.202:57386  192.168.32.202:4369
root    beam      2880  10    tcp4     192.168.32.202:50430  192.168.32.40:61473
root    beam      2880  11    tcp4     192.168.32.202:61991  192.168.32.201:63037
root    epmd      1160  3     tcp4     192.168.32.202:4369   *:*
root    epmd      1160  4     tcp4     192.168.32.202:4369   192.168.32.202:57386
root    inetd     1024  5     tcp4     192.168.32.202:22     *:*
root    sendmail  1002  4     tcp4     192.168.32.202:25     *:*
root    syslogd   942   6     udp4     192.168.32.202:514    *:*

Запуск процесса распределенного вычисления, оканчивающегося выдачей результатов. Получение списка присоединенных нод, остановка пула, выход из Erlang shell:

(poolmaster@conductor)13> pwnd:vStart[[1,4,-2,1]], [9,1,11,1], [3,-3,5,1], [1,1,1,1]], [1,5,1,-3]).
ResL = [{module,pwnd},{module,pwnd},{module,pwnd}]
PIDS = [<0.141.0>,<8877.51.0>,<8838.51.0>,<0.142.0>]
poolmaster@conductor: [<0.141.0>/1] V1=[1,4,-2,1] * V2=[1,5,1,-3] == [16,0,0,0]
poolmaster@conductor: [<0.142.0>/4] V1=[1,1,1,1] * V2=[1,5,1,-3] == [0,0,0,4]
crunchpool@j1: [<8838.51.0>/3] V1=[3,-3,5,1] * V2=[1,5,1,-3] == [0,0,-10,0]
crunchpool@j2: [<8877.51.0>/2] V1=[9,1,11,1] * V2=[1,5,1,-3] == [0,22,0,0]
[16,22,-10,4]
(poolmaster@conductor)14> nodes().
[crunchpool@j1,crunchpool@j2]
(poolmaster@conductor)15> pool:stop().
stopped
(poolmaster@conductor)16> nodes().
[]
(poolmaster@conductor)17> q().
ok
(poolmaster@conductor)18>
[root@conductor ~/crunching]#

В примере выше задан ключ -rsh ssh, сообщающий Erlang VM, что доступ к другим нодам должен осуществляться по протоколу SSH. В этой демонстрации доступ к узлам производится по SSH с применением аутентификации через публичный ключ. Можно также использовать другие методы аутентификации, например, Kerberos. Мастер-нода автоматически запускает Erlang VM удаленно на каждом из ведомых узлов пула. Т. о. налицо простота ввода новых мощностей в кластер. В идеале можно было бы свести к минимуму объем работы по конфигурированию окружения (ОС, VM etc.) и построить кластер на основе diskless узлов, загружающихся через LAN Boot, тем самым повысив надежность.

Исходя из последнего вывода на stdout смею предположить, что запуск (pspawn) процессов в пуле в отсутствие нагрузки происходит на разных серверах циклически (in a round-robin fashion), в противном случае учитывается загруженность каждого индивидуального вычислительного узла и производится балансировка нагрузки, т. е. запуск каждого нового процесса (именно Erlang процесса в Erlang VM, а не процесса ОС, aka "green process") на одном из наименее загруженных узлов пула.


Исходный код

-module(pwnd).

-export([startSay/0, say/2]).
-export([vSum/2, vMul/2, vVec/3, vStart/2, vSpawn/1,
       vReduce/2, vScatter/5, vWorker/1]).

-import(lists).

vSum([], []) ->
   [];
vSum([H1 | T1], [H2 | T2]) ->
   [H1 + H2] ++ vSum(T1, T2).

vMul([], []) ->
   [];
vMul([H1 | T1], [H2 | T2]) ->
   [H1 * H2] ++ vMul(T1, T2).

vVec(N, I, EL)->
   ZERO = lists:map(fun(X)->0*X end, lists:seq(1, N)),
   lists:sublist(ZERO, I - 1) ++
   [EL] ++ lists:sublist(ZERO, N - I).

vWorker(PID) ->
   receive
        {V1, V2, N} ->
             P = vVec(length(V1),N,lists:sum(vMul(V1,V2))),
             io:format("~w: [~w/~w] V1=~w * V2=~w == ~w~n",
             [node(), self(), N, V1, V2, P]),
             PID ! {vector, P}
   end.

vScatter([], _V, [], 0, _TOT) ->
   [];
vScatter([MH | MT], V, [PID | REST], N, TOT) ->
   PID ! {MH, V, TOT - N + 1},
   vScatter(MT, V, REST, N - 1, TOT).

vReduce(RES, 0) ->
   RES;
vReduce(RES, N) ->
   receive
        {vector, V} ->
             vReduce(vSum(RES, V), N - 1)
   end.

vSpawn([]) ->
   [];
vSpawn([_H | T]) ->
   [pool:pspawn_link(
         pwnd,
         vWorker,
         [self()]
   )] ++ vSpawn(T).

vStart(M, V) ->
   % Distribute code
   {Mod, Bin, Filename} = code:get_object_code(?MODULE),
   {ResL, []} = rpc:multicall(
        code,
        load_binary,
        [Mod, Filename, Bin]
   ),
   io:format("ResL = ~w~n", [ResL]),
   ZERO = lists:map(fun(X)->0*X end,
        lists:seq(1, length(V))
   ),
   PIDS = vSpawn(lists:seq(1, length(V))),
   io:format("PIDS = ~w~n", [PIDS]),
   vScatter(M, V, PIDS, length(PIDS), length(PIDS)),
   vReduce(ZERO, length(V)).

Заключение
Таким образом, отредактировав всего пару конфигурационных файлов и заменив spawn на pspawn, мы получили гибкое распределенное приложение, способное реагировать на изменения кода.
Для повышения общей производительности есть много вариантов. Например, компиляция BEAM модулей с нативным кодом (HiPE, изначально являлся проектом Upsala University). На тестовой non-SMP одноядерной платформе на базе Celeron 2400 MHz производительность перемножения векторов с количеством элементов ~108 при запуске HiPE модуля увеличилась на 30%. Есть возможность использования портов и драйверов (ports and linked-in drivers) для запуска приложений, написанных на С и других языках. Можно задействовать BLAS. Есть и другие варианты оптимизации.

Ниже, в коде, функция vStart/2 использует автоматический механизм передачи и загрузки модулей (двоичного кода) на узлы кластера. Поэтому отпадает необходимость при изменении программы распространять ее на каждую машину вручную. Кроме того в Erlang реализована возможность горячей замены кода (hotswap) без остановки выполнения приложения, при этом в памяти может существовать 2 версии модуля. Это позволяет существенно снизить или совсем исключить простои, которые являются необходимой мерой при разработке приложений, например, на С, Java и т. д. (подробнее). Некоторые системы, написанные разными разработчиками на Erlang, работают годами и регулярно обновляются.

В 2005 году был представлен проект ERESYE, позволяющий программировать Artificial Intelligence (искусственную разумность:) и экспертные системы в Erlang. Есть пример использования этой технологии.

Т. о. можно наблюдать преимущества функционального программирования в среде Erlang/OTP, в котором за долгую историю языка воплощено и продолжает активно внедряться множество продвинутых технологий, которым даже подражают другие языки, например, Scala (Erlang vs. Scala) и т. д.


Trivia
При загрузке кода в ваш редактор отступы могут сбиться. Для vim в командном режиме перейдите в начало файла (gg) и наберите на клавиатуре команду =G
Это откорректирует отступы.


Источники

• Giving a taste of a parallelization
• High performance technical computing with erlang