aboutsummaryrefslogblamecommitdiffstats
path: root/lib/megaco/examples/meas/megaco_codec_mstone2.erl
blob: f3588f2e3d468526d908797bfe7db66ed3dde5bd (plain) (tree)















































































































































































































































































































































































































                                                                                   
%%
%% %CopyrightBegin%
%% 
%% Copyright Ericsson AB 2006-2009. All Rights Reserved.
%% 
%% The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved online at http://www.erlang.org/.
%% 
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%% 
%% %CopyrightEnd%
%%

%%
%%----------------------------------------------------------------------
%% 
%% megaco_codec_mstone2:start().
%% megaco_codec_mstone2:start_no_drv().
%% megaco_codec_mstone2:start_only_drv().
%% 
%%----------------------------------------------------------------------
%% Purpose: mstone 2 measurement
%%          This module implement a simple performence measurment case.
%%          The architecture is as followes:
%%          - One loader process: 
%%            It keeps a list of all codec combinations, including
%%            all the messages (in a list) for each codec. 
%%            Initially it creates a timer (finished) (circa 10 minutes). 
%%            It spawns a worker process for each codec config (it also 
%%            creates a monitor to each process so it knows when they 
%%            exit). When the result comes in from a process (in the 
%%            form of a DOWN message), spawns a new worker process for 
%%            this codec config and update's the statistics.
%%            When the finished timer expires, it will stop respawing
%%            the worker processes, and instead just wait for them all
%%            to finish. 
%%            The test is finished by printing the statistics.
%%          - A worker process for each codec combination.
%%            This process is spawned by the loader process. It receives
%%            at start a list of messages. It shall decode and then 
%%            encode each message. When all messages has been processed
%%            it exits (normally).
%%----------------------------------------------------------------------

-module(megaco_codec_mstone2).


%% Exports
-export([
	 start/0,          start/1, 
	 start_flex/0,     start_flex/1, 
	 start_no_drv/0,   start_no_drv/1, 
	 start_only_drv/0, start_only_drv/1
	]).


%%%----------------------------------------------------------------------
%%% Macros
%%%----------------------------------------------------------------------

-define(LIB, megaco_codec_mstone_lib).

-ifndef(MSTONE_TIME).
-define(MSTONE_TIME, 10).
-endif.
-define(MSTONE_RUN_TIME, timer:minutes(?MSTONE_TIME)).

-ifndef(MSTONE_VERSION3).
-define(MSTONE_VERSION3, v3).
-endif.
-define(VERSION3, ?MSTONE_VERSION3).

-ifndef(MSTONE_CODECS).
-define(MSTONE_CODECS, megaco_codec_transform:codecs()).
-endif.

-define(DEFAULT_MESSAGE_PACKAGE, megaco_codec_transform:default_message_package()).
-define(DEFAULT_FACTOR,          1).
-define(DEFAULT_DRV_INCLUDE,     ignore).

