%%%------------------------------------------------------------------- %%% @author anlicheng %%% @copyright (C) 2024, %%% @doc %%% 资源协调器,用来充分利用网络资源 %%% @end %%% Created : 04. 6月 2024 10:55 %%%------------------------------------------------------------------- -module(sdlan_network_coordinator). -author("anlicheng"). -behaviour(gen_server). %% API -export([start_link/0]). -export([checkout/0, attach/2]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(SERVER, ?MODULE). -record(state, { release_count = 0, network_map = #{} %% {NetworkPid => ThrottleKey :: atom()} }). %%%=================================================================== %%% API %%%=================================================================== -spec attach(NetworkPid :: pid(), ThrottleKey :: atom()) -> no_return(). attach(NetworkPid, ThrottleKey) when is_pid(NetworkPid), is_atom(ThrottleKey) -> gen_server:cast(?SERVER, {attach, NetworkPid, ThrottleKey}). -spec checkout() -> ok | error. checkout() -> gen_server:call(?SERVER, checkout). %% @doc Spawns the server and registers the local name (unique) -spec(start_link() -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %% @private %% @doc Initializes the server -spec(init(Args :: term()) -> {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | {stop, Reason :: term()} | ignore). init([]) -> %% 让渡资源定时器 erlang:start_timer(100, self(), release_ticker), {ok, #state{release_count = 0}}. %% @private %% @doc Handling call messages -spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, State :: #state{}) -> {reply, Reply :: term(), NewState :: #state{}} | {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | {stop, Reason :: term(), NewState :: #state{}}). handle_call(checkout, _From, State = #state{release_count = Count}) -> case Count > 0 of true -> {reply, ok, State#state{release_count = Count - 1}}; false -> {reply, error, State} end. %% @private %% @doc Handling cast messages -spec(handle_cast(Request :: term(), State :: #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). handle_cast({attach, NetworkPid, ThrottleKey}, State = #state{network_map = NetworkMap}) -> monitor(process, NetworkPid), {noreply, State#state{network_map = maps:put(NetworkPid, ThrottleKey, NetworkMap)}}. %% @private %% @doc Handling all non call/cast messages -spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). handle_info({timeout, _, release_ticker}, State = #state{network_map = ChannelMap}) -> %% 让渡资源定时器 erlang:start_timer(100, self(), release_ticker), AccReleaseCount = lists:foldl(fun(ThrottleKey, Acc) -> case throttle:peek(sdlan_network, ThrottleKey) of {ok, RestCount, LeftToReset} -> {ok, NetworkBindWidth} = application:get_env(sdlan, network_bind_width), NeedCount = erlang:ceil(NetworkBindWidth / 1000 * LeftToReset), Acc + max(0, RestCount - NeedCount); {limit_exceeded, 0, _} -> Acc end end, 0, maps:keys(ChannelMap)), % logger:debug("[sdlan_network_coordinator] can release count is: ~p", [AccReleaseCount]), {noreply, State#state{release_count = AccReleaseCount}}; handle_info({'DOWN', _, process, NetworkPid, Reason}, State = #state{network_map = NetworkMap}) -> logger:debug("[sdlan_network_coordinator] network_pid close with reason: ~p", [Reason]), {noreply, State#state{network_map = maps:remove(NetworkPid, NetworkMap)}}. %% @private %% @doc This function is called by a gen_server when it is about to %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_server terminates %% with Reason. The return value is ignored. -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), State :: #state{}) -> term()). terminate(_Reason, _State = #state{}) -> ok. %% @private %% @doc Convert process state when code is changed -spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{}, Extra :: term()) -> {ok, NewState :: #state{}} | {error, Reason :: term()}). code_change(_OldVsn, State = #state{}, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%===================================================================