Initial Release

.gitignore

@@ -0,0 +1,8 @@
+erl_crash.dump
+*.beam
+ebin
+priv/*.args
+results
+*.log
+.otp.plt
+.skel.plt

LICENCE

@@ -0,0 +1,24 @@
+Copyright (c) 2012, University of St Andrews
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+   * Redistributions of source code must retain the above copyright
+     notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above copyright
+     notice, this list of conditions and the following disclaimer in the
+     documentation and/or other materials provided with the distribution.
+   * Neither the name of the University of St Andrews nor the
+     names of its contributors may be used to endorse or promote products
+     derived from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE UNIVERSITY OF ST ANDREWS BE LIABLE FOR ANY
+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

Makefile

@@ -0,0 +1,52 @@
+.PHONY : compile console typecheck typer clean
+
+all: compile
+
+clean:
+	@./rebar clean
+	@rm -f doc/skel.aux doc/skel.bbl doc/skel.blg doc/skel.fdb_latexmk doc/skel.fls
+	@rm -f doc/skel.out doc/skel.synctex.gz doc/skel.pdf doc/skel.log
+	@rm -f ./.skel.plt ./.otp.plt skel.tar.gz
+
+compile: src/*.erl
+	@./rebar compile
+
+console: compile
+	@exec erl -args_file ./priv/default.args
+
+examples: compile
+	@echo "==> skel (examples)"
+	@erlc +debug_info -I include -o ebin examples/*.erl
+
+typecheck: compile .skel.plt
+	@echo "==> skel (typecheck)"
+	@dialyzer --no_check_plt --plt ./.skel.plt -c ebin -Wrace_conditions
+
+typer: compile .skel.plt
+	@echo "==> skel (typer)"
+	@typer --plt ./.skel.plt --show -I include -pa ebin -r src
+
+pdf: doc/skel.pdf
+	@echo "==> skel (pdf)"
+
+package: skel.tar.gz
+	@echo "==> skel (package)"
+
+todo:
+	@echo "==> skel (todo)"
+	@grep -ir "TODO" --exclude Makefile -- .
+
+skel.tar.gz: compile
+	@tar -czf skel.tar.gz -C .. --exclude "skel.tar.gz" skel
+
+.otp.plt:
+	@echo "==> otp (plt) # This takes a while, go grab a coffee"
+	@dialyzer --build_plt --output_plt ./.otp.plt --apps erts kernel stdlib debugger et tools
+
+.skel.plt: .otp.plt compile
+	@echo "==> skel (plt)"
+	@dialyzer --add_to_plt --plt ./.otp.plt --output_plt ./.skel.plt -c ebin
+
+doc/skel.pdf: doc/skel.tex
+	@pdflatex -interaction=batchmode -output-directory=./doc skel.tex
+	@pdflatex -interaction=batchmode -output-directory=./doc skel.tex

README.md

@@ -0,0 +1,27 @@
+skel
+====
+
+A Streaming Process-based Skeleton Library for Erlang
+
+Usage:
+------
+
+`make` to compile the library source
+
+`make examples` to compile the library and the examples
+
+`make console` to compile and get a console (`make examples console` will give you a console with the examples recompiled too.)
+
+`make typecheck` to typecheck
+
+`make typer` to get inferred types (only of use during development for methods without type specs)
+
+`make clean` to clear up the repo.
+
+`make todo` to see some todos
+
+
+Copyright
+---------
+
+BSD 3-clause licence in LICENCE

examples/fib.erl

@@ -0,0 +1,63 @@
+-module(fib).
+
+-compile(export_all).
+
+p(Input) ->
+  _ = fib(22) + fib(20),
+  Input.
+
+f(Input) ->
+  _ = fib(24),
+  Input.
+
+f_prime(Input) ->
+  _ = fib(22),
+  Input.
+
+decomp(Input) ->
+  lists:duplicate(4, Input).
+
+recomp([Input|_]) ->
+  Input.
+
+% Computational Payload.
+% All that matters is that it has predictable runtime and 100% CPU utilisation
+fib(X) when X =< 1 ->
+  X;
+fib(X) ->
+  fib(X-1) + fib(X-2) + fib(X-3) + fib(X-4).
+
+run_all_examples() ->
+  [run_examples(X) || X <- [8,6,4,2,1]].
+
+run_examples(X) ->
+  erlang:system_flag(schedulers_online, X),
+  Ntimes = 10,
+  Input = lists:duplicate(1024, xx),
+  io:format("------~nRunning Examples on ~p cores. Please be Patient. ~n", [X]),
+  io:format("Example 4: ~p~n", [sk_profile:benchmark(fun ?MODULE:example4/1, [Input], Ntimes)]),
+  io:format("Example 3: ~p~n", [sk_profile:benchmark(fun ?MODULE:example3/1, [Input], Ntimes)]),
+  io:format("Example 2: ~p~n", [sk_profile:benchmark(fun ?MODULE:example2/1, [Input], Ntimes)]),
+  io:format("Example 1: ~p~n", [sk_profile:benchmark(fun ?MODULE:example1/1, [Input], Ntimes)]),
+  io:format("Done with examples on ~p cores.~n------~n", [X]).
+
+example1(Images) ->
+  [f(p(Im)) || Im <- Images].
+
+example2(Images) ->
+  skel:run([{seq, fun ?MODULE:p/1}, {seq, fun ?MODULE:f/1}], Images),
+  receive
+    {sink_results, Results} -> Results
+  end.
+
+example3(Images) ->
+  skel:run([{seq, fun ?MODULE:p/1}, {map, [{seq, fun ?MODULE:f_prime/1}], fun ?MODULE:decomp/1, fun ?MODULE:recomp/1}], Images),
+  receive
+    {sink_results, Results} -> Results
+  end.
+
+example4(Images) ->
+  skel:run([{farm, [{seq, fun ?MODULE:p/1}], 4}, {map, [{seq, fun ?MODULE:f_prime/1}], fun ?MODULE:decomp/1, fun ?MODULE:recomp/1}], Images),
+  receive
+    {sink_results, Results} -> Results
+  end.

examples/ifl.erl

@@ -0,0 +1,106 @@
+-module(ifl).
+
+-compile(export_all).
+
+-define(seq, {seq, fun ?MODULE:unary/1}).
+-define(runs, 1).
+
+payload() ->
+  fib(27).
+
+inputs() ->
+  lists:seq(1, 100000).
+
+fib(X) when X =< 1 ->
+  X;
+fib(X) ->
+  fib(X-1) + fib(X-2).
+
+decomp_for(Stages) ->
+  fun(X) ->
+    lists:duplicate(Stages, X)
+  end.
+
+recomp([X|_]) ->
+  X.
+
+unary(X) ->
+  payload(),
+  X.
+
+binary(X, _Y) ->
+  payload(),
+  X.
+
+test_pipe(Stages) ->
+  lists:duplicate(Stages, ?seq).
+
+test_farm(Stages) ->
+  [{farm, [?seq], Stages}].
+
+test_map(Stages) ->
+  [{map, [?seq], decomp_for(Stages), fun ?MODULE:recomp/1}].
+
+test_reduce(Stages) ->
+  [{reduce, fun ?MODULE:binary/2, decomp_for(Stages)}].
+
+
+benchmark() ->
+  {ok, VerboseLog} = file:open("ifl.verbose.10.log", [write]),
+  {ok, DataLog} = file:open("ifl.data.10.log", [write]),
+  io:format("Starting tests!~n"),
+  PayLoadResults = sk_profile:benchmark(fun ?MODULE:payload/0, [], 100),
+  io:format(VerboseLog, "Payload: ~n\t~w~n", [PayLoadResults]),
+  io:format(DataLog, "payload.dat: ~.5f~n", [proplists:get_value(mean, PayLoadResults)]),
+  [benchmark(PipeLine, Stages, DataLog, VerboseLog) ||
+    PipeLine <- [test_pipe, test_farm, test_map, test_reduce],
+    Stages <- [1,2,4,8,16]],
+  io:format("Done Tests!~n").
+
+benchmark(PipeLine, Stages, DataLog, VerboseLog) ->
+  Results = [bm_both(PipeLine, Stages, Schedulers) || Schedulers <- [1,2,4,8]],
+  data_log(PipeLine, Stages, Results, standard_io),
+  data_log(PipeLine, Stages, Results, DataLog),
+  verbose_log(PipeLine, Stages, Results, VerboseLog).
+
+bm_both(PipeLine, Stages, Schedulers) ->
+  erlang:system_flag(schedulers_online, Schedulers),
+  PipeLineX = ?MODULE:PipeLine(Stages),
+  {Schedulers, bm_sequential(PipeLineX), bm_parallel(PipeLineX)}.
+
+bm_parallel(PipeLine) ->
+  Inputs = inputs(),
+  sk_profile:benchmark(fun() ->
+    sk_assembler:run(PipeLine, Inputs),
+    receive
+      {sink_results, Results} -> Results
+    end
+  end, [], ?runs).
+
+bm_sequential(PipeLine) ->
+  Inputs = inputs(),
+  sk_profile:benchmark(fun() ->
+    sk_assembler_sequential:run(PipeLine, Inputs),
+    receive
+      {sink_results, Results} -> Results
+    end
+  end, [], ?runs).
+
+data_log(PipeLine, Stages, Results, DataLog) ->
+  io:format(DataLog, "~w.~w.dat: ", [PipeLine, Stages]),
+  [io:format(DataLog, "~B ~.5f ", [Schedulers, speedup(SeqResults, ParResults)]) ||
+    {Schedulers, SeqResults, ParResults} <- Results],
+  io:format(DataLog, "~n", []).
+
+verbose_log(PipeLine, Stages, Results, VerboseLog) ->
+  [io:format(VerboseLog,
+             "PL: ~w; Stages: ~w; SEQUENTIAL; Schedulers: ~w~n\t~w~n" ++
+             "PL: ~w; Stages: ~w; PARALLEL  ; Schedulers: ~w~n\t~w~n",
+             [PipeLine, Stages, Schedulers, SeqResults] ++
+             [PipeLine, Stages, Schedulers, ParResults]) ||
+    {Schedulers, SeqResults, ParResults} <- Results].
+
+speedup(SeqResults, ParResults) ->
+  proplists:get_value(mean, SeqResults) / proplists:get_value(mean, ParResults).
+
+

examples/imb.erl

@@ -0,0 +1,95 @@
+-module(imb).
+-compile(export_all).
+
+-include("imb.hrl").
+
+run(Benchmark) ->
+    run(Benchmark, 0).
+
+run(Benchmark, Processes) ->
+    BenchmarkFun = benchmark_decode(Benchmark, Processes),
+    Repetitions = repetition_generate(),
+    Data = data_generate(),
+    FormatBenchmark = format_benchmark_decode(Benchmark),
+    FormatHeading = format_heading_decode(Benchmark),
+    Headings = heading_decode(Benchmark),
+    print_heading(FormatHeading, Headings),
+    [exec_benchmark(BenchmarkFun, R, D, FormatBenchmark) || {R, D} <- lists:zip(Repetitions, Data)].
+
+heading_decode(sendrecv) ->
+    ["#bytes", "#repetitions", "t_min[usec]", "t_max[usec]", "t_avg[usec]", "Mbytes/sec"];
+heading_decode(alltoall) ->
+    ["#bytes", "#repetitions", "t_min[usec]", "t_max[usec]", "t_avg[usec]", "Mbytes/sec"];
+heading_decode(_) ->
+    ["#bytes", "#repetitions", "t[usec]", "Mbytes/sec"].
+
+format_heading_decode(sendrecv) ->
+    "~15s ~15s ~15s ~15s ~15s ~15s~n";
+format_heading_decode(alltoall) ->
+    "~15s ~15s ~15s ~15s ~15s ~15s~n";
+format_heading_decode(_) ->
+    "~15s ~15s ~15s ~15s~n".
+
+format_benchmark_decode(sendrecv) ->
+    "~15w ~15w ~15.2f ~15.2f ~15.2f ~15.2f~n";
+format_benchmark_decode(alltoall) ->
+    "~15w ~15w ~15.2f ~15.2f ~15.2f ~15.2f~n";
+format_benchmark_decode(_) ->
+    "~15w ~15w ~15.2f ~15.2f~n".
+
+benchmark_decode(pingpong, _) ->
+    fun(R, D) -> pingpong:run(R, D) end;
+benchmark_decode(pingping, _) ->
+    fun(R, D) -> pingping:run(R, D) end;
+benchmark_decode(sendrecv, P) when P > 0 ->
+    fun(R, D) -> sendrecv:run(R, D, P) end;
+benchmark_decode(alltoall, P) when P > 0 ->
+    fun(R, D) -> alltoall:run(R, D, P) end.
+
+exec_benchmark(BenchmarkFun, R, D, Format) ->
+    Time = BenchmarkFun(R, D),
+    if
+        is_list(Time) ->
+            TimeMin = lists:min(Time),
+            TimeMax = lists:max(Time),
+            TimeAvg = lists:sum(Time) / length(Time), 
+            MbytesPerSec = bandwidth_calc(D, TimeAvg),
+            io:format(Format, [size(D), R, TimeMin, TimeMax, TimeAvg, MbytesPerSec]);
+        is_float(Time) ->
+            MbytesPerSec = bandwidth_calc(D, Time),
+            io:format(Format, [size(D), R, Time, MbytesPerSec])
+    end.
+
+bandwidth_calc(Data, Time) ->
+    Megabytes = (size(Data) / math:pow(2, 20)),
+    Seconds = (Time * 1.0e-6),
+    Megabytes / Seconds.
+
+print_heading(Format, Headings) ->
+    io:format(Format, Headings).
+
+repetition_generate() ->
+    Thousands = [1000 || _ <- lists:seq(1, 17)],
+    lists:append(Thousands, [640, 320, 160, 80, 40, 20, 10]).
+
+data_generate() ->
+    [bytes_generate(0) | [bytes_generate(round(math:pow(2, X))) || X <- lists:seq(0, ?ITERATIONS - 2)]].
+
+bytes_generate(Size) ->
+    bytes_generate(Size, []).
+
+bytes_generate(0, Bytes) ->
+    list_to_binary(Bytes);
+bytes_generate(Size, Bytes) ->
+    bytes_generate(Size - 1, [1 | Bytes]).
+
+time_microseconds() ->
+    {MS, S, US} = now(),
+    (MS * 1.0e+12) + (S * 1.0e+6) + US.
+
+finalize(P) ->
+    receive
+        {done, P} -> 
+            ?TRACE("DONE: pid: ~p from: ~p ~n", [self(), P]),
+            ok
+    end.