Articles tagged message passing

Boost message passing between Erlang nodes

By Roberto Ostinelli / April 7, 2009

Message passing between Erlang nodes is considerably slower than within the same node. This is normal, and is due to the fact that messages sent between nodes are actually copied from the area of the sender to that of the receiver, then sent over from one node to the other via TCP/IP.

I was getting interested when a server could achieve amazing performance over a single node, performance which got much lower once the server got distributed over two Erlang nodes. I therefore tried a small benchmark test to somehow measure the difference in message passing speed within/across Erlang nodes.

The machine used in this benchmark is an apple macbook running Leopard, with 2.0GHz Intel Core 2 Duo and 2GB of DDR3 RAM memory.

This benchmark is a very simple ping request/pong response test. A process 'A' sends a ping message to process 'B', which replies with a pong message.

Code for the pong process:

ERLANG

% starts pong
start_pong() ->
    register(flood_pong, spawn(fun() -> pong_loop() end)).

pong_loop() ->
    receive
        {{Sender, SenderNode}, Any} ->
            % pong back
            {Sender, SenderNode} ! {pong, Any},
            pong_loop();
        shutdown ->
            io:format("pong shutdown~n",[]);
        _Ignore ->
            pong_loop()
    after 60000 ->
        io:format("pong timeout, shutdown~n",[])
    end.

This very simple code basically starts a process and registers it with the name flood_pong. You can see that to any request with format:

ERLANG

{{Sender, SenderNode}, Any}

flood_pong will reply with a message with format:

ERLANG

{pong, Any}

The code for the ping process, i.e. the one that starts the ping requests, is the following:

ERLANG

% start ping
start_ping(PongNode, Num) ->
    register(flood_ping, spawn(fun() -> ping_loop(Num, now(), Num) end)),
    send(PongNode, Num).

send(_PongNode, 0) ->
    ok;
send(PongNode, Num) ->
    % send a spawned ping
    spawn(fun() -> {flood_pong, PongNode} ! {{flood_ping, node()}, ping_request} end),
    send(PongNode, Num - 1).

ping_loop(Num, Start, 0) ->
    T = timer:now_diff(now(), Start),
    io:format("RECEIVED ALL ~p in ~p ms [~p/min]~n",[Num, T, (Num*60000000/T)]);
ping_loop(Num, Start, Count) ->
    receive
        {pong, _PingBack} ->
            ping_loop(Num, Start, Count-1);
        _Received ->
            ping_loop(Num, Start, Count)
    after 10000 ->
        io:format("ping timeout, missing ~p pong, shutdown~n",[Count])
    end.

This code does two things:

Starts and registers a flood_ping process, which will be responsible to await and count the incoming pong replies. This allows it to compute the time needed for all the pong replies to get back to the node where the ping requests where first sent.
Spawns the ping requests [one process per ping request].

Let's give it a try. First, we have to start up two Erlang nodes, so I fire up terminal and enter:

BASH

$ roberto$ erl +K true +P 500000 -name 'one@rob.loc' -setcookie asd

and in another terminal window:

BASH

$ roberto$ erl +K true +P 500000 -name 'two@rob.loc' -setcookie asd

This initializes two erlang nodes, 'one@rob.loc' and 'two@rob.loc', both with kernel polling enabled and a maximum number of processes per node set to 500,000.

I need to ensure that the nodes can see each other, so from 'one@rob.loc' I ping 'two@rob.loc':

ERLANG

(one@rob.loc)1>net_adm:ping('two@rob.loc').
pong
(one@rob.loc)2>nodes().
['two@rob.loc']

Now let's perform a message passing speed benchmark for messages sent and received within the same Erlang node. First, I start up the pong process:

ERLANG

(one@rob.loc)3> flood1:start_pong().
true

