%%% @doc Runs the fuse server in the system %%% @private -module(fuse_server). -behaviour(gen_server). -ifdef(PULSE). -include_lib("pulse_otp/include/pulse_otp.hrl"). -endif. %% Lifetime API -export([start_link/0]). %% Operational API -export([ ask/2, install/2, melt/1, remove/1, reset/1, run/3]). %% Callbacks -export([code_change/3, handle_call/3, handle_cast/2, handle_info/2, init/1, terminate/2]). %% Private -export([sync/0, q_melts/0]). -define(TAB, fuse_state). -record(state, { fuses = [] }). -record(fuse, { name :: atom(), intensity :: integer(), period :: integer(), heal_time :: integer(), melt_history = [], timer_ref = none }). -ifdef(EQC_TESTING). -define(TIME, fuse_time_mock). -else. -define(TIME, fuse_time). -endif. %% ------ %% @doc Start up the manager server for the fuse system %% This is assumed to be called by (@see fuse_sup). The `Timing' parameter controls how the system manages timing. %% @end start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). %% ------ %% @doc install/2 installs a new fuse into the running system %% Install a new fuse under `Name' with options given by `Opts'. %% We assume `Opts' are already in the right verified and validated format. %% @end install(Name, Opts) -> %% Assume options are already verified Fuse = init_state(Name, Opts), gen_server:call(?MODULE, {install, Fuse}). %% @doc ask/2 asks about the current given fuse state in a given context setting %% The documentation is (@see fuse:ask/1) %% @end -spec ask(atom(), fuse:fuse_context()) -> ok | blown | {error, not_found}. ask(Name, sync) -> gen_server:call(?MODULE, {ask, Name}); ask(Name, async_dirty) -> ask_(Name). %% ask_/1 is the real ask function. ask_(Name) -> StatsPlugin = application:get_env(fuse, stats_plugin, fuse_stats_ets), try ets:lookup_element(?TAB, Name, 2) of ok -> _ = StatsPlugin:increment(Name, ok), ok; blown -> _ = StatsPlugin:increment(Name, blown), blown catch error:badarg -> {error, not_found} end. %% @doc reset/1 resets the fuse %% The documentation is (@see fuse:reset/1) %% @end -spec reset(atom()) -> ok | {error, not_found}. reset(Name) -> gen_server:call(?MODULE, {reset, Name}). %% @doc melt/2 melts the fuse at a given point in time %% For documentation, (@see fuse:melt/2) %% @end -spec melt(Name) -> ok when Name :: atom(). melt(Name) -> gen_server:call(?MODULE, {melt, Name}). %% @doc remove/1 removes the fuse %% The documentation is (@see fuse:remove/1) %% @end -spec remove(atom()) -> ok | {error, not_found}. remove(Name) -> gen_server:call(?MODULE, {remove, Name}). %% sync/0 syncs the server. For internal use only in tests %% @private sync() -> gen_server:call(?MODULE, sync). q_melts() -> gen_server:call(?MODULE, q_melts). %% run/3 runs a thunk under a given fuse in a given context %% @doc Documentation is (@see fuse:run/3) %% @end %% @private -spec run(Name, fun(() -> {ok, Result} | {melt, Result}), fuse:fuse_context()) -> {ok, Result} | blown | {error, not_found} when Name :: atom(), Result :: any(). run(Name, Func, Context) -> case ask(Name, Context) of blown -> blown; ok -> case Func() of {ok, Result} -> {ok, Result}; {melt, Result} -> melt(Name), {ok, Result} end; {error, Reason} -> {error, Reason} end. %% @private init([]) -> _ = ets:new(?TAB, [named_table, protected, set, {read_concurrency, true}, {keypos, 1}]), {ok, #state{ }}. %% @private handle_call({install, #fuse { name = Name } = Fuse}, _From, #state { fuses = Fs } = State) -> case lists:keytake(Name, #fuse.name, Fs) of false -> install_metrics(Fuse), fix(Fuse); {value, OldFuse, _Otherfuses} -> fix(OldFuse), _ = reset_timer(OldFuse), %% For effect only ok end, {reply, ok, State#state { fuses = lists:keystore(Name, #fuse.name, Fs, Fuse)}}; handle_call({reset, Name}, _From, State) -> {Reply, State2} = handle_reset(Name, State, reset), {reply, Reply, State2}; handle_call({remove, Name}, _From, State) -> {Reply, State2} = handle_remove(Name, State), {reply, Reply, State2}; handle_call({melt, Name}, _From, State) -> Now = ?TIME:monotonic_time(), {Res, State2} = with_fuse(Name, State, fun(F) -> add_restart(Now, F) end), case Res of ok -> StatsPlugin = application:get_env(fuse, stats_plugin, fuse_stats_ets), _ = StatsPlugin:increment(Name, melt), {reply, ok, State2}; not_found -> {reply, ok, State2} end; handle_call({ask, Name}, _F, State) -> {reply, ask_(Name), State}; handle_call(sync, _F, State) -> {reply, ok, State}; handle_call(q_melts, _From, #state { fuses = Fs } = State) -> {reply, [{N, Ms} || #fuse { name = N, melt_history = Ms } <- Fs], State}; handle_call(_M, _F, State) -> {reply, {error, unknown}, State}. %% @private handle_cast(_M, State) -> {noreply, State}. %% @private handle_info({reset, Name}, State) -> {_Reply, State2} = handle_reset(Name, State, timeout), {noreply, State2}; handle_info(_M, State) -> {noreply, State}. %% @private terminate(_Reason, _State) -> ok. %% @private code_change(_OldVsn, State, _Extra) -> {ok, State}. %%% Internal functions %%% ------ handle_reset(Name, State, ResetType) -> Reset = fun(F) -> case ResetType of reset -> fix(F), NewF = reset_timer(F), {ok, NewF#fuse { melt_history = [] }}; timeout -> fix(F), {ok, F#fuse { melt_history = [], timer_ref = none }} end end, {Res, State2} = with_fuse(Name, State, Reset), case Res of ok -> {ok, State2}; not_found -> {{error, not_found}, State2} end. handle_remove(Name, #state { fuses = Fs } = State) -> case lists:keytake(Name, #fuse.name, Fs) of false -> {{error, not_found}, State}; {value, F, OtherFs} -> delete(F), {ok, State#state { fuses = OtherFs }} end. init_state(Name, {{standard, MaxR, MaxT}, {reset, Reset}}) -> NativePeriod = ?TIME:convert_time_unit(MaxT, milli_seconds, native), #fuse { name = Name, intensity = MaxR, period = NativePeriod, heal_time = Reset }. with_fuse(Name, #state { fuses = Fs} = State, Fun) -> case lists:keytake(Name, #fuse.name, Fs) of false -> {not_found, State}; {value, F, OtherFs} -> {R, FF} = Fun(F), {R, State#state { fuses = [FF | OtherFs] }} end. add_restart(Now, #fuse { intensity = I, period = Period, melt_history = R, heal_time = Heal, name = Name } = Fuse) -> R1 = add_restart_([Now | R], Now, Period), NewF = Fuse#fuse { melt_history = R1 }, case length(R1) of CurI when CurI =< I -> {ok, NewF}; _ -> blow(Fuse), TRef = ?TIME:send_after(Heal, self(), {reset, Name}), {ok, NewF#fuse { timer_ref = TRef }} end. add_restart_([R|Restarts], Now, Period) -> case in_period(R, Now, Period) of true -> [R|add_restart_(Restarts, Now, Period)]; false -> [] end; add_restart_([], _, _) -> []. in_period(Time, Now, Period) when (Now - Time) > Period -> false; in_period(_, _, _) -> true. blow(#fuse { name = Name }) -> ets:insert(?TAB, {Name, blown}), fuse_event:notify({Name, blown}), ok. fix(#fuse { name = Name }) -> ets:insert(?TAB, {Name, ok}), fuse_event:notify({Name, ok}), ok. delete(#fuse { name = Name }) -> ets:delete(?TAB, Name), fuse_event:notify({Name, removed}), ok. install_metrics(#fuse { name = N }) -> StatsPlugin = application:get_env(fuse, stats_plugin, fuse_stats_ets), _ = StatsPlugin:init(N), ok. reset_timer(#fuse { timer_ref = none } = F) -> F; reset_timer(#fuse { timer_ref = TRef } = F) -> _ = ?TIME:cancel_timer(TRef), F#fuse { timer_ref = none }.