%%% -------------------------------------------------------
%%% @author huangyongxing@yeah.net
%%% @doc
%%% 在线资源监控管理器
%%% 主要功能：
%%%   在线人数统计、CPU schedulers动态调节
%%%   角色进程定期GC
%%%   数据库心跳、资源异常日志记录
%%% @end
%%% -------------------------------------------------------
-module(online_monitor).
-behaviour(gen_server).
-author("huangyongxing@yeah.net").
%% API
-export([
        start_link/0
        ,get_node_online_num/0
    ]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
%% only test
-export([
        check_scheduler_change/1
        ,gm_check_mem_gc/0
        ,gm_do_monitor/0
        ,gm_role_mem_gc/0
    ]).
-include("common.hrl").
-include("server.hrl").
-include("node.hrl").
-include("clusters.hrl").
% -include_lib("stdlib/include/ms_transform.hrl").

-define(SERVER, ?MODULE).

-define(CHECK_INTERVAL,   30).                  % 统计时间间隔30s
-define(WRITE_DB_INTERVAL, 60).                 % 写入数据库时间间隔60s

%% 执行GC的内存上限，超出则不执行，避免引发内存暴涨，被系统kill掉进程
-define(GC_MEM_LIMIT, 104857600).

-record(state, {
        timer = undefined
        ,schedulers_online = 0                  % 当前设置调度器数量
        ,schedulers = 0                         % 最大调度器数量
        ,online_before_change = 0               % 上次调整调度器数量时的在线数
        ,last_write_online = 0                  % 上次记录在线数到数据库的时间
        ,last_db_heartbeat = 0                  % 上次数据库心跳时间
        ,last_check_mem = 0                     % 上次检查内存时间
        ,last_role_gc = 0                       % 上次角色进程gc时间
        ,cluster_type = 0                       % 节点类型：跨服|游戏服
        ,is_master_node = false                 % 是否游戏服主节点
    }).

%%%===================================================================
%%% API
%%%===================================================================

%% 启动服务
start_link() ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

%% 检查调整调度器数量(测试命令，传递的可能是非真实在线数)
check_scheduler_change(OnlineNum) ->
    gen_server:cast(?SERVER, {check_scheduler_change, OnlineNum}).

%% -------------------------------------------------------------------
%% API - GM操作（DEBUG专用，不要调用）
%% -------------------------------------------------------------------

%% GM操作：执行monitor各种处理
gm_do_monitor() ->
    ?SERVER ! do_monitor.

%% GM操作：手动执行GC
gm_check_mem_gc() ->
    spawn(fun() -> do_check_mem_gc(3000000) end).

%% GM操作：手动执行角色进程内存GC
gm_role_mem_gc() ->
    spawn(fun() -> role_mem_gc() end).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

init(_) ->
    process_flag(trap_exit, true),
    Ref         = erlang:start_timer(?CHECK_INTERVAL * 1000, self(), <<>>),
    ClusterType = clusters_api:get_cluster_type(),
    Schedulers  = erlang:system_info(schedulers),
    IsMasterNode=
    case ClusterType of
        ?CLUSTER_NODE   ->
            OnlineNum = get_node_online_num(),
            SchedulersOnline = calc_schedulers_match(OnlineNum, Schedulers),
            erlang:system_flag(schedulers_online, SchedulersOnline),
            node_mgr:is_master_node();
        ?CLUSTER_CENTER ->
            erlang:system_flag(schedulers_online, erlang:min(8, Schedulers)),
            false
    end,
    State = #state{
        timer = Ref
        ,cluster_type = ClusterType
        ,schedulers_online = erlang:system_info(schedulers_online)
        ,schedulers = Schedulers
        ,is_master_node = IsMasterNode
    },
    {ok, State}.

