Erlang Tutorial — Part 2

John J. Camilleri
TDA383/DIT390 Concurrent Programming
HT2016 LP1

Simple merge sort

See documentation for lists:split/2 - http://erlang.org/doc/man/lists.html#split-2

sort([]) -> [] ;
sort([X]) -> [X] ;
sort(Xs) ->
  Len = length(Xs),
  {Part1, Part2} = lists:split(Len div 2, Xs),
  merge(sort(Part1), sort(Part2)).

merge([], Ys) -> Ys ;
merge(Xs, []) -> Xs ;
merge(Xs, Ys) ->
  [XH|XT] = Xs,
  [YH|YT] = Ys,
  if
    XH > YH -> [YH] ++ merge(Xs, YT) ;
    true    -> [XH] ++ merge(XT, Ys)
  end.

Can sort a numerical list: [4,9,3,7,2,-1,5,0,8,98]
What about a list of tuples?
- [{person, "Alice", 30}, {person, "Bob", 21}, {person, "Charlie", 99}, {person, "Donna", 5}]
- [{student, "Alice", 30}, {teacher, "Bob", 21}, {engineer, "Charlie", 99}, {doctor, "Donna", 5}]

Make it concurrent

Spawn each sub-sort into a new process

sort(Xs) ->
  Len = length(Xs),
  {Part1, Part2} = lists:split(Len div 2, Xs),
  Pid = self(),
  spawn(fun() -> Pid ! {result, sort(Part1)} end),
  spawn(fun() -> Pid ! {result, sort(Part2)} end),
  receive {result, Res1} -> ok end,
  receive {result, Res2} -> ok end,
  merge(Res1, Res2).

This isn't very useful when we want the answer synchronously

Math server

Send a function and operands
Send client pid
Note that just because the code is defined in the same module, it does not mean that it's run by the same processes!
The function math_server/0 is used internally and shouldn't be exported from module

start_server() ->
  Pid = spawn(fun() -> math_server() end),
  catch unregister(math),
  register(math, Pid).

% Server
math_server() ->
  receive
    {compute, Fun, Args, From} ->
      spawn(fun() ->
        Ans = apply(Fun, Args),
        timer:sleep(round(Ans)*10), % artificial wait
        From ! {answer, Ans}
      end)
  end,
  math_server().

% Client
power(Num, Exp) ->
  math ! {compute, fun(X,Y) -> round(math:pow(X, Y)) end, [Num, Exp], self()},
  receive
    {answer, Ans} -> io:fwrite("~p ^ ~p = ~p\n", [Num, Exp, Ans])
  end.

Registering processes

Send messages to an agreed upon name (atom) rather than a process ID
Only one process per registered name
Builtin functions (BIFs):
- register/2
- unregister/1
- whereis/1
Shell command: regs() to see list of registered processes

Mixing up request and response

Send a bunch of calculations and wait for them all:

powers(Nums, Exp) ->
  Fun = fun(X,Y) -> round(math:pow(X, Y)) end,
  [ math ! {compute, Fun, [Num, Exp], self()} || Num <- Nums ],
  Ans = [ receive {answer, Ans} -> Ans end || _ <- Nums ],
  io:fwrite("~p ^ ~p = ~p\n", [Nums, Exp, Ans]).

Try the following:
- foo:powers([1,2,3,4], 3)
- foo:powers([7,0,5,2], 3)
- foo:powers([4,3,2,1], 3)
We are not receiving the right messages in the right order!
Each request should be tagged with a unique reference
The client then receives by pattern matching on this reference
Use make_ref() to create a unique reference

% Server
math_server() ->
  receive
    {compute, Fun, Args, From, Ref} ->
      spawn(fun() ->
        Ans = apply(Fun, Args),
        timer:sleep(round(Ans)*10),
        From ! {answer, Ref, Ans}
      end)
  end,
  math_server().

