OcamlP3l на эрланге

[basp]

Если JoCaml кажется просто забавным трюком, то в OcamlP3l уже видится что-то более серьезное.
Это попытка ввести шаблоны параллелизма, а фактически - функции высшего порядка(map,foldl,...) для независимых процессов, скрывая при этом особенности межпроцессного взаимодействия.
Ниже приведена наивная реализация шаблонов(skeleton) OcamlP3l на эрланге. В OcamlP3l есть еще такой интересный параметр - уровень(degree) параллелизма - число параллельных процессов в вычислении, соответственно при уровне==1 вычисление превращается в последовательное. У меня этого, конечно, нет.
Пример работы шаблона "ферма" на OcamlP3l

let farm_worker _ = fun x -> x * x;;
let print_result x = print_int x; print_newline();;
let compute = parfun (fun () -> (farm (seq(farm_worker),4)));;
pardo(fun () ->
let is = P3lstream.of_list [1;2;3;4;5;6;7;8] in
let s' = compute is in P3lstream.iter print_result s';
);;
И на эрланге:

farm(fun(X) -> X*X end, lists:seq(1,10)).

В коде еще используется модуль utils: foreach,map с векторами, y-комбинатор и сбор заданного числа ответов в список/вектор.


-module(ep3l).
-import(lists, [map/2,foreach/2, foldl/3]).
-import(erlang, [make_tuple/2]).

-compile(export_all).

testReduce() ->
reduce(fun(X,Y) -> X + Y end, [1,2,3,4]).

reduceworker(Seq,A,B,To) ->
R = Seq(A,B),
To ! {from, self(), R}.

reduce(Seq, X) ->
Self = self(),
foldl(fun(A,B) ->
spawn(?MODULE, reduceworker, [Seq,A,B,Self]),
receive {from, _From, Res} ->
Res
end
end,
hd(X),
tl(X)).

testMapvector() ->
mapvector(fun(X) -> X*X end,
{1,2,3,4,5}).

mapworker(Seq,{Pos,Arg},To) ->
Msg = Seq(Arg),
To ! {from, self(), {Pos,Msg}}.

mapvector(Seq, X) ->
Self = self(),
Num = utils:foreach(fun(I,A) ->
spawn(?MODULE, mapworker, [Seq,{I,A},Self])
end,
X),
utils:collect(Num,make_tuple(Num,0)).

testLoop() ->
loop(fun(X) -> (X rem 5 =/= 0) end,
fun(X) -> X+1 end,
[3,7,10,14]).

body(Cond, Seq, Arg, To) ->
R = Seq(Arg),
C = Cond(R),
if C ->
body(Cond, Seq, R, To); true ->
To ! {from, self(), R}
end.

loop(Cond, Seq, X) ->
Self = self(),
Num = foldl(fun(A,I) ->
spawn(?MODULE, body, [Cond,Seq,A,Self]),
I+1
end,
0,
X),
utils:collect(Num,[]).

testPipeline() ->
pipeline([fun(X) -> X*X end,
fun(X) -> X*2 end,
fun(X) -> X+1 end],
3).

pipe(Seq, X, Master) ->
Res = Seq(X),
Master ! {from, self(), Res}.

pipeline(Seqs,X) ->
Self = self(),
foldl(fun(A,Res) ->
Pid = spawn(?MODULE, pipe, [A,Res,Self]),
receive
{from, Pid, NewRes} ->
NewRes
end
end,
X,
Seqs).

testFarm() ->
farm(fun(X) -> X-1 end, lists:seq(1,10)).

worker(Seq,Arg,To) ->
Msg = Seq(Arg),
To ! {from, self(), Msg}.

farm(Worker, List) ->
Self = self(),
Num = foldl(fun(A,I) ->
spawn(?MODULE, worker, [Worker,A,Self]),
I+1
end,
0,
List),
utils:collect(Num,[]).