Distributed applications (global)
Ulf Wiger
etxuwig@REDACTED
Fri Mar 10 14:00:44 CET 2000
Sean,
I've made a hack to the dist_ac.erl module. I haven't tried it,
mind you, but it does compile. I thought you'd like to try it.
What I've done is to try to make sure that dist_ac will always
respond to the {dist_ac_weight, _, _, _} messages.
/Uffe
> diff dist_ac.erl .../otp-r6b_1/lib/kernel-2.4.3.1/src/dist_ac.erl
545a546,575
> case keysearch(Name, #appl.name, S#state.appls) of
> {value, Appl} ->
> Id = Appl#appl.id,
> case Id of
> run_waiting ->
> {?DIST_AC, Node} ! {dist_ac_weight, Name, 0, node()},
> {noreply, S};
> undefined ->
> {noreply,
> S#state{tmp_locals = [{Name, Weight, Node} |
> S#state.tmp_locals]}};
> {takeover, _} ->
> {noreply,
> S#state{tmp_locals = [{Name, Weight, Node} |
> S#state.tmp_locals]}};
> {failover, _} ->
> {noreply,
> S#state{tmp_locals = [{Name, Weight, Node} |
> S#state.tmp_locals]}};
> _ ->
> MyWeight = get_cached_weight(Name, S),
> {?DIST_AC, Node} ! {dist_ac_weight, Name, MyWeight,
node()},
> NTWs = keyreplaceadd(Name, 1, S#state.tmp_weights,
> {Name, MyWeight}),
> {noreply, S#state{tmp_weights = NTWs}}
> end;
> _ ->
> {noreply,
> S#state{tmp_locals = [{Name, Weight, Node} |
S#state.tmp_locals]}}
> end;
547,550d576
< %% UW patch: broke out the code into its own function
< NewS = answer_weight(Name, Weight, Node, S),
< {noreply, NewS};
<
786,788d811
< do_wait_dist_start(Node, Appl, Name, Nodes, PermittedNodes, S, Type).
<
< do_wait_dist_start(Node, Appl, Name, Nodes, PermittedNodes, S, Type) ->
803a827
> monitor_node(Node, false),
806,813c830
< do_wait_dist_start(Node, Appl, Name,
< Nodes, PermittedNodes, S, Type);
< {dist_ac_weight, Name1, Weight1, Node1} ->
< %% UW 000310: Node1 is busy working on another app, and wants
< %% to query us about weight. We must respond to avoid deadlock.
< NewS = answer_weight(Name1, Weight1, Node1, S),
< do_wait_dist_start(Node, Appl, Name,
< Nodes, PermittedNodes, NewS, Type);
---
> wait_dist_start(Node, Appl, Name, Nodes, PermittedNodes, S,
Type);
817,822c834,838
< filter(
< fun({Name2, _Weight, Node2}) when Node2 == Node,
< Name2 == Name -> false;
< (_) -> true
< end,
< S#state.tmp_locals),
---
> filter(fun({Name2, _Weight, Node2}) when Node2 == Node,
> Name2 == Name ->
false;
> (_) -> true
> end,
> S#state.tmp_locals),
836,841d851
< {dist_ac_weight, Name1, Weight1, Node1} ->
< %% UW 000310: Node1 is busy working on another app, and wants
< %% to query us about weight. We must respond to avoid deadlock.
< NewS = answer_weight(Name1, Weight1, Node1, S),
< wait_dist_start2(Appl, Name,
< Nodes, PermittedNodes, NewS, Type);
1468,1498d1477
< %%% UW Patch 000310: This function is called from 3 different
< %%% places (handle_info/2, do_wait_dist_start/7, and wait_dist_start2/6).
<
< answer_weight(Name, Weight, Node, S) ->
< case keysearch(Name, #appl.name, S#state.appls) of
< {value, Appl} ->
< Id = Appl#appl.id,
< case Id of
< run_waiting ->
< {?DIST_AC, Node} ! {dist_ac_weight, Name, 0, node()},
< S;
< undefined ->
< S#state{tmp_locals = [{Name, Weight, Node} |
< S#state.tmp_locals]};
< {takeover, _} ->
< S#state{tmp_locals = [{Name, Weight, Node} |
< S#state.tmp_locals]};
< {failover, _} ->
< S#state{tmp_locals = [{Name, Weight, Node} |
< S#state.tmp_locals]};
< _ ->
< MyWeight = get_cached_weight(Name, S),
< {?DIST_AC, Node} !
< {dist_ac_weight, Name, MyWeight, node()},
< NTWs = keyreplaceadd(Name, 1, S#state.tmp_weights,
< {Name, MyWeight}),
< S#state{tmp_weights = NTWs}
< end;
< _ ->
< S#state{tmp_locals = [{Name, Weight, Node} |
S#state.tmp_locals]}
< end.
On Fri, 10 Mar 2000, Sean Hinde wrote:
Sean>Thanks Uffe,
Sean>
Sean>Here is the trace for the two remaining nodes, I'll also bring in the
Sean>commercial support guys and let you all know the outcome.
Sean>
Sean>> Could you call process_info(whereis(dist_ac)) and post the
Sean>> result? You could also try
Sean>> erlang:process_display(whereis(dist_ac), backtrace).
Sean>
Sean>Cheers,
Sean>Sean
Sean>
Sean>
--
Ulf Wiger, Chief Designer AXD 301 <ulf.wiger@REDACTED>
Ericsson Telecom AB tfn: +46 8 719 81 95
Varuvägen 9, Älvsjö mob: +46 70 519 81 95
S-126 25 Stockholm, Sweden fax: +46 8 719 43 44
-------------- next part --------------
%% ``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(dist_ac).
-vsn('$Revision: /main/release/9').
-behaviour(gen_server).
%% External exports
-export([start_link/0, load_application/2,
takeover_application/2, permit_application/2, permit_only_loaded_application/2]).
-export([get_known_nodes/0]).
%% Internal exports
-export([init/1, handle_cast/2, handle_call/3, handle_info/2, terminate/2,
send_timeout/3]).
-export([info/0]).
-import(lists, [zf/2, filter/2, map/2, foreach/2, foldl/3, mapfoldl/3,
keysearch/3, keydelete/3, keyreplace/4, member/2]).
-define(AC, application_controller).
-define(DIST_AC, ?MODULE).
-define(LOCK_ID, ?MODULE).
%% This is the protocol version for the dist_ac protcol (between nodes)
-define(vsn, 1).
%%%-----------------------------------------------------------------
%%% This module implements the default Distributed Applications
%%% Controller. It is possible to write other controllers, when
%%% the functionality in this module are not sufficient.
%%% The process cooperates with the application_controller.
%%%-----------------------------------------------------------------
%%-----------------------------------------------------------------
%% Naming conventions:
%% Appl = #appl
%% AppName = atom()
%%-----------------------------------------------------------------
-record(state, {appls = [], tmp_locals = [], remote_started = [],
known = [], started = [], tmp_weights = [],
dist_loaded = [], t_reqs = [], s_reqs = [], p_reqs = []}).
%%-----------------------------------------------------------------
%% appls = [#appl()] - these are the applications we control
%% tmp_locals = [{AppName, Weight, node()}] - tmp, info part of
%% application startup for some distrib appls,
%% not yet handled.
%% remote_started = [{Node, AppName}] - info on apps started before
%% we were started
%% known = [Node] - These are the nodes known to us
%% started = [AppName] - An ordered list of started applications
%% (reversed start order)
%% tmp_weight = [{AppName, MyWeight}] - tmp, if we're forced to
%% send a dist_ac_weight message before we're prepared to,
%% we remember the weight we sent here, so we can use
%% it in the dist_ac_weight msgs later.
%% dist_loaded = {{Name, Node}, HisNodes, Permission} - info on
%% application loaded on other nodes (and own node)
%% t_reqs = [{AppName, From}] - processes waiting for takeover
%% to complete.
%% s_reqs = [{AppName, From}] - processes waiting for stop
%% to complete.
%% p_reqs = [{From, AppName, Bool, [Node]] - outstanding permit.
%% Nodes is a list of nodes we're still waiting for.
%%-----------------------------------------------------------------
-record(appl, {name, id, restart_time = 0, nodes = [], run = []}).
%%-----------------------------------------------------------------
%% id = local | undefined | {distributed, node()} | waiting | run_waiting |
%% {failover, Node} | {takeover, Node}
%% local : local application
%% undefined : not yet started
%% {distributed, Node} : running on another node, we're standby
%% {failover, Node} : failover from Node
%% {takeover, Node} : takeover from Node
%% waiting : other node went down, we're waiting for a timeout
%% to takeover it. From = pid() | undefined
%% run_waiting : we have decided to start the app; wait for the
%% AC result
%%-----------------------------------------------------------------
start_link() ->
case gen_server:start_link({local, ?DIST_AC}, ?MODULE, [], []) of
{ok, Pid} ->
gen_server:cast(?DIST_AC, init_sync),
{ok, Pid};
Else ->
Else
end.
%%-----------------------------------------------------------------
%% Func: load_application(AppName, DistNodes)
%% Args: AppName = atom()
%% DistNodes = default | {AppName, Time, [node() | {node()...}]}
%% Purpose: Notifies the dist_ac about distributed nodes for an
%% application. DistNodes overrides the kernel 'distributed'
%% parameter.
%% Returns: ok | {error, Reason}
%%-----------------------------------------------------------------
load_application(AppName, DistNodes) ->
gen_server:call(?DIST_AC, {load_application, AppName, DistNodes}, infinity).
takeover_application(AppName, RestartType) ->
case validRestartType(RestartType) of
true ->
wait_for_sync_dacs(),
Nodes = get_nodes(AppName),
global:trans(
{?LOCK_ID, self()},
fun() ->
gen_server:call(
?DIST_AC,
{takeover_application, AppName, RestartType},
infinity)
end,
Nodes);
false ->
{error, {invalid_restart_type, RestartType}}
end.
%%-----------------------------------------------------------------
%% This function controls which applications are permitted to run. If
%% an application X runs when this function is called as
%% permit_application(X, false), it is moved to another node where it
%% is permitted to run (distributed applications only). If there is
%% no such node, the application is stopped. (I.e. local applications
%% are always stopped, and distributed applications with no other node
%% alive are stopped as well.) If later a call to
%% permit_application(X, true) is made, X is restarted.
%% For example, suppose applications app1 and app2 are started and
%% running.
%% If we evaluate
%% permit_application(app2, false)
%% app2 is stopped and app1 only is running.
%% If we now evaluate
%% permit_application(app2, true),
%% permit_application(app3, true)
%% app2 is restarted, but not app3, since it hasn't been started by a
%% call to start_application.
%%-----------------------------------------------------------------
permit_application(AppName, Bool) ->
wait_for_sync_dacs(),
Nodes = get_nodes(AppName),
LockId = {?LOCK_ID, self()},
global:trans(
LockId,
fun() ->
gen_server:call(?DIST_AC,
{permit_application, AppName, Bool, LockId, started},
infinity)
end).
permit_only_loaded_application(AppName, Bool) ->
wait_for_sync_dacs(),
Nodes = get_nodes(AppName),
LockId = {?LOCK_ID, self()},
global:trans(
LockId,
fun() ->
gen_server:call(?DIST_AC,
{permit_application, AppName, Bool, LockId, only_loaded},
infinity)
end).
get_nodes(AppName) ->
gen_server:call(?DIST_AC, {get_nodes, AppName}, infinity).
get_known_nodes() ->
gen_server:call(?DIST_AC, get_known_nodes).
%%%-----------------------------------------------------------------
%%% call-back functions from gen_server
%%%-----------------------------------------------------------------
init([]) ->
process_flag(trap_exit, true),
{ok, #state{}}.
sync_dacs(Appls) ->
Res = global:trans({?LOCK_ID, sync_dacs},
fun() ->
Nodes = introduce_me(nodes(), Appls),
wait_dacs(Nodes, [node()], Appls, [])
end),
ets:insert(ac_tab, {sync_dacs, ok}),
Res.
introduce_me(Nodes, Appls) ->
Msg = {dist_ac_new_node, ?vsn, node(), Appls, []},
filter(fun(Node) ->
%% This handles nodes without DACs
case rpc:call(Node, erlang, whereis, [?DIST_AC]) of
Pid when pid(Pid) ->
Pid ! Msg,
true;
_ ->
false
end
end, Nodes).
wait_dacs([Node | Nodes], KnownNodes, Appls, RStarted) ->
monitor_node(Node, true),
receive
%% HisAppls /= [] is the case when our node connects to a running system
%%
%% It is always the responsibility of newer versions to understand
%% older versions of the protocol. As we don't have any older
%% versions (that are supposed to work with this version), we
%% don't handle version mismatch here.
{dist_ac_new_node, Vsn, Node, HisAppls, HisStarted} ->
monitor_node(Node, false),
NRStarted = RStarted ++ HisStarted,
NAppls = dist_merge(Appls, HisAppls, Node),
wait_dacs(Nodes, [Node | KnownNodes], NAppls, NRStarted);
{nodedown, Node} ->
monitor_node(Node, false),
wait_dacs(Nodes, KnownNodes, Appls, RStarted)
end;
wait_dacs([], KnownNodes, Appls, RStarted) ->
{KnownNodes, Appls, RStarted}.
info() ->
gen_server:call(?DIST_AC, info).
%%-----------------------------------------------------------------
%% All functions that can affect which applications are running
%% execute within a global lock, to ensure that they are not
%% executing at the same time as sync_dacs. However, to avoid a
%% deadlock situation where e.g. permit_application gets the lock
%% before sync_dacs, this function is used to ensure that the local
%% sync_dacs always gets the lock first of all. The lock is still
%% used to not interfere with sync_dacs on other nodes.
%%-----------------------------------------------------------------
wait_for_sync_dacs() ->
case catch ets:lookup(ac_tab, sync_dacs) of
[{sync_dacs, ok}] -> ok;
_ ->
receive after 100 -> ok end,
wait_for_sync_dacs()
end.
handle_cast(init_sync, _S) ->
%% When the dist_ac is started, it receives this msg, and gets into
%% the receive loop. 'go' is sent from the kernel_config proc when
%% all nodes that should be pinged has been pinged. The reason for this
%% is that dist_ac syncs with the other nodes at start-up. That is,
%% it does _not_ handle partitioned nets! The other nodes tries to call
%% the local name dist_ac, which means that this name must be registered
%% before the distribution. But it can't sync until after the distribution
%% is started. Therefore, this 'go'-thing.
receive
go -> ok
end,
Appls =
case application:get_env(kernel, distributed) of
{ok, D} -> dist_check(D);
undefined -> []
end,
dist_take_control(Appls),
net_kernel:monitor_nodes(true), % we're really just interested in nodedowns.
{Known, NAppls, RStarted} = sync_dacs(Appls),
{noreply, #state{appls = NAppls, known = Known, remote_started = RStarted}}.
handle_call(info, _From, S) ->
{reply, S, S};
handle_call({load_application, AppName, DistNodes}, _From, S) ->
Appls = S#state.appls,
case catch dist_replace(DistNodes, AppName, Appls) of
{error, Error} ->
{reply, {error, Error}, S};
{'EXIT', R} ->
{stop, R, {error, R}, S};
NAppls ->
NewS = case dist_find_nodes(NAppls, AppName) of
[] -> % No distrib nodes; we ignore it
S;
_Nodes ->
ensure_take_control(AppName, Appls),
{ok, S2} = load(AppName, S#state{appls = NAppls}),
S2
end,
{reply, ok, NewS}
end;
handle_call({takeover_application, AppName, RestartType}, From, S) ->
Appls = S#state.appls,
case keysearch(AppName, #appl.name, Appls) of
{value, Appl} when element(1, Appl#appl.id) == distributed ->
{distributed, Node} = Appl#appl.id,
ac_takeover(req, AppName, Node, RestartType),
NAppl = Appl#appl{id = takeover},
NAppls = keyreplace(AppName, #appl.name, Appls, NAppl),
TR = S#state.t_reqs,
{noreply, S#state{appls = NAppls,
t_reqs = [{AppName, From} | TR]}};
{value, #appl{id = local}} ->
{reply, {error, {already_running_locally, AppName}}, S};
_ ->
{reply, {error, {not_running_distributed, AppName}}, S}
end;
handle_call({permit_application, AppName, Bool, LockId, StartInfo}, From, S) ->
case lists:keysearch(AppName, #appl.name, Appls = S#state.appls) of
false ->
%% This one covers the case with permit for non-distributed
%% applications. This shouldn't be handled like this, and not
%% here, but we have to be backwards-compatible.
case application_controller:get_loaded(AppName) of
{true, _} when Bool == false ->
ac_stop_it(AppName),
{reply, ok, S};
{true, _} when Bool == true ->
ac_start_it(req, AppName),
{reply, ok, S};
false ->
{reply, {error, {not_loaded, AppName}}, S}
end;
{value, _} ->
NAppls = dist_update_run(S#state.appls, AppName, node(), Bool),
NewS = S#state{appls = NAppls},
%% Check if the application is running
IsRunning = keysearch(AppName, #appl.name, NAppls),
IsMyApp = case IsRunning of
{value, #appl{id = local}} -> true;
_ -> false
end,
%% Tell everyone about the new permission
Nodes = dist_flat_nodes(NAppls, AppName),
Msg = {dist_ac_new_permission, node(), AppName, Bool, IsMyApp},
send_msg(Msg, Nodes),
case StartInfo of
only_loaded ->
{reply, ok, NewS};
started ->
permit(Bool, IsRunning, AppName, From, NewS, LockId)
end
end;
%%-----------------------------------------------------------------
%% The distributed parameter is changed. Update the parameters
%% but the applications are actually not moved to other nodes
%% even if they should.
%%-----------------------------------------------------------------
handle_call({distribution_changed, NewDistribution}, _From, S) ->
Appls = S#state.appls,
NewAppls = dist_change_update(Appls, NewDistribution),
NewS = S#state{appls = NewAppls},
{reply, ok, NewS};
handle_call({get_nodes, AppName}, _From, S) ->
Alive = intersection(dist_flat_nodes(S#state.appls, AppName),
S#state.known),
{reply, Alive, S};
handle_call(get_known_nodes, _From, S) ->
{reply, S#state.known, S}.
handle_info({ac_load_application_req, AppName}, S) ->
{ok, NewS} = load(AppName, S),
?AC ! {ac_load_application_reply, AppName, ok},
{noreply, NewS};
handle_info({ac_application_unloaded, AppName}, S) ->
{ok, NewS} = unload(AppName, S),
{noreply, NewS};
handle_info({ac_start_application_req, AppName}, S) ->
%% We must decide if we or another node should start the application
Lock = {?LOCK_ID, self()},
case global:set_lock(Lock, [node()], 1) of
true ->
S2 = case catch start_appl(AppName, S, reply) of
{ok, NewS, _} ->
NewS;
{error, R} ->
?AC ! {ac_start_application_reply, AppName, {error,R}},
S
end,
global:del_lock(Lock),
{noreply, S2};
false ->
send_after(100, {ac_start_application_req, AppName}),
{noreply, S}
end;
handle_info({ac_application_run, AppName, Res}, S) ->
%% We ordered a start, and here's the result. Tell all other nodes.
Appls = S#state.appls,
Nodes = S#state.known,
%% Send this to _all_ known nodes, as any node could sync
%% on this app (not only nodes that can run it).
send_msg({dist_ac_app_started, node(), AppName, Res}, Nodes),
NId = case Res of
ok -> local;
{error, R} -> undefined
end,
{value, Appl} = keysearch(AppName, #appl.name, Appls),
%% Check if we have somebody waiting for the takeover result
NTReqs = del_t_reqs(AppName, S#state.t_reqs, Res),
NAppl = Appl#appl{id = NId},
NAppls = keyreplace(AppName, #appl.name, Appls, NAppl),
{noreply, S#state{appls = NAppls, t_reqs = NTReqs}};
handle_info({ac_application_not_run, AppName}, S) ->
%% We ordered a stop, and now it has stopped
{value, Appl} = keysearch(AppName, #appl.name, Appls = S#state.appls),
%% Check if we have somebody waiting for the takeover result;
%% if somebody called stop just before takeover was handled,
NTReqs = del_t_reqs(AppName, S#state.t_reqs, {error, stopped}),
%% Check if we have somebody waiting for stop to return
SReqs = filter(fun({Name, From2}) when Name == AppName ->
gen_server:reply(From2, ok),
false;
(_) ->
true
end, S#state.s_reqs),
send_msg({dist_ac_app_stopped, AppName}, S#state.known),
NAppl = Appl#appl{id = undefined},
NAppls = keyreplace(AppName, #appl.name, Appls, NAppl),
{noreply, S#state{appls = NAppls, t_reqs = NTReqs, s_reqs = SReqs}};
handle_info({ac_application_stopped, AppName}, S) ->
%% Somebody called application:stop - reset state as it was before
%% the application was started.
{value, Appl} = keysearch(AppName, #appl.name, Appls = S#state.appls),
%% Check if we have somebody waiting for the takeover result;
%% if somebody called stop just before takeover was handled,
NTReqs = del_t_reqs(AppName, S#state.t_reqs, {error, stopped}),
%% Check if we have somebody waiting for stop to return
SReqs = filter(fun({Name, From2}) when Name == AppName ->
gen_server:reply(From2, ok),
false;
(_) ->
true
end, S#state.s_reqs),
send_msg({dist_ac_app_stopped, AppName}, S#state.known),
NAppl = Appl#appl{id = undefined},
NAppls = keyreplace(AppName, #appl.name, Appls, NAppl),
Started = lists:delete(AppName, S#state.started),
{noreply, S#state{appls = NAppls, started = Started,
t_reqs = NTReqs, s_reqs = SReqs}};
%%-----------------------------------------------------------------
%% A new node gets running.
%% Send him info about our started distributed applications.
%%-----------------------------------------------------------------
handle_info({dist_ac_new_node, Vsn, Node, HisAppls, []}, S) ->
Appls = S#state.appls,
MyStarted = zf(fun(Appl) when Appl#appl.id == local ->
{true, {node(), Appl#appl.name}};
(_) ->
false
end, Appls),
{?DIST_AC, Node} ! {dist_ac_new_node, ?vsn, node(), Appls, MyStarted},
NAppls = dist_merge(Appls, HisAppls, Node),
{noreply, S#state{appls = NAppls, known = [Node | S#state.known]}};
handle_info({dist_ac_app_started, Node, Name, Res}, S) ->
case lists:member(Name, S#state.started) of
true ->
Appls = S#state.appls,
{value, Appl} = keysearch(Name, #appl.name, Appls),
NId = case Appl#appl.id of
_ when element(1, Res) == error ->
%% Start of appl on some node failed.
%% Set Id to undefined. That node will have
%% to take some actions, e.g. reboot
undefined;
{distributed, _} ->
%% Another node tookover from some node. Update
%% appl list.
{distributed, Node};
local ->
%% Another node tookover from me; stop my application
%% and update the running list.
{distributed, Node};
_ ->
%% Another node started appl. Update appl list.
{distributed, Node}
end,
ac_started(req, Name, Node),
NAppl = Appl#appl{id = NId},
NAppls = keyreplace(Name, #appl.name, Appls, NAppl),
TmpWeights = keydelete_all(Name, 1, S#state.tmp_weights),
NewS = S#state{appls = NAppls, tmp_weights = TmpWeights},
NPermitReq = req_del_permit_false(NewS#state.p_reqs, Name),
case catch req_start_app(NewS#state{p_reqs = NPermitReq}, Name) of
{error, R} ->
{stop, R};
{ok, NewS2} ->
{noreply, NewS2}
end;
false ->
%% The app has not been started at this node yet; remember this in
%% remote started.
NRStarted = [{Node, Name} | S#state.remote_started],
{noreply, S#state{remote_started = NRStarted}}
end;
handle_info({dist_ac_app_stopped, AppName}, S) ->
{value, Appl} = keysearch(AppName, #appl.name, Appls = S#state.appls),
NAppl = Appl#appl{id = undefined},
NAppls = keyreplace(AppName, #appl.name, Appls, NAppl),
RStarted = keydelete(AppName, 2, S#state.remote_started),
{noreply, S#state{appls = NAppls, remote_started = RStarted}};
handle_info({dist_ac_weight, Name, Weight, Node}, S) ->
%% This means another node starts up, and will eventually take over
%% this appl. We have a situation like: {Name, [{Node}, node()]}
%% Node sends us this msg, and we must respond. It doesn't really
%% matter what we send him; but it must be a dist_ac_weight msg.
%% Another situation is {Name, [RNode, {node()}, Node]}.
%%
%% Yet another situation is that the node where Name was running crashed,
%% and Node has got the nodedown message, but we haven't. In this case,
%% we must send a correct weight to Node. i.e. the same weight that
%% we'll send to him later, when we get the nodedown message.
%% UW patch: broke out the code into its own function
NewS = answer_weight(Name, Weight, Node, S),
{noreply, NewS};
%%-----------------------------------------------------------------
%% A node died. Check if we should takeover some applications.
%%-----------------------------------------------------------------
handle_info({nodedown, Node}, S) ->
AppNames = dist_get_runnable(S#state.appls),
HisAppls = filter(fun(#appl{name = Name, id = {distributed, N}})
when Node == N -> lists:member(Name, AppNames);
(_) -> false
end,
S#state.appls),
Appls2 = zf(fun(Appl) when Appl#appl.id == {distributed, Node} ->
case lists:member(Appl#appl.name, AppNames) of
true ->
{true, Appl#appl{id = {failover, Node}}};
false ->
ac_not_running(Appl#appl.name),
{true, Appl#appl{id = undefined}}
end;
(_) ->
true
end,
S#state.appls),
RStarted = filter(fun({Node2, _Name}) when Node2 == Node -> false;
(_) -> true
end,
S#state.remote_started),
Appls3 = dist_del_node(Appls2, Node),
{NPermitReq, Appls4, SReqs} = req_del_node(S, Node, Appls3),
NKnown = lists:delete(Node, S#state.known),
NewS = S#state{appls = Appls4, p_reqs = NPermitReq, known = NKnown,
s_reqs = SReqs,
remote_started = RStarted},
restart_appls(HisAppls),
{noreply, NewS};
handle_info({dist_ac_app_loaded, Node, Name, HisNodes, Permission, HeKnowsMe},
S) ->
Nodes = dist_find_nodes(Appls = S#state.appls, Name),
case is_loaded(Name, S) of
true ->
case equal_nodes(Nodes, HisNodes) of
true ->
NAppls = dist_update_run(Appls, Name, Node, Permission),
if
HeKnowsMe == false ->
%% We've got it loaded, but he doesn't know -
%% he's a new node connecting to us.
Msg = {dist_ac_app_loaded, node(), Name,
Nodes, dist_is_runnable(Appls, Name), true},
{?DIST_AC, Node} ! Msg;
true ->
ok
end,
{noreply, S#state{appls = NAppls}};
false ->
exit({distribution_mismatch, Name, Node})
end;
false ->
Load =[{{Name, Node}, HisNodes, Permission} | S#state.dist_loaded],
{noreply, S#state{dist_loaded = Load}}
end;
handle_info({dist_ac_app_unloaded, Node, Name}, S) ->
Appls = dist_update_run(S#state.appls, Name, Node, undefined),
Load = keydelete({Name, Node}, 1, S#state.dist_loaded),
{noreply, S#state{appls = Appls, dist_loaded = Load}};
handle_info({dist_ac_new_permission, Node, AppName, false, IsHisApp}, S) ->
Appls = dist_update_run(S#state.appls, AppName, Node, false),
NewS = S#state{appls =Appls},
case dist_is_runnable(Appls, AppName) of
true when IsHisApp == true ->
case catch start_appl(AppName, NewS, req) of
{ok, NewS2, _} ->
{noreply, NewS2};
{error, R} -> % if app was permanent, AC will shutdown the node
{noreply, NewS}
end;
_ ->
{noreply, NewS}
end;
handle_info({dist_ac_new_permission, Node, AppName, true, _IsHisApp}, S) ->
Appls = dist_update_run(S#state.appls, AppName, Node, true),
{noreply, S#state{appls = Appls}};
handle_info({internal_restart_appl, Name}, S) ->
case restart_appl(Name, S) of
{error, R} ->
{stop, {error, R}, S};
NewS ->
{noreply, NewS}
end;
handle_info(_, S) ->
{noreply, S}.
terminate(Reason, S) ->
ok.
%%%-----------------------------------------------------------------
%%% Internal functions
%%%-----------------------------------------------------------------
load(AppName, S) ->
Appls0 = S#state.appls,
%% Get the dist specification for the app on other nodes
DistLoaded = get_dist_loaded(AppName, Load1 = S#state.dist_loaded),
%% Get the local dist specification
Nodes = dist_find_nodes(Appls0, AppName),
FNodes = flat_nodes(Nodes),
%% Update dists spec with our local permission
Permission = get_default_permission(AppName),
Appls1 = dist_update_run(Appls0, AppName, node(), Permission),
%% Compare the local spec with other nodes's specs
%% If equal, update our spec with his current permission
{LoadedNodes, Appls2} =
mapfoldl(
fun({Node, HisNodes, HisPermission}, Appls) ->
case equal_nodes(Nodes, HisNodes) of
true ->
{Node, dist_update_run(Appls, AppName,
Node, HisPermission)};
_ ->
exit({distribution_mismatch, AppName, Node})
end
end, Appls1, DistLoaded),
Load2 = del_dist_loaded(AppName, Load1),
%% Tell all Nodes about the new appl loaded, and its permission.
foreach(fun(Node) when Node /= node() ->
Msg = {dist_ac_app_loaded, node(), AppName,
Nodes, Permission, member(Node, LoadedNodes)},
{?DIST_AC, Node} ! Msg;
(_) -> ok
end, FNodes),
{ok, S#state{appls = Appls2, dist_loaded = Load2}}.
ensure_take_control(AppName, Appls) ->
%% Check if this is a new application that we don't control yet
case keysearch(AppName, #appl.name, Appls) of
{value, _} -> % we have control
ok;
false -> % take control!
%% Note: this works because this is executed within a
%% synchronous call. I.e. we get the control *before*
%% application:load returns. (otherwise application:start
%% could be called before we got the chance to take control)
%% The only reason we have to bother about this is because
%% we have to be backwards compatible in the sense that all
%% apps don't have to be specified in the 'distributed' parameter,
%% but may be implicitly 'distributed' by a call to
%% application:load.
application_controller:control_application(AppName)
end.
unload(AppName, S) ->
Appls = S#state.appls,
Nodes = dist_flat_nodes(Appls, AppName),
%% Tell all ACs in DistNodes about the unloaded appl
Msg = {dist_ac_app_unloaded, node(), AppName},
send_msg(Msg, Nodes),
{value, Appl} = keysearch(AppName, #appl.name, Appls),
NAppl = Appl#appl{id = undefined, run = []},
{ok, S#state{appls = keyreplace(AppName, #appl.name, Appls, NAppl)}}.
start_appl(AppName, S, Type) ->
%% Get nodes, and check if App is loaded on all involved nodes.
%% If it is loaded everywhere, we know that we have the same picture
%% of the nodes; otherwise the load wouldn't have succeeded.
{value, Appl} = keysearch(AppName, #appl.name, Appls = S#state.appls),
case Appl#appl.id of
local ->
%% UW 990913: we've already started the app
%% this could happen if ac_start_application_req was resent.
{ok,S,false};
_ ->
{Id, IsWaiting} = case dist_get_all_nodes(Appl) of
{ok, DistNodes, PermittedNodes} ->
start_distributed(Appl, AppName, DistNodes,
PermittedNodes, S, Type);
Error -> throw(Error)
end,
NAppl = Appl#appl{id = Id},
NAppls = keyreplaceadd(AppName, #appl.name, Appls, NAppl),
{ok, NewS} = req_start_app(S#state{appls = NAppls}, AppName),
TmpLocals = keydelete_all(AppName, 1, NewS#state.tmp_locals),
TmpWeights = keydelete_all(AppName, 1, NewS#state.tmp_weights),
RStarted = keydelete(AppName, 2, S#state.remote_started),
Started = replaceadd(AppName, NewS#state.started),
{ok,
NewS#state{started = Started, tmp_locals = TmpLocals,
tmp_weights = TmpWeights, remote_started = RStarted},
IsWaiting}
end.
start_distributed(Appl, Name, Nodes, PermittedNodes, S, Type) ->
case find_start_node(Nodes, PermittedNodes, Name, S) of
{ok, Node} when Node == node() ->
case Appl#appl.id of
{failover, FoNode} when Type == req ->
ac_failover(Name, FoNode, undefined);
{distributed, Node2} when Type == req ->
ac_takeover(req, Name, Node2, undefined);
_ when Type == reply ->
case lists:keysearch(Name, 2, S#state.remote_started) of
{value, {Node3, _}} ->
ac_takeover(reply, Name, Node3, undefined);
_ ->
ac_start_it(Type, Name)
end;
_ ->
ac_start_it(Type, Name)
end,
{run_waiting, true};
{already_started, Node} ->
ac_started(Type, Name, Node),
{{distributed, Node}, false};
{ok, Node} ->
case keysearch(Name, #appl.name, S#state.appls) of
{value, #appl{id = {distributed, Node}}} ->
ac_started(Type, Name, Node),
{{distributed, Node}, false};
_ ->
wait_dist_start(Node, Appl, Name, Nodes,
PermittedNodes, S, Type)
end;
not_started ->
wait_dist_start2(Appl, Name, Nodes, PermittedNodes, S, Type);
no_permission ->
ac_not_started(Type, Name),
{undefined, false}
end.
wait_dist_start(Node, Appl, Name, Nodes, PermittedNodes, S, Type) ->
monitor_node(Node, true),
do_wait_dist_start(Node, Appl, Name, Nodes, PermittedNodes, S, Type).
do_wait_dist_start(Node, Appl, Name, Nodes, PermittedNodes, S, Type) ->
receive
{dist_ac_app_started, Node, Name, ok} ->
ac_started(Type, Name, Node),
monitor_node(Node, false),
{{distributed, Node}, false};
{dist_ac_app_started, Node, Name, {error, R}} ->
ac_error(Type, Name, {Node, R}),
monitor_node(Node, false),
{Appl#appl.id, false};
{dist_ac_weight, Name, _Weigth, Node} ->
%% This is the situation: {Name, [RNode, {Node}, node()]}
%% and permit(false) is called on RNode, and we sent the
%% weigth first. Node handled it in handle_info, and
%% now we must send him a weigth msg. We can use any weigth;
%% he wins anyway.
{?DIST_AC, Node} !
{dist_ac_weight, Name, get_cached_weight(Name, S), node()},
do_wait_dist_start(Node, Appl, Name,
Nodes, PermittedNodes, S, Type);
{dist_ac_weight, Name1, Weight1, Node1} ->
%% UW 000310: Node1 is busy working on another app, and wants
%% to query us about weight. We must respond to avoid deadlock.
NewS = answer_weight(Name1, Weight1, Node1, S),
do_wait_dist_start(Node, Appl, Name,
Nodes, PermittedNodes, NewS, Type);
{nodedown, Node} ->
monitor_node(Node, false),
TmpLocals =
filter(
fun({Name2, _Weight, Node2}) when Node2 == Node,
Name2 == Name -> false;
(_) -> true
end,
S#state.tmp_locals),
NewS = S#state{tmp_locals = TmpLocals},
start_distributed(Appl, Name, Nodes,
lists:delete(Node, PermittedNodes), NewS, Type)
end.
wait_dist_start2(Appl, Name, Nodes, PermittedNodes, S, Type) ->
receive
{dist_ac_app_started, Node, Name, ok} ->
ac_started(Type, Name, Node),
{{distributed, Node}, false};
{dist_ac_app_started, Node, Name, {error, R}} ->
ac_error(Type, Name, {Node, R}),
{Appl#appl.id, false};
{dist_ac_weight, Name1, Weight1, Node1} ->
%% UW 000310: Node1 is busy working on another app, and wants
%% to query us about weight. We must respond to avoid deadlock.
NewS = answer_weight(Name1, Weight1, Node1, S),
wait_dist_start2(Appl, Name,
Nodes, PermittedNodes, NewS, Type);
{nodedown, Node} ->
%% A node went down, try to start the app again - there may not
%% be any more nodes to wait for.
TmpLocals =
filter(fun({Name2, _Weight, Node2}) when Node2 == Node,
Name2 == Name -> false;
(_) -> true
end,
S#state.tmp_locals),
NewS = S#state{tmp_locals = TmpLocals},
start_distributed(Appl, Name, Nodes,
lists:delete(Node, PermittedNodes), NewS, Type)
end.
ac_start_it(reply, Name) ->
?AC ! {ac_start_application_reply, Name, start_it};
ac_start_it(req, Name) ->
?AC ! {ac_change_application_req, Name, start_it}.
ac_started(reply, Name, Node) ->
?AC ! {ac_start_application_reply, Name, {started, Node}};
ac_started(req, Name, Node) ->
?AC ! {ac_change_application_req, Name, {started, Node}}.
ac_error(reply, Name, Error) ->
?AC ! {ac_start_application_reply, Name, {error, Error}};
ac_error(req, Name, Error) ->
ok.
ac_not_started(reply, Name) ->
?AC ! {ac_start_application_reply, Name, not_started};
ac_not_started(req, Name) ->
?AC ! {ac_change_application_req, Name, stop_it}.
ac_stop_it(Name) ->
?AC ! {ac_change_application_req, Name, stop_it}.
ac_takeover(reply, Name, Node, RestartType) ->
?AC ! {ac_start_application_reply, Name, {takeover, Node}};
ac_takeover(req, Name, Node, RestartType) ->
?AC ! {ac_change_application_req, Name,
{takeover, Node, RestartType}}.
ac_failover(Name, Node, RestartType) ->
?AC ! {ac_change_application_req, Name,
{failover, Node, RestartType}}.
ac_not_running(Name) ->
?AC ! {ac_change_application_req, Name, not_running}.
restart_appls(Appls) ->
foreach(fun(Appl) ->
AppName = Appl#appl.name,
send_after(Appl#appl.restart_time,
{internal_restart_appl, AppName})
end, lists:reverse(Appls)).
restart_appl(AppName, S) ->
case keysearch(AppName, #appl.name, S#state.appls) of
{value, Appl} when element(1, Appl#appl.id) == failover ->
case catch start_appl(AppName, S, req) of
{ok, NewS, _} ->
NewS;
{error, R} ->
error_logger:error_msg(R),
S
end;
_ ->
S
end.
%% permit(ShouldBeRunning, IsRunning, ...)
permit(false, {value, #appl{id = undefined}}, AppName, From, S, LockId) ->
{reply, ok, S}; % It's not running
permit(false, {value, #appl{id = Id}}, _AppName, _From, S, LockId)
when element(1, Id) == distributed ->
%% It is running at another node already
{reply, ok, S};
permit(false, {value, _}, AppName, From, S, LockId) ->
%% It is a distributed application
%% Check if there is any runnable node
case dist_get_runnable_nodes(S#state.appls, AppName) of
[] ->
%% There is no runnable node; stop application
ac_stop_it(AppName),
SReqs = [{AppName, From} | S#state.s_reqs],
{noreply, S#state{s_reqs = SReqs}};
Nodes ->
%% Delete all outstanding 'permit true' requests.
PR = req_del_permit_true(S#state.p_reqs, AppName),
NPReqs = [{From, AppName, false, Nodes} | PR],
{noreply, S#state{p_reqs = NPReqs}}
end;
permit(true, {value, #appl{id = local}}, AppName, From, S, LockId) ->
{reply, ok, S};
permit(true, _, AppName, From, S, LockId) ->
case catch start_appl(AppName, S, req) of
{_ErrorTag, {not_running, App}} ->
%% Delete all outstanding 'permit false' requests
PR = req_del_permit_false(S#state.p_reqs, AppName),
NPReqs = [{false, AppName, true, App} | PR],
{reply, ok, S#state{p_reqs = NPReqs}};
{ok, NewS, true} ->
%% We have ordered a start or a takeover; we must not return
%% until the app is running.
TR = NewS#state.t_reqs,
%% Delete the lock, so others may start the app
global:del_lock(LockId),
{noreply, NewS#state{t_reqs = [{AppName, From} | TR]}};
{ok, NewS, false} ->
{reply, ok, S};
{_ErrorTag, R} ->
{stop, R, {error, R}, S}
end.
do_start_appls(StartApps, S) ->
SortedStartApps = StartApps,
Appls = S#state.appls,
{ok, foldl(
fun(AppName, NewS) ->
case catch start_appl(AppName, NewS, req) of
{error, R} ->
throw({{error, NewS}, R});
{ok, NewS2, _} ->
NewS2
end
end, S#state{appls = Appls}, lists:reverse(SortedStartApps))}.
%%-----------------------------------------------------------------
%% Nodes = [node() | {node(), ..., node()}]
%% A list in priority order. If it is a tuple, we may pick any of
%% them. This decision is made by all nodes in the list, and all
%% nodes choose the same. This is accomplished in the following
%% way: all Nodes send to all others a msg which tells how many
%% applications each node has started. The one with least no of
%% appls starts this one.
%%-----------------------------------------------------------------
find_start_node(Nodes, PermittedNodes, Name, S) ->
AllNodes = intersection(flat_nodes(Nodes), PermittedNodes),
case lists:member(node(), AllNodes) of
true ->
Weight = get_cached_weight(Name, S),
find_start_node(Nodes, Name, S, Weight, AllNodes);
false ->
case keysearch(Name, 2, S#state.remote_started) of
{value, {Node, _Name}} ->
{already_started, Node};
_ when AllNodes /= [] ->
not_started;
_ ->
no_permission
end
end.
find_start_node([AnyNodes | Nodes], Name, S, Weight, AllNodes)
when tuple(AnyNodes) ->
case find_any_node(tuple_to_list(AnyNodes), Name, S, Weight, AllNodes) of
false -> find_start_node(Nodes, Name, S, Weight, AllNodes);
Res -> Res
end;
find_start_node([Node | Nodes], Name, S, Weight, AllNodes) ->
case lists:member(Node, AllNodes) of
true ->
case keysearch(Name, #appl.name, S#state.appls) of
{value, #appl{id = {distributed, Node}}} ->
{already_started, Node};
_ ->
case keysearch(Name, 2, S#state.remote_started) of
{value, {Node, _Name}} ->
{already_started, Node};
_ ->
{ok, Node}
end
end;
false -> find_start_node(Nodes, Name, S, Weight, AllNodes)
end;
find_start_node([], _Name, _S, _Weight, _AllNodes) ->
not_started.
%%-----------------------------------------------------------------
%% First of all, check if the application is already running
%% somewhere in AnyNodes; in that case we shall not move it!
%%-----------------------------------------------------------------
find_any_node(AnyNodes, Name, S, Weight, AllNodes) ->
case check_running(Name, S, intersection(AnyNodes, AllNodes)) of
{already_started, Node} -> {already_started, Node};
false ->
%% Synchronize with all other nodes.
send_nodes(AllNodes, {dist_ac_weight, Name, Weight, node()}),
Answers = [{Weight, node()} |
collect_answers(AllNodes, Name, S, [])],
%% Make a decision (the same at every node) (smallest weight wins)
find_alive_node(lists:sort(Answers),
intersection(AnyNodes, S#state.known))
end.
%%-----------------------------------------------------------------
%% Check if another node started the appl before we got alive.
%% If so, check if the node is one of AnyNodes.
%%-----------------------------------------------------------------
check_running(Name, #state{remote_started = RStarted,
appls = Appls}, AnyNodes) ->
case keysearch(Name, 2, RStarted) of
{value, {Node, _Name}} ->
case lists:member(Node, AnyNodes) of
true -> {already_started, Node};
false -> false
end;
false ->
case keysearch(Name, #appl.name, Appls) of
{value, #appl{id = {distributed, Node}}} ->
case lists:member(Node, AnyNodes) of
true -> {already_started, Node};
false -> false
end;
_ ->
false
end
end.
find_alive_node([{_, Node} | Nodes], AliveNodes) ->
case lists:member(Node, AliveNodes) of
true -> {ok, Node};
false -> find_alive_node(Nodes, AliveNodes)
end;
find_alive_node([], _AliveNodes) ->
false.
%%-----------------------------------------------------------------
%% First, check if the node's msg is buffered (received in our
%% main loop). Otherwise, wait for msg or nodedown.
%% We have sent the dist_ac_weight message, and will wait for it
%% to be received here (or a nodedown). This implies that a
%% dist_ac must *always* be prepared to get this messages, and to
%% send it to us.
%%-----------------------------------------------------------------
collect_answers([Node | Nodes], Name, S, Res) when Node /= node() ->
case keysearch(Node, 3, S#state.tmp_locals) of
{value, {Name, Weight, Node}} ->
collect_answers(Nodes, Name, S, [{Weight, Node} | Res]);
_ ->
monitor_node(Node, true),
receive
{dist_ac_weight, Name, Weight, Node} ->
monitor_node(Node, false),
collect_answers(Nodes, Name, S, [{Weight, Node} | Res]);
{nodedown, Node} ->
monitor_node(Node, false),
collect_answers(Nodes, Name, S, Res)
end
end;
collect_answers([Node | Nodes], Name, S, Res) ->
collect_answers(Nodes, Name, S, Res);
collect_answers([], _Name, _S, Res) ->
Res.
send_nodes(Nodes, Msg) ->
FlatNodes = flat_nodes(Nodes),
foreach(fun(Node) when Node /= node() -> {?DIST_AC, Node} ! Msg;
(Node) -> ok
end, FlatNodes).
send_after(Time, Msg) when integer(Time), Time >= 0 ->
spawn_link(?MODULE, send_timeout, [self(), Time, Msg]);
send_after(_,_) -> % infinity
ok.
send_timeout(To, Time, Msg) ->
receive
after Time -> To ! Msg
end.
send_msg(Msg, Nodes) ->
foreach(fun(Node) when Node /= node() -> {?DIST_AC, Node} ! Msg;
(_) -> ok
end, Nodes).
replaceadd(Item, List) ->
case member(Item, List) of
true -> List;
false -> [Item | List]
end.
keyreplaceadd(Key, Pos, List, New) ->
case keysearch(Key, Pos, List) of
{value, _} -> keyreplace(Key, Pos, List, New);
_ -> [New | List]
end.
keydelete_all(Key, N, [H|T]) when element(N, H) == Key ->
keydelete_all(Key, N, T);
keydelete_all(Key, N, [H|T]) ->
[H|keydelete_all(Key, N, T)];
keydelete_all(Key, N, []) -> [].
-ifdef(NOTUSED).
keysearchdelete(Key, Pos, List) ->
ksd(Key, Pos, List, []).
ksd(Key, Pos, [H | T], Rest) when element(Pos, H) == Key ->
{value, H, Rest ++ T};
ksd(Key, Pos, [H | T], Rest) ->
ksd(Key, Pos, T, [H | Rest]);
ksd(_Key, _Pos, [], _Rest) ->
false.
get_new_appl(Name, [{application, Name, App} | _]) ->
{ok, {application, Name, App}};
get_new_appl(Name, [_ | T]) -> get_new_appl(Name, T);
get_new_appl(Name, []) -> false.
-endif.
equal_nodes([H | T1], [H | T2]) when atom(H) ->
equal_nodes(T1, T2);
equal_nodes([H1 | T1], [H2 | T2]) when tuple(H1), tuple(H2) ->
case equal(tuple_to_list(H1), tuple_to_list(H2)) of
true -> equal_nodes(T1, T2);
false -> false
end;
equal_nodes([], []) -> true;
equal_nodes(_, _) -> false.
equal([H | T] , S) ->
case lists:member(H, S) of
true -> equal(T, lists:delete(H, S));
false -> false
end;
equal([], []) -> true;
equal(_, _) -> false.
flat_nodes(Nodes) when list(Nodes) ->
foldl(fun(Node, Res) when atom(Node) -> [Node | Res];
(Tuple, Res) when tuple(Tuple) -> tuple_to_list(Tuple) ++ Res
end, [], Nodes);
flat_nodes(Nodes) ->
throw({error, {badarg, Nodes}}).
get_cached_weight(Name, S) ->
case lists:keysearch(Name, 1, S#state.tmp_weights) of
{value, {_, W}} -> W;
_ -> get_weight()
end.
%% Simple weight; just count the number of applications running.
get_weight() ->
length(application:which_applications()).
get_dist_loaded(Name, [{{Name, Node}, HisNodes, Permission} | T]) ->
[{Node, HisNodes, Permission} | get_dist_loaded(Name, T)];
get_dist_loaded(Name, [H | T]) ->
get_dist_loaded(Name, T);
get_dist_loaded(Name, []) ->
[].
del_dist_loaded(Name, [{{Name, Node}, HisNodes, Permission} | T]) ->
del_dist_loaded(Name, T);
del_dist_loaded(Name, [H | T]) ->
[H | del_dist_loaded(Name, T)];
del_dist_loaded(Name, []) ->
[].
req_start_app(State, Name) ->
{ok, foldl(
fun({false, AppName, true, Name2}, S) when Name == Name2 ->
PR = keydelete(AppName, 2, S#state.p_reqs),
NS = S#state{p_reqs = PR},
case catch do_start_appls([AppName], NS) of
{_ErrorTag, {not_running, App}} ->
NRequests = [{false, AppName, true, App} | PR],
S#state{p_reqs = NRequests};
{ok, NewS} ->
NewS;
{_ErrorTag, R} ->
throw({error, R})
end;
(_, S) ->
S
end, State, State#state.p_reqs)}.
req_del_permit_true(Reqs, Name) ->
filter(fun({From, Name2, true, _}) when Name2 == Name ->
gen_server:reply(From, ok),
false;
(_) ->
true
end, Reqs).
req_del_permit_false(Reqs, Name) ->
filter(fun({From, Name2, false, _Nodes}) when Name2 == Name ->
gen_server:reply(From, ok),
false;
(_) ->
true
end, Reqs).
req_del_node(S, Node, Appls) ->
check_waiting(S#state.p_reqs, S, Node, Appls, [], S#state.s_reqs).
del_t_reqs(AppName, TReqs, Res) ->
lists:filter(fun({AN, From}) when AppName == AN ->
gen_server:reply(From, Res),
false;
(_) ->
true
end,
TReqs).
check_waiting([{From, AppName, false, Nodes} | Reqs],
S, Node, Appls, Res, SReqs) ->
case lists:delete(Node, Nodes) of
[] ->
ac_stop_it(AppName),
NSReqs = [{AppName, From} | SReqs],
check_waiting(Reqs, Node, S, Appls, Res, NSReqs);
NNodes ->
check_waiting(Reqs, Node, S, Appls,
[{From, AppName, false, NNodes} | Res], SReqs)
end;
check_waiting([H | Reqs], S, Node, Appls, Res, SReqs) ->
check_waiting(Reqs, Node, S, Appls, [H | Res], SReqs);
check_waiting([], Node, S, Appls, Res, SReqs) ->
{Res, Appls, SReqs}.
intersection(Nodes1, Nodes2) ->
Nodes1 -- (Nodes1 -- Nodes2).
get_default_permission(AppName) ->
case application:get_env(kernel, permissions) of
{ok, Permissions} ->
case keysearch(AppName, 1, Permissions) of
{value, {_, true}} -> true;
{value, {_, false}} -> false;
{value, {_, X}} -> exit({bad_permission, {AppName, X}});
false -> true
end;
undefined -> true
end.
%%-----------------------------------------------------------------
%% ADT dist() - info on how an application is distributed
%% dist() = [{AppName, Time, DistNodes, [{Node, Runnable}]}]
%% Time = int() >= 0 | infinity
%% Nodes = [node() | {node()...}]
%% Runnable = true | false | undefined
%% An appl may not be started if any Runnable is undefined;
%% i.e. the appl must be loaded on all Nodes.
%%-----------------------------------------------------------------
dist_check([{AppName, Nodes} | T]) ->
P = get_default_permission(AppName),
[#appl{name = AppName, nodes = Nodes, run = [{node(), P}]} | dist_check(T)];
dist_check([{AppName, Time, Nodes} | T]) when integer(Time), Time >= 0 ->
P = get_default_permission(AppName),
[#appl{name = AppName, restart_time = Time, nodes = Nodes,
run = [{node(), P}]} | dist_check(T)];
dist_check([{AppName, infinity, Nodes} | T]) ->
P = get_default_permission(AppName),
[#appl{name = AppName, restart_time = infinity,
nodes = Nodes, run = [{node(), P}]} |
dist_check(T)];
dist_check([_ | T]) ->
dist_check(T);
dist_check([]) ->
[].
dist_take_control(Appls) ->
foreach(fun(#appl{name = AppName}) ->
application_controller:control_application(AppName)
end, Appls).
dist_replace(default, _Name, Appls) -> Appls;
dist_replace({AppName, Nodes}, AppName, Appls) ->
Run = map(fun(Node) -> {Node, undefined} end, flat_nodes(Nodes)),
keyreplaceadd(AppName, #appl.name, Appls,
#appl{name = AppName, restart_time = 0,
nodes = Nodes, run = Run});
dist_replace({AppName, Time, Nodes}, AppName, Appls)
when integer(Time), Time >= 0 ->
Run = map(fun(Node) -> {Node, undefined} end, flat_nodes(Nodes)),
keyreplaceadd(AppName, #appl.name, Appls,
#appl{name = AppName, restart_time = Time,
nodes = Nodes, run = Run});
dist_replace(Bad, _Name, _Appls) ->
throw({error, {bad_distribution_spec, Bad}}).
dist_update_run(Appls, AppName, Node, Permission) ->
map(fun(Appl) when Appl#appl.name == AppName ->
Run = Appl#appl.run,
NRun = keyreplaceadd(Node, 1, Run, {Node, Permission}),
Appl#appl{run = NRun};
(Appl) ->
Appl
end, Appls).
dist_change_update(Appls, []) ->
Appls;
dist_change_update(Appls, [{AppName, NewNodes} | NewDist]) ->
NewAppls = do_dist_change_update(Appls, AppName, 0, NewNodes),
dist_change_update(NewAppls, NewDist);
dist_change_update(Appls, [{AppName, NewTime, NewNodes} | NewDist]) ->
NewAppls = do_dist_change_update(Appls, AppName, NewTime, NewNodes),
dist_change_update(NewAppls, NewDist).
do_dist_change_update(Appls, AppName, NewTime, NewNodes) ->
map(fun(Appl) when Appl#appl.name == AppName ->
Appl#appl{restart_time = NewTime, nodes = NewNodes};
(Appl) ->
Appl
end, Appls).
%% Merge his Permissions with mine.
dist_merge(MyAppls, HisAppls, HisNode) ->
zf(fun(Appl) ->
#appl{name = AppName, nodes = Nodes, run = Run} = Appl,
% HeIsMember = lists:member(HisNode, flat_nodes(Nodes)),
HeIsMember = true,
case keysearch(AppName, #appl.name, HisAppls) of
{value, #appl{run = HisRun}} when HeIsMember == true ->
case keysearch(HisNode, 1, HisRun) of
{value, Val} -> % He has it loaded
NRun = keyreplaceadd(HisNode, 1, Run, Val),
{true, Appl#appl{run = NRun}};
false -> % He hasn't loaded it yet
Val = {HisNode, undefined},
{true, Appl#appl{run = [Val | Run]}}
end;
_ ->
true
end
end, MyAppls).
dist_get_runnable_nodes(Appls, AppName) ->
case keysearch(AppName, #appl.name, Appls) of
{value, #appl{run = Run}} ->
zf(fun({Node, true}) -> {true, Node};
(_) -> false
end, Run);
false ->
[]
end.
dist_is_runnable(Appls, AppName) ->
case keysearch(AppName, #appl.name, Appls) of
{value, #appl{run = Run}} ->
case keysearch(node(), 1, Run) of
{value, {_, true}} -> true;
_ -> false
end;
false ->
false
end.
is_loaded(AppName, #state{appls = Appls}) ->
case keysearch(AppName, #appl.name, Appls) of
{value, #appl{run = Run}} ->
case keysearch(node(), 1, Run) of
{value, {_Node, undefined}} -> false;
{value, _} -> true;
false -> false
end;
false ->
false
end.
dist_get_runnable(Appls) ->
zf(fun(#appl{name = AppName, run = Run}) ->
case keysearch(node(), 1, Run) of
{value, {_, true}} -> {true, AppName};
_ -> false
end
end, Appls).
dist_get_all_nodes(#appl{name = AppName, nodes = Nodes, run = Run}) ->
case check_nodes(Run, [], []) of
{Res, []} -> {ok, Nodes, Res};
{_Res, BadNodes} ->
{error, {app_not_loaded, AppName, BadNodes}}
end.
check_nodes([{Node, undefined} | T], Res, BadNodes) ->
check_nodes(T, Res, [Node | BadNodes]);
check_nodes([{Node, true} | T], Res, BadNodes) ->
check_nodes(T, [Node | Res], BadNodes);
check_nodes([{Node, false} | T], Res, BadNodes) ->
check_nodes(T, Res, BadNodes);
check_nodes([], Res, BadNodes) ->
{Res, BadNodes}.
-ifdef(NOTUSED).
dist_find_time([#appl{name = Name, restart_time = Time} |_], Name) -> Time;
dist_find_time([_ | T], Name) -> dist_find_time(T, Name);
dist_find_time([], Name) -> 0.
-endif.
%% Find all nodes that can run the app (even if they're not permitted
%% to right now).
dist_find_nodes([#appl{name = Name, nodes = Nodes} |_], Name) -> Nodes;
dist_find_nodes([_ | T], Name) -> dist_find_nodes(T, Name);
dist_find_nodes([], Name) -> [].
dist_flat_nodes(Appls, Name) ->
flat_nodes(dist_find_nodes(Appls, Name)).
dist_del_node(Appls, Node) ->
map(fun(Appl) ->
NRun = filter(fun({N, _Runnable}) when N == Node -> false;
(_) -> true
end, Appl#appl.run),
Appl#appl{run = NRun};
(X) ->
X
end, Appls).
validRestartType(permanent) -> true;
validRestartType(temporary) -> true;
validRestartType(transient) -> true;
validRestartType(RestartType) -> false.
%%% UW Patch 000310: This function is called from 3 different
%%% places (handle_info/2, do_wait_dist_start/7, and wait_dist_start2/6).
answer_weight(Name, Weight, Node, S) ->
case keysearch(Name, #appl.name, S#state.appls) of
{value, Appl} ->
Id = Appl#appl.id,
case Id of
run_waiting ->
{?DIST_AC, Node} ! {dist_ac_weight, Name, 0, node()},
S;
undefined ->
S#state{tmp_locals = [{Name, Weight, Node} |
S#state.tmp_locals]};
{takeover, _} ->
S#state{tmp_locals = [{Name, Weight, Node} |
S#state.tmp_locals]};
{failover, _} ->
S#state{tmp_locals = [{Name, Weight, Node} |
S#state.tmp_locals]};
_ ->
MyWeight = get_cached_weight(Name, S),
{?DIST_AC, Node} !
{dist_ac_weight, Name, MyWeight, node()},
NTWs = keyreplaceadd(Name, 1, S#state.tmp_weights,
{Name, MyWeight}),
S#state{tmp_weights = NTWs}
end;
_ ->
S#state{tmp_locals = [{Name, Weight, Node} | S#state.tmp_locals]}
end.
More information about the erlang-questions
mailing list