gonzalobf/gen_udp_benchmark.erl

## gen_udp_benchmark.erl
-module(gen_udp_benchmark).
% gen_udp:send benchmark
%
% This set of benchmarks runs gen_udp:send with a number of different values and
% configurations. It demonstrates that the time taken to run gen_udp:send changes
% depending on the number of messages the caller has in its queue.
%
% During each gen_udp execution:
%  - no more messages are added to the caller's queue
%  - the time taken to execute gen_udp:send remains constant when running in a
%    loop. This confirms the first call takes as much as the rest of the calls
%    (mean and median very similar).
%
% It can be observed that the execution time for gen_udp:send grows linearly with the number of messages in the queue, suggesting that
% gen_udp:send is scanning all the messages in the queue of the caller.
%

-behaviour(gen_server).

-export([
    bench/0,
    bench/3,
    start/3,
    send/2,
    init/1,
    handle_cast/2,
    handle_call/3,
    handle_info/2
]).

bench() ->
    QueuedMessages = [100000, 1000000, 10000000],
    OnOff = [on_heap, off_heap],
    % Sends multiple messages in a loop and count how much each one takes
    MessagesToSend = [1, 10, 100, 1000],
    [
        bench(QM, OO, MTS)
     || MTS <- MessagesToSend,
        OO <- OnOff,
        QM <- QueuedMessages
    ],
    ok.

bench(QueuedMessages, OnOff, MessagesToSend) ->
    {Pid, Ref} = spawn_monitor(
        fun() ->
            {ok, Pid} = start(QueuedMessages, OnOff, MessagesToSend),
            send(Pid, QueuedMessages),
            receive
                done -> ok
            end,
            exit(Pid, kill)
        end
    ),
    % Wait to finish
    receive
        {'DOWN', Ref, process, Pid, normal} -> ok
    end.

start(QueuedMessages, OnOff, MessagesToSend) ->
    State = #{
        messages => QueuedMessages, caller => self(), heap => OnOff, msg_to_send => MessagesToSend
    },
    gen_server:start(?MODULE, State, [{spawn_opt, [{message_queue_data, OnOff}]}]).

send(Pid, QueuedMessages) ->
    gen_server:cast(Pid, send),
    % Create a big binary to send it over and queue a lot of messages in the process queue
    % Give the binary enough size so a ref is sent
    BigBinary = <<<<"a">> || _ <- lists:seq(1, 10000)>>,
    [Pid ! BigBinary || _ <- lists:seq(1, QueuedMessages)],
    ok.

init(State) ->
    erlang:process_flag(priority, high),
    {ok, Socket} = gen_udp:open(
        8780,
        [
            {active, true},
            binary,
            {sndbuf, 5000000}
        ]
    ),
    {ok, State#{socket => Socket}}.

handle_cast(send, State) ->
    #{
        socket := Socket,
        messages := Messages,
        caller := Caller,
        heap := OnOff,
        msg_to_send := MsgToSend
    } = State,
    % Wait until we have all the expected messages in the queue, this won't change the messages in the queue
    sleep_until_messages(Messages),

    % Send a udp message and time it. There is a receive inside gen_udp:send that
    % will need to iterate all queue looking for inet_reply
    Times = [send_message(Socket) || _ <- lists:seq(1, MsgToSend)],

    Min = lists:min(Times),
    Max = lists:max(Times),
    Mean = lists:sum(Times) / length(Times),
    Median = median(Times),
    Metrics = io_lib:format("~p (ms/message)", [
        [{min, Min}, {max, Max}, {mean, Mean}, {median, Median}]
    ]),
    io:format("~p msgs in queue, using ~p, sending ~p msg to gen_udp took -> ~s ~n", [
        Messages, OnOff, MsgToSend, Metrics
    ]),
    gen_udp:close(Socket),
    Caller ! done,
    {noreply, State}.

send_message(Socket) ->
    Before = os:system_time(microsecond),
    gen_udp:send(Socket, {127, 0, 0, 1}, 8789, <<"hey there!">>),
    After = os:system_time(microsecond),
    (After - Before) / 1000.

sleep_until_messages(Expected) ->
    timer:sleep(1000),
    {message_queue_len, Len} = erlang:process_info(self(), message_queue_len),
    case Expected > Len of
        true -> sleep_until_messages(Expected);
        false -> ok
    end.

median([]) ->
    [];
median(L) ->
    Sorted = lists:sort(L),
    Length = length(L),
    case Length rem 2 of
        0 ->
            [X, Y] = lists:sublist(Sorted, round(Length / 2), 2),
            (X + Y) / 2;
        1 ->
            lists:nth(round((Length + 1) / 2), Sorted)
    end.

handle_call(_, _, State) ->
    {stop, ok, State}.

