ttb.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,001 行 · 第 1/2 页

	{init_node,Node,MetaFile,PI,Traci} ->
	    erlang:monitor_node(Node,true),
	    MetaPid = 
		case rpc:call(Node,
			      observer_backend,
			      ttb_init_node,
			      [MetaFile,PI,Traci]) of
		    {ok,MP} ->
			MP;
		    {badrpc,nodedown} ->
			%% We will get a nodedown message
			undefined
		end,
	    loop(dict:store(Node,{MetaFile,MetaPid},NodeInfo));
	{get_nodes,Sender} ->
	    Sender ! {?MODULE,dict:fetch_keys(NodeInfo)},
	    loop(NodeInfo);
	{write_trace_info,Key,What} ->
	    dict:fold(fun(Node,{_MetaFile,MetaPid},_) -> 
			      rpc:call(Node,observer_backend,
				       ttb_write_trace_info,[MetaPid,Key,What])
		      end,
		      ok,
		      NodeInfo),
	    loop(NodeInfo);
	{nodedown,Node} ->
	    loop(dict:erase(Node,NodeInfo));
	{stop,nofetch,Sender} ->
	    dict:fold(
	      fun(Node,{_,MetaPid},_) -> 
		      rpc:call(Node,observer_backend,ttb_stop,[MetaPid])
	      end,
	      ok,
	      NodeInfo),
	    dbg:stop_clear(),
	    ets:delete(?history_table),
	    Sender ! {?MODULE,stopped};
	{stop,FetchOrFormat,Sender} ->
	    Localhost = host(node()),
	    Dir = ?upload_dir++ts(),
 	    file:make_dir(Dir),
	    %% The nodes are traversed twice here because
	    %% the meta tracing in observer_backend must be
	    %% stopped before dbg is stopped, and dbg must
	    %% be stopped before the trace logs are moved orelse
	    %% windows complains.
	    AllNodesAndMeta = 
		dict:fold(
		  fun(Node,{MetaFile,MetaPid},Nodes) -> 
			  rpc:call(Node,observer_backend,ttb_stop,[MetaPid]),
			  [{Node,MetaFile}|Nodes]
		  end,
		  [],
		  NodeInfo),
	    dbg:stop_clear(),
	    AllNodes = 
		lists:map(
		  fun({Node,MetaFile}) ->
			  spawn(fun() -> fetch(Localhost,Dir,Node,MetaFile) end),
			  Node
		  end,
		  AllNodesAndMeta),
	    ets:delete(?history_table),
	    wait_for_fetch(AllNodes),
	    io:format("Stored logs in ~s~n",[filename:absname(Dir)]),
	    case FetchOrFormat of
		format -> format(Dir);
		fetch -> ok
	    end,
	    Sender ! {?MODULE,stopped}
         ?get_status
    end.

get_nodes() ->
    ?MODULE ! {get_nodes,self()},
    receive {?MODULE,Nodes} -> Nodes end.

ts() ->
    {{Y,M,D},{H,Min,S}} = calendar:now_to_local_time(now()),
    io_lib:format("-~4.4.0w~2.2.0w~2.2.0w-~2.2.0w~2.2.0w~2.2.0w",
		  [Y,M,D,H,Min,S]).



fetch(Localhost,Dir,Node,MetaFile) ->
    case host(Node) of
	Localhost -> % same host, just move the files
	    Files = rpc:call(Node,observer_backend,ttb_get_filenames,[MetaFile]),
	    lists:foreach(
	      fun(File0) ->
		      File = filename:join(Dir,filename:basename(File0)),
		      file:rename(File0,File)
	      end,
	      Files);
	_Otherhost ->
	    {ok, LSock} = gen_tcp:listen(0, [binary,{packet,2},{active,false}]),
	    {ok,Port} = inet:port(LSock),
	    rpc:cast(Node,observer_backend,ttb_fetch,
		     [MetaFile,{Port,Localhost}]),
	    {ok, Sock} = gen_tcp:accept(LSock),
	    receive_files(Dir,Sock,undefined),
	    ok = gen_tcp:close(LSock),
	    ok = gen_tcp:close(Sock)
    end,
    ?MODULE ! {fetch_complete,Node}.

