Reading Notes #3: Ring Benchmark

I have been working my way through Joe Armstrong's Programming in Erlang. One of the exercises in chapter 8 is to write a "ring benchmark" in Erlang and any other language, compare the results and publish them.

Write a ring benchmark. Create N processes in a ring. Send a message round the ring M times so that a total of N * M messages get sent. Time how long this takes for different values of N and M. Write a similar program in some other programming language you are familiar with. Compare the results. Write a blog, and publish the results on the Internet!

Here are my benchmarks written in Erlang and Python respectively. I have presumed that the author meant sending messages synchronously; i.e. a message has to go around the whole ring before the next can be sent.

The Erlang version was easy to write. After all the language is designed to use a processes and messages paradigm of programming. Here is the source code:

view plainprint?

1. -module(pring).  
2.   
3. %%  
4. %% Exported Functions  
5. %%  
6. -export([run/2]).  
7.   
8. -ifdef(debug).  
9. -define(DEBUG(Format, Args), io:format(Format, Args)).  
10. -else.  
11. -define(DEBUG(Format, Args), void).  
12. -endif.  
13.   
14. %%  
15. %% API Functions  
16. %%  
17. run(N, M) when is_integer(N), is_integer(M), N > 0, M > 0 ->  
18.     time_it(fun() -> start(N, M) end).  
19.   
20. %%  
21. %% Local Functions  
22. %%  
23. time_it(F) when is_function(F) ->  
24.     statistics(runtime),  
25.     statistics(wall_clock),  
26.     Result = F(),  
27.     {_, CPUTime} = statistics(runtime),  
28.     {_, ElapsedTime} = statistics(wall_clock),  
29.     {Result, CPUTime/1000, ElapsedTime/1000}.  
30.   
31. start(N, M) when is_integer(N), is_integer(M), N > 0, M > 0 ->  
32.     This = self(),  
33.     First = spawn(fun() -> loop(This, This) end),  
34.     ?DEBUG("Creating ~p for N = ~p~n", [First, N]),  
35.     Last = create(First, First, N - 1),  
36.     First ! {realign, First, Last},  
37.     First ! {relay, message, M, This},  
38.     receive  
39.         {done, message} ->  
40.             N * M  
41.     end.  
42.   
43. create(First, Previous, N) when is_pid(First), is_pid(Previous), is_integer(N), N > 0 ->  
44.     Pid = spawn(fun() -> loop(First, Previous) end),  
45.     ?DEBUG("Creating ~p for N = ~p~n", [Pid, N]),  
46.     create(First, Pid, N - 1);  
47.   
48. create(First, Previous, 0) when is_pid(First), is_pid(Previous) ->  
49.     Previous.  
50.   
51. send_relay_message(Target, Message, M, ReportBack) when is_pid(Target), is_integer(M), is_pid(ReportBack) ->  
52.     ?DEBUG("~p => ~p (M = ~p) ~n", [self(), Target, M + 1]),  
53.     Target ! {relay, Message, M, ReportBack}.  
54.   
55. loop(First, Previous) when is_pid(First), is_pid(Previous) ->  
56.     This = self(),  
57.     receive  
58.         {realign, NewFirst, NewPrevious} when is_pid(NewFirst), is_pid(NewPrevious) ->  
59.             ?DEBUG("I am ~p Realigning: First ~p to ~p, Previous ~p to ~p~n", [This, First, NewFirst, Previous, NewPrevious]),  
60.             loop(NewFirst, NewPrevious);  
61.   
62.         {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =:= First, M > 0 ->  
63.             send_relay_message(Previous, Message, M - 1, ReportBack),  
64.             loop(First, Previous);  
65.   
66.         {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =:= First, M =:= 0 ->  
67.             ?DEBUG("I am ~p All rounds of relay over~n", [This]),  
68.             ReportBack ! {done, Message},  
69.             void;  
70.   
71.         {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =/= First, M > 0 ->  
72.             send_relay_message(Previous, Message, M, ReportBack),  
73.             loop(First, Previous);  
74.   
75.         {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =/= First, M =:= 0 ->  
76.             send_relay_message(Previous, Message, M, ReportBack),  
77.             void;  
78.   
79.         _Other ->  
80.             ?DEBUG("I am ~p I don't understand ~p~n", [This, _Other]),  
81.             loop(First, Previous)  
82.     end.  

-module(pring). %% %% Exported Functions %% -export([run/2]). -ifdef(debug). -define(DEBUG(Format, Args), io:format(Format, Args)). -else. -define(DEBUG(Format, Args), void). -endif. %% %% API Functions %% run(N, M) when is_integer(N), is_integer(M), N > 0, M > 0 -> time_it(fun() -> start(N, M) end). %% %% Local Functions %% time_it(F) when is_function(F) -> statistics(runtime), statistics(wall_clock), Result = F(), {_, CPUTime} = statistics(runtime), {_, ElapsedTime} = statistics(wall_clock), {Result, CPUTime/1000, ElapsedTime/1000}. start(N, M) when is_integer(N), is_integer(M), N > 0, M > 0 -> This = self(), First = spawn(fun() -> loop(This, This) end), ?DEBUG("Creating ~p for N = ~p~n", [First, N]), Last = create(First, First, N - 1), First ! {realign, First, Last}, First ! {relay, message, M, This}, receive {done, message} -> N * M end. create(First, Previous, N) when is_pid(First), is_pid(Previous), is_integer(N), N > 0 -> Pid = spawn(fun() -> loop(First, Previous) end), ?DEBUG("Creating ~p for N = ~p~n", [Pid, N]), create(First, Pid, N - 1); create(First, Previous, 0) when is_pid(First), is_pid(Previous) -> Previous. send_relay_message(Target, Message, M, ReportBack) when is_pid(Target), is_integer(M), is_pid(ReportBack) -> ?DEBUG("~p => ~p (M = ~p) ~n", [self(), Target, M + 1]), Target ! {relay, Message, M, ReportBack}. loop(First, Previous) when is_pid(First), is_pid(Previous) -> This = self(), receive {realign, NewFirst, NewPrevious} when is_pid(NewFirst), is_pid(NewPrevious) -> ?DEBUG("I am ~p Realigning: First ~p to ~p, Previous ~p to ~p~n", [This, First, NewFirst, Previous, NewPrevious]), loop(NewFirst, NewPrevious); {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =:= First, M > 0 -> send_relay_message(Previous, Message, M - 1, ReportBack), loop(First, Previous); {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =:= First, M =:= 0 -> ?DEBUG("I am ~p All rounds of relay over~n", [This]), ReportBack ! {done, Message}, void; {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =/= First, M > 0 -> send_relay_message(Previous, Message, M, ReportBack), loop(First, Previous); {relay, Message, M, ReportBack} when is_integer(M), is_pid(ReportBack), This =/= First, M =:= 0 -> send_relay_message(Previous, Message, M, ReportBack), void; _Other -> ?DEBUG("I am ~p I don't understand ~p~n", [This, _Other]), loop(First, Previous) end.
I wrote the "other programming language" version in Python. The benchmark code makes use of Python's multiprocessing module (available in version 2.6 and above). Since I couldn't find any Erlang-style way to directly send a message to a process I chose to use a set of Queue objects to accomplish message passing. Each process has access to two queues - an "in" queue to receive messages and an "out" queue to relay them to the next process in the ring. Each process' "out" queue serves as the "in" for the next. After constructing N - 1 processes, a "plug" process is created to complete the process ring.

view plainprint?

1. from multiprocessing import Process, Pipe, Queue  
2. import os, sys, argparse, time  
3.   
4. def create(first, target, in_queue, n):  
5.     if n == 0:  
6.         return in_queue  
7.   
8.     out_queue = Queue()  
9.     process = Process(target = target, args = (first, in_queue, out_queue,))  
10.     process.start()  
11.     return create(first, target, out_queue, n - 1)  
12.   
13. def send_relay_message(queue, message, m, sender):  
14.     queue.put(('relay', message, m,))  
15.   
16. def loop(first, in_queue, out_queue):  
17.     this = os.getpid()  
18.   
19.     while True:  
20.         input_data = in_queue.get()  
21.         message_type = input_data[0]  
22.   
23.         if message_type == 'realign':  
24.             new_first = input_data[1]  
25.             first = new_first  
26.   
27.         elif message_type == 'relay':  
28.             (message, m) = input_data[1:]  
29.             if (this == first) and (m > 0):  
30.                 send_relay_message(out_queue, message, m - 1, this)  
31.   
32.             elif (this == first) and (m == 0):  
33.                 return  
34.   
35.             elif (this != first) and (m > 0):  
36.                 send_relay_message(out_queue, message, m, this)  
37.   
38.             elif (this != first) and (m == 0):  
39.                 send_relay_message(out_queue, message, m, this)  
40.                 return  
41.   
42.             else:  
43.                 print "I am %s. I do not understand %s" % (this, input_data)  
44.   
45. if __name__ == '__main__':  
46.     parser = argparse.ArgumentParser(description = 'Run an N * M ring benchmark.')  
47.     parser.add_argument('-n', '--processes', help ='Number of Processes')  
48.     parser.add_argument('-m', '--messages', help = 'Number of Messages')  
49.     args = parser.parse_args(sys.argv[1:] or [])  
50.   
51.     n, m = map(int, (args.processes, args.messages))  
52.     start = time.time()  
53.   
54.     this = os.getpid()  
55.     first_in_queue = Queue()  
56.     first_out_queue = Queue()  
57.     first = Process(target = loop, args = [this, first_in_queue, first_out_queue])  
58.     first.start()  
59.   
60.     last_out_queue = create(first.pid, loop, first_out_queue, n - 2)  
61.     plug = Process(target = loop, args = [first, last_out_queue, first_in_queue])  
62.     plug.start()  
63.   
64.     first_in_queue.put(('realign', first.pid))  
65.     first_in_queue.put(('relay', 'hello', m))  
66.     first.join()  
67.     plug.join()  
68.   
69.     end = time.time()  
70.     print (end - start)  

from multiprocessing import Process, Pipe, Queue import os, sys, argparse, time def create(first, target, in_queue, n): if n == 0: return in_queue out_queue = Queue() process = Process(target = target, args = (first, in_queue, out_queue,)) process.start() return create(first, target, out_queue, n - 1) def send_relay_message(queue, message, m, sender): queue.put(('relay', message, m,)) def loop(first, in_queue, out_queue): this = os.getpid() while True: input_data = in_queue.get() message_type = input_data[0] if message_type == 'realign': new_first = input_data[1] first = new_first elif message_type == 'relay': (message, m) = input_data[1:] if (this == first) and (m > 0): send_relay_message(out_queue, message, m - 1, this) elif (this == first) and (m == 0): return elif (this != first) and (m > 0): send_relay_message(out_queue, message, m, this) elif (this != first) and (m == 0): send_relay_message(out_queue, message, m, this) return else: print "I am %s. I do not understand %s" % (this, input_data) if __name__ == '__main__': parser = argparse.ArgumentParser(description = 'Run an N * M ring benchmark.') parser.add_argument('-n', '--processes', help ='Number of Processes') parser.add_argument('-m', '--messages', help = 'Number of Messages') args = parser.parse_args(sys.argv[1:] or []) n, m = map(int, (args.processes, args.messages)) start = time.time() this = os.getpid() first_in_queue = Queue() first_out_queue = Queue() first = Process(target = loop, args = [this, first_in_queue, first_out_queue]) first.start() last_out_queue = create(first.pid, loop, first_out_queue, n - 2) plug = Process(target = loop, args = [first, last_out_queue, first_in_queue]) plug.start() first_in_queue.put(('realign', first.pid)) first_in_queue.put(('relay', 'hello', m)) first.join() plug.join() end = time.time() print (end - start)
I am not entirely convinced that Python's processes are the equivalent of Erlang ones. Certainly threads are not the answer given their use of shared memory. Perhaps the difficulty in creating lightweight, shared-nothing processes in other languages is what the author wanted to illustrate using the example.

The results are shown below. The program was executed on a machine with a dual core Pentium, 2GB memory and running Ubuntu Jaunty.

Please note that:

• I could not increase the number of Python processes to the order of 1000 (10^3). When I tried to do this the program crashed with a number of errors.
• I got a curious QueueFeederThread exception when I executed the process for N = 100, M = 10000.

N	M	N * M	(Wall) Time in seconds
			Erlang	Python
10	10	10^2	0.0	0.0732
10	100	10^3	0.003	0.1795
100	100	10^4	0.036	2.3976
100	1000	10^5	0.242	12.2252
1000	1000	10^6	1.877	-
1000	10000	10^7	18.593	-
10000	10000	10^8	197.477	-