handle_cast({check_scheduler_change, OnlineNum}, State = #state{
        cluster_type = ?CLUSTER_NODE
        ,schedulers_online = SchedulersOnline
        ,schedulers = Schedulers
        ,online_before_change = OnlineBeforeChange
    }) ->
    {NewSchedulersOnline, NewOnlineBeforeChange} =
    schedulers_change(OnlineNum, OnlineBeforeChange, SchedulersOnline, Schedulers),
    {noreply, State#state{schedulers_online = NewSchedulersOnline, online_before_change = NewOnlineBeforeChange}};
handle_cast({check_scheduler_change, _}, State) ->
    {noreply, State};

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

handle_call(_Request, _From, State) ->
    {noreply, State}.

handle_info({timeout, Ref, <<>>}, State = #state{timer = Ref}) ->
    NewTimer = erlang:start_timer(?CHECK_INTERVAL * 1000, self(), <<>>),
    NewState = State#state{timer = NewTimer},
    handle_info('do_monitor', NewState);

%% 执行在线统计更新与资源监控调节
handle_info('do_monitor', State = #state{
        last_write_online = LastWriteOnline
        ,last_db_heartbeat = LastDBHeartbeat
        ,last_check_mem = LastCheckMem
        ,last_role_gc = LastRoleGC
        ,cluster_type = ClusterType
        ,is_master_node = IsMasterNode
    }) ->
    {NewWriteOnline, NewDBHeartbeat, NewCheckMem, NewRoleGC} =
    check_do_monitor(ClusterType, IsMasterNode, LastWriteOnline, LastDBHeartbeat, LastCheckMem, LastRoleGC),
    NewState = State#state{
        last_write_online = NewWriteOnline
        ,last_db_heartbeat = NewDBHeartbeat
        ,last_check_mem = NewCheckMem
        ,last_role_gc = NewRoleGC
    },
    OnlineNum = get_node_online_num(),
    handle_cast({check_scheduler_change, OnlineNum}, NewState);

handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra)->
    {ok, State}.

%%%===================================================================
%%% Internal functions
%%%===================================================================

%% 检查时间间隔是否合法
check_interval(NowTime, _LastTime = 0, Interval) ->
    BootTime = game_env:get(boot_time),
    FixBootTime =
    case is_integer(BootTime) of
        true  -> BootTime;
        false -> 0
    end,
    (NowTime - FixBootTime) >= Interval;
check_interval(NowTime, LastTime, Interval) ->
    (NowTime - LastTime) >= Interval.

%% -------------------------------------------------------
%% 调度器数量检测修改
%% -------------------------------------------------------
schedulers_change(OnlineNum, OnlineBeforeChange, SchedulersOnline, Schedulers) ->
    NewSchedulersOnline = calc_schedulers_match(OnlineNum, Schedulers),
    if
        NewSchedulersOnline =:= SchedulersOnline ->
            {SchedulersOnline, OnlineBeforeChange};
        abs(OnlineNum - OnlineBeforeChange) >= 30 ->
            %% 变化超过30人在线，才处理，避免在线数在临界线上下波动时频繁切换调度器数量
            ?INFO("schedulers online change [~w => ~w], online change [~w => ~w], total [~w]",
                [SchedulersOnline, NewSchedulersOnline, OnlineBeforeChange, OnlineNum, Schedulers]
            ),
            erlang:system_flag(schedulers_online, NewSchedulersOnline),
            {NewSchedulersOnline, OnlineNum};
        true ->
            {SchedulersOnline, OnlineBeforeChange}
    end.

calc_schedulers_match(OnlineNum, Schedulers) ->
    erlang:min(
        if
            OnlineNum =<  300 -> 1;
            OnlineNum =<  600 -> 2;
            OnlineNum =< 1000 -> 4;
            OnlineNum =< 1300 -> 6;
            true              -> 8
        end,
        Schedulers
    ).

%% -------------------------------------------------------
%% 在线记录
%% -------------------------------------------------------

%% {NewWriteOnline, NewDBHeartbeat, NewCheckMem}
check_do_monitor(ClusterType, IsMasterNode, LastWriteOnline, LastDBHeartbeat, LastCheckMem, LastRoleGC) ->
    NowTime = utime:unixtime(),
    {NewDBHeartbeat, NeedDoQuery} = check_db_keepalive_query(NowTime, LastDBHeartbeat),
    {NewCheckMem, NeedGC} = check_mem_gc(NowTime, LastCheckMem),
    {NewRoleGC, NeedRoleGC} = check_role_gc(ClusterType, NowTime, LastRoleGC),
    {NewWriteOnline, NeedWriteDB} = check_write_online(ClusterType, NowTime, LastWriteOnline),
    proc_lib:spawn_link(
        fun() ->
                catch log_online(ClusterType, IsMasterNode, NeedWriteDB, NowTime),
                NeedDoQuery andalso db:keepalive_query(),
                NeedGC andalso do_check_mem_gc(30000000),
                NeedRoleGC andalso role_mem_gc()
        end),
    {NewWriteOnline, NewDBHeartbeat, NewCheckMem, NewRoleGC}.