-ifndef(MSTONE_RUNNER_MIN_HEAP_SZ).
%% -define(MSTONE_RUNNER_MIN_HEAP_SZ,  16#7fff).
-define(MSTONE_RUNNER_MIN_HEAP_SZ,  16#ffff).
%% -define(MSTONE_RUNNER_MIN_HEAP_SZ, 16#17ffe).
%% -define(MSTONE_RUNNER_MIN_HEAP_SZ, 16#1ffff).
%% -define(MSTONE_RUNNER_OPTS, [link]).
-endif.
-define(MSTONE_RUNNER_OPTS, 
        [link, {min_heap_size, ?MSTONE_RUNNER_MIN_HEAP_SZ}]).


%%%----------------------------------------------------------------------
%%% Records
%%%----------------------------------------------------------------------

-record(codec_data, {ref, mod, config = [], msgs = []}).

-record(state, {timer, running = [], idle = [], flex_handler, flex_conf}).


%%%----------------------------------------------------------------------
%%% API
%%%----------------------------------------------------------------------

start() ->
    start(?DEFAULT_MESSAGE_PACKAGE).

start([MessagePackage]) ->
    do_start(MessagePackage, ?DEFAULT_DRV_INCLUDE);
start(MessagePackage) ->
    do_start(MessagePackage, ?DEFAULT_DRV_INCLUDE).


start_flex() ->
    start_flex(?DEFAULT_MESSAGE_PACKAGE).

start_flex([MessagePackage]) ->
    do_start(MessagePackage, flex);
start_flex(MessagePackage) ->
    do_start(MessagePackage, flex).


start_no_drv() ->
    start_no_drv(?DEFAULT_MESSAGE_PACKAGE).

start_no_drv([MessagePackage]) ->
    do_start(MessagePackage, no_drv);
start_no_drv(MessagePackage) ->
    do_start(MessagePackage, no_drv).


start_only_drv() ->
    start_only_drv(?DEFAULT_MESSAGE_PACKAGE).

start_only_drv([MessagePackage]) ->
    do_start(MessagePackage, only_drv);
start_only_drv(MessagePackage) ->
    do_start(MessagePackage, only_drv).


do_start(MessagePackageRaw, DrvInclude) ->
%%     io:format("do_start -> entry with"
%% 	      "~n   MessagePackageRaw: ~p"
%% 	      "~n   DrvInclude:        ~p"
%% 	      "~n", [MessagePackageRaw, DrvInclude]), 
    MessagePackage = parse_message_package(MessagePackageRaw),
    mstone_init(MessagePackage, DrvInclude).
    
parse_message_package(MessagePackageRaw) when is_list(MessagePackageRaw) ->
    list_to_atom(MessagePackageRaw);
parse_message_package(MessagePackage) when is_atom(MessagePackage) ->
    MessagePackage;
parse_message_package(BadMessagePackage) ->
    throw({error, {bad_message_package, BadMessagePackage}}).


mstone_init(MessagePackage, DrvInclude) ->
    io:format("~n", []),
    ?LIB:set_default_sched_bind(), 
    ?LIB:display_os_info(),
    ?LIB:display_system_info(),
    ?LIB:display_app_info(),
    io:format("~n", []),
    Ref = erlang:monitor(process, 
			 spawn(fun() -> 
				       loader(MessagePackage, DrvInclude) 
			       end)),
    receive
	{'DOWN', Ref, process, _Pid, {done, Result}} ->
	    display_result(Result);
	{'DOWN', Ref, process, _Pid, Result} ->
	    io:format("Unexpected result:~n~p~n", [Result]),
	    ok
    end.


%%%----------------------------------------------------------------------
%%% Internal functions
%%%----------------------------------------------------------------------

display_result(Result) ->
    {value, {worker_cnt, WC}} = lists:keysearch(worker_cnt, 1, Result),
    CodecStat = 
	[{Mod, Conf, Cnt} || {{codec_cnt, Mod, Conf}, Cnt} <- Result],
    MStone = lists:sum([Cnt || {_, _, Cnt} <- CodecStat]),
    io:format("Number of procs spawned: ~w~n"
	      "MStone:                  ~w~n"
	      "~n", [WC, MStone]),
    display_worker_result(lists:keysort(3, CodecStat)),
    ok.
    
display_worker_result([]) ->
    io:format("~n", []);
display_worker_result([{Mod, Conf, Cnt}|Res]) ->
    io:format("~s: ~w~n", [image_of(Mod, Conf), Cnt]),
    display_worker_result(Res).

image_of(megaco_per_bin_encoder, Conf) ->
    bin_image("per_bin", Conf);
image_of(megaco_ber_bin_encoder, Conf) ->
    bin_image("ber_bin", Conf);
image_of(megaco_pretty_text_encoder, Conf) ->
    text_image("pretty", Conf);
image_of(megaco_compact_text_encoder, Conf) ->
    text_image("compact", Conf);
image_of(megaco_erl_dist_encoder, Conf) ->
    erl_image("erl_dist", Conf).

bin_image(Bin, Conf) ->
    Drv = 
	case lists:member(driver, Conf) of
	    true ->
		[driver];
	    false ->
		[]
	end,
    Nat = 
	case lists:member(native, Conf) of
	    true ->
		[native];
	    false ->
		[]
	end,
    io_lib:format("~s ~w", [Bin, Drv ++ Nat]).

text_image(Txt, Conf) ->
    Flex = 
	case lists:keysearch(flex, 1, Conf) of
	    false ->
		[];
	    _ ->
		[flex]
	end,
    io_lib:format("~s ~w", [Txt, Flex]).

erl_image(Erl, Conf) ->
    MC = 
	case lists:member(megaco_compressed, Conf) of
	    true ->
		[megaco_compressed];
	    false ->
		[]
	end,
    C = 
	case lists:member(compressed, Conf) of
	    true ->
		[compressed];
	    false ->
		[]
	end,
    io_lib:format("~s ~w", [Erl, MC ++ C]).
    

%%%----------------------------------------------------------------------

loader(MessagePackage, DrvInclude) ->
    loader(?MSTONE_CODECS, MessagePackage, DrvInclude).


%% Codecs is a list of megaco codec shortnames: 
%%
%%    pretty | compact | ber | per | erlang
%%

loader(Codecs, MessagePackage, DrvInclude) ->
    process_flag(trap_exit, true),
    case (catch init(Codecs, MessagePackage, DrvInclude)) of
	{ok, State} ->
	    loader_loop(running, State);
	Error ->
	    exit(Error)
    end.

init(Codecs, MessagePackage, DrvInclude) ->
    ets:new(mstone, [set, private, named_table, {keypos, 1}]),
    ets:insert(mstone, {worker_cnt, 0}),
    {Pid, FlexConf} = ?LIB:start_flex_scanner(),
    io:format("prepare messages", []),
    EMessages = ?LIB:expanded_messages(MessagePackage, Codecs, DrvInclude), 
    io:format("~ninit codec data", []),
    CodecData = init_codec_data(EMessages, FlexConf),
    Timer = erlang:send_after(?MSTONE_RUN_TIME, self(), mstone_finished), 
    io:format("~n", []),
    {ok, #state{timer        = Timer, 
		idle         = CodecData, 
		flex_handler = Pid, 
		flex_conf    = FlexConf}}.

init_codec_data(EMsgs, FlexConf) ->
    [init_codec_data(Codec, Mod, Conf, Msgs, FlexConf) || 
	{Codec, Mod, Conf, Msgs} <- EMsgs].

init_codec_data(Codec, Mod, Conf0, Msgs0, FlexConf) 
  when is_atom(Codec) andalso 
       is_atom(Mod)   andalso 
       is_list(Conf0) andalso 
       is_list(Msgs0)  ->
    io:format(".", []),
    Conf = [{version3,?VERSION3}|init_codec_conf(FlexConf, Conf0)], 
    Msgs = [?LIB:detect_version(Mod, Conf, Bin) || {_, Bin} <- Msgs0],
    ets:insert(mstone, {{codec_cnt, Mod, Conf}, 0}),
    #codec_data{mod = Mod, config = Conf, msgs = Msgs}.
    

init_codec_conf(FlexConf, [flex_scanner]) ->
    FlexConf;
init_codec_conf(_, Conf) ->
    Conf.


%% -- Main loop --

loader_loop(finishing, #state{flex_handler = Pid, running = []}) ->	
    %% we are done
    io:format("~n", []),
    ?LIB:stop_flex_scanner(Pid),
    exit({done, lists:sort(ets:tab2list(mstone))});

loader_loop(finishing, State) ->
    receive
	{'DOWN', Ref, process, _Pid, {mstone_done, Codec, Conf, Cnt}} ->
	    loader_loop(finishing, done_worker(Ref, Codec, Conf, Cnt, State))
    end;

loader_loop(running, #state{idle = []} = State) ->	    
    receive
	mstone_finished ->
	    loader_loop(finishing, State);

	{'DOWN', Ref, process, _Pid, {mstone_done, Codec, Conf, Cnt}} ->
	    loader_loop(running, done_worker(Ref, Codec, Conf, Cnt, State))
    end;

loader_loop(running, State) ->	
    receive
	mstone_finished ->
	    loader_loop(finishing, State);

	{'DOWN', Ref, process, _Pid, {mstone_done, Codec, Conf, Cnt}} ->
	    State2 = done_worker(Ref, Codec, Conf, Cnt, State),
	    loader_loop(running, State2)

    after 0 ->
	    loader_loop(running, start_worker(State))
    end.

done_worker(Ref, Codec, Conf, Cnt,
	    #state{running = Running, idle = Idle} = State) ->
    %% io:format("worker ~w ~w done with ~w~n", [Codec, Conf, Cnt]),
    ets:update_counter(mstone, worker_cnt, 1),
    ets:update_counter(mstone, {codec_cnt, Codec, Conf}, Cnt),
    Running2 = lists:keydelete(Ref, #codec_data.ref, Running),
    CD = Running -- Running2,
    State#state{running = Running2, idle = lists:append(Idle, CD)}.

start_worker(#state{running = Running, idle = [H|T]} = State) ->
    #codec_data{mod = Codec, config = Conf, msgs = Msgs} = H,
    Worker = fun() -> worker(Codec, Conf, Msgs, 0) end, 
    Ref    = erlang:monitor(process, spawn(Worker)),
    CD = H#codec_data{ref = Ref},
    State#state{running = [CD | Running], idle = T}.


%%%----------------------------------------------------------------------

worker(Codec, Conf, [], Cnt) ->
    exit({mstone_done, Codec, Conf, Cnt});
worker(Codec, Conf, [{V, Msg}|Msgs], Cnt) ->
    work(Codec, Conf, V, Msg),
    worker(Codec, Conf, Msgs, Cnt + 1).

work(Codec, Conf, V, M) ->
    case (catch apply(Codec, decode_message, [Conf, V, M])) of
        {ok, Msg} ->
            case (catch apply(Codec, encode_message, [Conf, V, Msg])) of
                {ok, Bin} when is_binary(Bin) ->
                    ok;
                EncodeError ->
		    emsg("failed encoding message: ~n~p", [EncodeError]),
                    exit({mstone_worker_encode_failure, EncodeError})
            end;
        DecodeError ->
            emsg("failed decoding message: ~n~p", [DecodeError]),
	    exit({mstone_worker_decode_failure, DecodeError})
    end.
    


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

emsg(F, A) ->
    error_logger:error_msg(F ++ "~n", A).