Перемножение матрицы на вектор с использованием пула (pool(3)) серверов Erlang

Создание процессов

Для создания нового процесса служит несколько встроенных функций (spawn и её аналоги). Функции возвращают идентификатор процесса, который может использоваться, например, для отправки сообщений вновь созданному процессу. В интерактивной консоли erl можно получить список процессов и другую информацию посредством вызова функций processes(). и i(). Соответственно.

 

Отправка и приём сообщений

Erlang использует для отправки сообщения синтаксис с восклицательным знаком: ИдПроцесса! Сообщение. Приём сообщения — то есть извлечение его из очереди («почтового ящика») процесса — выполняется с помощью receive-выражений, обычно записываемых следующим образом^[90]:

receive образец1 when охрана1 -> выражение11, выражение12,...; образец2 when охрана2 -> выражение21, выражение22,...;... образецN when охранаN -> выражениеN1, выражениеN2,...; НесвязаннаяПеременнаяДляОстальныхСообщений -> выражение1, выражение2,...end

Встретив такое выражение, интерпретатор последовательно просматривает сообщения из очереди. Каждое сообщение интерпретатор сопоставляет с образцом и, если оно удовлетворяет образцу, вычисляются соответствующие выражения. Когда все сообщения перебраны, и подходящего не оказалось, процесс блокируется в ожидании новых сообщений, после чего перебор очереди повторяется. Если в receive-выражении отсутствует образец, которому удовлетворяет любое сообщение, такое выражение называется выборочным receive-выражением^[

Процесс можно связать с другим, в результате чего между процессами устанавливается двунаправленное соединение (англ. link). В случае, если один из процессов завершается ненормально, всем связанным с ним процессам передаётся сигнал выхода (англ. exit signal). Процессы, получившие сигнал, завершаются, распространяя сигнал дальше. Сигнал выхода является кортежем, элементами которого являются атом 'EXIT' (выход), идентификатор завершившегося процесса и причину завершения процесса. Причина завершения передаётся по цепочке завершающихся процессов.

Процесс может осуществить перехват ошибки (англ. trapping errors), если у него установлен флаг перехвата выходаТакой процесс получает сигналы выхода связанных с ним процессов в виде обычных сообщений с той же структурой кортежа. Перехваченный сигнала выхода более не передаётся связанным с процессом-перехватчиком процессам. Сигнал выхода с причиной — атомом normal (нормальное завершение процесса) не вызывает завершения связанного процесса. Если же причина — атом kill, процесс завершается безусловно (независимо от флага перехвата выхода), а связанным с ним процессам в качестве причины отправляется атом killed, что даёт им возможность среагировать.

В Erlang есть возможность установить и однонаправленное соединение. При завершении наблюдаемого процесса процесс-наблюдатель получает сообщение с указанием причины завершения.

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

 

Перемножение матрицы на вектор с использованием пула (pool(3)) серверов Erlang

Код:

-module(pwnd).

 

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

 

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)).

 

Выполнение:

После старта пула

ключ -rsh ssh, сообщающий Erlang VM, что доступ к другим узлам должен осуществляться по протоколу SSH. В этой демонстрации был использован доступ к узлам по SSH с использованием публичных ключей. Можно также использовать другие методы аутентификации, например, Kerberos. Мастер-нод автоматически запускает Erlang VM удаленно на каждом из ведомых узлов пула.