receive_files(Dir,Sock,Fd) ->
    case gen_tcp:recv(Sock, 0) of
	{ok, <<0,Bin/binary>>} ->
	    file:write(Fd,Bin),
	    receive_files(Dir,Sock,Fd);
	{ok, <<1,Bin/binary>>} ->
	    File0 = binary_to_list(Bin),
	    File = filename:join(Dir,File0),
	    {ok,Fd1} = file:open(File,[raw,write]),
	    receive_files(Dir,Sock,Fd1);
	{error, closed} ->
	    ok = file:close(Fd)
    end.    

host(Node) ->
    [_name,Host] = string:tokens(atom_to_list(Node),"@"),
    Host.


wait_for_fetch([]) ->
    ok;
wait_for_fetch(Nodes) ->
    receive 
	{fetch_complete,Node} ->
	    wait_for_fetch(lists:delete(Node,Nodes))
    end.

%%% - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
%%% TRACE INFORMATION FILE
%%% ======================
%%% The trace information file has the same name as the trace log,
%%% but with the extension ".ti". It contains process information,
%%% trace information and any data the user writes with the
%%% function write_trace_info/2.
%%%
%%% The file is read during formatting of trace logs, and all data
%%% except process information is included in the handler function.
%%% - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

write_info(Nodes,PI,Traci) ->
    lists:foreach(fun({N,{local,C,_},F}) -> 
			  MetaFile = F ++ ".ti",
			  file:delete(MetaFile),
			  Traci1 = [{node,N},{file,C}|Traci],
			  {ok,Port} = dbg:get_tracer(N),
			  ?MODULE ! 
			      {init_node, N, {local,MetaFile,Port}, PI, Traci1};
		     ({N,C,F}) -> 
			  MetaFile = F ++ ".ti",
			  Traci1 = [{node,N},{file,C}|Traci],
			  ?MODULE ! {init_node, N, MetaFile, PI, Traci1}
		  end,
		  Nodes).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% Format binary trace logs
format(Files) ->
    format(Files,[]).
format(Files,Opt) ->
    {Out,Handler} = format_opt(Opt),
    ets:new(?MODULE,[named_table]),
    format(Files,Out,Handler).
format(File,Out,Handler) when is_list(File), is_integer(hd(File)) ->
    Files = 
	case filelib:is_dir(File) of
	    true ->  % will merge all files in the directory
		MetaFiles = filelib:wildcard(filename:join(File,"*.ti")),
		lists:map(fun(M) ->
				  Sub = string:left(M,length(M)-3),
				  case filelib:is_file(Sub) of
				      true -> Sub;
				      false -> Sub++".*.wrp"
				  end
			  end,
			  MetaFiles);
	    false -> % format one file
		[File]
	end,
    format(Files,Out,Handler);
format(Files,Out,Handler) when is_list(Files), is_list(hd(Files)) ->
    StopDbg = case whereis(dbg) of
		  undefined -> true;
		  _ -> false
	      end,
    Details = lists:foldl(fun(File,Acc) -> [prepare(File,Handler)|Acc] end,
			  [],Files),
    Fd = get_fd(Out),
    R = do_format(Fd,Details),
    file:close(Fd),
    ets:delete(?MODULE),
    case StopDbg of
	true -> dbg:stop_clear();
	false -> ok
    end,
    R.

prepare(File,Handler) ->
    {Traci,Proci} = read_traci(File),
    Node = get_node(Traci),
    lists:foreach(fun({Pid,PI}) ->
			  %% The last definition for a Pid will overwrite
			  %% any previous definitions. That should be what
			  %% we want (we will get the registered name for
			  %% the process, rather than the initial call if
			  %% both are present in the list).
			  ets:insert(?MODULE,{Pid,PI,Node})
		  end,Proci),
    FileOrWrap = get_file(File,Traci),
    Handler1 = get_handler(Handler,Traci),
    {FileOrWrap,Traci,Handler1}.

format_opt(Opt) ->
    Out = case lists:keysearch(out,1,Opt) of
	      {value,{out,O}} -> O;
	      _ -> standard_io
	  end,
    Handler = case lists:keysearch(handler,1,Opt) of
	      {value,{handler,H}} -> H;
	      _ -> undefined
	  end,
    {Out,Handler}.


read_traci(File) ->
    MetaFile = get_metafile(File),
    case file:read_file(MetaFile) of
	{ok,B} -> 
	    interpret_binary(B,dict:new(),[]);
	_ -> 
	    io:format("Warning: no meta data file: ~s~n",[MetaFile]),
	    {dict:new(),[]}
    end.

