et_collector.erl

OTP是开放电信平台的简称

report(TH, end_of_trace) when record(TH, table_handle) ->
    {ok, TH};
report(_, Bad) ->
    exit({bad_event, Bad}).

report_event(CollectorPid, DetailLevel, FromTo, Label, Contents) ->
    report_event(CollectorPid, DetailLevel, FromTo, FromTo, Label, Contents).

report_event(CollectorPid, DetailLevel, From, To, Label, Contents)
  when integer(DetailLevel), DetailLevel >= 0, DetailLevel =< 100, list(Contents) ->
    TS= erlang:now(),
    E = #event{detail_level = DetailLevel,
               trace_ts     = TS,
               event_ts     = TS,
               from         = From,
               to           = To, 
               label        = Label, 
               contents     = Contents},
    report(CollectorPid, E).

%%----------------------------------------------------------------------
%% make_key(Type, Stuff) -> Key
%%
%% Makes a key out of an event record or an old key
%%
%% Type = record(table_handle) | trace_ts | event_ts
%% Stuff = record(event) | Key
%% Key = record(event_ts) | record(trace_ts)
%%----------------------------------------------------------------------

make_key(TH, Stuff) when record(TH, table_handle) ->
    make_key(TH#table_handle.event_order, Stuff);
make_key(trace_ts, Stuff) ->
    if
        record(Stuff, event) ->
            #event{trace_ts = R, event_ts = P} = Stuff,
            #trace_ts{trace_ts = R, event_ts = P};
        record(Stuff, trace_ts) ->
            Stuff;
        record(Stuff, event_ts) ->
            #event_ts{trace_ts = R, event_ts = P} = Stuff,
            #trace_ts{trace_ts = R, event_ts = P}
    end;
make_key(event_ts, Stuff) ->
    if
        record(Stuff, event) ->
            #event{trace_ts = R, event_ts = P} = Stuff,
            #event_ts{trace_ts = R, event_ts = P};
        record(Stuff, event_ts) ->
            Stuff;
        record(Stuff, trace_ts) ->
            #trace_ts{trace_ts = R, event_ts = P} = Stuff,
            #event_ts{trace_ts = R, event_ts = P}
    end.

%%----------------------------------------------------------------------
%% get_table_handle(CollectorPid) -> Handle
%%
%% Return a table handle
%%
%% CollectorPid = pid()
%% Handle = record(table_handle)
%%----------------------------------------------------------------------

get_table_handle(CollectorPid) when pid(CollectorPid) ->
    call(CollectorPid, get_table_handle).

%%----------------------------------------------------------------------
%% get_global_pid() -> CollectorPid | exit(Reason)
%%
%% Return a the identity of the globally registered collector
%% if there is any
%%
%% CollectorPid = pid()
%% Reason = term()
%%----------------------------------------------------------------------

get_global_pid() ->
    case global:whereis_name(?MODULE) of
        CollectorPid when pid(CollectorPid) ->
            CollectorPid;
        undefined ->
            exit(global_collector_not_started)
    end.

%%----------------------------------------------------------------------
%% change_pattern(CollectorPid, RawPattern) -> {old_pattern, TracePattern}
%%
%% Change active trace pattern globally on all trace nodes
%%
%% CollectorPid = pid()
%% RawPattern = {report_module(), extended_dbg_match_spec()}
%% report_module() = atom() | undefined
%% extended_dbg_match_spec()() = detail_level() | dbg_match_spec()
%% RawPattern = detail_level()
%% detail_level() = min | max | integer(X) when X =< 0, X >= 100
%% TracePattern = {report_module(), dbg_match_spec_match_spec()}
%%----------------------------------------------------------------------

change_pattern(CollectorPid, RawPattern) ->
    Pattern = et_selector:make_pattern(RawPattern),
    call(CollectorPid, {change_pattern, Pattern}).

%%----------------------------------------------------------------------
%% dict_insert(CollectorPid, {filter, collector}, FilterFun) -> ok
%% dict_insert(CollectorPid, {subscriber, SubscriberPid}, Void) -> ok
%% dict_insert(CollectorPid, Key, Val) -> ok
%% 
%% Insert a dictionary entry
%% and send a {et, {dict_insert, Key, Val}} tuple
%% to all registered subscribers.
%%
%% If the entry is a new subscriber, it will imply that
%% the new subscriber process first will get one message
%% for each already stored dictionary entry, before it
%% and all old subscribers will get this particular entry.
%% The collector process links to and then supervises the
%% subscriber process. If the subscriber process dies it
%% will imply that it gets unregistered as with a normal
%% dict_delete/2.
%%
%% CollectorPid = pid()
%% FilterFun = filter_fun() 
%% SubscriberPid = pid()
%% Void = term()
%% Key = term()
%% Val = term()
%%----------------------------------------------------------------------

dict_insert(CollectorPid, Key = {filter, Name}, Fun) ->
    if
	atom(Name), function(Fun) ->
	    call(CollectorPid, {dict_insert, Key, Fun});
	true ->
	    exit({badarg, Key})
    end;
dict_insert(CollectorPid, Key = {subscriber, Pid}, Val) ->
    if
	pid(Pid) ->
	    call(CollectorPid, {dict_insert, Key, Val});
	true ->
	    exit({badarg, Key})
    end;
dict_insert(CollectorPid, Key, Val) ->
    call(CollectorPid, {dict_insert, Key, Val}).

%%----------------------------------------------------------------------
%% dict_lookup(CollectorPid, Key) -> [Val]
%%
%% Lookup a dictionary entry and return zero or one value
%% 
%% CollectorPid = pid()
%% Key = term()
%% Val = term()
%%----------------------------------------------------------------------

dict_lookup(CollectorPid, Key) ->
    call(CollectorPid, {dict_lookup, Key}).

%%----------------------------------------------------------------------
%% Ddict_delete(CollectorPid, Key) -> ok
%%
%% elete a dictionary entry
%% and send a {et, {dict_delete, Key}} tuple
%% to all registered subscribers.
%%
%% If the deleted entry is a registered subscriber, it will
%% imply that the subscriber process gets is unregistered as
%% subscriber as well as it gets it final message.
%%
%% dict_delete(CollectorPid, {subscriber, SubscriberPid})
%% dict_delete(CollectorPid, Key)
%% 
%% CollectorPid = pid()
%% SubscriberPid = pid()
%% Key = term()
%%----------------------------------------------------------------------

dict_delete(CollectorPid, Key) ->
    call(CollectorPid, {dict_delete, Key}).

%%----------------------------------------------------------------------
%% dict_match(CollectorPid, Pattern) -> [Match]
%%
%% Match some dictionary entries
%% 
%% CollectorPid = pid()
%% Pattern = '_' | {key_pattern(), val_pattern()}
%% key_pattern() = ets_match_object_pattern()
%% val_pattern() = ets_match_object_pattern()
%% Match = {key(), val()}
%% key() = term()
%% val() = term()
%%----------------------------------------------------------------------

dict_match(CollectorPid, Pattern)  ->
    call(CollectorPid, {dict_match, Pattern}).

%%----------------------------------------------------------------------
%% multicast(_CollectorPid, Msg) -> ok
%%
%% Sends a message to all registered subscribers
%%
%% CollectorPid = pid()
%% Msg = term()
%%----------------------------------------------------------------------

multicast(_CollectorPid, Msg = {dict_insert, _Key, _Val}) ->
    exit({badarg, Msg});
multicast(_CollectorPid, Msg = {dict_delete, _Key}) ->
    exit({badarg, Msg});
multicast(CollectorPid, Msg) ->
    call(CollectorPid, {multicast, Msg}).

%%----------------------------------------------------------------------
%% start_trace_client(CollectorPid, Type, Parameters) ->
%%     file_loaded | {trace_client_pid, pid()} | exit(Reason)
%%
%% Load raw Erlang trace from a file, port or process.
%% 
%% Type       = dbg_trace_client_type()
%% Parameters = dbg_trace_client_parameters()
%% Pid        = dbg_trace_client_pid()
%%----------------------------------------------------------------------

start_trace_client(CollectorPid, Type, FileName) when Type == event_file ->
    load_event_file(CollectorPid, FileName);
start_trace_client(CollectorPid, Type, FileName) when Type == file -> 
    WaitFor = {make_ref(), end_of_trace},
    EventFun = fun(E, {ReplyTo, {ok, TH}}) -> {ReplyTo, report(TH, E)} end,
    EndFun = fun({ReplyTo, {ok, _TH}}) -> ReplyTo ! WaitFor, ReplyTo  end,
    Spec = trace_spec_wrapper(EventFun, EndFun, {self(), {ok, CollectorPid}}),
    Pid = dbg:trace_client(Type, FileName, Spec),
    unlink(Pid),
    Ref = erlang:monitor(process, Pid),
    receive
        WaitFor -> 
	    erlang:demonitor(Ref),
	    receive 
		{'DOWN', Ref, _, _, _} ->
		    file_loaded
	    after 0 ->
		    file_loaded
	    end;
        {'DOWN', Ref, _, _, Reason} ->
            exit(Reason)
    end;
start_trace_client(CollectorPid, Type, Parameters) ->
    EventFun = fun(Event, {ok, TH}) -> report(TH, Event) end,
    EndFun   = fun(Acc) -> Acc end,
    Spec = trace_spec_wrapper(EventFun, EndFun, {ok, CollectorPid}),
    Pid = dbg:trace_client(Type, Parameters, Spec),
    CollectorPid ! {register_trace_client, Pid},
    unlink(Pid),
    {trace_client_pid, Pid}.

trace_spec_wrapper(EventFun, EndFun, EventInitialAcc)
  when function(EventFun), function(EndFun) ->
    {fun(Trace, Acc) -> 
             case Trace == end_of_trace of
                 true  -> EndFun(Acc);
                 false -> EventFun(Trace,  Acc)
             end
     end,
     EventInitialAcc}.

start_trace_port(Parameters) ->
    dbg:tracer(port, dbg:trace_port(ip, Parameters)).

%%----------------------------------------------------------------------
%% iterate(Handle, Prev, Limit) ->
%%     iterate(Handle, Prev, Limit, undefined, Prev)
%%     
%% Iterates over the currently stored events
%% 
%% Short for iterate/5.
%%----------------------------------------------------------------------

iterate(Handle, Prev, Limit) ->
    iterate(Handle, Prev, Limit, undefined, Prev).

%%----------------------------------------------------------------------
%% iterate(Handle, Prev, Limit, Fun, Acc) -> NewAcc
%%
%% Iterates over the currently stored events and apply a function for
%% each event. The iteration may be performed forwards or backwards
%% and may be limited to a maximum number of events (abs(Limit)).
%%
%% Handle = collector_pid() | table_handle()
%% Prev = first | last | event_key()
%% Limit = done() | forward() | backward()
%% collector_pid() = pid()
%% table_handle() = record(table_handle)
%% event_key() = 
%% done() = 0
%% forward() = infinity | integer(X) where X > 0
%% backward() = '-infinity' | integer(X) where X < 0
%% Fun = fun(Event, Acc) -> NewAcc
%% Acc = NewAcc = term()
%%----------------------------------------------------------------------

iterate(_, _, Limit, _, Acc) when Limit == 0 ->
    Acc;
iterate(CollectorPid, Prev, Limit, Fun, Acc) when pid(CollectorPid) ->
    case get_table_handle(CollectorPid) of
        {ok, TH} when record(TH, table_handle) ->
            iterate(TH, Prev, Limit, Fun, Acc);
        {error, Reason} ->
            exit(Reason)
    end;
iterate(TH, Prev, Limit, Fun, Acc) when record(TH, table_handle) ->
    if
        Limit == infinity ->
            next_iterate(TH, Prev, Limit, Fun, Acc);
        integer(Limit), Limit > 0 ->
            next_iterate(TH, Prev, Limit, Fun, Acc);
        Limit == '-infinity' ->
            prev_iterate(TH, Prev, Limit, Fun, Acc);
        integer(Limit), Limit < 0 ->
            prev_iterate(TH, Prev, Limit, Fun, Acc)
    end.    
    
next_iterate(TH, Prev = first, Limit, Fun, Acc) ->
    Tab = TH#table_handle.event_tab,
    case catch ets:first(Tab) of
        '$end_of_table' ->
            Acc;
        {'EXIT', _} = Error ->
            io:format("~p(~p): First ~p~n", [?MODULE, ?LINE, Error]),
            iterate(TH#table_handle.collector_pid, Prev, Limit, Fun, Acc);
        First ->
            lookup_and_apply(TH, Prev, First, Limit, -1, Fun, Acc)
    end;
next_iterate(TH, Prev = last, Limit, Fun, Acc) ->
    Tab = TH#table_handle.event_tab,
    case catch ets:last(Tab) of
        '$end_of_table' ->
            Acc;
        {'EXIT', _} = Error ->
            io:format("~p(~p): Last ~p~n", [?MODULE, ?LINE, Error]),
            iterate(TH#table_handle.collector_pid, Prev, Limit, Fun, Acc);
        Last ->
            lookup_and_apply(TH, Prev, Last, Limit, -1, Fun, Acc)
    end;
next_iterate(TH, Prev, Limit, Fun, Acc) ->
    Tab = TH#table_handle.event_tab,
    Key = make_key(TH, Prev),
    case catch ets:next(Tab, Key) of
        '$end_of_table' ->
            Acc;
        {'EXIT', _} = Error ->
            io:format("~p(~p): Next ~p -> ~p~n", [?MODULE, ?LINE, Key, Error]),
            iterate(TH#table_handle.collector_pid, Prev, Limit, Fun, Acc);
        Next ->
            lookup_and_apply(TH, Prev, Next, Limit, -1, Fun, Acc)
    end.

prev_iterate(TH, Prev = first, Limit, Fun, Acc) ->
    Tab = TH#table_handle.event_tab,
    case catch ets:first(Tab) of
        '$end_of_table' ->
            Acc;
        {'EXIT', _} = Error ->
            io:format("~p(~p): First ~p~n", [?MODULE, ?LINE, Error]),
            iterate(TH#table_handle.collector_pid, Prev, Limit, Fun, Acc);
        First ->
            lookup_and_apply(TH, Prev, First, Limit, 1, Fun, Acc)
    end;
prev_iterate(TH, Prev = last, Limit, Fun, Acc) ->
    Tab = TH#table_handle.event_tab,
    case catch ets:last(Tab) of
        '$end_of_table' ->
            Acc;
        {'EXIT', _} = Error ->
            io:format("~p(~p): Last ~p~n", [?MODULE, ?LINE, Error]),
            iterate(TH#table_handle.collector_pid, Prev, Limit, Fun, Acc);
        Last ->
            lookup_and_apply(TH, Prev, Last, Limit, 1, Fun, Acc)
    end;
prev_iterate(TH, Prev, Limit, Fun, Acc) ->
    Tab = TH#table_handle.event_tab,
    Key = make_key(TH, Prev),
    case catch ets:prev(Tab, Key) of
        '$end_of_table' ->
            Acc;
        {'EXIT', _} = Error ->
            io:format("~p(~p): Prev ~p -> ~p~n", [?MODULE, ?LINE, Key, Error]),
            iterate(TH#table_handle.collector_pid, Prev, Limit, Fun, Acc);
        Next ->
            lookup_and_apply(TH, Prev, Next, Limit, 1, Fun, Acc)
    end.

lookup_and_apply(TH, _Prev, Next, Limit, Incr, Fun, _Acc) when Fun == undefined ->
    Limit2 = incr(Limit, Incr),
    iterate(TH, Next, Limit2, Fun, Next);   
lookup_and_apply(TH, Prev, Next, Limit, Incr, Fun, Acc) ->
    Tab = TH#table_handle.event_tab,
    case catch ets:lookup_element(Tab, Next, 2) of
        {'EXIT', _} ->
            iterate(TH#table_handle.collector_pid, Prev, Limit, Fun, Acc);
        E when record(E, event) ->
            Acc2 = Fun(E, Acc),