check_write_online(?CLUSTER_NODE, NowTime, LastWriteOnline) ->
    case check_interval(NowTime, LastWriteOnline, 60) of
        true  -> {NowTime, true};
        false -> {LastWriteOnline, false}
    end;
check_write_online(_, _, LastWriteOnline) ->
    {LastWriteOnline, false}.

log_online(?CLUSTER_NODE, true, NeedWriteDB, NowTime) ->
    case NeedWriteDB of
        true ->
            {TrueOnline, HostingNum} = lib_php_api:get_online_num(),
            Sql = util:fmt_list(
                "INSERT INTO `log_online` (`total`,`online`,`timestamp`,`onhook`) VALUES ('~w','~w','~w','~w')",
                [TrueOnline, [TrueOnline], NowTime, HostingNum]
            ),
            catch db:execute(Sql);
        false -> ignore
    end;
log_online(_ClusterType, _IsMasterNode, _NeedWriteDB, _NowTime) ->
    ignore.

%% 获取本节点在线人数
get_node_online_num() ->
    case ets:info(?ETS_ONLINE, size) of
        undefined -> 0;
        Size -> Size
    end.

%% -------------------------------------------------------
%% 数据库心跳
%% -------------------------------------------------------

%% check_db_keepalive_query(NowTime， LastDBHeartbeat) -> {NewLastDBHeartTime, NeedDoQuery}.
check_db_keepalive_query(NowTime, LastDBHeartbeat) ->
    DbCheckInterval = 1800,
    case check_interval(NowTime, LastDBHeartbeat, DbCheckInterval) of
        true  ->    % 间隔超过1800秒，则执行心跳检查
            {NowTime, true};
        false ->
            {LastDBHeartbeat, false}
    end.

%% -------------------------------------------------------
%% 角色进程定期gc
%% -------------------------------------------------------

check_role_gc(?CLUSTER_NODE, NowTime, LastRoleGC) ->
    case check_interval(NowTime, LastRoleGC, 120) of
        true  -> {NowTime, true};
        false -> {LastRoleGC, false}
    end;
check_role_gc(_, _, LastRoleGC) ->
    {LastRoleGC, false}.

%% 角色进程自动gc
role_mem_gc() ->
    Pids = lib_player:get_online_pids(),
    [
        begin
                case is_pid(Pid) andalso is_process_alive(Pid) of
                    true  ->
                        {message_queue_len, MessageLen} = erlang:process_info(Pid, message_queue_len),
                        case MessageLen < 10 of
                            true  -> erlang:garbage_collect(Pid);
                            false -> ignore
                        end;
                    false ->
                        ignore
                end
        end || Pid <- Pids
    ],
    ok.

%% -------------------------------------------------------
%% 内存监测自动gc
%% -------------------------------------------------------

%% 检查内存消耗 -> {NewLastCheckTime, NeedGC}
%% 单进程内存上限30M
check_mem_gc(NowTime, LastCheckMem) ->
    case check_interval(NowTime, LastCheckMem, 1200) of
        true  ->    % 每1200秒检查一次
            {NowTime, true};
        false ->
            {LastCheckMem, false}
    end.

%% 检查内存偏高（异常可疑度高）的进程，记录日志便于分析，并尝试进行GC
do_check_mem_gc(MemLim) ->
    Processes = erlang:processes(),
    do_check_mem_gc(MemLim, Processes, 1).

