%%%-------------------------------------------------------------------
%%% File    : gcpFlowControl.erl
%%% Author  : EAB/UPD/AV
%%% Description : Implements overload protection.
%%%-------------------------------------------------------------------
-module(gcpFlowControl).
-id('24/190 55-CNA 113 033 Ux').
-vsn('/main/R1A/14').
-date('2005-05-04').
-author('uabasve').
%%% ----------------------------------------------------------
%%% %CCaseTemplateFile: module.erl %
%%% %CCaseTemplateId: 16/002 01-FEA 202 714 Ux, Rev: /main/4 %
%%%
%%% Copyright (C) 2001-2005 by Ericsson Telecom AB
%%% SE-126 25  STOCKHOLM
%%% SWEDEN, tel int + 46 8 719 0000
%%%
%%% The program may be used and/or copied only with the written
%%% permission from Ericsson Telecom AB, or in accordance with
%%% the terms and conditions stipulated in the agreement/contract
%%% under which the program has been supplied.
%%%
%%% All rights reserved
%%%
%%%
%%% ----------------------------------------------------------
%%% #1.    REVISION LOG
%%% ----------------------------------------------------------
%%% Rev        Date       Name        What
%%% --------   --------   --------    ------------------------
%%% R1A/1-2    05-02-07   ejojmjn     Copied from EAS R7A/11.
%%% R1A/3-14   05-03-14   uabasve     Clean.
%%%--------------------------------------------------------------------

-include_lib("megaco/include/megaco.hrl").
-include_lib("megaco/include/megaco_message_v1.hrl").
-include("gcp.hrl").

-export([send_request/4,     %% user send from gcpInterface
         receive_reply/2,    %% from callback in gcpTransaction
         init_ets_tables/1,
         init_data/2]).

-define(PRIO_INFINITY, 16).
-define(MIN_WINDOW, 10).
-define(MAX_WINDOW, 100).

-define(BUCKET_MAX, 100).
-define(BUCKET_THRESH_HIGH, 80).
-define(BUCKET_THRESH_LOW, 20).

-define(ALLOW_TIMEOUT, 1000).

%% Holds counters for flow control in GCP
-record(gcpFlowControlTable,
	{key,
	 window        = 50,
	 available     = 50,
	 bucket        = 0,
	 q             = 0,
	 sent          = 0,   %% Counts all attempts
	 rejectable    = 0,   %% Counts rejectable attempts
	 t95,
	 errors        = 0,
	 rejects       = 0,
	 replies       = 0}).

-record(gcpFlowControlBitmap,
        {key,
         count = 0}).

%%====================================================================
%% External functions
%%====================================================================

%%--------------------------------------------------------------------
%% Function: send_request/4
%%
%% Output: ok | {error, Reason}
%%--------------------------------------------------------------------

send_request(ActiveLink, TimerOptions, ActionRequests, UserData) ->
    #gcpActiveLinkTable{key = Key,
                        conn_handle = ConnHandle}
        = ActiveLink,
    Prio = prio(ActionRequests),
    incr(Key, sent),
    case allow(Key, Prio) of
        {true, Timestamp} ->
            grant_request(user_data(ConnHandle),
                          Key,
                          Prio,
                          Timestamp,
                          ConnHandle,
                          TimerOptions,
                          ActionRequests,
                          UserData);
        false ->
            {error, rejected}
    end.

%%--------------------------------------------------------------------
%% Function: receive_reply/2
%% Description:
%%--------------------------------------------------------------------

receive_reply(Key, Timestamp) ->
    incr(Key, available),
    incr(Key, replies),
    release(Key),
    report_time(Key, Timestamp).

%%--------------------------------------------------------------------
%% Func: init_ets_tables/1
%%
%% Returns: ok
%%--------------------------------------------------------------------

