xprof_core

%% -*- 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/1, 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, name, last_ts, hdr_ref,
                window_size, max_duration, ignore_recursion,
                capture_spec, capture_id=0, capture_counter=0}).

-define(ONE_SEC, 1000000). %% Second in microseconds
-define(WINDOW_SIZE, 10*60). %% 10 min window size
%% The largest duration value that can be stored in the HDR histogram in ms
-define(MAX_DURATION, 30*1000).

%% @doc Starts new process registered localy.
-spec start_link(xprof_core:mfa_spec()) -> {ok, pid()}.
start_link(MFASpec) ->
    Name = xprof_core_lib:mfaspec2atom(MFASpec),
    gen_server:start_link({local, Name}, ?MODULE, [MFASpec, Name], []).

%% @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 = {M,F,A}, Threshold, Limit) ->
    lager:info("Capturing ~p calls to ~w:~w/~w that exceed ~p ms:",
               [Limit, M, F, A, Threshold]),

    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([MFASpec, Name]) ->
    MaxDuration = application:get_env(xprof, max_duration, ?MAX_DURATION) * 1000,
    IgnoreRecursion = application:get_env(xprof, ignore_recursion, true),
    {ok, HDR} = init_storage(Name, MaxDuration),
    %% add trace pattern with args capturing turned off
    capture_args_trace_off(MFASpec),
    {ok, #state{mfa = MFASpec,
                hdr_ref = HDR,
                name = Name,
                last_ts = os:timestamp(),
                window_size = ?WINDOW_SIZE,
                max_duration = MaxDuration,
                ignore_recursion = IgnoreRecursion}, 1000}.

handle_call({capture, Threshold, Limit}, _From,
            State = #state{mfa = MFA}) ->
    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(MFA),
    {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 = MFA}) ->
    capture_args_trace_off(MFA),
    #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) ->
    lager:warning("Received unknown message: ~p", [Request]),
    {reply, ignored, State}.

handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info({trace_ts, Pid, call, _MFA, Args, StartTime}, State) ->
    put_ts_args(Pid, StartTime, Args, State#state.ignore_recursion),

    {Timeout, NewState} = maybe_make_snapshot(State),
    {noreply, NewState, Timeout};
handle_info({trace_ts, Pid, Tag, _MFA, RetOrExc, EndTime}, State)
  when Tag =:= return_from;
       Tag =:= exception_from ->

    NewState = case get_ts_args(Pid, State#state.ignore_recursion) of
                   undefined ->
                       State;
                   {StartTime, Args} ->
                       CallTime = timer:now_diff(EndTime, StartTime),
                       record_results(Pid, CallTime, Args, {Tag, RetOrExc}, State)
               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, MaxDuration) ->
    ets:new(Name, [public, named_table]),
    ets:insert(Name, {capture_spec, -1, -1, -1, -1}),
    hdr_histogram:open(MaxDuration * 1000, 3).

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 ->
            save_snapshot(NowTS, State),
            remove_outdated_snapshots(Name, xprof_core_lib:now2epoch(NowTS)-WindSize),
            {calc_next_timeout(DiffMicro), State#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{name=Name, hdr_ref=Ref}) ->
    Epoch = xprof_core_lib:now2epoch(NowTS),
    ets:insert(Name, [{{sec, Epoch}, get_current_hist_stats(Ref, Epoch)}]),
    hdr_histogram:reset(Ref).

get_current_hist_stats(HistRef, Time) ->
    [{time, Time},
     {min,      hdr_histogram:min(HistRef)},
     {mean,     hdr_histogram:mean(HistRef)},
     %%{median,   hdr_histogram:median(HistRef)},
     {max,      hdr_histogram:max(HistRef)},
     %%{stddev,   hdr_histogram:stddev(HistRef)},
     %%{p25,      hdr_histogram:percentile(HistRef,25.0)},
     {p50,      hdr_histogram:percentile(HistRef,50.0)},
     {p75,      hdr_histogram:percentile(HistRef,75.0)},
     {p90,      hdr_histogram:percentile(HistRef,90.0)},
     {p99,      hdr_histogram:percentile(HistRef,99.0)},
     %%{p9999999, hdr_histogram:percentile(HistRef,99.9999)},
     %%{memsize,  hdr_histogram:get_memory_size(HistRef)},
     {count,    hdr_histogram:get_total_count(HistRef)}].

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}).

put_ts_args(Pid, StartTime, Args, IgnoreRecursion) ->
    %% Count the depth of recursion in this process
    CD = case get({Pid, call_depth}) of
             undefined -> 1;
             CallDepth -> CallDepth + 1
         end,
    put({Pid, call_depth}, CD),
    case CD =:= 1 orelse not IgnoreRecursion of
        true ->
            put({Pid, CD, ts_args}, {StartTime, Args});
        false ->
            ok
    end.

%% @doc If IgnoreRecursion is true only return start time and args
%% of the outermost call
get_ts_args(Pid, IgnoreRecursion) ->
    case get({Pid, call_depth}) of
        undefined ->
            %% we missed the call of this function
            undefined;
        1 ->
            erase({Pid, call_depth}),
            erase({Pid, 1, ts_args});
        CD when CD > 1 ->
            put({Pid, call_depth}, CD - 1),
            case IgnoreRecursion of
                false ->
                    erase({Pid, CD, ts_args});
                true ->
                    undefined
            end
    end.


record_results(Pid, CallTime, Args, Res,
               State = #state{mfa = MFA,
                              name = Name,
                              hdr_ref = Ref,
                              max_duration = MaxDuration,
                              capture_spec = CaptureSpec,
                              capture_counter = Count}) ->
    if CallTime > MaxDuration ->
            lager:error("Call ~p took ~p ms that is larger than the maximum "
                        "that can be stored (~p ms)",
                        [Name, CallTime/1000, MaxDuration div 1000]),
            ok = hdr_histogram:record(Ref, MaxDuration);
       true ->
            ok = hdr_histogram:record(Ref, CallTime)
    end,

    case CaptureSpec of
        {Threshold, Limit}
          when CallTime > Threshold * 1000 andalso
               Count =< Limit andalso
               Args =/= arity ->
            ets:insert(Name, {{args_res, Count},
                              {Pid, CallTime, Args, Res}}),
            %% reached limit - turn off args tracing
            Count =:= Limit andalso
                capture_args_trace_off(MFA),
            State#state{capture_counter = Count + 1};
        _ ->
            State
    end.

-spec capture_args_trace_on(xprof_core:mfa_spec()) -> any().
capture_args_trace_on({MFAId, {_MSOff, MSOn}}) ->
    erlang:trace_pattern(MFAId, MSOn, [local]).

-spec capture_args_trace_off(xprof_core:mfa_spec()) -> any().
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]).