get_metafile(File) ->
    case filename:rootname(File,".wrp") of
	File -> File++".ti";
	Wrap -> filename:rootname(Wrap)++".ti"
    end.


interpret_binary(<<>>,Dict,P) ->
    {Dict,lists:reverse(P)};
interpret_binary(B,Dict,P) ->
    {Term,Rest} = get_term(B),
    {Dict1,P1} = 
	case Term of
	    {pid,PI} ->
		{Dict,[PI|P]};
	    {Key,Val} ->
		{dict:update(Key,fun(Val0) -> [Val|Val0] end, [Val], Dict),P}
	end,
    interpret_binary(Rest,Dict1,P1).

get_fd(Out) ->
    case Out of
	standard_io ->
	    Out;
	_file ->
	    file:delete(Out),
	    case file:open(Out,[append]) of
		{ok,Fd} -> Fd;
		Error -> exit(Error)
	    end
    end.

get_node(Traci) ->
    case dict:find(node,Traci) of
	{ok,[Node]} -> Node;
	error -> unknown
    end.

get_file(File,Traci) ->
    case dict:find(file,Traci) of
	{ok,[Client]} ->
	    check_client(Client,File);
	error ->
	    check_exists(File)
    end.

check_client(Client,File) when is_list(Client) ->
    check_exists(File);
check_client(Client,File) when is_tuple(Client),element(2,Client)==wrap ->
    Root = filename:rootname(File,".wrp"),
    case filename:extension(Root) of
	".*" -> 
	    Part1 = filename:rootname(Root,"*"),
	    setelement(1,Client,Part1);
	_ -> 
	    check_exists(File)
    end.