I then need to start the ping requests, and I do so by issuing the start_ping/2 function, which has per arguments the name of the recipient node of the ping requests, and the number of requests to be performed. I therefore set the first parameter to node() and the second to 200,000 [we'll always try out 200,000 ping requests].

ERLANG

(one@rob.loc)4> flood1:start_ping(node(), 200000).
ok
RECEIVED ALL 200000 in 2257727 ms [5315080.166911234/min]

I see therefore that within the same Erlang node, this benchmark gives a result of an average of 5.3 million ping/pong messages per minute. Quite a rush.

Let's now try the same thing with pong running on 'two@rob.loc', and ping requests coming from 'one@rob.loc'. I start the pong process on 'two@rob.loc':

ERLANG

(two@rob.loc)1> flood1:start_pong().

Then I issue the ping request from 'one@rob.loc':

ERLANG

(one@rob.loc)5> flood1:start_ping('two@rob.loc', 200000).
ok
RECEIVED ALL 200000 in 16727643 ms [717375.4246189974/min]

I see therefore that on two different Erlang nodes, this benchmark gives a result of an average of 700 thousands ping/pong messages per minute.

This is around 7 times a loss in performance when compared to the same benchmark run on a single Erlang node. This is to be expected, still it does mean that a system heavily dependant on message passing and running on a single Erlang node would be considerably faster than one running on distributed nodes (one of the things Erlang is great at doing).

At this point, I tried reducing the overhead in message passing, by using UDP instead of Erlang's native functionalities. This was no success at all, UDP having all the troubles it's known for (packet loss, ordering, double packet sending, ...) without me getting any interesting speed improvement.

So I got this thought: what if Erlang's message passing overhead is so demanding, due to TCP/IP, that actually queuing messages sent to processes running on the same node and sending them altogether would increase the overall passing speed? TCP/IP already does provide such mechanisms, but what if performing this queuing at Erlang application level could bring significant improvements to infra-nodes message passing?

This is the picture. On both Erlang nodes I run a gen_server (which I called qr) which performs queuing and routing of messages sent from one node to the other. Instead of sending messages directly from a process A on node 'one@rob.loc' to a process B on node 'two@rob.loc', process A sends a routing request to the qr running on 'one@rob.loc', which will then send it to the qr running on 'two@rob.loc', which will finally forward it to B. Fact is, the sending qr will queue messages until a certain number of them for the same node have piled up, or a timeout period has expired.

I start by defining the gen_server API to route a message. Here it is:

ERLANG

% Function() -> void()
% Description: Gets a routing request
route({ToPid, ToNode}, Message) ->
    case ToNode =:= node() of
        true ->
            % send directly
            ToPid ! Message;
        false ->
            % queue
            gen_server:cast(?SERVER, {{queue, ToNode}, {ToPid, Message}})
    end.

You can see that we need to pass the Pid of the destination process, as well as the name of the Node where this process is running, together with the Message that we need to send.

In case the node is remote, this will cast a message to qr, which will handle it:

ERLANG

% add incoming routing request to queue
handle_cast({{queue, DestNode}, {ToPid, Message}}, Queue) ->
    % to pid
    Msg = {route, ToPid, Message},
    % get if node exists in queue
    case lists:keysearch(DestNode, 1, Queue) of
        false ->
            % add node
            NewQueue = [{DestNode, {now(), [Msg]}}|Queue];
        {value, {DestNode, {CreationTime, MsgList}}} ->
            % check if queue is long enough
            case length(MsgList) >= ?QUEUELENGTH of
                true ->
                    % queue of a node is of maximum lenght, send routing message
                    {?QR, DestNode} ! {queue_route, [Msg|MsgList]},
                    % empty queue for node
                    NewQueue = lists:keydelete(DestNode, 1, Queue);
                false ->
                    % add message to queue list and replace node
                    NewQueue = lists:keyreplace(DestNode, 1, Queue, {DestNode, {CreationTime, [Msg|MsgList]}})
            end
    end,
    % purge timeout
    PurgedQueue = purge_queue_selective(NewQueue),
    {noreply, PurgedQueue, 200};

[...]

% Function -> PurgedQueue
% Description: Loop queue and purge only nodes on timeout.
purge_queue_selective(Queue) ->
    % check timeout, send and remove element if needed
    FilterFun = fun({DestNode, {CreationTime, MsgList}}) ->
        case timer:now_diff(now(), CreationTime) > ?QUEUETIMEOUT of
            true ->
                % timeout for a node, send routing message
                % io:format("send selective: {?QR, ~p} ! {queue_route, ~p}~n",[DestNode, MsgList]),
                {?QR, DestNode} ! {queue_route, MsgList},
                % delete node from queue
                false;
            false ->
                true
        end
    end,
    % return cleaned queue
    lists:filter(FilterFun, Queue).

The mecanism is quite simple. When a message to be sent to a remote node arrives to qr, we build a list of key tuples [which will be passed along as gen_server State variable] of format:

ERLANG

{DestNode, {CreationTime, MsgList}}

DestNode, i.e. the destination node, is the tuple key. CreationTime is a parameter which allows to know when a timeout for the messages to DestNode has passed. MsgList is a list of tuples of the messages to be sent to DestNode, in format:

ERLANG

{route, ToPid, Message}

We handle this key list to check when to send all the queued messages to the other node, which we do:

when a certain timeout has passed for that node [computed using CreationTime];
when the number of messages to be sent has reached a certain size.

We see that the gen_server cast also establishes an overall timeout of 200 ms, which will fire when no other messages are received by qr and thus we need to send the remaining messages out. This timeout is handled with a handle_info call:

ERLANG

% timeout on a cast message, purge queue
handle_info(timeout, Queue) ->
    % purge timeout
    PurgedQueue = purge_queue(Queue),
    % return
    {noreply, PurgedQueue};

[...]

% Function -> PurgedQueue
% Description: Loop queue and purge all nodes.
purge_queue(Queue) ->
    % sent to all remaining
    FilterFun = fun({DestNode, {_CreationTime, MsgList}}) ->
        % send routing message
        {?QR, DestNode} ! {queue_route, MsgList}
    end,
    lists:foreach(FilterFun, Queue),
    % return an empty queue
    [].

As we can see from here over, a message sent from a qr on one node to a qr of another node has format:

ERLANG

{queue_route, MsgList}

With MsgList being a list of the individual messages in format:

ERLANG

{route, ToPid, Message}

When a qr of a node receives such a message, it handles it:

ERLANG

handle_info({queue_route, MsgList}, State) ->
    RouteFun = fun({route, ToPid, Message}) ->
        % local send to node
        ToPid ! Message
    end,
    lists:foreach(RouteFun, MsgList),
    % return
    {noreply, State, 200};

Ok, this basically is it. Let's now see if this does improve message passing between nodes.

For this, we need to redifine the pong_loop so that it does not send messages directly to the process, but routes its requests to qr instead.

New code for the pong process:

ERLANG

% starts pong
start_pong() ->
    register(flood_pong, spawn(fun() -> pong_loop() end)).

pong_loop() ->
    receive
        {{Sender, SenderNode}, Any} ->
            % pong back
            qr:route({Sender, SenderNode}, {pong, Any}),    % < === ONLY LINE CHANGED ===
            pong_loop();
        shutdown ->
            io:format("pong shutdown~n",[]);
        _Ignore ->
            pong_loop()
    after 30000 ->
        io:format("pong timeout, shutdown~n",[])
    end.

The same goes for the ping process:

ERLANG

% start ping
start_ping(PongNode, Num) ->
    register(flood_ping, spawn(fun() -> ping_loop(Num, now(), Num) end)),
    send(PongNode, Num).

send(_PongNode, 0) ->
    ok;
send(PongNode, Num) ->
    % send a spawned ping                               % === \/ ONLY LINE CHANGED ===
    spawn(fun() -> qr:route({flood_pong, PongNode}, {{flood_ping, node()}, ping_request}) end),
    send(PongNode, Num - 1).

ping_loop(Num, Start, 0) ->
    T = timer:now_diff(now(), Start),
    io:format("RECEIVED ALL ~p in ~p ms [~p/min]~n",[Num, T, (Num*60000000/T)]);
ping_loop(Num, Start, Count) ->
    receive
        {pong, _PingBack} ->
            ping_loop(Num, Start, Count-1);
        _Received ->
            ping_loop(Num, Start, Count)
    after 10000 ->
        io:format("ping timeout, missing ~p pong, shutdown~n",[Count])
    end.

We are all set. Let's try to immediately perform a test on two different Erlang nodes, pong running on 'two@rob.loc', and ping requests coming from 'one@rob.loc'. I start qr and the pong process on 'two@rob.loc':

ERLANG

(two@rob.loc)55> qr:start_link().
{ok,}
(two@rob.loc)56> flood2:start_pong().
true

Then I start qr and issue the ping request from 'one@rob.loc':

ERLANG

(one@rob.loc)44> qr:start_link().
{ok,}
(one@rob.loc)45> flood2:start_ping('two@rob.loc', 200000).
ok
RECEIVED ALL 200000 in 5690225 ms [2108879.7015935224/min]

I see therefore that on two different Erlang nodes, this benchmark gives a result of an average of 2.1 million ping/pong messages per minute, 3 times as much as without message queuing. This is interesting.

Just to be on the safe side, I try the same benchmark for messages sent via qr to processes running on the same Erlang node.

ERLANG

{ok,}
(two@rob.loc)50> flood2:start_pong().
true
(two@rob.loc)51> flood2:start_ping(node(), 200000).
ok
RECEIVED ALL 200000 in 2264539 ms [5299091.779828035/min]

I see therefore that within the same Erlang node, this benchmark gives a result of an average of 5.3 million ping/pong messages per minute, the same that we had without the queuing mechanism.

To summarize:

without queuing mechanism:
- same Erlang node: 5.3 million messages/min;
- different Erlang nodes: 700 K messages/min.
with queuing mechanism:
- same Erlang node: 5.3 million messages/min;
- different Erlang nodes: 2.1 million messages/min.

The complete code to run this on your machine is available here. This whole 'queuing idea' is still an experiment, and I'd be more than delighted to hear your feedback, to see whether you are getting the same results, you know how to improve the concept or the code, or you have any considerations at all you would like to share.

UPDATE [April 17th, 2009]

Due to a comment here below from Adam Bregenzer, who performed a test on a Linux box and had very different results, I've decided to perform some testing on 3 different OS. Here is a summary of these tests, but please bear in mind that these results are definitely non exhaustive.

1. macbook running Leopard OSX, with 2.0GHz Intel Core 2 Duo and 2GB of DDR3 RAM memory.

on the same Erlang node: 5.3 million messages/min;
on different Erlang nodes:
- without queuing mechanism: 700 K messages/min.
- with queuing mechanism: 2.1 million messages/min.

2. Ubuntu 8.10 Linux, 64-bit system with 2.5GHz Intel Core 2 Duo [Adam's test]

on the same Erlang node: 11.7 million messages/min;
on different Erlang nodes:
- without queuing mechanism: 4.8 million messages/min.
- with queuing mechanism: 3.6 million messages/min.

3. Ubuntu 9.04 beta Linux, 32-bit system on a VM with 1 CPU 2.0GHz and 512MB of RAM memory.

on the same Erlang node: 4.6 million messages/min;
on different Erlang nodes:
- without queuing mechanism: 1.6 million messages/min.
- with queuing mechanism: 960K messages/min.

4. Windows XP, with 3.0GHz Intel Core 2 Duo E8400 and 2.5GB of DDR3 RAM memory.

on the same Erlang node: 24.7 million messages/min;
on different Erlang nodes:
- without queuing mechanism: 2.2 million messages/min.
- with queuing mechanism: 5.1 million messages/min.

As it is normal with Erlang, results are extremely different depending on many factor, one of which seems to be the OS. On the non-exhaustive tests here above, the qr mechanism does:

improve message passing performance between Erlang nodes in the OSX and Windows environment;
decrease message passing performance between Erlang nodes in the Ubuntu Linux environment.

I do not have the insights to understand which characteristics of each OS is switching results in this manner: TCP/IP flow, threads, I/O. Though, I'm currently deploying a similar qr mechanism in my applications, which can be activated if necessary based on individual benchmarking tests.

UPDATE [June 9th, 2009]

I have had the availability of two HP ProLiant DL380, running Ubuntu 8.04 Server 32 bit, therefore I decided to take a shot also to test the same bench on different machines.

Here is a summary of the results of these tests. As usual, please bear in mind that these results are definitely non exhaustive.

on the same Erlang node: 5.3 million messages/min [on both machines];
on two different Erlang nodes, both running on the same machine:
- without queuing mechanism: 1.7 K million messages/min.
- with queuing mechanism: 3.1 million messages/min.
on two different Erlang nodes, each running on a different machine:
- without queuing mechanism: 1.7 K million messages/min.
- with queuing mechanism: 3.2 million messages/min.

This seems to state that the qr mechanism does improve message passing performance between Erlang nodes, also on the Ubuntu 8.04 environment.

I've also updated the original code, so that the destination node identifier is now extracted from the recipient Pid and thus it is not necessary to pass it as variable. This code can be found here.