From eb08b6bae01924bdbaeafee6b48caf9342edf8d9 Mon Sep 17 00:00:00 2001 From: Ery Lee Date: Sat, 6 Dec 2014 23:10:40 +0800 Subject: [PATCH] 0.2 --- apps/emqtt/src/emqtt_client_monitor.erl | 85 -- apps/emqtt/src/gen_server2.erl | 1234 ----------------------- apps/emqtt/src/priority_queue.erl | 194 ---- apps/emqtt/src/supervisor2.erl | 1232 ---------------------- 4 files changed, 2745 deletions(-) delete mode 100644 apps/emqtt/src/emqtt_client_monitor.erl delete mode 100644 apps/emqtt/src/gen_server2.erl delete mode 100644 apps/emqtt/src/priority_queue.erl delete mode 100644 apps/emqtt/src/supervisor2.erl diff --git a/apps/emqtt/src/emqtt_client_monitor.erl b/apps/emqtt/src/emqtt_client_monitor.erl deleted file mode 100644 index 193156a85..000000000 --- a/apps/emqtt/src/emqtt_client_monitor.erl +++ /dev/null @@ -1,85 +0,0 @@ -%% The contents of this file are subject to the Mozilla Public License -%% Version 1.1 (the "License"); you may not use this file except in -%% compliance with the License. You may obtain a copy of the License at -%% http://www.mozilla.org/MPL/ -%% -%% Software distributed under the License is distributed on an "AS IS" -%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the -%% License for the specific language governing rights and limitations -%% under the License. -%% -%% The Original Code is eMQTT -%% -%% The Initial Developer of the Original Code is -%% Copyright (C) 2012 Ery Lee All Rights Reserved. - --module(emqtt_client_monitor). - --include("emqtt.hrl"). - --include("emqtt_internal.hrl"). - --include_lib("elog/include/elog.hrl"). - --export([start_link/0, mon/1]). - --behaviour(gen_server). - --export([init/1, - handle_call/3, - handle_cast/2, - handle_info/2, - terminate/2, - code_change/3]). - --record(state, {}). - -mon(Client) when is_pid(Client) -> - gen_server2:cast(?MODULE, {monitor, Client}). - -%%---------------------------------------------------------------------------- - -start_link() -> - gen_server2:start_link({local, ?MODULE}, ?MODULE, [], []). - -init([]) -> - ets:new(clientmon, [set, protected, named_table]), - ets:new(clientmon_reverse, [set, protected, named_table]), - ?INFO("~p is started.", [?MODULE]), - {ok, #state{}}. - -handle_call(Req, _From, State) -> - {stop, {badreq, Req}, State}. - -handle_cast({monitor, Client}, State) -> - Ref = erlang:monitor(process, Client), - ets:insert(clientmon, {Client, Ref}), - ets:insert(clientmon_reverse, {Ref, Client}), - {noreply, State}; - -handle_cast(Msg, State) -> - {stop, {badmsg, Msg}, State}. - -handle_info({'DOWN', MRef, _Type, _Object, _Info}, State) -> - case ets:lookup(clientmon_reverse, MRef) of - [{_, Client}] -> - emqtt_router:down(Client), - ets:delete(clientmon, Client), - ets:delete(clientmon_reverse, MRef); - [] -> - ignore - end, - {noreply, State}; - -handle_info(Info, State) -> - {stop, {badinfo, Info},State}. - - -terminate(_Reason, _State) -> - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - - - diff --git a/apps/emqtt/src/gen_server2.erl b/apps/emqtt/src/gen_server2.erl deleted file mode 100644 index 78bbbe062..000000000 --- a/apps/emqtt/src/gen_server2.erl +++ /dev/null @@ -1,1234 +0,0 @@ -%% This file is a copy of gen_server.erl from the R13B-1 Erlang/OTP -%% distribution, with the following modifications: -%% -%% 1) the module name is gen_server2 -%% -%% 2) more efficient handling of selective receives in callbacks -%% gen_server2 processes drain their message queue into an internal -%% buffer before invoking any callback module functions. Messages are -%% dequeued from the buffer for processing. Thus the effective message -%% queue of a gen_server2 process is the concatenation of the internal -%% buffer and the real message queue. -%% As a result of the draining, any selective receive invoked inside a -%% callback is less likely to have to scan a large message queue. -%% -%% 3) gen_server2:cast is guaranteed to be order-preserving -%% The original code could reorder messages when communicating with a -%% process on a remote node that was not currently connected. -%% -%% 4) The callback module can optionally implement prioritise_call/3, -%% prioritise_cast/2 and prioritise_info/2. These functions take -%% Message, From and State or just Message and State and return a -%% single integer representing the priority attached to the message. -%% Messages with higher priorities are processed before requests with -%% lower priorities. The default priority is 0. -%% -%% 5) The callback module can optionally implement -%% handle_pre_hibernate/1 and handle_post_hibernate/1. These will be -%% called immediately prior to and post hibernation, respectively. If -%% handle_pre_hibernate returns {hibernate, NewState} then the process -%% will hibernate. If the module does not implement -%% handle_pre_hibernate/1 then the default action is to hibernate. -%% -%% 6) init can return a 4th arg, {backoff, InitialTimeout, -%% MinimumTimeout, DesiredHibernatePeriod} (all in milliseconds, -%% 'infinity' does not make sense here). Then, on all callbacks which -%% can return a timeout (including init), timeout can be -%% 'hibernate'. When this is the case, the current timeout value will -%% be used (initially, the InitialTimeout supplied from init). After -%% this timeout has occurred, hibernation will occur as normal. Upon -%% awaking, a new current timeout value will be calculated. -%% -%% The purpose is that the gen_server2 takes care of adjusting the -%% current timeout value such that the process will increase the -%% timeout value repeatedly if it is unable to sleep for the -%% DesiredHibernatePeriod. If it is able to sleep for the -%% DesiredHibernatePeriod it will decrease the current timeout down to -%% the MinimumTimeout, so that the process is put to sleep sooner (and -%% hopefully stays asleep for longer). In short, should a process -%% using this receive a burst of messages, it should not hibernate -%% between those messages, but as the messages become less frequent, -%% the process will not only hibernate, it will do so sooner after -%% each message. -%% -%% When using this backoff mechanism, normal timeout values (i.e. not -%% 'hibernate') can still be used, and if they are used then the -%% handle_info(timeout, State) will be called as normal. In this case, -%% returning 'hibernate' from handle_info(timeout, State) will not -%% hibernate the process immediately, as it would if backoff wasn't -%% being used. Instead it'll wait for the current timeout as described -%% above. -%% -%% 7) The callback module can return from any of the handle_* -%% functions, a {become, Module, State} triple, or a {become, Module, -%% State, Timeout} quadruple. This allows the gen_server to -%% dynamically change the callback module. The State is the new state -%% which will be passed into any of the callback functions in the new -%% module. Note there is no form also encompassing a reply, thus if -%% you wish to reply in handle_call/3 and change the callback module, -%% you need to use gen_server2:reply/2 to issue the reply manually. -%% -%% 8) The callback module can optionally implement -%% format_message_queue/2 which is the equivalent of format_status/2 -%% but where the second argument is specifically the priority_queue -%% which contains the prioritised message_queue. - -%% All modifications are (C) 2009-2012 VMware, Inc. - -%% ``The contents of this file are subject to the Erlang Public License, -%% Version 1.1, (the "License"); you may not use this file except in -%% compliance with the License. You should have received a copy of the -%% Erlang Public License along with this software. If not, it can be -%% retrieved via the world wide web at http://www.erlang.org/. -%% -%% Software distributed under the License is distributed on an "AS IS" -%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See -%% the License for the specific language governing rights and limitations -%% under the License. -%% -%% The Initial Developer of the Original Code is Ericsson Utvecklings AB. -%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings -%% AB. All Rights Reserved.'' -%% -%% $Id$ -%% --module(gen_server2). - -%%% --------------------------------------------------- -%%% -%%% The idea behind THIS server is that the user module -%%% provides (different) functions to handle different -%%% kind of inputs. -%%% If the Parent process terminates the Module:terminate/2 -%%% function is called. -%%% -%%% The user module should export: -%%% -%%% init(Args) -%%% ==> {ok, State} -%%% {ok, State, Timeout} -%%% {ok, State, Timeout, Backoff} -%%% ignore -%%% {stop, Reason} -%%% -%%% handle_call(Msg, {From, Tag}, State) -%%% -%%% ==> {reply, Reply, State} -%%% {reply, Reply, State, Timeout} -%%% {noreply, State} -%%% {noreply, State, Timeout} -%%% {stop, Reason, Reply, State} -%%% Reason = normal | shutdown | Term terminate(State) is called -%%% -%%% handle_cast(Msg, State) -%%% -%%% ==> {noreply, State} -%%% {noreply, State, Timeout} -%%% {stop, Reason, State} -%%% Reason = normal | shutdown | Term terminate(State) is called -%%% -%%% handle_info(Info, State) Info is e.g. {'EXIT', P, R}, {nodedown, N}, ... -%%% -%%% ==> {noreply, State} -%%% {noreply, State, Timeout} -%%% {stop, Reason, State} -%%% Reason = normal | shutdown | Term, terminate(State) is called -%%% -%%% terminate(Reason, State) Let the user module clean up -%%% Reason = normal | shutdown | {shutdown, Term} | Term -%%% always called when server terminates -%%% -%%% ==> ok | Term -%%% -%%% handle_pre_hibernate(State) -%%% -%%% ==> {hibernate, State} -%%% {stop, Reason, State} -%%% Reason = normal | shutdown | Term, terminate(State) is called -%%% -%%% handle_post_hibernate(State) -%%% -%%% ==> {noreply, State} -%%% {stop, Reason, State} -%%% Reason = normal | shutdown | Term, terminate(State) is called -%%% -%%% The work flow (of the server) can be described as follows: -%%% -%%% User module Generic -%%% ----------- ------- -%%% start -----> start -%%% init <----- . -%%% -%%% loop -%%% handle_call <----- . -%%% -----> reply -%%% -%%% handle_cast <----- . -%%% -%%% handle_info <----- . -%%% -%%% terminate <----- . -%%% -%%% -----> reply -%%% -%%% -%%% --------------------------------------------------- - -%% API --export([start/3, start/4, - start_link/3, start_link/4, - call/2, call/3, - cast/2, reply/2, - abcast/2, abcast/3, - multi_call/2, multi_call/3, multi_call/4, - enter_loop/3, enter_loop/4, enter_loop/5, enter_loop/6, wake_hib/1]). - -%% System exports --export([system_continue/3, - system_terminate/4, - system_code_change/4, - format_status/2]). - -%% Internal exports --export([init_it/6]). - --import(error_logger, [format/2]). - -%% State record --record(gs2_state, {parent, name, state, mod, time, - timeout_state, queue, debug, prioritise_call, - prioritise_cast, prioritise_info}). - --ifdef(use_specs). - -%%%========================================================================= -%%% Specs. These exist only to shut up dialyzer's warnings -%%%========================================================================= - --type(gs2_state() :: #gs2_state{}). - --spec(handle_common_termination/3 :: - (any(), atom(), gs2_state()) -> no_return()). --spec(hibernate/1 :: (gs2_state()) -> no_return()). --spec(pre_hibernate/1 :: (gs2_state()) -> no_return()). --spec(system_terminate/4 :: (_, _, _, gs2_state()) -> no_return()). - --type(millis() :: non_neg_integer()). - -%%%========================================================================= -%%% API -%%%========================================================================= - --callback init(Args :: term()) -> - {ok, State :: term()} | - {ok, State :: term(), timeout() | hibernate} | - {ok, State :: term(), timeout() | hibernate, - {backoff, millis(), millis(), millis()}} | - ignore | - {stop, Reason :: term()}. --callback handle_call(Request :: term(), From :: {pid(), Tag :: term()}, - State :: term()) -> - {reply, Reply :: term(), NewState :: term()} | - {reply, Reply :: term(), NewState :: term(), timeout() | hibernate} | - {noreply, NewState :: term()} | - {noreply, NewState :: term(), timeout() | hibernate} | - {stop, Reason :: term(), - Reply :: term(), NewState :: term()}. --callback handle_cast(Request :: term(), State :: term()) -> - {noreply, NewState :: term()} | - {noreply, NewState :: term(), timeout() | hibernate} | - {stop, Reason :: term(), NewState :: term()}. --callback handle_info(Info :: term(), State :: term()) -> - {noreply, NewState :: term()} | - {noreply, NewState :: term(), timeout() | hibernate} | - {stop, Reason :: term(), NewState :: term()}. --callback terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), - State :: term()) -> - ok | term(). --callback code_change(OldVsn :: (term() | {down, term()}), State :: term(), - Extra :: term()) -> - {ok, NewState :: term()} | {error, Reason :: term()}. - -%% It's not possible to define "optional" -callbacks, so putting specs -%% for handle_pre_hibernate/1 and handle_post_hibernate/1 will result -%% in warnings (the same applied for the behaviour_info before). - --else. - --export([behaviour_info/1]). - -behaviour_info(callbacks) -> - [{init,1},{handle_call,3},{handle_cast,2},{handle_info,2}, - {terminate,2},{code_change,3}]; -behaviour_info(_Other) -> - undefined. - --endif. - -%%% ----------------------------------------------------------------- -%%% Starts a generic server. -%%% start(Mod, Args, Options) -%%% start(Name, Mod, Args, Options) -%%% start_link(Mod, Args, Options) -%%% start_link(Name, Mod, Args, Options) where: -%%% Name ::= {local, atom()} | {global, atom()} -%%% Mod ::= atom(), callback module implementing the 'real' server -%%% Args ::= term(), init arguments (to Mod:init/1) -%%% Options ::= [{timeout, Timeout} | {debug, [Flag]}] -%%% Flag ::= trace | log | {logfile, File} | statistics | debug -%%% (debug == log && statistics) -%%% Returns: {ok, Pid} | -%%% {error, {already_started, Pid}} | -%%% {error, Reason} -%%% ----------------------------------------------------------------- -start(Mod, Args, Options) -> - gen:start(?MODULE, nolink, Mod, Args, Options). - -start(Name, Mod, Args, Options) -> - gen:start(?MODULE, nolink, Name, Mod, Args, Options). - -start_link(Mod, Args, Options) -> - gen:start(?MODULE, link, Mod, Args, Options). - -start_link(Name, Mod, Args, Options) -> - gen:start(?MODULE, link, Name, Mod, Args, Options). - - -%% ----------------------------------------------------------------- -%% Make a call to a generic server. -%% If the server is located at another node, that node will -%% be monitored. -%% If the client is trapping exits and is linked server termination -%% is handled here (? Shall we do that here (or rely on timeouts) ?). -%% ----------------------------------------------------------------- -call(Name, Request) -> - case catch gen:call(Name, '$gen_call', Request) of - {ok,Res} -> - Res; - {'EXIT',Reason} -> - exit({Reason, {?MODULE, call, [Name, Request]}}) - end. - -call(Name, Request, Timeout) -> - case catch gen:call(Name, '$gen_call', Request, Timeout) of - {ok,Res} -> - Res; - {'EXIT',Reason} -> - exit({Reason, {?MODULE, call, [Name, Request, Timeout]}}) - end. - -%% ----------------------------------------------------------------- -%% Make a cast to a generic server. -%% ----------------------------------------------------------------- -cast({global,Name}, Request) -> - catch global:send(Name, cast_msg(Request)), - ok; -cast({Name,Node}=Dest, Request) when is_atom(Name), is_atom(Node) -> - do_cast(Dest, Request); -cast(Dest, Request) when is_atom(Dest) -> - do_cast(Dest, Request); -cast(Dest, Request) when is_pid(Dest) -> - do_cast(Dest, Request). - -do_cast(Dest, Request) -> - do_send(Dest, cast_msg(Request)), - ok. - -cast_msg(Request) -> {'$gen_cast',Request}. - -%% ----------------------------------------------------------------- -%% Send a reply to the client. -%% ----------------------------------------------------------------- -reply({To, Tag}, Reply) -> - catch To ! {Tag, Reply}. - -%% ----------------------------------------------------------------- -%% Asyncronous broadcast, returns nothing, it's just send'n pray -%% ----------------------------------------------------------------- -abcast(Name, Request) when is_atom(Name) -> - do_abcast([node() | nodes()], Name, cast_msg(Request)). - -abcast(Nodes, Name, Request) when is_list(Nodes), is_atom(Name) -> - do_abcast(Nodes, Name, cast_msg(Request)). - -do_abcast([Node|Nodes], Name, Msg) when is_atom(Node) -> - do_send({Name,Node},Msg), - do_abcast(Nodes, Name, Msg); -do_abcast([], _,_) -> abcast. - -%%% ----------------------------------------------------------------- -%%% Make a call to servers at several nodes. -%%% Returns: {[Replies],[BadNodes]} -%%% A Timeout can be given -%%% -%%% A middleman process is used in case late answers arrives after -%%% the timeout. If they would be allowed to glog the callers message -%%% queue, it would probably become confused. Late answers will -%%% now arrive to the terminated middleman and so be discarded. -%%% ----------------------------------------------------------------- -multi_call(Name, Req) - when is_atom(Name) -> - do_multi_call([node() | nodes()], Name, Req, infinity). - -multi_call(Nodes, Name, Req) - when is_list(Nodes), is_atom(Name) -> - do_multi_call(Nodes, Name, Req, infinity). - -multi_call(Nodes, Name, Req, infinity) -> - do_multi_call(Nodes, Name, Req, infinity); -multi_call(Nodes, Name, Req, Timeout) - when is_list(Nodes), is_atom(Name), is_integer(Timeout), Timeout >= 0 -> - do_multi_call(Nodes, Name, Req, Timeout). - - -%%----------------------------------------------------------------- -%% enter_loop(Mod, Options, State, , , ) ->_ -%% -%% Description: Makes an existing process into a gen_server. -%% The calling process will enter the gen_server receive -%% loop and become a gen_server process. -%% The process *must* have been started using one of the -%% start functions in proc_lib, see proc_lib(3). -%% The user is responsible for any initialization of the -%% process, including registering a name for it. -%%----------------------------------------------------------------- -enter_loop(Mod, Options, State) -> - enter_loop(Mod, Options, State, self(), infinity, undefined). - -enter_loop(Mod, Options, State, Backoff = {backoff, _, _ , _}) -> - enter_loop(Mod, Options, State, self(), infinity, Backoff); - -enter_loop(Mod, Options, State, ServerName = {_, _}) -> - enter_loop(Mod, Options, State, ServerName, infinity, undefined); - -enter_loop(Mod, Options, State, Timeout) -> - enter_loop(Mod, Options, State, self(), Timeout, undefined). - -enter_loop(Mod, Options, State, ServerName, Backoff = {backoff, _, _, _}) -> - enter_loop(Mod, Options, State, ServerName, infinity, Backoff); - -enter_loop(Mod, Options, State, ServerName, Timeout) -> - enter_loop(Mod, Options, State, ServerName, Timeout, undefined). - -enter_loop(Mod, Options, State, ServerName, Timeout, Backoff) -> - Name = get_proc_name(ServerName), - Parent = get_parent(), - Debug = debug_options(Name, Options), - Queue = priority_queue:new(), - Backoff1 = extend_backoff(Backoff), - loop(find_prioritisers( - #gs2_state { parent = Parent, name = Name, state = State, - mod = Mod, time = Timeout, timeout_state = Backoff1, - queue = Queue, debug = Debug })). - -%%%======================================================================== -%%% Gen-callback functions -%%%======================================================================== - -%%% --------------------------------------------------- -%%% Initiate the new process. -%%% Register the name using the Rfunc function -%%% Calls the Mod:init/Args function. -%%% Finally an acknowledge is sent to Parent and the main -%%% loop is entered. -%%% --------------------------------------------------- -init_it(Starter, self, Name, Mod, Args, Options) -> - init_it(Starter, self(), Name, Mod, Args, Options); -init_it(Starter, Parent, Name0, Mod, Args, Options) -> - Name = name(Name0), - Debug = debug_options(Name, Options), - Queue = priority_queue:new(), - GS2State = find_prioritisers( - #gs2_state { parent = Parent, - name = Name, - mod = Mod, - queue = Queue, - debug = Debug }), - case catch Mod:init(Args) of - {ok, State} -> - proc_lib:init_ack(Starter, {ok, self()}), - loop(GS2State #gs2_state { state = State, - time = infinity, - timeout_state = undefined }); - {ok, State, Timeout} -> - proc_lib:init_ack(Starter, {ok, self()}), - loop(GS2State #gs2_state { state = State, - time = Timeout, - timeout_state = undefined }); - {ok, State, Timeout, Backoff = {backoff, _, _, _}} -> - Backoff1 = extend_backoff(Backoff), - proc_lib:init_ack(Starter, {ok, self()}), - loop(GS2State #gs2_state { state = State, - time = Timeout, - timeout_state = Backoff1 }); - {stop, Reason} -> - %% For consistency, we must make sure that the - %% registered name (if any) is unregistered before - %% the parent process is notified about the failure. - %% (Otherwise, the parent process could get - %% an 'already_started' error if it immediately - %% tried starting the process again.) - unregister_name(Name0), - proc_lib:init_ack(Starter, {error, Reason}), - exit(Reason); - ignore -> - unregister_name(Name0), - proc_lib:init_ack(Starter, ignore), - exit(normal); - {'EXIT', Reason} -> - unregister_name(Name0), - proc_lib:init_ack(Starter, {error, Reason}), - exit(Reason); - Else -> - Error = {bad_return_value, Else}, - proc_lib:init_ack(Starter, {error, Error}), - exit(Error) - end. - -name({local,Name}) -> Name; -name({global,Name}) -> Name; -%% name(Pid) when is_pid(Pid) -> Pid; -%% when R12 goes away, drop the line beneath and uncomment the line above -name(Name) -> Name. - -unregister_name({local,Name}) -> - _ = (catch unregister(Name)); -unregister_name({global,Name}) -> - _ = global:unregister_name(Name); -unregister_name(Pid) when is_pid(Pid) -> - Pid; -%% Under R12 let's just ignore it, as we have a single term as Name. -%% On R13 it will never get here, as we get tuple with 'local/global' atom. -unregister_name(_Name) -> ok. - -extend_backoff(undefined) -> - undefined; -extend_backoff({backoff, InitialTimeout, MinimumTimeout, DesiredHibPeriod}) -> - {backoff, InitialTimeout, MinimumTimeout, DesiredHibPeriod, now()}. - -%%%======================================================================== -%%% Internal functions -%%%======================================================================== -%%% --------------------------------------------------- -%%% The MAIN loop. -%%% --------------------------------------------------- -loop(GS2State = #gs2_state { time = hibernate, - timeout_state = undefined }) -> - pre_hibernate(GS2State); -loop(GS2State) -> - process_next_msg(drain(GS2State)). - -drain(GS2State) -> - receive - Input -> drain(in(Input, GS2State)) - after 0 -> GS2State - end. - -process_next_msg(GS2State = #gs2_state { time = Time, - timeout_state = TimeoutState, - queue = Queue }) -> - case priority_queue:out(Queue) of - {{value, Msg}, Queue1} -> - process_msg(Msg, GS2State #gs2_state { queue = Queue1 }); - {empty, Queue1} -> - {Time1, HibOnTimeout} - = case {Time, TimeoutState} of - {hibernate, {backoff, Current, _Min, _Desired, _RSt}} -> - {Current, true}; - {hibernate, _} -> - %% wake_hib/7 will set Time to hibernate. If - %% we were woken and didn't receive a msg - %% then we will get here and need a sensible - %% value for Time1, otherwise we crash. - %% R13B1 always waits infinitely when waking - %% from hibernation, so that's what we do - %% here too. - {infinity, false}; - _ -> {Time, false} - end, - receive - Input -> - %% Time could be 'hibernate' here, so *don't* call loop - process_next_msg( - drain(in(Input, GS2State #gs2_state { queue = Queue1 }))) - after Time1 -> - case HibOnTimeout of - true -> - pre_hibernate( - GS2State #gs2_state { queue = Queue1 }); - false -> - process_msg(timeout, - GS2State #gs2_state { queue = Queue1 }) - end - end - end. - -wake_hib(GS2State = #gs2_state { timeout_state = TS }) -> - TimeoutState1 = case TS of - undefined -> - undefined; - {SleptAt, TimeoutState} -> - adjust_timeout_state(SleptAt, now(), TimeoutState) - end, - post_hibernate( - drain(GS2State #gs2_state { timeout_state = TimeoutState1 })). - -hibernate(GS2State = #gs2_state { timeout_state = TimeoutState }) -> - TS = case TimeoutState of - undefined -> undefined; - {backoff, _, _, _, _} -> {now(), TimeoutState} - end, - proc_lib:hibernate(?MODULE, wake_hib, - [GS2State #gs2_state { timeout_state = TS }]). - -pre_hibernate(GS2State = #gs2_state { state = State, - mod = Mod }) -> - case erlang:function_exported(Mod, handle_pre_hibernate, 1) of - true -> - case catch Mod:handle_pre_hibernate(State) of - {hibernate, NState} -> - hibernate(GS2State #gs2_state { state = NState } ); - Reply -> - handle_common_termination(Reply, pre_hibernate, GS2State) - end; - false -> - hibernate(GS2State) - end. - -post_hibernate(GS2State = #gs2_state { state = State, - mod = Mod }) -> - case erlang:function_exported(Mod, handle_post_hibernate, 1) of - true -> - case catch Mod:handle_post_hibernate(State) of - {noreply, NState} -> - process_next_msg(GS2State #gs2_state { state = NState, - time = infinity }); - {noreply, NState, Time} -> - process_next_msg(GS2State #gs2_state { state = NState, - time = Time }); - Reply -> - handle_common_termination(Reply, post_hibernate, GS2State) - end; - false -> - %% use hibernate here, not infinity. This matches - %% R13B. The key is that we should be able to get through - %% to process_msg calling sys:handle_system_msg with Time - %% still set to hibernate, iff that msg is the very msg - %% that woke us up (or the first msg we receive after - %% waking up). - process_next_msg(GS2State #gs2_state { time = hibernate }) - end. - -adjust_timeout_state(SleptAt, AwokeAt, {backoff, CurrentTO, MinimumTO, - DesiredHibPeriod, RandomState}) -> - NapLengthMicros = timer:now_diff(AwokeAt, SleptAt), - CurrentMicros = CurrentTO * 1000, - MinimumMicros = MinimumTO * 1000, - DesiredHibMicros = DesiredHibPeriod * 1000, - GapBetweenMessagesMicros = NapLengthMicros + CurrentMicros, - Base = - %% If enough time has passed between the last two messages then we - %% should consider sleeping sooner. Otherwise stay awake longer. - case GapBetweenMessagesMicros > (MinimumMicros + DesiredHibMicros) of - true -> lists:max([MinimumTO, CurrentTO div 2]); - false -> CurrentTO - end, - {Extra, RandomState1} = random:uniform_s(Base, RandomState), - CurrentTO1 = Base + Extra, - {backoff, CurrentTO1, MinimumTO, DesiredHibPeriod, RandomState1}. - -in({'$gen_cast', Msg} = Input, - GS2State = #gs2_state { prioritise_cast = PC }) -> - in(Input, PC(Msg, GS2State), GS2State); -in({'$gen_call', From, Msg} = Input, - GS2State = #gs2_state { prioritise_call = PC }) -> - in(Input, PC(Msg, From, GS2State), GS2State); -in({'EXIT', Parent, _R} = Input, GS2State = #gs2_state { parent = Parent }) -> - in(Input, infinity, GS2State); -in({system, _From, _Req} = Input, GS2State) -> - in(Input, infinity, GS2State); -in(Input, GS2State = #gs2_state { prioritise_info = PI }) -> - in(Input, PI(Input, GS2State), GS2State). - -in(Input, Priority, GS2State = #gs2_state { queue = Queue }) -> - GS2State # gs2_state { queue = priority_queue:in(Input, Priority, Queue) }. - -process_msg({system, From, Req}, - GS2State = #gs2_state { parent = Parent, debug = Debug }) -> - %% gen_server puts Hib on the end as the 7th arg, but that version - %% of the fun seems not to be documented so leaving out for now. - sys:handle_system_msg(Req, From, Parent, ?MODULE, Debug, GS2State); -process_msg({'EXIT', Parent, Reason} = Msg, - GS2State = #gs2_state { parent = Parent }) -> - terminate(Reason, Msg, GS2State); -process_msg(Msg, GS2State = #gs2_state { debug = [] }) -> - handle_msg(Msg, GS2State); -process_msg(Msg, GS2State = #gs2_state { name = Name, debug = Debug }) -> - Debug1 = sys:handle_debug(Debug, fun print_event/3, Name, {in, Msg}), - handle_msg(Msg, GS2State #gs2_state { debug = Debug1 }). - -%%% --------------------------------------------------- -%%% Send/recive functions -%%% --------------------------------------------------- -do_send(Dest, Msg) -> - catch erlang:send(Dest, Msg). - -do_multi_call(Nodes, Name, Req, infinity) -> - Tag = make_ref(), - Monitors = send_nodes(Nodes, Name, Tag, Req), - rec_nodes(Tag, Monitors, Name, undefined); -do_multi_call(Nodes, Name, Req, Timeout) -> - Tag = make_ref(), - Caller = self(), - Receiver = - spawn( - fun () -> - %% Middleman process. Should be unsensitive to regular - %% exit signals. The sychronization is needed in case - %% the receiver would exit before the caller started - %% the monitor. - process_flag(trap_exit, true), - Mref = erlang:monitor(process, Caller), - receive - {Caller,Tag} -> - Monitors = send_nodes(Nodes, Name, Tag, Req), - TimerId = erlang:start_timer(Timeout, self(), ok), - Result = rec_nodes(Tag, Monitors, Name, TimerId), - exit({self(),Tag,Result}); - {'DOWN',Mref,_,_,_} -> - %% Caller died before sending us the go-ahead. - %% Give up silently. - exit(normal) - end - end), - Mref = erlang:monitor(process, Receiver), - Receiver ! {self(),Tag}, - receive - {'DOWN',Mref,_,_,{Receiver,Tag,Result}} -> - Result; - {'DOWN',Mref,_,_,Reason} -> - %% The middleman code failed. Or someone did - %% exit(_, kill) on the middleman process => Reason==killed - exit(Reason) - end. - -send_nodes(Nodes, Name, Tag, Req) -> - send_nodes(Nodes, Name, Tag, Req, []). - -send_nodes([Node|Tail], Name, Tag, Req, Monitors) - when is_atom(Node) -> - Monitor = start_monitor(Node, Name), - %% Handle non-existing names in rec_nodes. - catch {Name, Node} ! {'$gen_call', {self(), {Tag, Node}}, Req}, - send_nodes(Tail, Name, Tag, Req, [Monitor | Monitors]); -send_nodes([_Node|Tail], Name, Tag, Req, Monitors) -> - %% Skip non-atom Node - send_nodes(Tail, Name, Tag, Req, Monitors); -send_nodes([], _Name, _Tag, _Req, Monitors) -> - Monitors. - -%% Against old nodes: -%% If no reply has been delivered within 2 secs. (per node) check that -%% the server really exists and wait for ever for the answer. -%% -%% Against contemporary nodes: -%% Wait for reply, server 'DOWN', or timeout from TimerId. - -rec_nodes(Tag, Nodes, Name, TimerId) -> - rec_nodes(Tag, Nodes, Name, [], [], 2000, TimerId). - -rec_nodes(Tag, [{N,R}|Tail], Name, Badnodes, Replies, Time, TimerId ) -> - receive - {'DOWN', R, _, _, _} -> - rec_nodes(Tag, Tail, Name, [N|Badnodes], Replies, Time, TimerId); - {{Tag, N}, Reply} -> %% Tag is bound !!! - unmonitor(R), - rec_nodes(Tag, Tail, Name, Badnodes, - [{N,Reply}|Replies], Time, TimerId); - {timeout, TimerId, _} -> - unmonitor(R), - %% Collect all replies that already have arrived - rec_nodes_rest(Tag, Tail, Name, [N|Badnodes], Replies) - end; -rec_nodes(Tag, [N|Tail], Name, Badnodes, Replies, Time, TimerId) -> - %% R6 node - receive - {nodedown, N} -> - monitor_node(N, false), - rec_nodes(Tag, Tail, Name, [N|Badnodes], Replies, 2000, TimerId); - {{Tag, N}, Reply} -> %% Tag is bound !!! - receive {nodedown, N} -> ok after 0 -> ok end, - monitor_node(N, false), - rec_nodes(Tag, Tail, Name, Badnodes, - [{N,Reply}|Replies], 2000, TimerId); - {timeout, TimerId, _} -> - receive {nodedown, N} -> ok after 0 -> ok end, - monitor_node(N, false), - %% Collect all replies that already have arrived - rec_nodes_rest(Tag, Tail, Name, [N | Badnodes], Replies) - after Time -> - case rpc:call(N, erlang, whereis, [Name]) of - Pid when is_pid(Pid) -> % It exists try again. - rec_nodes(Tag, [N|Tail], Name, Badnodes, - Replies, infinity, TimerId); - _ -> % badnode - receive {nodedown, N} -> ok after 0 -> ok end, - monitor_node(N, false), - rec_nodes(Tag, Tail, Name, [N|Badnodes], - Replies, 2000, TimerId) - end - end; -rec_nodes(_, [], _, Badnodes, Replies, _, TimerId) -> - case catch erlang:cancel_timer(TimerId) of - false -> % It has already sent it's message - receive - {timeout, TimerId, _} -> ok - after 0 -> - ok - end; - _ -> % Timer was cancelled, or TimerId was 'undefined' - ok - end, - {Replies, Badnodes}. - -%% Collect all replies that already have arrived -rec_nodes_rest(Tag, [{N,R}|Tail], Name, Badnodes, Replies) -> - receive - {'DOWN', R, _, _, _} -> - rec_nodes_rest(Tag, Tail, Name, [N|Badnodes], Replies); - {{Tag, N}, Reply} -> %% Tag is bound !!! - unmonitor(R), - rec_nodes_rest(Tag, Tail, Name, Badnodes, [{N,Reply}|Replies]) - after 0 -> - unmonitor(R), - rec_nodes_rest(Tag, Tail, Name, [N|Badnodes], Replies) - end; -rec_nodes_rest(Tag, [N|Tail], Name, Badnodes, Replies) -> - %% R6 node - receive - {nodedown, N} -> - monitor_node(N, false), - rec_nodes_rest(Tag, Tail, Name, [N|Badnodes], Replies); - {{Tag, N}, Reply} -> %% Tag is bound !!! - receive {nodedown, N} -> ok after 0 -> ok end, - monitor_node(N, false), - rec_nodes_rest(Tag, Tail, Name, Badnodes, [{N,Reply}|Replies]) - after 0 -> - receive {nodedown, N} -> ok after 0 -> ok end, - monitor_node(N, false), - rec_nodes_rest(Tag, Tail, Name, [N|Badnodes], Replies) - end; -rec_nodes_rest(_Tag, [], _Name, Badnodes, Replies) -> - {Replies, Badnodes}. - - -%%% --------------------------------------------------- -%%% Monitor functions -%%% --------------------------------------------------- - -start_monitor(Node, Name) when is_atom(Node), is_atom(Name) -> - if node() =:= nonode@nohost, Node =/= nonode@nohost -> - Ref = make_ref(), - self() ! {'DOWN', Ref, process, {Name, Node}, noconnection}, - {Node, Ref}; - true -> - case catch erlang:monitor(process, {Name, Node}) of - {'EXIT', _} -> - %% Remote node is R6 - monitor_node(Node, true), - Node; - Ref when is_reference(Ref) -> - {Node, Ref} - end - end. - -%% Cancels a monitor started with Ref=erlang:monitor(_, _). -unmonitor(Ref) when is_reference(Ref) -> - erlang:demonitor(Ref), - receive - {'DOWN', Ref, _, _, _} -> - true - after 0 -> - true - end. - -%%% --------------------------------------------------- -%%% Message handling functions -%%% --------------------------------------------------- - -dispatch({'$gen_cast', Msg}, Mod, State) -> - Mod:handle_cast(Msg, State); -dispatch(Info, Mod, State) -> - Mod:handle_info(Info, State). - -common_reply(_Name, From, Reply, _NState, [] = _Debug) -> - reply(From, Reply), - []; -common_reply(Name, From, Reply, NState, Debug) -> - reply(Name, From, Reply, NState, Debug). - -common_debug([] = _Debug, _Func, _Info, _Event) -> - []; -common_debug(Debug, Func, Info, Event) -> - sys:handle_debug(Debug, Func, Info, Event). - -handle_msg({'$gen_call', From, Msg}, GS2State = #gs2_state { mod = Mod, - state = State, - name = Name, - debug = Debug }) -> - case catch Mod:handle_call(Msg, From, State) of - {reply, Reply, NState} -> - Debug1 = common_reply(Name, From, Reply, NState, Debug), - loop(GS2State #gs2_state { state = NState, - time = infinity, - debug = Debug1 }); - {reply, Reply, NState, Time1} -> - Debug1 = common_reply(Name, From, Reply, NState, Debug), - loop(GS2State #gs2_state { state = NState, - time = Time1, - debug = Debug1}); - {noreply, NState} -> - Debug1 = common_debug(Debug, fun print_event/3, Name, - {noreply, NState}), - loop(GS2State #gs2_state {state = NState, - time = infinity, - debug = Debug1}); - {noreply, NState, Time1} -> - Debug1 = common_debug(Debug, fun print_event/3, Name, - {noreply, NState}), - loop(GS2State #gs2_state {state = NState, - time = Time1, - debug = Debug1}); - {stop, Reason, Reply, NState} -> - {'EXIT', R} = - (catch terminate(Reason, Msg, - GS2State #gs2_state { state = NState })), - reply(Name, From, Reply, NState, Debug), - exit(R); - Other -> - handle_common_reply(Other, Msg, GS2State) - end; -handle_msg(Msg, GS2State = #gs2_state { mod = Mod, state = State }) -> - Reply = (catch dispatch(Msg, Mod, State)), - handle_common_reply(Reply, Msg, GS2State). - -handle_common_reply(Reply, Msg, GS2State = #gs2_state { name = Name, - debug = Debug}) -> - case Reply of - {noreply, NState} -> - Debug1 = common_debug(Debug, fun print_event/3, Name, - {noreply, NState}), - loop(GS2State #gs2_state { state = NState, - time = infinity, - debug = Debug1 }); - {noreply, NState, Time1} -> - Debug1 = common_debug(Debug, fun print_event/3, Name, - {noreply, NState}), - loop(GS2State #gs2_state { state = NState, - time = Time1, - debug = Debug1 }); - {become, Mod, NState} -> - Debug1 = common_debug(Debug, fun print_event/3, Name, - {become, Mod, NState}), - loop(find_prioritisers( - GS2State #gs2_state { mod = Mod, - state = NState, - time = infinity, - debug = Debug1 })); - {become, Mod, NState, Time1} -> - Debug1 = common_debug(Debug, fun print_event/3, Name, - {become, Mod, NState}), - loop(find_prioritisers( - GS2State #gs2_state { mod = Mod, - state = NState, - time = Time1, - debug = Debug1 })); - _ -> - handle_common_termination(Reply, Msg, GS2State) - end. - -handle_common_termination(Reply, Msg, GS2State) -> - case Reply of - {stop, Reason, NState} -> - terminate(Reason, Msg, GS2State #gs2_state { state = NState }); - {'EXIT', What} -> - terminate(What, Msg, GS2State); - _ -> - terminate({bad_return_value, Reply}, Msg, GS2State) - end. - -reply(Name, {To, Tag}, Reply, State, Debug) -> - reply({To, Tag}, Reply), - sys:handle_debug( - Debug, fun print_event/3, Name, {out, Reply, To, State}). - - -%%----------------------------------------------------------------- -%% Callback functions for system messages handling. -%%----------------------------------------------------------------- -system_continue(Parent, Debug, GS2State) -> - loop(GS2State #gs2_state { parent = Parent, debug = Debug }). - -system_terminate(Reason, _Parent, Debug, GS2State) -> - terminate(Reason, [], GS2State #gs2_state { debug = Debug }). - -system_code_change(GS2State = #gs2_state { mod = Mod, - state = State }, - _Module, OldVsn, Extra) -> - case catch Mod:code_change(OldVsn, State, Extra) of - {ok, NewState} -> - NewGS2State = find_prioritisers( - GS2State #gs2_state { state = NewState }), - {ok, [NewGS2State]}; - Else -> - Else - end. - -%%----------------------------------------------------------------- -%% Format debug messages. Print them as the call-back module sees -%% them, not as the real erlang messages. Use trace for that. -%%----------------------------------------------------------------- -print_event(Dev, {in, Msg}, Name) -> - case Msg of - {'$gen_call', {From, _Tag}, Call} -> - io:format(Dev, "*DBG* ~p got call ~p from ~w~n", - [Name, Call, From]); - {'$gen_cast', Cast} -> - io:format(Dev, "*DBG* ~p got cast ~p~n", - [Name, Cast]); - _ -> - io:format(Dev, "*DBG* ~p got ~p~n", [Name, Msg]) - end; -print_event(Dev, {out, Msg, To, State}, Name) -> - io:format(Dev, "*DBG* ~p sent ~p to ~w, new state ~w~n", - [Name, Msg, To, State]); -print_event(Dev, {noreply, State}, Name) -> - io:format(Dev, "*DBG* ~p new state ~w~n", [Name, State]); -print_event(Dev, Event, Name) -> - io:format(Dev, "*DBG* ~p dbg ~p~n", [Name, Event]). - - -%%% --------------------------------------------------- -%%% Terminate the server. -%%% --------------------------------------------------- - -terminate(Reason, Msg, #gs2_state { name = Name, - mod = Mod, - state = State, - debug = Debug }) -> - case catch Mod:terminate(Reason, State) of - {'EXIT', R} -> - error_info(R, Reason, Name, Msg, State, Debug), - exit(R); - _ -> - case Reason of - normal -> - exit(normal); - shutdown -> - exit(shutdown); - {shutdown,_}=Shutdown -> - exit(Shutdown); - _ -> - error_info(Reason, undefined, Name, Msg, State, Debug), - exit(Reason) - end - end. - -error_info(_Reason, _RootCause, application_controller, _Msg, _State, _Debug) -> - %% OTP-5811 Don't send an error report if it's the system process - %% application_controller which is terminating - let init take care - %% of it instead - ok; -error_info(Reason, RootCause, Name, Msg, State, Debug) -> - Reason1 = error_reason(Reason), - Fmt = - "** Generic server ~p terminating~n" - "** Last message in was ~p~n" - "** When Server state == ~p~n" - "** Reason for termination == ~n** ~p~n", - case RootCause of - undefined -> format(Fmt, [Name, Msg, State, Reason1]); - _ -> format(Fmt ++ "** In 'terminate' callback " - "with reason ==~n** ~p~n", - [Name, Msg, State, Reason1, - error_reason(RootCause)]) - end, - sys:print_log(Debug), - ok. - -error_reason({undef,[{M,F,A}|MFAs]} = Reason) -> - case code:is_loaded(M) of - false -> {'module could not be loaded',[{M,F,A}|MFAs]}; - _ -> case erlang:function_exported(M, F, length(A)) of - true -> Reason; - false -> {'function not exported',[{M,F,A}|MFAs]} - end - end; -error_reason(Reason) -> - Reason. - -%%% --------------------------------------------------- -%%% Misc. functions. -%%% --------------------------------------------------- - -opt(Op, [{Op, Value}|_]) -> - {ok, Value}; -opt(Op, [_|Options]) -> - opt(Op, Options); -opt(_, []) -> - false. - -debug_options(Name, Opts) -> - case opt(debug, Opts) of - {ok, Options} -> dbg_options(Name, Options); - _ -> dbg_options(Name, []) - end. - -dbg_options(Name, []) -> - Opts = - case init:get_argument(generic_debug) of - error -> - []; - _ -> - [log, statistics] - end, - dbg_opts(Name, Opts); -dbg_options(Name, Opts) -> - dbg_opts(Name, Opts). - -dbg_opts(Name, Opts) -> - case catch sys:debug_options(Opts) of - {'EXIT',_} -> - format("~p: ignoring erroneous debug options - ~p~n", - [Name, Opts]), - []; - Dbg -> - Dbg - end. - -get_proc_name(Pid) when is_pid(Pid) -> - Pid; -get_proc_name({local, Name}) -> - case process_info(self(), registered_name) of - {registered_name, Name} -> - Name; - {registered_name, _Name} -> - exit(process_not_registered); - [] -> - exit(process_not_registered) - end; -get_proc_name({global, Name}) -> - case whereis_name(Name) of - undefined -> - exit(process_not_registered_globally); - Pid when Pid =:= self() -> - Name; - _Pid -> - exit(process_not_registered_globally) - end. - -get_parent() -> - case get('$ancestors') of - [Parent | _] when is_pid(Parent)-> - Parent; - [Parent | _] when is_atom(Parent)-> - name_to_pid(Parent); - _ -> - exit(process_was_not_started_by_proc_lib) - end. - -name_to_pid(Name) -> - case whereis(Name) of - undefined -> - case whereis_name(Name) of - undefined -> - exit(could_not_find_registerd_name); - Pid -> - Pid - end; - Pid -> - Pid - end. - -whereis_name(Name) -> - case ets:lookup(global_names, Name) of - [{_Name, Pid, _Method, _RPid, _Ref}] -> - if node(Pid) == node() -> - case is_process_alive(Pid) of - true -> Pid; - false -> undefined - end; - true -> - Pid - end; - [] -> undefined - end. - -find_prioritisers(GS2State = #gs2_state { mod = Mod }) -> - PrioriCall = function_exported_or_default( - Mod, 'prioritise_call', 3, - fun (_Msg, _From, _State) -> 0 end), - PrioriCast = function_exported_or_default(Mod, 'prioritise_cast', 2, - fun (_Msg, _State) -> 0 end), - PrioriInfo = function_exported_or_default(Mod, 'prioritise_info', 2, - fun (_Msg, _State) -> 0 end), - GS2State #gs2_state { prioritise_call = PrioriCall, - prioritise_cast = PrioriCast, - prioritise_info = PrioriInfo }. - -function_exported_or_default(Mod, Fun, Arity, Default) -> - case erlang:function_exported(Mod, Fun, Arity) of - true -> case Arity of - 2 -> fun (Msg, GS2State = #gs2_state { state = State }) -> - case catch Mod:Fun(Msg, State) of - Res when is_integer(Res) -> - Res; - Err -> - handle_common_termination(Err, Msg, GS2State) - end - end; - 3 -> fun (Msg, From, GS2State = #gs2_state { state = State }) -> - case catch Mod:Fun(Msg, From, State) of - Res when is_integer(Res) -> - Res; - Err -> - handle_common_termination(Err, Msg, GS2State) - end - end - end; - false -> Default - end. - -%%----------------------------------------------------------------- -%% Status information -%%----------------------------------------------------------------- -format_status(Opt, StatusData) -> - [PDict, SysState, Parent, Debug, - #gs2_state{name = Name, state = State, mod = Mod, queue = Queue}] = - StatusData, - NameTag = if is_pid(Name) -> - pid_to_list(Name); - is_atom(Name) -> - Name - end, - Header = lists:concat(["Status for generic server ", NameTag]), - Log = sys:get_debug(log, Debug, []), - Specfic = callback(Mod, format_status, [Opt, [PDict, State]], - fun () -> [{data, [{"State", State}]}] end), - Messages = callback(Mod, format_message_queue, [Opt, Queue], - fun () -> priority_queue:to_list(Queue) end), - [{header, Header}, - {data, [{"Status", SysState}, - {"Parent", Parent}, - {"Logged events", Log}, - {"Queued messages", Messages}]} | - Specfic]. - -callback(Mod, FunName, Args, DefaultThunk) -> - case erlang:function_exported(Mod, FunName, length(Args)) of - true -> case catch apply(Mod, FunName, Args) of - {'EXIT', _} -> DefaultThunk(); - Success -> Success - end; - false -> DefaultThunk() - end. diff --git a/apps/emqtt/src/priority_queue.erl b/apps/emqtt/src/priority_queue.erl deleted file mode 100644 index 780fa2e92..000000000 --- a/apps/emqtt/src/priority_queue.erl +++ /dev/null @@ -1,194 +0,0 @@ -%% The contents of this file are subject to the Mozilla Public License -%% Version 1.1 (the "License"); you may not use this file except in -%% compliance with the License. You may obtain a copy of the License -%% at http://www.mozilla.org/MPL/ -%% -%% Software distributed under the License is distributed on an "AS IS" -%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See -%% the License for the specific language governing rights and -%% limitations under the License. -%% -%% The Original Code is RabbitMQ. -%% -%% The Initial Developer of the Original Code is VMware, Inc. -%% Copyright (c) 2007-2012 VMware, Inc. All rights reserved. -%% - -%% Priority queues have essentially the same interface as ordinary -%% queues, except that a) there is an in/3 that takes a priority, and -%% b) we have only implemented the core API we need. -%% -%% Priorities should be integers - the higher the value the higher the -%% priority - but we don't actually check that. -%% -%% in/2 inserts items with priority 0. -%% -%% We optimise the case where a priority queue is being used just like -%% an ordinary queue. When that is the case we represent the priority -%% queue as an ordinary queue. We could just call into the 'queue' -%% module for that, but for efficiency we implement the relevant -%% functions directly in here, thus saving on inter-module calls and -%% eliminating a level of boxing. -%% -%% When the queue contains items with non-zero priorities, it is -%% represented as a sorted kv list with the inverted Priority as the -%% key and an ordinary queue as the value. Here again we use our own -%% ordinary queue implemention for efficiency, often making recursive -%% calls into the same function knowing that ordinary queues represent -%% a base case. - - --module(priority_queue). - --export([new/0, is_queue/1, is_empty/1, len/1, to_list/1, in/2, in/3, - out/1, join/2]). - -%%---------------------------------------------------------------------------- - --ifdef(use_specs). - --export_type([q/0]). - --type(q() :: pqueue()). --type(priority() :: integer() | 'infinity'). --type(squeue() :: {queue, [any()], [any()]}). --type(pqueue() :: squeue() | {pqueue, [{priority(), squeue()}]}). - --spec(new/0 :: () -> pqueue()). --spec(is_queue/1 :: (any()) -> boolean()). --spec(is_empty/1 :: (pqueue()) -> boolean()). --spec(len/1 :: (pqueue()) -> non_neg_integer()). --spec(to_list/1 :: (pqueue()) -> [{priority(), any()}]). --spec(in/2 :: (any(), pqueue()) -> pqueue()). --spec(in/3 :: (any(), priority(), pqueue()) -> pqueue()). --spec(out/1 :: (pqueue()) -> {empty | {value, any()}, pqueue()}). --spec(join/2 :: (pqueue(), pqueue()) -> pqueue()). - --endif. - -%%---------------------------------------------------------------------------- - -new() -> - {queue, [], []}. - -is_queue({queue, R, F}) when is_list(R), is_list(F) -> - true; -is_queue({pqueue, Queues}) when is_list(Queues) -> - lists:all(fun ({infinity, Q}) -> is_queue(Q); - ({P, Q}) -> is_integer(P) andalso is_queue(Q) - end, Queues); -is_queue(_) -> - false. - -is_empty({queue, [], []}) -> - true; -is_empty(_) -> - false. - -len({queue, R, F}) when is_list(R), is_list(F) -> - length(R) + length(F); -len({pqueue, Queues}) -> - lists:sum([len(Q) || {_, Q} <- Queues]). - -to_list({queue, In, Out}) when is_list(In), is_list(Out) -> - [{0, V} || V <- Out ++ lists:reverse(In, [])]; -to_list({pqueue, Queues}) -> - [{maybe_negate_priority(P), V} || {P, Q} <- Queues, - {0, V} <- to_list(Q)]. - -in(Item, Q) -> - in(Item, 0, Q). - -in(X, 0, {queue, [_] = In, []}) -> - {queue, [X], In}; -in(X, 0, {queue, In, Out}) when is_list(In), is_list(Out) -> - {queue, [X|In], Out}; -in(X, Priority, _Q = {queue, [], []}) -> - in(X, Priority, {pqueue, []}); -in(X, Priority, Q = {queue, _, _}) -> - in(X, Priority, {pqueue, [{0, Q}]}); -in(X, Priority, {pqueue, Queues}) -> - P = maybe_negate_priority(Priority), - {pqueue, case lists:keysearch(P, 1, Queues) of - {value, {_, Q}} -> - lists:keyreplace(P, 1, Queues, {P, in(X, Q)}); - false when P == infinity -> - [{P, {queue, [X], []}} | Queues]; - false -> - case Queues of - [{infinity, InfQueue} | Queues1] -> - [{infinity, InfQueue} | - lists:keysort(1, [{P, {queue, [X], []}} | Queues1])]; - _ -> - lists:keysort(1, [{P, {queue, [X], []}} | Queues]) - end - end}. - -out({queue, [], []} = Q) -> - {empty, Q}; -out({queue, [V], []}) -> - {{value, V}, {queue, [], []}}; -out({queue, [Y|In], []}) -> - [V|Out] = lists:reverse(In, []), - {{value, V}, {queue, [Y], Out}}; -out({queue, In, [V]}) when is_list(In) -> - {{value,V}, r2f(In)}; -out({queue, In,[V|Out]}) when is_list(In) -> - {{value, V}, {queue, In, Out}}; -out({pqueue, [{P, Q} | Queues]}) -> - {R, Q1} = out(Q), - NewQ = case is_empty(Q1) of - true -> case Queues of - [] -> {queue, [], []}; - [{0, OnlyQ}] -> OnlyQ; - [_|_] -> {pqueue, Queues} - end; - false -> {pqueue, [{P, Q1} | Queues]} - end, - {R, NewQ}. - -join(A, {queue, [], []}) -> - A; -join({queue, [], []}, B) -> - B; -join({queue, AIn, AOut}, {queue, BIn, BOut}) -> - {queue, BIn, AOut ++ lists:reverse(AIn, BOut)}; -join(A = {queue, _, _}, {pqueue, BPQ}) -> - {Pre, Post} = - lists:splitwith(fun ({P, _}) -> P < 0 orelse P == infinity end, BPQ), - Post1 = case Post of - [] -> [ {0, A} ]; - [ {0, ZeroQueue} | Rest ] -> [ {0, join(A, ZeroQueue)} | Rest ]; - _ -> [ {0, A} | Post ] - end, - {pqueue, Pre ++ Post1}; -join({pqueue, APQ}, B = {queue, _, _}) -> - {Pre, Post} = - lists:splitwith(fun ({P, _}) -> P < 0 orelse P == infinity end, APQ), - Post1 = case Post of - [] -> [ {0, B} ]; - [ {0, ZeroQueue} | Rest ] -> [ {0, join(ZeroQueue, B)} | Rest ]; - _ -> [ {0, B} | Post ] - end, - {pqueue, Pre ++ Post1}; -join({pqueue, APQ}, {pqueue, BPQ}) -> - {pqueue, merge(APQ, BPQ, [])}. - -merge([], BPQ, Acc) -> - lists:reverse(Acc, BPQ); -merge(APQ, [], Acc) -> - lists:reverse(Acc, APQ); -merge([{P, A}|As], [{P, B}|Bs], Acc) -> - merge(As, Bs, [ {P, join(A, B)} | Acc ]); -merge([{PA, A}|As], Bs = [{PB, _}|_], Acc) when PA < PB orelse PA == infinity -> - merge(As, Bs, [ {PA, A} | Acc ]); -merge(As = [{_, _}|_], [{PB, B}|Bs], Acc) -> - merge(As, Bs, [ {PB, B} | Acc ]). - -r2f([]) -> {queue, [], []}; -r2f([_] = R) -> {queue, [], R}; -r2f([X,Y]) -> {queue, [X], [Y]}; -r2f([X,Y|R]) -> {queue, [X,Y], lists:reverse(R, [])}. - -maybe_negate_priority(infinity) -> infinity; -maybe_negate_priority(P) -> -P. diff --git a/apps/emqtt/src/supervisor2.erl b/apps/emqtt/src/supervisor2.erl deleted file mode 100644 index 5af38573f..000000000 --- a/apps/emqtt/src/supervisor2.erl +++ /dev/null @@ -1,1232 +0,0 @@ -%% This file is a copy of supervisor.erl from the R13B-3 Erlang/OTP -%% distribution, with the following modifications: -%% -%% 1) the module name is supervisor2 -%% -%% 2) there is a new strategy called -%% simple_one_for_one_terminate. This is exactly the same as for -%% simple_one_for_one, except that children *are* explicitly -%% terminated as per the shutdown component of the child_spec. -%% -%% 3) child specifications can contain, as the restart type, a tuple -%% {permanent, Delay} | {transient, Delay} | {intrinsic, Delay} -%% where Delay >= 0 (see point (4) below for intrinsic). The delay, -%% in seconds, indicates what should happen if a child, upon being -%% restarted, exceeds the MaxT and MaxR parameters. Thus, if a -%% child exits, it is restarted as normal. If it exits sufficiently -%% quickly and often to exceed the boundaries set by the MaxT and -%% MaxR parameters, and a Delay is specified, then rather than -%% stopping the supervisor, the supervisor instead continues and -%% tries to start up the child again, Delay seconds later. -%% -%% Note that you can never restart more frequently than the MaxT -%% and MaxR parameters allow: i.e. you must wait until *both* the -%% Delay has passed *and* the MaxT and MaxR parameters allow the -%% child to be restarted. -%% -%% Also note that the Delay is a *minimum*. There is no guarantee -%% that the child will be restarted within that time, especially if -%% other processes are dying and being restarted at the same time - -%% essentially we have to wait for the delay to have passed and for -%% the MaxT and MaxR parameters to permit the child to be -%% restarted. This may require waiting for longer than Delay. -%% -%% Sometimes, you may wish for a transient or intrinsic child to -%% exit abnormally so that it gets restarted, but still log -%% nothing. gen_server will log any exit reason other than -%% 'normal', 'shutdown' or {'shutdown', _}. Thus the exit reason of -%% {'shutdown', 'restart'} is interpreted to mean you wish the -%% child to be restarted according to the delay parameters, but -%% gen_server will not log the error. Thus from gen_server's -%% perspective it's a normal exit, whilst from supervisor's -%% perspective, it's an abnormal exit. -%% -%% 4) Added an 'intrinsic' restart type. Like the transient type, this -%% type means the child should only be restarted if the child exits -%% abnormally. Unlike the transient type, if the child exits -%% normally, the supervisor itself also exits normally. If the -%% child is a supervisor and it exits normally (i.e. with reason of -%% 'shutdown') then the child's parent also exits normally. -%% -%% 5) normal, and {shutdown, _} exit reasons are all treated the same -%% (i.e. are regarded as normal exits) -%% -%% All modifications are (C) 2010-2012 VMware, Inc. -%% -%% %CopyrightBegin% -%% -%% Copyright Ericsson AB 1996-2009. All Rights Reserved. -%% -%% The contents of this file are subject to the Erlang Public License, -%% Version 1.1, (the "License"); you may not use this file except in -%% compliance with the License. You should have received a copy of the -%% Erlang Public License along with this software. If not, it can be -%% retrieved online at http://www.erlang.org/. -%% -%% Software distributed under the License is distributed on an "AS IS" -%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See -%% the License for the specific language governing rights and limitations -%% under the License. -%% -%% %CopyrightEnd% -%% --module(supervisor2). - --behaviour(gen_server). - -%% External exports --export([start_link/2,start_link/3, - start_child/2, restart_child/2, - delete_child/2, terminate_child/2, - which_children/1, find_child/2, - check_childspecs/1]). - -%% Internal exports --export([init/1, handle_call/3, handle_info/2, terminate/2, code_change/3]). --export([handle_cast/2]). - --define(DICT, dict). - --record(state, {name, - strategy, - children = [], - dynamics = ?DICT:new(), - intensity, - period, - restarts = [], - module, - args}). - --record(child, {pid = undefined, % pid is undefined when child is not running - name, - mfa, - restart_type, - shutdown, - child_type, - modules = []}). - --define(is_simple(State), State#state.strategy =:= simple_one_for_one orelse - State#state.strategy =:= simple_one_for_one_terminate). --define(is_terminate_simple(State), - State#state.strategy =:= simple_one_for_one_terminate). - --ifdef(use_specs). - -%%-------------------------------------------------------------------------- -%% Types -%%-------------------------------------------------------------------------- - --export_type([child_spec/0, startchild_ret/0, strategy/0, sup_name/0]). - --type child() :: 'undefined' | pid(). --type child_id() :: term(). --type mfargs() :: {M :: module(), F :: atom(), A :: [term()] | undefined}. --type modules() :: [module()] | 'dynamic'. --type delay() :: non_neg_integer(). --type restart() :: 'permanent' | 'transient' | 'temporary' | 'intrinsic' - | {'permanent', delay()} | {'transient', delay()} - | {'intrinsic', delay()}. --type shutdown() :: 'brutal_kill' | timeout(). --type worker() :: 'worker' | 'supervisor'. --type sup_name() :: {'local', Name :: atom()} | {'global', Name :: atom()}. --type sup_ref() :: (Name :: atom()) - | {Name :: atom(), Node :: node()} - | {'global', Name :: atom()} - | pid(). --type child_spec() :: {Id :: child_id(), - StartFunc :: mfargs(), - Restart :: restart(), - Shutdown :: shutdown(), - Type :: worker(), - Modules :: modules()}. - - --type strategy() :: 'one_for_all' | 'one_for_one' - | 'rest_for_one' | 'simple_one_for_one' - | 'simple_one_for_one_terminate'. - --type child_rec() :: #child{pid :: child() | {restarting,pid()} | [pid()], - name :: child_id(), - mfa :: mfargs(), - restart_type :: restart(), - shutdown :: shutdown(), - child_type :: worker(), - modules :: modules()}. - --type state() :: #state{strategy :: strategy(), - children :: [child_rec()], - dynamics :: ?DICT(), - intensity :: non_neg_integer(), - period :: pos_integer()}. - -%%-------------------------------------------------------------------------- -%% Callback behaviour -%%-------------------------------------------------------------------------- - --callback init(Args :: term()) -> - {ok, {{RestartStrategy :: strategy(), - MaxR :: non_neg_integer(), - MaxT :: non_neg_integer()}, - [ChildSpec :: child_spec()]}} - | ignore. - -%%-------------------------------------------------------------------------- -%% Specs -%%-------------------------------------------------------------------------- - --type startchild_err() :: 'already_present' - | {'already_started', Child :: child()} | term(). --type startchild_ret() :: {'ok', Child :: child()} - | {'ok', Child :: child(), Info :: term()} - | {'error', startchild_err()}. - --spec start_child(SupRef, ChildSpec) -> startchild_ret() when - SupRef :: sup_ref(), - ChildSpec :: child_spec() | (List :: [term()]). - --spec restart_child(SupRef, Id) -> Result when - SupRef :: sup_ref(), - Id :: child_id(), - Result :: {'ok', Child :: child()} - | {'ok', Child :: child(), Info :: term()} - | {'error', Error}, - Error :: 'running' | 'not_found' | 'simple_one_for_one' | term(). - --spec delete_child(SupRef, Id) -> Result when - SupRef :: sup_ref(), - Id :: child_id(), - Result :: 'ok' | {'error', Error}, - Error :: 'running' | 'not_found' | 'simple_one_for_one'. - --spec terminate_child(SupRef, Id) -> Result when - SupRef :: sup_ref(), - Id :: pid() | child_id(), - Result :: 'ok' | {'error', Error}, - Error :: 'not_found' | 'simple_one_for_one'. - --spec which_children(SupRef) -> [{Id,Child,Type,Modules}] when - SupRef :: sup_ref(), - Id :: child_id() | 'undefined', - Child :: child(), - Type :: worker(), - Modules :: modules(). - --spec check_childspecs(ChildSpecs) -> Result when - ChildSpecs :: [child_spec()], - Result :: 'ok' | {'error', Error :: term()}. - --type init_sup_name() :: sup_name() | 'self'. - --type stop_rsn() :: 'shutdown' | {'bad_return', {module(),'init', term()}} - | {'bad_start_spec', term()} | {'start_spec', term()} - | {'supervisor_data', term()}. - --spec init({init_sup_name(), module(), [term()]}) -> - {'ok', state()} | 'ignore' | {'stop', stop_rsn()}. - --type call() :: 'which_children'. --spec handle_call(call(), term(), state()) -> {'reply', term(), state()}. - --spec handle_cast('null', state()) -> {'noreply', state()}. - --spec handle_info(term(), state()) -> - {'noreply', state()} | {'stop', 'shutdown', state()}. - --spec terminate(term(), state()) -> 'ok'. - --spec code_change(term(), state(), term()) -> - {'ok', state()} | {'error', term()}. - --else. - --export([behaviour_info/1]). - -behaviour_info(callbacks) -> - [{init,1}]; -behaviour_info(_Other) -> - undefined. - --endif. - -%%% --------------------------------------------------- -%%% This is a general process supervisor built upon gen_server.erl. -%%% Servers/processes should/could also be built using gen_server.erl. -%%% SupName = {local, atom()} | {global, atom()}. -%%% --------------------------------------------------- -start_link(Mod, Args) -> - gen_server:start_link(?MODULE, {self, Mod, Args}, []). - -start_link(SupName, Mod, Args) -> - gen_server:start_link(SupName, ?MODULE, {SupName, Mod, Args}, []). - -%%% --------------------------------------------------- -%%% Interface functions. -%%% --------------------------------------------------- -start_child(Supervisor, ChildSpec) -> - call(Supervisor, {start_child, ChildSpec}). - -restart_child(Supervisor, Name) -> - call(Supervisor, {restart_child, Name}). - -delete_child(Supervisor, Name) -> - call(Supervisor, {delete_child, Name}). - -%%----------------------------------------------------------------- -%% Func: terminate_child/2 -%% Returns: ok | {error, Reason} -%% Note that the child is *always* terminated in some -%% way (maybe killed). -%%----------------------------------------------------------------- -terminate_child(Supervisor, Name) -> - call(Supervisor, {terminate_child, Name}). - -which_children(Supervisor) -> - call(Supervisor, which_children). - -find_child(Supervisor, Name) -> - [Pid || {Name1, Pid, _Type, _Modules} <- which_children(Supervisor), - Name1 =:= Name]. - -call(Supervisor, Req) -> - gen_server:call(Supervisor, Req, infinity). - -check_childspecs(ChildSpecs) when is_list(ChildSpecs) -> - case check_startspec(ChildSpecs) of - {ok, _} -> ok; - Error -> {error, Error} - end; -check_childspecs(X) -> {error, {badarg, X}}. - -%%% --------------------------------------------------- -%%% -%%% Initialize the supervisor. -%%% -%%% --------------------------------------------------- -init({SupName, Mod, Args}) -> - process_flag(trap_exit, true), - case Mod:init(Args) of - {ok, {SupFlags, StartSpec}} -> - case init_state(SupName, SupFlags, Mod, Args) of - {ok, State} when ?is_simple(State) -> - init_dynamic(State, StartSpec); - {ok, State} -> - init_children(State, StartSpec); - Error -> - {stop, {supervisor_data, Error}} - end; - ignore -> - ignore; - Error -> - {stop, {bad_return, {Mod, init, Error}}} - end. - -init_children(State, StartSpec) -> - SupName = State#state.name, - case check_startspec(StartSpec) of - {ok, Children} -> - case start_children(Children, SupName) of - {ok, NChildren} -> - {ok, State#state{children = NChildren}}; - {error, NChildren} -> - terminate_children(NChildren, SupName), - {stop, shutdown} - end; - Error -> - {stop, {start_spec, Error}} - end. - -init_dynamic(State, [StartSpec]) -> - case check_startspec([StartSpec]) of - {ok, Children} -> - {ok, State#state{children = Children}}; - Error -> - {stop, {start_spec, Error}} - end; -init_dynamic(_State, StartSpec) -> - {stop, {bad_start_spec, StartSpec}}. - -%%----------------------------------------------------------------- -%% Func: start_children/2 -%% Args: Children = [#child] in start order -%% SupName = {local, atom()} | {global, atom()} | {pid(),Mod} -%% Purpose: Start all children. The new list contains #child's -%% with pids. -%% Returns: {ok, NChildren} | {error, NChildren} -%% NChildren = [#child] in termination order (reversed -%% start order) -%%----------------------------------------------------------------- -start_children(Children, SupName) -> start_children(Children, [], SupName). - -start_children([Child|Chs], NChildren, SupName) -> - case do_start_child(SupName, Child) of - {ok, Pid} -> - start_children(Chs, [Child#child{pid = Pid}|NChildren], SupName); - {ok, Pid, _Extra} -> - start_children(Chs, [Child#child{pid = Pid}|NChildren], SupName); - {error, Reason} -> - report_error(start_error, Reason, Child, SupName), - {error, lists:reverse(Chs) ++ [Child | NChildren]} - end; -start_children([], NChildren, _SupName) -> - {ok, NChildren}. - -do_start_child(SupName, Child) -> - #child{mfa = {M, F, A}} = Child, - case catch apply(M, F, A) of - {ok, Pid} when is_pid(Pid) -> - NChild = Child#child{pid = Pid}, - report_progress(NChild, SupName), - {ok, Pid}; - {ok, Pid, Extra} when is_pid(Pid) -> - NChild = Child#child{pid = Pid}, - report_progress(NChild, SupName), - {ok, Pid, Extra}; - ignore -> - {ok, undefined}; - {error, What} -> {error, What}; - What -> {error, What} - end. - -do_start_child_i(M, F, A) -> - case catch apply(M, F, A) of - {ok, Pid} when is_pid(Pid) -> - {ok, Pid}; - {ok, Pid, Extra} when is_pid(Pid) -> - {ok, Pid, Extra}; - ignore -> - {ok, undefined}; - {error, Error} -> - {error, Error}; - What -> - {error, What} - end. - - -%%% --------------------------------------------------- -%%% -%%% Callback functions. -%%% -%%% --------------------------------------------------- -handle_call({start_child, EArgs}, _From, State) when ?is_simple(State) -> - #child{mfa = {M, F, A}} = hd(State#state.children), - Args = A ++ EArgs, - case do_start_child_i(M, F, Args) of - {ok, undefined} -> - {reply, {ok, undefined}, State}; - {ok, Pid} -> - NState = State#state{dynamics = - ?DICT:store(Pid, Args, State#state.dynamics)}, - {reply, {ok, Pid}, NState}; - {ok, Pid, Extra} -> - NState = State#state{dynamics = - ?DICT:store(Pid, Args, State#state.dynamics)}, - {reply, {ok, Pid, Extra}, NState}; - What -> - {reply, What, State} - end; - -%%% The requests terminate_child, delete_child and restart_child are -%%% invalid for simple_one_for_one and simple_one_for_one_terminate -%%% supervisors. -handle_call({_Req, _Data}, _From, State) when ?is_simple(State) -> - {reply, {error, State#state.strategy}, State}; - -handle_call({start_child, ChildSpec}, _From, State) -> - case check_childspec(ChildSpec) of - {ok, Child} -> - {Resp, NState} = handle_start_child(Child, State), - {reply, Resp, NState}; - What -> - {reply, {error, What}, State} - end; - -handle_call({restart_child, Name}, _From, State) -> - case get_child(Name, State) of - {value, Child} when Child#child.pid =:= undefined -> - case do_start_child(State#state.name, Child) of - {ok, Pid} -> - NState = replace_child(Child#child{pid = Pid}, State), - {reply, {ok, Pid}, NState}; - {ok, Pid, Extra} -> - NState = replace_child(Child#child{pid = Pid}, State), - {reply, {ok, Pid, Extra}, NState}; - Error -> - {reply, Error, State} - end; - {value, _} -> - {reply, {error, running}, State}; - _ -> - {reply, {error, not_found}, State} - end; - -handle_call({delete_child, Name}, _From, State) -> - case get_child(Name, State) of - {value, Child} when Child#child.pid =:= undefined -> - NState = remove_child(Child, State), - {reply, ok, NState}; - {value, _} -> - {reply, {error, running}, State}; - _ -> - {reply, {error, not_found}, State} - end; - -handle_call({terminate_child, Name}, _From, State) -> - case get_child(Name, State) of - {value, Child} -> - NChild = do_terminate(Child, State#state.name), - {reply, ok, replace_child(NChild, State)}; - _ -> - {reply, {error, not_found}, State} - end; - -handle_call(which_children, _From, State) when ?is_simple(State) -> - [#child{child_type = CT, modules = Mods}] = State#state.children, - Reply = lists:map(fun ({Pid, _}) -> {undefined, Pid, CT, Mods} end, - ?DICT:to_list(State#state.dynamics)), - {reply, Reply, State}; - -handle_call(which_children, _From, State) -> - Resp = - lists:map(fun (#child{pid = Pid, name = Name, - child_type = ChildType, modules = Mods}) -> - {Name, Pid, ChildType, Mods} - end, - State#state.children), - {reply, Resp, State}. - -%%% Hopefully cause a function-clause as there is no API function -%%% that utilizes cast. -handle_cast(null, State) -> - error_logger:error_msg("ERROR: Supervisor received cast-message 'null'~n", - []), - - {noreply, State}. - -handle_info({delayed_restart, {RestartType, Reason, Child}}, State) - when ?is_simple(State) -> - {ok, NState} = do_restart(RestartType, Reason, Child, State), - {noreply, NState}; -handle_info({delayed_restart, {RestartType, Reason, Child}}, State) -> - case get_child(Child#child.name, State) of - {value, Child1} -> - {ok, NState} = do_restart(RestartType, Reason, Child1, State), - {noreply, NState}; - _ -> - {noreply, State} - end; - -%% -%% Take care of terminated children. -%% -handle_info({'EXIT', Pid, Reason}, State) -> - case restart_child(Pid, Reason, State) of - {ok, State1} -> - {noreply, State1}; - {shutdown, State1} -> - {stop, shutdown, State1} - end; - -handle_info(Msg, State) -> - error_logger:error_msg("Supervisor received unexpected message: ~p~n", - [Msg]), - {noreply, State}. -%% -%% Terminate this server. -%% -terminate(_Reason, State) when ?is_terminate_simple(State) -> - terminate_simple_children( - hd(State#state.children), State#state.dynamics, State#state.name), - ok; -terminate(_Reason, State) -> - terminate_children(State#state.children, State#state.name), - ok. - -%% -%% Change code for the supervisor. -%% Call the new call-back module and fetch the new start specification. -%% Combine the new spec. with the old. If the new start spec. is -%% not valid the code change will not succeed. -%% Use the old Args as argument to Module:init/1. -%% NOTE: This requires that the init function of the call-back module -%% does not have any side effects. -%% -code_change(_, State, _) -> - case (State#state.module):init(State#state.args) of - {ok, {SupFlags, StartSpec}} -> - case catch check_flags(SupFlags) of - ok -> - {Strategy, MaxIntensity, Period} = SupFlags, - update_childspec(State#state{strategy = Strategy, - intensity = MaxIntensity, - period = Period}, - StartSpec); - Error -> - {error, Error} - end; - ignore -> - {ok, State}; - Error -> - Error - end. - -check_flags({Strategy, MaxIntensity, Period}) -> - validStrategy(Strategy), - validIntensity(MaxIntensity), - validPeriod(Period), - ok; -check_flags(What) -> - {bad_flags, What}. - -update_childspec(State, StartSpec) when ?is_simple(State) -> - case check_startspec(StartSpec) of - {ok, [Child]} -> - {ok, State#state{children = [Child]}}; - Error -> - {error, Error} - end; - -update_childspec(State, StartSpec) -> - case check_startspec(StartSpec) of - {ok, Children} -> - OldC = State#state.children, % In reverse start order ! - NewC = update_childspec1(OldC, Children, []), - {ok, State#state{children = NewC}}; - Error -> - {error, Error} - end. - -update_childspec1([Child|OldC], Children, KeepOld) -> - case update_chsp(Child, Children) of - {ok,NewChildren} -> - update_childspec1(OldC, NewChildren, KeepOld); - false -> - update_childspec1(OldC, Children, [Child|KeepOld]) - end; -update_childspec1([], Children, KeepOld) -> - % Return them in (keeped) reverse start order. - lists:reverse(Children ++ KeepOld). - -update_chsp(OldCh, Children) -> - case lists:map(fun (Ch) when OldCh#child.name =:= Ch#child.name -> - Ch#child{pid = OldCh#child.pid}; - (Ch) -> - Ch - end, - Children) of - Children -> - false; % OldCh not found in new spec. - NewC -> - {ok, NewC} - end. - -%%% --------------------------------------------------- -%%% Start a new child. -%%% --------------------------------------------------- - -handle_start_child(Child, State) -> - case get_child(Child#child.name, State) of - false -> - case do_start_child(State#state.name, Child) of - {ok, Pid} -> - Children = State#state.children, - {{ok, Pid}, - State#state{children = - [Child#child{pid = Pid}|Children]}}; - {ok, Pid, Extra} -> - Children = State#state.children, - {{ok, Pid, Extra}, - State#state{children = - [Child#child{pid = Pid}|Children]}}; - {error, What} -> - {{error, {What, Child}}, State} - end; - {value, OldChild} when OldChild#child.pid =/= undefined -> - {{error, {already_started, OldChild#child.pid}}, State}; - {value, _OldChild} -> - {{error, already_present}, State} - end. - -%%% --------------------------------------------------- -%%% Restart. A process has terminated. -%%% Returns: {ok, #state} | {shutdown, #state} -%%% --------------------------------------------------- - -restart_child(Pid, Reason, State) when ?is_simple(State) -> - case ?DICT:find(Pid, State#state.dynamics) of - {ok, Args} -> - [Child] = State#state.children, - RestartType = Child#child.restart_type, - {M, F, _} = Child#child.mfa, - NChild = Child#child{pid = Pid, mfa = {M, F, Args}}, - do_restart(RestartType, Reason, NChild, State); - error -> - {ok, State} - end; -restart_child(Pid, Reason, State) -> - Children = State#state.children, - case lists:keysearch(Pid, #child.pid, Children) of - {value, Child} -> - RestartType = Child#child.restart_type, - do_restart(RestartType, Reason, Child, State); - _ -> - {ok, State} - end. - -do_restart({permanent = RestartType, Delay}, Reason, Child, State) -> - do_restart_delay({RestartType, Delay}, Reason, Child, State); -do_restart(permanent, Reason, Child, State) -> - report_error(child_terminated, Reason, Child, State#state.name), - restart(Child, State); -do_restart(Type, normal, Child, State) -> - del_child_and_maybe_shutdown(Type, Child, State); -do_restart({RestartType, Delay}, {shutdown, restart} = Reason, Child, State) - when RestartType =:= transient orelse RestartType =:= intrinsic -> - do_restart_delay({RestartType, Delay}, Reason, Child, State); -do_restart(Type, {shutdown, _}, Child, State) -> - del_child_and_maybe_shutdown(Type, Child, State); -do_restart(Type, shutdown, Child = #child{child_type = supervisor}, State) -> - del_child_and_maybe_shutdown(Type, Child, State); -do_restart({RestartType, Delay}, Reason, Child, State) - when RestartType =:= transient orelse RestartType =:= intrinsic -> - do_restart_delay({RestartType, Delay}, Reason, Child, State); -do_restart(Type, Reason, Child, State) when Type =:= transient orelse - Type =:= intrinsic -> - report_error(child_terminated, Reason, Child, State#state.name), - restart(Child, State); -do_restart(temporary, Reason, Child, State) -> - report_error(child_terminated, Reason, Child, State#state.name), - NState = state_del_child(Child, State), - {ok, NState}. - -do_restart_delay({RestartType, Delay}, Reason, Child, State) -> - case restart1(Child, State) of - {ok, NState} -> - {ok, NState}; - {terminate, NState} -> - _TRef = erlang:send_after(trunc(Delay*1000), self(), - {delayed_restart, - {{RestartType, Delay}, Reason, Child}}), - {ok, state_del_child(Child, NState)} - end. - -del_child_and_maybe_shutdown(intrinsic, Child, State) -> - {shutdown, state_del_child(Child, State)}; -del_child_and_maybe_shutdown({intrinsic, _Delay}, Child, State) -> - {shutdown, state_del_child(Child, State)}; -del_child_and_maybe_shutdown(_, Child, State) -> - {ok, state_del_child(Child, State)}. - -restart(Child, State) -> - case add_restart(State) of - {ok, NState} -> - restart(NState#state.strategy, Child, NState, fun restart/2); - {terminate, NState} -> - report_error(shutdown, reached_max_restart_intensity, - Child, State#state.name), - {shutdown, state_del_child(Child, NState)} - end. - -restart1(Child, State) -> - case add_restart(State) of - {ok, NState} -> - restart(NState#state.strategy, Child, NState, fun restart1/2); - {terminate, _NState} -> - %% we've reached the max restart intensity, but the - %% add_restart will have added to the restarts - %% field. Given we don't want to die here, we need to go - %% back to the old restarts field otherwise we'll never - %% attempt to restart later. - {terminate, State} - end. - -restart(Strategy, Child, State, Restart) - when Strategy =:= simple_one_for_one orelse - Strategy =:= simple_one_for_one_terminate -> - #child{mfa = {M, F, A}} = Child, - Dynamics = ?DICT:erase(Child#child.pid, State#state.dynamics), - case do_start_child_i(M, F, A) of - {ok, undefined} -> - {ok, State}; - {ok, Pid} -> - NState = State#state{dynamics = ?DICT:store(Pid, A, Dynamics)}, - {ok, NState}; - {ok, Pid, _Extra} -> - NState = State#state{dynamics = ?DICT:store(Pid, A, Dynamics)}, - {ok, NState}; - {error, Error} -> - report_error(start_error, Error, Child, State#state.name), - Restart(Child, State) - end; -restart(one_for_one, Child, State, Restart) -> - case do_start_child(State#state.name, Child) of - {ok, Pid} -> - NState = replace_child(Child#child{pid = Pid}, State), - {ok, NState}; - {ok, Pid, _Extra} -> - NState = replace_child(Child#child{pid = Pid}, State), - {ok, NState}; - {error, Reason} -> - report_error(start_error, Reason, Child, State#state.name), - Restart(Child, State) - end; -restart(rest_for_one, Child, State, Restart) -> - {ChAfter, ChBefore} = split_child(Child#child.pid, State#state.children), - ChAfter2 = terminate_children(ChAfter, State#state.name), - case start_children(ChAfter2, State#state.name) of - {ok, ChAfter3} -> - {ok, State#state{children = ChAfter3 ++ ChBefore}}; - {error, ChAfter3} -> - Restart(Child, State#state{children = ChAfter3 ++ ChBefore}) - end; -restart(one_for_all, Child, State, Restart) -> - Children1 = del_child(Child#child.pid, State#state.children), - Children2 = terminate_children(Children1, State#state.name), - case start_children(Children2, State#state.name) of - {ok, NChs} -> - {ok, State#state{children = NChs}}; - {error, NChs} -> - Restart(Child, State#state{children = NChs}) - end. - -%%----------------------------------------------------------------- -%% Func: terminate_children/2 -%% Args: Children = [#child] in termination order -%% SupName = {local, atom()} | {global, atom()} | {pid(),Mod} -%% Returns: NChildren = [#child] in -%% startup order (reversed termination order) -%%----------------------------------------------------------------- -terminate_children(Children, SupName) -> - terminate_children(Children, SupName, []). - -terminate_children([Child | Children], SupName, Res) -> - NChild = do_terminate(Child, SupName), - terminate_children(Children, SupName, [NChild | Res]); -terminate_children([], _SupName, Res) -> - Res. - -terminate_simple_children(Child, Dynamics, SupName) -> - Pids = dict:fold(fun (Pid, _Args, Pids) -> - erlang:monitor(process, Pid), - unlink(Pid), - exit(Pid, child_exit_reason(Child)), - [Pid | Pids] - end, [], Dynamics), - TimeoutMsg = {timeout, make_ref()}, - TRef = timeout_start(Child, TimeoutMsg), - {Replies, Timedout} = - lists:foldl( - fun (_Pid, {Replies, Timedout}) -> - {Pid1, Reason1, Timedout1} = - receive - TimeoutMsg -> - Remaining = Pids -- [P || {P, _} <- Replies], - [exit(P, kill) || P <- Remaining], - receive - {'DOWN', _MRef, process, Pid, Reason} -> - {Pid, Reason, true} - end; - {'DOWN', _MRef, process, Pid, Reason} -> - {Pid, Reason, Timedout} - end, - {[{Pid1, child_res(Child, Reason1, Timedout1)} | Replies], - Timedout1} - end, {[], false}, Pids), - timeout_stop(Child, TRef, TimeoutMsg, Timedout), - ReportError = shutdown_error_reporter(SupName), - Report = fun(_, ok) -> ok; - (Pid, {error, R}) -> ReportError(R, Child#child{pid = Pid}) - end, - [receive - {'EXIT', Pid, Reason} -> - Report(Pid, child_res(Child, Reason, Timedout)) - after - 0 -> Report(Pid, Reply) - end || {Pid, Reply} <- Replies], - ok. - -child_exit_reason(#child{shutdown = brutal_kill}) -> kill; -child_exit_reason(#child{}) -> shutdown. - -child_res(#child{shutdown=brutal_kill}, killed, false) -> ok; -child_res(#child{}, shutdown, false) -> ok; -child_res(#child{restart_type=permanent}, normal, false) -> {error, normal}; -child_res(#child{restart_type={permanent,_}},normal, false) -> {error, normal}; -child_res(#child{}, normal, false) -> ok; -child_res(#child{}, R, _) -> {error, R}. - -timeout_start(#child{shutdown = Time}, Msg) when is_integer(Time) -> - erlang:send_after(Time, self(), Msg); -timeout_start(#child{}, _Msg) -> - ok. - -timeout_stop(#child{shutdown = Time}, TRef, Msg, false) when is_integer(Time) -> - erlang:cancel_timer(TRef), - receive - Msg -> ok - after - 0 -> ok - end; -timeout_stop(#child{}, _TRef, _Msg, _Timedout) -> - ok. - -do_terminate(Child, SupName) when Child#child.pid =/= undefined -> - ReportError = shutdown_error_reporter(SupName), - case shutdown(Child#child.pid, Child#child.shutdown) of - ok -> - ok; - {error, normal} -> - case Child#child.restart_type of - permanent -> ReportError(normal, Child); - {permanent, _Delay} -> ReportError(normal, Child); - _ -> ok - end; - {error, OtherReason} -> - ReportError(OtherReason, Child) - end, - Child#child{pid = undefined}; -do_terminate(Child, _SupName) -> - Child. - -%%----------------------------------------------------------------- -%% Shutdowns a child. We must check the EXIT value -%% of the child, because it might have died with another reason than -%% the wanted. In that case we want to report the error. We put a -%% monitor on the child an check for the 'DOWN' message instead of -%% checking for the 'EXIT' message, because if we check the 'EXIT' -%% message a "naughty" child, who does unlink(Sup), could hang the -%% supervisor. -%% Returns: ok | {error, OtherReason} (this should be reported) -%%----------------------------------------------------------------- -shutdown(Pid, brutal_kill) -> - - case monitor_child(Pid) of - ok -> - exit(Pid, kill), - receive - {'DOWN', _MRef, process, Pid, killed} -> - ok; - {'DOWN', _MRef, process, Pid, OtherReason} -> - {error, OtherReason} - end; - {error, Reason} -> - {error, Reason} - end; - -shutdown(Pid, Time) -> - - case monitor_child(Pid) of - ok -> - exit(Pid, shutdown), %% Try to shutdown gracefully - receive - {'DOWN', _MRef, process, Pid, shutdown} -> - ok; - {'DOWN', _MRef, process, Pid, OtherReason} -> - {error, OtherReason} - after Time -> - exit(Pid, kill), %% Force termination. - receive - {'DOWN', _MRef, process, Pid, OtherReason} -> - {error, OtherReason} - end - end; - {error, Reason} -> - {error, Reason} - end. - -%% Help function to shutdown/2 switches from link to monitor approach -monitor_child(Pid) -> - - %% Do the monitor operation first so that if the child dies - %% before the monitoring is done causing a 'DOWN'-message with - %% reason noproc, we will get the real reason in the 'EXIT'-message - %% unless a naughty child has already done unlink... - erlang:monitor(process, Pid), - unlink(Pid), - - receive - %% If the child dies before the unlik we must empty - %% the mail-box of the 'EXIT'-message and the 'DOWN'-message. - {'EXIT', Pid, Reason} -> - receive - {'DOWN', _, process, Pid, _} -> - {error, Reason} - end - after 0 -> - %% If a naughty child did unlink and the child dies before - %% monitor the result will be that shutdown/2 receives a - %% 'DOWN'-message with reason noproc. - %% If the child should die after the unlink there - %% will be a 'DOWN'-message with a correct reason - %% that will be handled in shutdown/2. - ok - end. - - -%%----------------------------------------------------------------- -%% Child/State manipulating functions. -%%----------------------------------------------------------------- -state_del_child(#child{pid = Pid}, State) when ?is_simple(State) -> - NDynamics = ?DICT:erase(Pid, State#state.dynamics), - State#state{dynamics = NDynamics}; -state_del_child(Child, State) -> - NChildren = del_child(Child#child.name, State#state.children), - State#state{children = NChildren}. - -del_child(Name, [Ch|Chs]) when Ch#child.name =:= Name -> - [Ch#child{pid = undefined} | Chs]; -del_child(Pid, [Ch|Chs]) when Ch#child.pid =:= Pid -> - [Ch#child{pid = undefined} | Chs]; -del_child(Name, [Ch|Chs]) -> - [Ch|del_child(Name, Chs)]; -del_child(_, []) -> - []. - -%% Chs = [S4, S3, Ch, S1, S0] -%% Ret: {[S4, S3, Ch], [S1, S0]} -split_child(Name, Chs) -> - split_child(Name, Chs, []). - -split_child(Name, [Ch|Chs], After) when Ch#child.name =:= Name -> - {lists:reverse([Ch#child{pid = undefined} | After]), Chs}; -split_child(Pid, [Ch|Chs], After) when Ch#child.pid =:= Pid -> - {lists:reverse([Ch#child{pid = undefined} | After]), Chs}; -split_child(Name, [Ch|Chs], After) -> - split_child(Name, Chs, [Ch | After]); -split_child(_, [], After) -> - {lists:reverse(After), []}. - -get_child(Name, State) -> - lists:keysearch(Name, #child.name, State#state.children). -replace_child(Child, State) -> - Chs = do_replace_child(Child, State#state.children), - State#state{children = Chs}. - -do_replace_child(Child, [Ch|Chs]) when Ch#child.name =:= Child#child.name -> - [Child | Chs]; -do_replace_child(Child, [Ch|Chs]) -> - [Ch|do_replace_child(Child, Chs)]. - -remove_child(Child, State) -> - Chs = lists:keydelete(Child#child.name, #child.name, State#state.children), - State#state{children = Chs}. - -%%----------------------------------------------------------------- -%% Func: init_state/4 -%% Args: SupName = {local, atom()} | {global, atom()} | self -%% Type = {Strategy, MaxIntensity, Period} -%% Strategy = one_for_one | one_for_all | simple_one_for_one | -%% rest_for_one -%% MaxIntensity = integer() -%% Period = integer() -%% Mod :== atom() -%% Arsg :== term() -%% Purpose: Check that Type is of correct type (!) -%% Returns: {ok, #state} | Error -%%----------------------------------------------------------------- -init_state(SupName, Type, Mod, Args) -> - case catch init_state1(SupName, Type, Mod, Args) of - {ok, State} -> - {ok, State}; - Error -> - Error - end. - -init_state1(SupName, {Strategy, MaxIntensity, Period}, Mod, Args) -> - validStrategy(Strategy), - validIntensity(MaxIntensity), - validPeriod(Period), - {ok, #state{name = supname(SupName,Mod), - strategy = Strategy, - intensity = MaxIntensity, - period = Period, - module = Mod, - args = Args}}; -init_state1(_SupName, Type, _, _) -> - {invalid_type, Type}. - -validStrategy(simple_one_for_one_terminate) -> true; -validStrategy(simple_one_for_one) -> true; -validStrategy(one_for_one) -> true; -validStrategy(one_for_all) -> true; -validStrategy(rest_for_one) -> true; -validStrategy(What) -> throw({invalid_strategy, What}). - -validIntensity(Max) when is_integer(Max), - Max >= 0 -> true; -validIntensity(What) -> throw({invalid_intensity, What}). - -validPeriod(Period) when is_integer(Period), - Period > 0 -> true; -validPeriod(What) -> throw({invalid_period, What}). - -supname(self,Mod) -> {self(),Mod}; -supname(N,_) -> N. - -%%% ------------------------------------------------------ -%%% Check that the children start specification is valid. -%%% Shall be a six (6) tuple -%%% {Name, Func, RestartType, Shutdown, ChildType, Modules} -%%% where Name is an atom -%%% Func is {Mod, Fun, Args} == {atom, atom, list} -%%% RestartType is permanent | temporary | transient | -%%% intrinsic | {permanent, Delay} | -%%% {transient, Delay} | {intrinsic, Delay} -%% where Delay >= 0 -%%% Shutdown = integer() | infinity | brutal_kill -%%% ChildType = supervisor | worker -%%% Modules = [atom()] | dynamic -%%% Returns: {ok, [#child]} | Error -%%% ------------------------------------------------------ - -check_startspec(Children) -> check_startspec(Children, []). - -check_startspec([ChildSpec|T], Res) -> - case check_childspec(ChildSpec) of - {ok, Child} -> - case lists:keymember(Child#child.name, #child.name, Res) of - true -> {duplicate_child_name, Child#child.name}; - false -> check_startspec(T, [Child | Res]) - end; - Error -> Error - end; -check_startspec([], Res) -> - {ok, lists:reverse(Res)}. - -check_childspec({Name, Func, RestartType, Shutdown, ChildType, Mods}) -> - catch check_childspec(Name, Func, RestartType, Shutdown, ChildType, Mods); -check_childspec(X) -> {invalid_child_spec, X}. - -check_childspec(Name, Func, RestartType, Shutdown, ChildType, Mods) -> - validName(Name), - validFunc(Func), - validRestartType(RestartType), - validChildType(ChildType), - validShutdown(Shutdown, ChildType), - validMods(Mods), - {ok, #child{name = Name, mfa = Func, restart_type = RestartType, - shutdown = Shutdown, child_type = ChildType, modules = Mods}}. - -validChildType(supervisor) -> true; -validChildType(worker) -> true; -validChildType(What) -> throw({invalid_child_type, What}). - -validName(_Name) -> true. - -validFunc({M, F, A}) when is_atom(M), - is_atom(F), - is_list(A) -> true; -validFunc(Func) -> throw({invalid_mfa, Func}). - -validRestartType(permanent) -> true; -validRestartType(temporary) -> true; -validRestartType(transient) -> true; -validRestartType(intrinsic) -> true; -validRestartType({permanent, Delay}) -> validDelay(Delay); -validRestartType({intrinsic, Delay}) -> validDelay(Delay); -validRestartType({transient, Delay}) -> validDelay(Delay); -validRestartType(RestartType) -> throw({invalid_restart_type, - RestartType}). - -validDelay(Delay) when is_number(Delay), - Delay >= 0 -> true; -validDelay(What) -> throw({invalid_delay, What}). - -validShutdown(Shutdown, _) - when is_integer(Shutdown), Shutdown > 0 -> true; -validShutdown(infinity, supervisor) -> true; -validShutdown(brutal_kill, _) -> true; -validShutdown(Shutdown, _) -> throw({invalid_shutdown, Shutdown}). - -validMods(dynamic) -> true; -validMods(Mods) when is_list(Mods) -> - lists:foreach(fun (Mod) -> - if - is_atom(Mod) -> ok; - true -> throw({invalid_module, Mod}) - end - end, - Mods); -validMods(Mods) -> throw({invalid_modules, Mods}). - -%%% ------------------------------------------------------ -%%% Add a new restart and calculate if the max restart -%%% intensity has been reached (in that case the supervisor -%%% shall terminate). -%%% All restarts accured inside the period amount of seconds -%%% are kept in the #state.restarts list. -%%% Returns: {ok, State'} | {terminate, State'} -%%% ------------------------------------------------------ - -add_restart(State) -> - I = State#state.intensity, - P = State#state.period, - R = State#state.restarts, - Now = erlang:now(), - R1 = add_restart([Now|R], Now, P), - State1 = State#state{restarts = R1}, - case length(R1) of - CurI when CurI =< I -> - {ok, State1}; - _ -> - {terminate, State1} - end. - -add_restart([R|Restarts], Now, Period) -> - case inPeriod(R, Now, Period) of - true -> - [R|add_restart(Restarts, Now, Period)]; - _ -> - [] - end; -add_restart([], _, _) -> - []. - -inPeriod(Time, Now, Period) -> - case difference(Time, Now) of - T when T > Period -> - false; - _ -> - true - end. - -%% -%% Time = {MegaSecs, Secs, MicroSecs} (NOTE: MicroSecs is ignored) -%% Calculate the time elapsed in seconds between two timestamps. -%% If MegaSecs is equal just subtract Secs. -%% Else calculate the Mega difference and add the Secs difference, -%% note that Secs difference can be negative, e.g. -%% {827, 999999, 676} diff {828, 1, 653753} == > 2 secs. -%% -difference({TimeM, TimeS, _}, {CurM, CurS, _}) when CurM > TimeM -> - ((CurM - TimeM) * 1000000) + (CurS - TimeS); -difference({_, TimeS, _}, {_, CurS, _}) -> - CurS - TimeS. - -%%% ------------------------------------------------------ -%%% Error and progress reporting. -%%% ------------------------------------------------------ - -report_error(Error, Reason, Child, SupName) -> - ErrorMsg = [{supervisor, SupName}, - {errorContext, Error}, - {reason, Reason}, - {offender, extract_child(Child)}], - error_logger:error_report(supervisor_report, ErrorMsg). - -shutdown_error_reporter(SupName) -> - fun(Reason, Child) -> - report_error(shutdown_error, Reason, Child, SupName) - end. - -extract_child(Child) -> - [{pid, Child#child.pid}, - {name, Child#child.name}, - {mfa, Child#child.mfa}, - {restart_type, Child#child.restart_type}, - {shutdown, Child#child.shutdown}, - {child_type, Child#child.child_type}]. - -report_progress(Child, SupName) -> - Progress = [{supervisor, SupName}, - {started, extract_child(Child)}], - error_logger:info_report(progress, Progress).