do_check_mem_gc(MemLim, [Pid | Others], GCCnt) ->
    NewGCCnt =
    case is_pid(Pid) andalso is_process_alive(Pid)  of
        true ->
            case erlang:process_info(Pid, memory) of
                {memory, Mem} ->
                    case Mem > MemLim of
                        true ->
                            %% 写入日志记录
                            List = [
                                registered_name, current_function, initial_call,
                                status, message_queue_len,
                                total_heap_size, heap_size, stack_size, reductions
                            ],
                            ProcInfo = erlang:process_info(Pid, List),
                            InitCall = get_proc_initial_call(Pid),
                            ProcName =
                            case ProcInfo of
                                [{registered_name, []}      | _] -> proc_global_name(Pid);
                                [{registered_name, RegName} | _] -> RegName;
                                _ -> undefined
                            end,
                            Fmt =
                            case ProcInfo of
                                [_, _, _, _, {message_queue_len, MessageLen} | _] when MessageLen >= 10 ->
                                    NeedGC = false,
                                    %% 瞬时消息队列长度过长的进程不主动执行GC
                                    "message_queue_len too long, is not readly to execute GC (gc count:~w)~n"
                                    "pid:~w,proc init call:~w,mem:~w,ProcName:~w~n"
                                    "proc info:~p";
                                _ when InitCall =:= {mod_scene_agent, init, 1} ->
                                    NeedGC = false,
                                    %% 场景进程不主动执行GC，避免进程崩溃
                                    "This is scene process, is not readly to execute GC (gc count:~w)~n"
                                    "pid:~w,proc init call:~w,mem:~w,ProcName:~w~n"
                                    "proc info:~p";
                                [_, _, _, _Status = running | _] ->
                                    NeedGC = false,
                                    "This process running, is not readly to execute GC (gc count:~w)~n"
                                    "pid:~w,proc init call:~w,mem:~w,ProcName:~w~n"
                                    "proc info:~p";
                                _ when Mem >= ?GC_MEM_LIMIT ->
                                    %% 内存太大了，为防止ErlangVM被杀，权且放过，记录日志交由开发分析优化
                                    NeedGC = false,
                                    "This process cost too large memory, is not readly to execute GC (gc count:~w)~n"
                                    "pid:~w,proc init call:~w,mem:~w,ProcName:~w~n"
                                    "proc info:~p";
                                _ ->
                                    %% 其他进程尝试进行GC，减少内存占用
                                    NeedGC = true,
                                    %% 尤其老服可能存在明显的大量进程内存高，每处理N个做短暂延迟
                                    case GCCnt rem 10 == 0 of
                                        true  ->
                                            %% 实际上，下方garbage_collect使用的是阻塞式的，应该不需要做gc间隔延迟，
                                            %% 节点内存暴涨根源还在于进程内存占用不合理，
                                            %% 可能数据结构或算法、甚至于方案设计有问题，需要针对性做分析优化
                                            %% 但是基于历史原因也保留短暂的延迟，
                                            %% 或许一定程度上有助于调度器内存的管理，降低出问题概率
                                            timer:sleep(300);
                                        false -> ignore
                                    end,
                                    "process execute GC (gc count:~w)~n"
                                    "pid:~w,proc init call:~w,mem:~w,ProcName:~w~n"
                                    "proc info:~p"
                            end,
                            catch ?ERR(Fmt, [GCCnt, Pid, InitCall, Mem, ProcName, ProcInfo]),
                            case NeedGC of
                                true  ->
                                    erlang:garbage_collect(Pid),
                                    GCCnt + 1;
                                false ->
                                    GCCnt
                            end;
                        false ->
                            GCCnt
                    end;
                _ ->        % 进程已经不存在了
                    GCCnt
            end;
        false ->
            GCCnt
    end,
    do_check_mem_gc(MemLim, Others, NewGCCnt);
do_check_mem_gc(_MemLim, [], _GCCnt) ->
    ok.

%% 获取进程通过global模块注册的名称
proc_global_name(Pid) when is_pid(Pid) ->
    case ets:lookup(global_pid_names, Pid) of
        [{_, GlobalName}] ->
            {global, GlobalName};
        _ -> undefined
    end.

%% 计算标准otp behaviour进程的$initial_call函数
get_proc_initial_call(Pid) ->
    case erlang:process_info(Pid, dictionary) of
        {dictionary, Dictionary} ->
            case lists:keyfind('$initial_call', 1, Dictionary) of
                {_, MFA} -> MFA;
                _ -> undefined
            end;
        _ ->
            null
    end.