check_exists(File) ->
    case file:read_file_info(File) of
	{ok,#file_info{type=regular}} -> File;
	_ -> 
	    exit({error,no_file})
    end.
	

get_handler(Handler,Traci) ->
    case Handler of
	undefined -> 
	    case dict:find(handler,Traci) of
		{ok,[H]} -> H;
		error -> undefined
	    end;
	_ ->
	    Handler
    end.

do_format(Fd,Details) ->
    Clients = lists:foldl(fun({FileOrWrap,Traci,Handler},Acc) ->
				  [start_client(FileOrWrap,Traci,Handler)
				   |Acc]
			  end,[],Details),
    init_collector(Fd,Clients).


start_client(FileOrWrap,Traci,et) ->
    dbg:trace_client(file, FileOrWrap, 
		     {fun handler/2, 
		      {dict:to_list(Traci),{{ttb_et,handler},initial}}});
start_client(FileOrWrap,Traci,undefined) ->
    dbg:trace_client(file, FileOrWrap, 
		     {fun handler/2, 
		      {dict:to_list(Traci),{fun defaulthandler/4,initial}}});
start_client(FileOrWrap,Traci,Handler) ->
    dbg:trace_client(file, FileOrWrap, 
		     {fun handler/2, {dict:to_list(Traci),Handler}}).

handler(Trace,State) ->
    %% State here is only used for the initial state. The accumulated
    %% State is maintained by collector!!!
    receive 
	{get,Collector} -> Collector ! {self(),{Trace,State}};
	done -> ok
    end,
    State.

handler1(Trace,{Fd,{Traci,{Fun,State}}}) when is_function(Fun) ->
    {Traci,{Fun,Fun(Fd,Trace,Traci,State)}};
handler1(Trace,{Fd,{Traci,{{M,F},State}}}) when is_atom(M), is_atom(F) ->
    {Traci,{{M,F},M:F(Fd,Trace,Traci,State)}}.

defaulthandler(Fd,Trace,_Traci,initial) ->
    dbg:dhandler(Trace,Fd);
defaulthandler(_Fd,Trace,_Traci,State) ->
    dbg:dhandler(Trace,State).

init_collector(Fd,Clients) ->
    Collected = get_first(Clients),
    collector(Fd,sort(Collected)).

collector(Fd,[{_,{Client,{Trace,State}}}|Rest]) ->
    Trace1 = update_procinfo(Trace),
    State1 = handler1(Trace1,{Fd,State}),
    case get_next(Client,State1) of
	end_of_trace -> 
	    handler1(end_of_trace,{Fd,State1}),
	    collector(Fd,Rest);
	Next -> collector(Fd,sort([Next|Rest]))
    end;
collector(_Fd,[]) ->
    ok.

update_procinfo({drop,_N}=Trace) ->
    Trace;
update_procinfo(Trace) when element(1,Trace)==seq_trace ->
    Info = element(3,Trace),
    Info1 = 
	case Info of
	    {send, Serial, From, To, Msg} ->
		{send, Serial, get_procinfo(From), get_procinfo(To), Msg};
	    {'receive', Serial, From, To, Msg} ->
		{'receive', Serial, get_procinfo(From), get_procinfo(To), Msg};
	    {print, Serial, From, Void, UserInfo} ->
		{print, Serial, get_procinfo(From), Void, UserInfo};
	    Other ->
		Other
	end,
    setelement(3,Trace,Info1);
update_procinfo(Trace) when element(3,Trace)==send ->
    PI = get_procinfo(element(5,Trace)),
    setelement(5,Trace,PI);
update_procinfo(Trace) ->
    Pid = element(2,Trace),
    ProcInfo = get_procinfo(Pid),
    setelement(2,Trace,ProcInfo).

get_procinfo(Pid) when is_pid(Pid) ->
    case ets:lookup(?MODULE,Pid) of
	[PI] -> PI;
	[] -> Pid
    end;
get_procinfo(Name) when is_atom(Name) ->
    case ets:match_object(?MODULE,{'_',Name,node()}) of
	[PI] -> PI;
	[] -> Name
    end;
get_procinfo({Name,Node}) when is_atom(Name) ->
    case ets:match_object(?MODULE,{'_',Name,Node}) of
	[PI] -> PI;
	[] -> {Name,Node}
    end.	 

get_first([Client|Clients]) ->
    Client ! {get,self()},
    receive 
	{Client,{end_of_trace,_}} -> 
	    get_first(Clients);
	{Client,{Trace,_State}}=Next -> 
	    [{timestamp(Trace),Next}|get_first(Clients)]
    end;
get_first([]) -> [].

get_next(Client,State) when is_pid(Client) ->
    Client ! {get,self()},
    receive 
	{Client,{end_of_trace,_}} -> 
	    end_of_trace;
	{Client,{Trace,_OldState}} -> 
	    {timestamp(Trace),{Client,{Trace,State}}} % inserting new state!!
    end.

sort(List) ->
    lists:keysort(1,List).


timestamp(Trace) when element(1,Trace)=:=trace_ts;
		      element(1,Trace)=:=seq_trace,size(Trace)=:=4 ->
    element(size(Trace),Trace);
timestamp(_Trace) ->
%%    exit({error,{no_timestamp,Trace}}).
    0.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% common internal functions
to_list(Atom) when is_atom(Atom) -> [Atom];
to_list(List) when is_list(List) -> List.

write_binary(File,TermList) ->
    {ok,Fd} = file:open(File,[raw,append]),    
    %% Using the function implemented in observer_backend, only because
    %% is exists - so I don't have to write the same code twice.
    observer_backend:ttb_write_binary(Fd,TermList),
    file:close(Fd).

get_term(B) ->
    <<S:8, B2/binary>> = B,
    <<T:S/binary, Rest/binary>> = B2,
    case binary_to_term(T) of
	{'$size',Sz} -> 
	    %% size of the actual term was bigger than 8 bits
	    <<T1:Sz/binary, Rest1/binary>> = Rest,
	    {binary_to_term(T1),Rest1};
	Term ->
	    {Term,Rest}
    end.

display_warning(Item,Warning) ->
    io:format("Warning: {~w,~w}~n",[Warning,Item]).


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% Trace client which reads an IP port and puts data directly to a file.
%%% This is used when tracing remote nodes with no file system.
ip_to_file(Trace,{file,File}) ->
    Fun = dbg:trace_port(file,File), %File can be a filename or a wrap spec
    Port = Fun(),
    ip_to_file(Trace,Port);
ip_to_file({metadata,MetaFile,MetaData},Port) ->
    {ok,MetaFd} = file:open(MetaFile,[write,raw,append]),
    file:write(MetaFd,MetaData),
    file:close(MetaFd),
    Port;
ip_to_file(Trace,Port) ->
    B = term_to_binary(Trace),
    erlang:port_command(Port,B),
    Port.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% For debugging
dump_ti(File) ->
    {ok,B} = file:read_file(File),
    dump_ti(B,[]).

dump_ti(<<>>,Acc) ->
    lists:reverse(Acc);
dump_ti(B,Acc) ->
    {Term,Rest} = get_term(B),
    dump_ti(Rest,[Term|Acc]).