handle_info(_, State) ->
    {noreply, State}.
	-module(gen_udp_benchmark).
	% gen_udp:send benchmark
	%
	% This set of benchmarks runs gen_udp:send with a number of different values and
	% configurations. It demonstrates that the time taken to run gen_udp:send changes
	% depending on the number of messages the caller has in its queue.
	%
	% During each gen_udp execution:
	% - no more messages are added to the caller's queue
	% - the time taken to execute gen_udp:send remains constant when running in a
	% loop. This confirms the first call takes as much as the rest of the calls
	% (mean and median very similar).
	%
	% It can be observed that the execution time for gen_udp:send grows linearly with the number of messages in the queue, suggesting that
	% gen_udp:send is scanning all the messages in the queue of the caller.
	%

	-behaviour(gen_server).

	-export([
	bench/0,
	bench/3,
	start/3,
	send/2,
	init/1,
	handle_cast/2,
	handle_call/3,
	handle_info/2
	]).

	bench() ->
	QueuedMessages = [100000, 1000000, 10000000],
	OnOff = [on_heap, off_heap],
	% Sends multiple messages in a loop and count how much each one takes
	MessagesToSend = [1, 10, 100, 1000],
	[
	bench(QM, OO, MTS)
	\|\| MTS <- MessagesToSend,
	OO <- OnOff,
	QM <- QueuedMessages
	],
	ok.

	bench(QueuedMessages, OnOff, MessagesToSend) ->
	{Pid, Ref} = spawn_monitor(
	fun() ->
	{ok, Pid} = start(QueuedMessages, OnOff, MessagesToSend),
	send(Pid, QueuedMessages),
	receive
	done -> ok
	end,
	exit(Pid, kill)
	end
	),
	% Wait to finish
	receive
	{'DOWN', Ref, process, Pid, normal} -> ok
	end.

	start(QueuedMessages, OnOff, MessagesToSend) ->
	State = #{
	messages => QueuedMessages, caller => self(), heap => OnOff, msg_to_send => MessagesToSend
	},
	gen_server:start(?MODULE, State, [{spawn_opt, [{message_queue_data, OnOff}]}]).

	send(Pid, QueuedMessages) ->
	gen_server:cast(Pid, send),
	% Create a big binary to send it over and queue a lot of messages in the process queue
	% Give the binary enough size so a ref is sent
	BigBinary = <<<<"a">> \|\| _ <- lists:seq(1, 10000)>>,
	[Pid ! BigBinary \|\| _ <- lists:seq(1, QueuedMessages)],
	ok.

	init(State) ->
	erlang:process_flag(priority, high),
	{ok, Socket} = gen_udp:open(
	8780,
	[
	{active, true},
	binary,
	{sndbuf, 5000000}
	]
	),
	{ok, State#{socket => Socket}}.

	handle_cast(send, State) ->
	#{
	socket := Socket,
	messages := Messages,
	caller := Caller,
	heap := OnOff,
	msg_to_send := MsgToSend
	} = State,
	% Wait until we have all the expected messages in the queue, this won't change the messages in the queue
	sleep_until_messages(Messages),

	% Send a udp message and time it. There is a receive inside gen_udp:send that
	% will need to iterate all queue looking for inet_reply
	Times = [send_message(Socket) \|\| _ <- lists:seq(1, MsgToSend)],

	Min = lists:min(Times),
	Max = lists:max(Times),
	Mean = lists:sum(Times) / length(Times),
	Median = median(Times),
	Metrics = io_lib:format("~p (ms/message)", [
	[{min, Min}, {max, Max}, {mean, Mean}, {median, Median}]
	]),
	io:format("~p msgs in queue, using ~p, sending ~p msg to gen_udp took -> ~s ~n", [
	Messages, OnOff, MsgToSend, Metrics
	]),
	gen_udp:close(Socket),
	Caller ! done,
	{noreply, State}.

	send_message(Socket) ->
	Before = os:system_time(microsecond),
	gen_udp:send(Socket, {127, 0, 0, 1}, 8789, <<"hey there!">>),
	After = os:system_time(microsecond),
	(After - Before) / 1000.

	sleep_until_messages(Expected) ->
	timer:sleep(1000),
	{message_queue_len, Len} = erlang:process_info(self(), message_queue_len),
	case Expected > Len of
	true -> sleep_until_messages(Expected);
	false -> ok
	end.

	median([]) ->
	[];
	median(L) ->
	Sorted = lists:sort(L),
	Length = length(L),
	case Length rem 2 of
	0 ->
	[X, Y] = lists:sublist(Sorted, round(Length / 2), 2),
	(X + Y) / 2;
	1 ->
	lists:nth(round((Length + 1) / 2), Sorted)
	end.

	handle_call(_, _, State) ->
	{stop, ok, State}.

	handle_info(_, State) ->
	{noreply, State}.