% Client
powers(Nums, Exp) -> % e.g. Nums = [4,2,8], Exp = 2
  Refd = [ {make_ref(), Num} || Num <- Nums ], % Refd = [{#Ref<.>, 4}, {#Ref<.>, 2}, {#Ref<.>, 8}]
  Fun = fun(X,Y) -> round(math:pow(X, Y)) end,
  [ math ! {compute, Fun, [Num, Exp], self(), Ref} || {Ref, Num} <- Refd ], % Send messages, ignore return values
  Ans = [ receive {answer, Ref, Ans} -> Ans end || {Ref, _} <- Refd ], % Ans = [16,4,64]
  io:fwrite("~p ^ ~p = ~p\n", [Nums, Exp, Ans]).

Generic server

It's tedious to manually add own Pid and Refs to messages
Servers tend to have a common structure which we don't want to implement each time
Introduce genserver.erl

% genserver.erl
-module(genserver).
-export([start/3, request/2]).

%% Spawn a process and register it with a given atom
%% Function F should have arity 1
start(Atom, State, F) ->
  Pid = spawn(fun() -> loop(State, F) end),
  catch(unregister(Atom)),
  register(Atom, Pid),
  Pid.

loop(State, F) ->
  receive
    {request, From, Ref, Data} ->
      {reply, R, NewState} = F(State, Data),
      From!{result, Ref, R},
      loop(NewState, F)
  end.

%% Send a request to a Pid and wait for a response
request(Pid, Data) ->
  Ref = make_ref(),
  Pid!{request, self(), Ref, Data},
  receive
    {result, Ref, Result} ->
      Result
  after 3000 ->
    exit("Timeout")
  end.

Implementation using generic server

We supply our message handler function to genserver:start/3
We don't need to do any sending/receiving directly ourselves

% Start our server
start_server() ->
  genserver:start(math, nostate, fun math_server/2).

% Server message handling function (private)
math_server(State, {compute, Fun, Args}) ->
  Ans = apply(Fun, Args),
  {reply, Ans, State}.

% Client
powers(Nums, Exp) ->
  Fun = fun(X,Y) -> round(math:pow(X, Y)) end,
  Ans = [ genserver:request(math, {compute, Fun, [Num, Exp]}) || Num <- Nums ],
  io:fwrite("~p ^ ~p = ~p\n", [Nums, Exp, Ans]).

But in this example we have removed asynchronicity

Make it asynchronous again

The client needs to send its Pid
This time the server creates the ref and returns it after the first request
Server spawns a new process which sends result directly to client
The client then does a separate receive using the Ref from server

% Server
math_server(State, {compute, Fun, Args, From}) ->
  Ref = make_ref(),
  spawn(fun() ->
    Ans = apply(Fun, Args),
    timer:sleep(round(Ans)*10),
    From ! {answer, Ref, Ans}
  end),
  {reply, Ref, State}.

% Client
powers(Nums, Exp) ->
  Fun = fun(X,Y) -> round(math:pow(X, Y)) end,
  Refs = [ genserver:request(math, {compute, Fun, [Num, Exp], self()}) || Num <- Nums ],
  Ans = [ receive {answer, Ref, Ans} -> Ans end || Ref <- Refs ],
  io:fwrite("~p ^ ~p = ~p\n", [Nums, Exp, Ans]).

Records

"Convenient" syntax for treating tuples as records
Compile-time only

-record(person, {name="", born}). % Define the record type for compiler. Default values are optional.

X = #person{name="John", born=1986}), % Converted to: {person,"John",1986}
Name = X#person.name. % Name == "John"

Records cannot be updated in-place! Remember that variables are immutable.
You must create a copy of an existing record, specifying the fields you wish to change, and store it in a new variable:

X2 = X#person{name="Mary"}. % X2 == {person, "Mary", 1986}

Exercise

Consider the code below (names have been obfuscated).
Remember, hd/1 gives the head of a list and tl/1 gives the tail.

s(Xs) ->
  receive
    {alpha, X} -> s([X|Xs]) ;
    {beta,  P} -> P ! hd(Xs), s(tl(Xs))
  end.

demo() ->
  Pid = spawn(fun() -> s([]) end),
  Pid ! {alpha, "hello"},
  Pid ! {alpha, 467},
  Pid ! {alpha, {person, "Mary"}},
  Pid ! {beta, self()},
  receive
    Something -> io:fwrite("I received: ~p\n", [Something])
  end.

What is the function s/1 modelling?
What will happen when running demo/0? Will there be any output or errors?
Are there any potential problems with our implementation? How would you fix them?