Current section
Files
Jump to
Current section
Files
src/xprof_core_trace_handler.erl
%% -*- erlang-indent-level: 4;indent-tabs-mode: nil -*-
%% ex: ts=4 sw=4 et
%% @doc Gen server that tracks all calls to a particular function. It
%% registers itself localy under a atom that consists of MFA and xprof_
%% prefix. The same name is used to create public ETS table that holds entries
%% with call time stats for every second.
-module(xprof_core_trace_handler).
-behaviour(gen_server).
-export([start_link/3,
data/2,
capture/3,
capture_stop/1,
get_captured_data/2
]).
-export([trace_mfa_off/1]).
%% gen_server callbacks
-export([init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
terminate/2,
code_change/3]).
-record(state, {mfa_spec,
name,
last_ts,
window_size,
capture_spec,
capture_id = 0,
capture_counter = 0,
cb_mod,
cb_state}).
-define(ONE_SEC, 1000000). %% Second in microseconds
-define(WINDOW_SIZE, 10*60). %% 10 min window size
%% @doc Starts new process registered localy.
-spec start_link(xprof_core:cmd(), xprof_core:options(), module())
-> {ok, pid()} | {error, string()}.
start_link(_Cmd, Options, CmdCB) ->
MFASpec = proplists:get_value(mfa, Options),
Name = xprof_core_lib:mfaspec2atom(MFASpec),
gen_server:start_link({local, Name}, ?MODULE, [Name, Options, MFASpec, CmdCB], []).
%% @doc Returns histogram data for seconds that occured after FromEpoch.
-spec data(xprof_core:mfa_id(), non_neg_integer()) -> [proplists:proplist()] |
{error, not_found}.
data(MFA, FromEpoch) ->
Name = xprof_core_lib:mfa2atom(MFA),
try
ets:select(Name, [{
{{sec, '$1'},'$2'},
[{'>','$1',FromEpoch}],
['$2']
}])
catch
error:badarg ->
{error, not_found}
end.
%% @doc Starts capturing args and results from function calls that lasted longer
%% than specified time threshold.
-spec capture(xprof_core:mfa_id(), non_neg_integer(), non_neg_integer()) ->
{ok, non_neg_integer()} | {error, not_found}.
capture(MFA, Threshold, Limit) ->
Name = xprof_core_lib:mfa2atom(MFA),
try
gen_server:call(Name, {capture, Threshold, Limit})
catch
exit:{noproc, _} ->
{error, not_found}
end.
-spec capture_stop(xprof_core:mfa_id()) -> ok | {error, not_found | not_captured}.
capture_stop(MFA) ->
Name = xprof_core_lib:mfa2atom(MFA),
try
gen_server:call(Name, capture_stop)
catch
exit:{noproc, _} ->
{error, not_found}
end.
%% @doc
-spec get_captured_data(xprof_core:mfa_id(), non_neg_integer()) ->
{ok,
{Index :: non_neg_integer(),
Threshold :: non_neg_integer(),
OrigLimit :: non_neg_integer(),
HasMore :: boolean()
}, [tuple()]} |
{error, not_found}.
get_captured_data(MFA, Offset) when Offset >= 0 ->
Name = xprof_core_lib:mfa2atom(MFA),
try
[{capture_spec, Index, Threshold, Limit, OrigLimit}] =
ets:lookup(Name, capture_spec),
case Index of
-1 ->
%% no capturing was done for this MFA yet,
%% no need to traverse the table
{ok, {Index, Threshold, OrigLimit, _HasMore = false}, _Items = []};
_ ->
MS = [{
{{args_res, '$1'},
{'$2', '$3','$4','$5'}},
[{'>','$1',Offset}],
[{{'$1', '$2', '$3', '$4', '$5'}}]
}],
Items = lists:sort(ets:select(Name, MS)),
HasMore = Offset + length(Items) < Limit,
{ok, {Index, Threshold, OrigLimit, HasMore}, Items}
end
catch error:badarg ->
{error, not_found}
end.
%% gen_server callbacks
init([Name, Options, MFASpec, CmdCB]) ->
{ok, CBState} = CmdCB:init(Options, MFASpec),
init_storage(Name),
%% add trace pattern with args capturing turned off
case capture_args_trace_off(MFASpec) of
0 ->
%% should not happen after ensure_mfa
{stop, {no_match_mfaspec, MFASpec}};
_ ->
{ok, #state{mfa_spec = MFASpec,
name = Name,
last_ts = os:timestamp(),
window_size = ?WINDOW_SIZE,
cb_mod = CmdCB,
cb_state = CBState
}, 1000}
end.
handle_call({capture, Threshold, Limit}, _From,
State = #state{mfa_spec = MFASpec}) ->
NewId = State#state.capture_id + 1,
NewState = State#state{capture_spec = {Threshold, Limit},
capture_id = NewId,
capture_counter = 1},
init_new_capture_in_ets(NewState),
capture_args_trace_on(MFASpec),
{Timeout, NewState2} = maybe_make_snapshot(NewState),
{reply, {ok, NewId}, NewState2, Timeout};
handle_call(capture_stop, _From, State = #state{capture_spec = undefined}) ->
{Timeout, NewState} = maybe_make_snapshot(State),
{reply, {error, not_captured}, NewState, Timeout};
handle_call(capture_stop, _From, State = #state{mfa_spec = MFASpec}) ->
capture_args_trace_off(MFASpec),
#state{capture_spec = {Threshold, Limit},
name = Name,
capture_id = Id} = State,
%% update limit to otherwise has_more will return invalid value
%% we take ets count to make it idempotent in case 2 instances send stop cmd
Count = ets:select_count(Name, [{{{args_res, '_'}, '_'}, [], [true]}]),
NewState = State#state{capture_spec = {Threshold, Count}},
ets:insert(Name, {capture_spec, Id, Threshold, Count, Limit}),
{Timeout, NewState2} = maybe_make_snapshot(NewState),
{reply, ok, NewState2, Timeout};
handle_call(_Request, _From, State) ->
{reply, ignored, State}.
handle_cast(_Msg, State) ->
{noreply, State}.
handle_info({trace_ts, _Pid, _Tag, _MFA, _Args, _StartTime} = Msg,
State = #state{cb_mod = CmdCB,
capture_spec = CaptureSpec,
capture_counter = Count}) ->
CaptureThreshold =
case CaptureSpec of
{Threshold, Limit} when Limit >= Count ->
Threshold;
_ ->
undefined
end,
NewState =
case CmdCB:handle_event(Msg, CaptureThreshold, State#state.cb_state) of
ok -> State;
{ok, NewCBState} ->
State#state{cb_state = NewCBState};
{capture, Item} ->
capture_item(Item, State);
{capture, Item, NewCBState} ->
capture_item(Item, State#state{cb_state = NewCBState})
end,
{Timeout, NewState2} = maybe_make_snapshot(NewState),
{noreply, NewState2, Timeout};
handle_info(timeout, State) ->
{Timeout, NewState} = maybe_make_snapshot(State),
{noreply, NewState, Timeout};
handle_info(_Info, State) ->
{noreply, State}.
terminate(_Reason, _State) ->
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%% Internal functions
init_storage(Name) ->
ets:new(Name, [public, named_table]),
ets:insert(Name, {capture_spec, -1, -1, -1, -1}).
maybe_make_snapshot(State = #state{name=Name, last_ts=LastTS,
window_size=WindSize}) ->
NowTS = os:timestamp(),
case timer:now_diff(NowTS, LastTS) of
DiffMicro when DiffMicro >= ?ONE_SEC ->
NewState = save_snapshot(NowTS, State),
remove_outdated_snapshots(Name, xprof_core_lib:now2epoch(NowTS)-WindSize),
xprof_core_lib:notify_event_handler({snapshot, NowTS}),
{calc_next_timeout(DiffMicro), NewState#state{last_ts = NowTS}};
DiffMicro ->
{calc_next_timeout(DiffMicro), State}
end.
calc_next_timeout(DiffMicro) ->
DiffMilli = DiffMicro div 1000,
1000 - (DiffMilli rem 1000).
save_snapshot(NowTS, State = #state{name = Name, cb_mod = CmdCB, cb_state = CBState}) ->
Epoch = xprof_core_lib:now2epoch(NowTS),
{Snapshot, NewState} =
case CmdCB:take_snapshot(CBState) of
{Snapshot0, NewCBState} ->
{Snapshot0, State#state{cb_state = NewCBState}};
Snapshot0 ->
{Snapshot0, State}
end,
ets:insert(Name, {{sec, Epoch}, [{time, Epoch}|Snapshot]}),
NewState.
remove_outdated_snapshots(Name, TS) ->
ets:select_delete(Name,
[{
{{sec, '$1'},'_'},
[{'<','$1',TS}],
[true]
}]).
init_new_capture_in_ets(State) ->
#state{name=Name, capture_id=Id,
capture_spec={Threshold, Limit}} = State,
ets:select_delete(Name,
[{
{{args_res, '_'},'_'},
[],
[true]
}]),
ets:insert(Name, {capture_spec, Id, Threshold, Limit, Limit}).
capture_item(Item,
State = #state{mfa_spec = MFASpec,
name = Name,
capture_spec = {_Threshold, Limit},
capture_counter = Count}) ->
ets:insert(Name, {{args_res, Count}, Item}),
%% reached limit - turn off args tracing
Count =:= Limit andalso
capture_args_trace_off(MFASpec),
State#state{capture_counter = Count + 1}.
-spec capture_args_trace_on(xprof_core:mfa_spec()) -> non_neg_integer().
capture_args_trace_on({MFAId, {_MSOff, MSOn}}) ->
erlang:trace_pattern(MFAId, MSOn, [local]).
-spec capture_args_trace_off(xprof_core:mfa_spec()) -> non_neg_integer().
capture_args_trace_off({MFAId, {MSOff, _MSOn}}) ->
erlang:trace_pattern(MFAId, MSOff, [local]).
-spec trace_mfa_off(xprof_core:mfa_id()) -> any().
trace_mfa_off({M, F, '_'}) ->
%% FIXME: this will turn off tracing also
%% for the same function with a given arity
erlang:trace_pattern({M, F, '_'}, false, [local]);
trace_mfa_off(MFA) ->
erlang:trace_pattern(MFA, false, [local]).