init_ets_tables(Role) ->
    create_ets(Role, gcpFlowControlTable, #gcpFlowControlTable.key),
    create_ets(Role, gcpFlowControlBitmap, #gcpFlowControlBitmap.key),
    ok.

create_ets(Role, Table, Pos) when integer(Pos) ->
    create_ets(Role,
               Table,
               [named_table, ordered_set, public, {keypos, Pos}]);

create_ets(test, Table, ArgList) ->
    ets:new(Table, ArgList);
create_ets(Role, Table, ArgList) ->
    case ets:info(Table) of
        undefined ->
            sysCmd:ets_new(Table, ArgList);
        _ when Role == ch ->
            sysCmd:inherit_tables([Table]);
        _ when Role == om ->
            ok
    end.

%%--------------------------------------------------------------------
%% Func: init_data/2
%%--------------------------------------------------------------------

init_data(Key, T95) ->
    ets:insert(gcpFlowControlTable, #gcpFlowControlTable{key = Key,
                                                         t95 = T95}).

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

%%% ----------------------------------------------------------
%%%     incr
%%% ----------------------------------------------------------

cntr(Key, Field) ->
    incr(Key, Field, 0).

incr(Key, Field) ->
    incr(Key, Field, 1).

-define(INCR(Field),
        incr(Key, Field, X) -> upd_c(Key, {#gcpFlowControlTable.Field, X})).

?INCR(sent);
?INCR(replies);
?INCR(q);
?INCR(t95);
?INCR(errors);
?INCR(rejects);
?INCR(rejectable);
?INCR(window);
?INCR(available);

incr(Key, bucket, X)->
    upd_c(Key, {#gcpFlowControlTable.bucket, X, ?BUCKET_MAX, ?BUCKET_MAX}).

upd_c(Key, N) ->
    ets:update_counter(gcpFlowControlTable, Key, N).

%%% ----------------------------------------------------------
%%%     decr
%%%
%%%     Beware that decr is implemented as incr, care has to be taken
%%%     not to bungle things when max/min values are used.
%%% ----------------------------------------------------------

decr(Key, available, X) ->
    upd_c(Key, {#gcpFlowControlTable.available, -X});
decr(Key, window, X) ->
    upd_c(Key, {#gcpFlowControlTable.window, -X});
decr(Key, bucket, X) ->
    upd_c(Key, {#gcpFlowControlTable.bucket, -X, 0, 0}).

decr(Key, Field) ->
    decr(Key, Field, 1).

%%% ----------------------------------------------------------
%%%     allow
%%% ----------------------------------------------------------

allow(Key, ?PRIO_INFINITY) ->
    decr(Key, available),
    {true, now()};

allow(Key, Prio) ->
    incr(Key, rejectable),
    case decr(Key, available) of
        N when N > 0 ->
            {true, no_stamp};
        _ ->
            %% We did not send it, therefore incr available again
            incr(Key, available),
            queue(Key, Prio)
    end.

%%% ----------------------------------------------------------
%%%     queue
%%% ----------------------------------------------------------

queue(Key, Prio) ->
    incr(Key, q),
    T = {Key, Prio, now(), self()},
    ets:insert(gcpFlowControlBitmap, #gcpFlowControlBitmap{key = T}),
    wait(T).

%%% ----------------------------------------------------------
%%%     wait
%%% ----------------------------------------------------------

wait({Key, _Prio, _When, _Self} = T) ->
    receive
        allow ->
            ets:delete(gcpFlowControlBitmap, T),
            decr(Key, available),
            {true, no_stamp}
    after ?ALLOW_TIMEOUT ->
            timeout(T),
            adjust_window(Key),
            incr(Key, rejects),
            false
    end.

timeout(T) ->
    case ets:update_counter(gcpFlowControlBitmap, T, 1) of
        1 ->
            %% Got the lock: no one has released Key and sent 'allow'.
            ets:delete(gcpFlowControlBitmap, T),
            ok;
        _ ->
            %% A releasing process got the lock: 'allow' has been
            %% sent. Try to remove the message before proceeding.
            %% (This is to keep mdisp from complaining apparently.)
            ets:delete(gcpFlowControlBitmap, T),
            receive
                allow ->
                    ok
            after ?ALLOW_TIMEOUT ->
                    io:format("~p: errant allow: ~p~n", [?MODULE, T])
            end
    end.

%% Now, if we reject and our general response time is low
%% (i.e. low bucket) then we increase the window size.
adjust_window(Key) ->
    adjust_window(Key,
                  cntr(Key, bucket) < ?BUCKET_THRESH_LOW
                  andalso cntr(Key, window) < ?MAX_WINDOW).

adjust_window(Key, true) ->
    incr(Key, window),
    incr(Key, available),
    incr(Key, bucket, 20);
adjust_window(_, false) ->
    ok.

%%--------------------------------------------------------------------
%% Func: report_time/2
%%--------------------------------------------------------------------

report_time(_, no_stamp) ->
    ok;
report_time(Key, {MS, S, Ms})->
    {MegaSecs, Secs, MicroSecs} = now(),
    p(Key,
      MicroSecs - Ms + 1000000*(Secs - S + 1000000*(MegaSecs - MS)),
      cntr(Key, t95)).

%%% ----------------------------------------------------------
%%%     p
%%% ----------------------------------------------------------

p(Key, Time, T95) when Time =< T95 ->
    decr(Key, bucket);
p(Key, _Time, _T95) ->
    %% If we have a long response time, then increase the leaky
    %% bucket. If the bucket is over the high watermark and the window
    %% is not already at its minimum size, then decrease the window
    %% and available.
    case {cntr(Key, window), incr(Key, bucket, 20)} of
        {Window, Bucket} when Window > ?MIN_WINDOW,
                              Bucket > ?BUCKET_THRESH_HIGH ->
            decr(Key, window),
            decr(Key, available);
        _ ->
            ok
    end.

%%% ----------------------------------------------------------
%%%     release
%%% ----------------------------------------------------------

release(Key) ->
    %% The choice of the key below will cause ets:prev/2 to return
    %% the key with the highest priority which was queued most
    %% recently. This relies on the fact that integers sort before
    %% atoms, the atom 'prio' in this case. The atoms 'queued' and
    %% 'pid' are of no significance.
    release(Key, {Key, prio, queued, pid}).

%% This isn't a (FIFO) queue within each priority, but a (LIFO) stack.

release(Key, T) ->
    release(Key, cntr(Key, available), ets:prev(gcpFlowControlBitmap, T)).

%% Note that only keys on the same Key are matched.
release(Key, N, {Key, _Prio, _When, Pid} = T) when N > 0 ->
    case catch ets:update_counter(gcpFlowControlBitmap, T, 1) of
        1 ->
            Pid ! allow;
        _ ->
            %% Another process has released this key.
            release(Key, T)
    end;

release(_, _, _)->
    ok.

%%% ----------------------------------------------------------
%%%     user_data
%%% ----------------------------------------------------------

user_data(ConnHandle) ->
    case catch megaco:conn_info(ConnHandle, reply_data) of
        {'EXIT', _Reason} ->
            false;
        Rec ->
            {value, Rec}
    end.

%%% ----------------------------------------------------------
%%%     grant_request
%%% ----------------------------------------------------------

grant_request({value, Rec},
              Key, Prio, Time,
              ConnHandle, Options, ActionRequests, UserData) ->
    ReplyData = Rec#gcpReplyData{user_data = UserData,
                                 prio      = Prio,
                                 timestamp = Time},
    cast_rc(megaco:cast(ConnHandle,
                        ActionRequests,
                        [{reply_data, ReplyData} | Options]),
            Key,
            ActionRequests);

grant_request(false, Key, _, _, _, _, _, _) ->
    incr(Key, available),
    {error, reply_data}.

cast_rc(ok = Ok, _, _) ->
    Ok;
cast_rc({error, Reason}, Key, ActionRequests) ->
    incr(Key, available),
    gcpLib:error_report(?MODULE, send_request, [ActionRequests],
                        "send failed",
                        Reason),
    {error, {encode, Reason}}.

%%--------------------------------------------------------------------
%% Func: prio/1
%% Returns: The priority of the request
%%--------------------------------------------------------------------

prio([ActionRequest | _]) ->
    #'ActionRequest'{contextId = ContextId,
                     contextRequest = ContextRequest}
        = ActionRequest,
    prio(ContextId, ContextRequest).

prio(?megaco_choose_context_id, #'ContextRequest'{priority = Prio})
  when integer(Prio) ->
    Prio;
prio(_, _) ->
    ?PRIO_INFINITY.