aboutsummaryrefslogtreecommitdiffstats
path: root/lib/diameter/src
diff options
context:
space:
mode:
Diffstat (limited to 'lib/diameter/src')
-rw-r--r--lib/diameter/src/base/diameter.erl9
-rw-r--r--lib/diameter/src/base/diameter_config.erl25
-rw-r--r--lib/diameter/src/base/diameter_peer_fsm.erl38
-rw-r--r--lib/diameter/src/base/diameter_service.erl16
-rw-r--r--lib/diameter/src/base/diameter_watchdog.erl99
5 files changed, 147 insertions, 40 deletions
diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl
index d60510db7d..3e3a6be0ef 100644
--- a/lib/diameter/src/base/diameter.erl
+++ b/lib/diameter/src/base/diameter.erl
@@ -44,6 +44,7 @@
stop/0]).
-export_type([evaluable/0,
+ restriction/0,
sequence/0,
app_alias/0,
service_name/0,
@@ -284,11 +285,19 @@ call(SvcName, App, Message) ->
-type sequence()
:: {'Unsigned32'(), 0..32}.
+-type restriction()
+ :: false
+ | node
+ | nodes
+ | [node()]
+ | evaluable().
+
%% Options passed to start_service/2
-type service_opt()
:: capability()
| {application, [application_opt()]}
+ | {restrict_connections, restriction()}
| {sequence, sequence() | evaluable()}.
-type application_opt()
diff --git a/lib/diameter/src/base/diameter_config.erl b/lib/diameter/src/base/diameter_config.erl
index fc37ca8541..63d28f25a2 100644
--- a/lib/diameter/src/base/diameter_config.erl
+++ b/lib/diameter/src/base/diameter_config.erl
@@ -555,7 +555,8 @@ make_config(SvcName, Opts) ->
Os = split(Opts, fun opt/2, [{false, share_peers},
{false, use_shared_peers},
{false, monitor},
- {?NOMASK, sequence}]),
+ {?NOMASK, sequence},
+ {nodes, restrict_connections}]),
%% share_peers and use_shared_peers are currently undocumented.
#service{name = SvcName,
@@ -579,15 +580,33 @@ opt(monitor, P)
when is_pid(P) ->
P;
+opt(restrict_connections, T)
+ when T == node;
+ T == nodes;
+ T == [];
+ is_atom(hd(T)) ->
+ T;
+
+opt(restrict_connections = K, F) ->
+ try diameter_lib:eval(F) of %% no guarantee that it won't fail later
+ Nodes when is_list(Nodes) ->
+ F;
+ V ->
+ ?THROW({value, {K,V}})
+ catch
+ E:R ->
+ ?THROW({value, {K, E, R, ?STACK}})
+ end;
+
opt(sequence, {_,_} = T) ->
sequence(T);
-opt(sequence, F) ->
+opt(sequence = K, F) ->
try diameter_lib:eval(F) of
T -> sequence(T)
catch
E:R ->
- ?THROW({value, {sequence, E, R, ?STACK}})
+ ?THROW({value, {K, E, R, ?STACK}})
end;
opt(K, _) ->
diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index 8ce6ea847a..297a5d7709 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -60,6 +60,7 @@
-define(Q_KEY, q). %% transport start queue
-define(START_KEY, start). %% start of connected transport
-define(SEQUENCE_KEY, mask). %% mask for sequence numbers
+-define(RESTRICT_KEY, restrict). %% nodes for connection check
%% The default sequence mask.
-define(NOMASK, {0,32}).
@@ -126,7 +127,9 @@
%%% ---------------------------------------------------------------------------
-spec start(T, [Opt], #diameter_service{} %% from old code
- | {diameter:sequence(), #diameter_service{}})
+ | {diameter:sequence(),
+ diameter:restriction(),
+ #diameter_service{}})
-> pid()
when T :: {connect|accept, diameter:transport_ref()},
Opt :: diameter:transport_opt().
@@ -157,11 +160,11 @@ init(T) ->
gen_server:enter_loop(?MODULE, [], i(T)).
i({WPid, Type, Opts, #diameter_service{} = Svc}) -> %% from old code
- i({WPid, Type, Opts, {?NOMASK, Svc}});
+ i({WPid, Type, Opts, {?NOMASK, [node() | nodes()], Svc}});
-i({WPid, T, Opts, {Mask, #diameter_service{applications = Apps,
- capabilities = Caps}
- = Svc}}) ->
+i({WPid, T, Opts, {Mask, Nodes, #diameter_service{applications = Apps,
+ capabilities = Caps}
+ = Svc}}) ->
[] /= Apps orelse ?ERROR({no_apps, T, Opts}),
putr(?DWA_KEY, dwa(Caps)),
{M, Ref} = T,
@@ -169,6 +172,7 @@ i({WPid, T, Opts, {Mask, #diameter_service{applications = Apps,
{[Ts], Rest} = proplists:split(Opts, [capabilities_cb]),
putr(?CB_KEY, {Ref, [F || {_,F} <- Ts]}),
putr(?SEQUENCE_KEY, Mask),
+ putr(?RESTRICT_KEY, Nodes),
erlang:monitor(process, WPid),
{TPid, Addrs} = start_transport(T, Rest, Svc),
#state{parent = WPid,
@@ -990,15 +994,31 @@ dpa_timer() ->
%% Register a term and ensure it's not registered elsewhere. Note that
%% two process that simultaneously register the same term may well
%% both fail to do so this isn't foolproof.
+%%
+%% Everywhere is no longer everywhere, it's where a
+%% restrict_connections service_opt() specifies.
register_everywhere(T) ->
- diameter_reg:add_new(T)
- andalso unregistered(T).
+ reg(getr(?RESTRICT_KEY), T).
+
+reg(Nodes, T) ->
+ add(lists:member(node(), Nodes), T) andalso unregistered(Nodes, T).
+
+add(true, T) ->
+ diameter_reg:add_new(T);
+add(false, T) ->
+ diameter_reg:add(T).
-unregistered(T) ->
- {ResL, _} = rpc:multicall(?MODULE, match, [{node(), T}]),
+%% unregistered
+%%
+%% Ensure that the term in question isn't registered on other nodes.
+
+unregistered(Nodes, T) ->
+ {ResL, _} = rpc:multicall(Nodes, ?MODULE, match, [{node(), T}]),
lists:all(fun(L) -> [] == L end, ResL).
+%% match/1
+
match({Node, _})
when Node == node() ->
[];
diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl
index 91e7cbd996..b4e54cc9f9 100644
--- a/lib/diameter/src/base/diameter_service.erl
+++ b/lib/diameter/src/base/diameter_service.erl
@@ -110,6 +110,9 @@
%% The default sequence mask.
-define(NOMASK, {0,32}).
+%% The default restrict_connections.
+-define(RESTRICT, nodes).
+
%% Workaround for dialyzer's lack of understanding of match specs.
-type match(T)
:: T | '_' | '$1' | '$2' | '$3' | '$4'.
@@ -126,6 +129,7 @@
monitor = false :: false | pid(), %% process to die with
options
:: [{sequence, diameter:sequence()} %% sequence mask
+ | {restrict_connections, diameter:restriction()}
| {share_peers, boolean()} %% broadcast peers to remote nodes?
| {use_shared_peers, boolean()}]}).%% use broadcasted peers?
%% shared_peers reflects the peers broadcast from remote nodes. Note
@@ -500,6 +504,10 @@ handle_call(stop, _From, S) ->
handle_call(sequence, _From, #state{options = [{_, Mask} | _]} = S) ->
{reply, Mask, S};
+%% Watchdog is asking for the nodes restriction.
+handle_call(restriction, _From, #state{options = [_,_,_,{_,R} | _]} = S) ->
+ {reply, R, S};
+
handle_call(Req, From, S) ->
unexpected(handle_call, [Req, From], S),
{reply, nok, S}.
@@ -656,7 +664,8 @@ upgrade({state, Id, Svc, Name, Svc, PT, CT, SB, UB, SD, LD, MPid}) ->
monitor = MPid,
options = [{sequence, ?NOMASK},
{share_peers, SB},
- {use_shared_peers, UB}]},
+ {use_shared_peers, UB},
+ {restrict_connections, ?RESTRICT}]},
upgrade_insert(S),
S.
@@ -867,7 +876,10 @@ cfg_acc({_Ref, Type, _Opts} = T, {S, Acc})
service_options(Opts) ->
[{sequence, proplists:get_value(sequence, Opts, ?NOMASK)},
{share_peers, get_value(share_peers, Opts)},
- {use_shared_peers, get_value(use_shared_peers, Opts)}].
+ {use_shared_peers, get_value(use_shared_peers, Opts)},
+ {restrict_connections, proplists:get_value(restrict_connections,
+ Opts,
+ ?RESTRICT)}].
%% The order of options is significant since we match against the list.
mref(false = No) ->
diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl
index b37a1a10e9..d814f1afe2 100644
--- a/lib/diameter/src/base/diameter_watchdog.erl
+++ b/lib/diameter/src/base/diameter_watchdog.erl
@@ -57,8 +57,9 @@
parent = self() :: pid(),
transport :: pid() | undefined,
tref :: reference(), %% reference for current watchdog timer
- message_data, %% term passed into diameter_service with message
- sequence :: diameter:sequence()}). %% mask
+ message_data, %% term passed into diameter_service with message
+ sequence :: diameter:sequence(), %% mask
+ restrict :: {diameter:restriction(), boolean()}}).
%% start/2
%%
@@ -121,15 +122,18 @@ make_state({T, Pid, {RecvData,
putr(restart, {T, Opts, Svc}), %% save seeing it in trace
putr(dwr, dwr(Caps)), %%
{_,_} = Mask = call(Pid, sequence),
+ Restrict = call(Pid, restriction),
+ Nodes = restrict_nodes(Restrict),
#watchdog{parent = Pid,
transport = monitor(diameter_peer_fsm:start(T,
- Opts,
- {Mask, Svc})),
+ Opts,
+ {Mask, Nodes, Svc})),
tw = proplists:get_value(watchdog_timer,
Opts,
?DEFAULT_TW_INIT),
message_data = {RecvData, SvcName, Apps, Mask},
- sequence = Mask}.
+ sequence = Mask,
+ restrict = {Restrict, lists:member(node(), Nodes)}}.
%% Retrieve the sequence mask from the parent from the parent, rather
%% than having it passed into init/1, for upgrade reasons: the call to
@@ -160,9 +164,11 @@ handle_info(T, #watchdog{} = State) ->
{stop, {shutdown, T}, State}
end;
-handle_info(T, S) -> %% upgrade
- handle_info(T, #watchdog{} = list_to_tuple(tuple_to_list(S)
- ++ [?NOMASK])).
+handle_info(T, S) ->
+ handle_info(T, upgrade(S)).
+
+upgrade(S) ->
+ #watchdog{} = list_to_tuple(tuple_to_list(S) ++ [?NOMASK, {nodes, true}]).
event(#watchdog{status = T}, #watchdog{status = T}) ->
ok;
@@ -255,9 +261,10 @@ transition({close, TPid, _Reason}, #watchdog{transport = TPid}) ->
transition({open, TPid, Hosts, T} = Open,
#watchdog{transport = TPid,
status = initial,
- parent = Pid}
+ parent = Pid,
+ restrict = {_, R}}
= S) ->
- case okay(getr(restart), Hosts) of
+ case okay(getr(restart), Hosts, R) of
okay ->
open(Pid, {TPid, T}),
set_watchdog(S#watchdog{status = okay});
@@ -363,20 +370,24 @@ encode(Msg, Mask) ->
#diameter_packet{bin = Bin} = diameter_codec:encode(?BASE, Pkt),
Bin.
-%% okay/2
+%% okay/3
-okay({{accept, Ref}, _, _}, Hosts) ->
+okay({{accept, Ref}, _, _}, Hosts, Restrict) ->
T = {?MODULE, connection, Ref, Hosts},
diameter_reg:add(T),
- okay(diameter_reg:match(T));
+ if Restrict ->
+ okay(diameter_reg:match(T));
+ true ->
+ okay
+ end;
%% Register before matching so that at least one of two registering
-%% processes will match the other. (Which can't happen as long as
-%% diameter_peer_fsm guarantees at most one open connection to the same
-%% peer.)
+%% processes will match the other.
-okay({{connect, _}, _, _}, _) ->
+okay({{connect, _}, _, _}, _, _) ->
okay.
+%% okay/2
+
%% The peer hasn't been connected recently ...
okay([{_,P}]) ->
P = self(), %% assert
@@ -633,23 +644,40 @@ restart(#watchdog{transport = undefined} = S) ->
restart(S) ->
S.
+%% restart/2
+%%
%% Only restart the transport in the connecting case. For an accepting
-%% transport, we've registered the peer connection when leaving state
-%% initial and this is used by a new accepting process to realize that
-%% it's actually in state down rather then initial when receiving
-%% notification of an open connection.
+%% transport, there's no guarantee that an accepted connection in a
+%% restarted transport if from the peer we've lost contact with so
+%% have to be prepared for another watchdog to handle it. This is what
+%% the diameter_reg registration in this module is for: the peer
+%% connection is registered when leaving state initial and this is
+%% used by a new accepting watchdog to realize that it's actually in
+%% state down rather then initial when receiving notification of an
+%% open connection.
restart({{connect, _} = T, Opts, Svc}, #watchdog{parent = Pid,
- sequence = Mask}
+ sequence = Mask,
+ restrict = {R,_}}
= S) ->
Pid ! {reconnect, self()},
+ Nodes = restrict_nodes(R),
S#watchdog{transport = monitor(diameter_peer_fsm:start(T,
- Opts,
- {Mask, Svc}))};
+ Opts,
+ {Mask, Nodes, Svc})),
+ restrict = {R, lists:member(node(), Nodes)}};
+
+%% No restriction on the number of connections to the same peer: just
+%% die. Note that a state machine never enters state REOPEN in this
+%% case.
+restart({{accept, _}, _, _}, #watchdog{restrict = {_, false}}) ->
+ stop;
+
+%% Otherwise hang around until told to die.
restart({{accept, _}, _, _}, S) ->
S.
-%% Don't currently use Opts/Svc in the accept case but having them in
-%% the process dictionary is helpful if the process dies unexpectedly.
+
+%% Don't currently use Opts/Svc in the accept case.
%% dwr/1
@@ -659,3 +687,22 @@ dwr(#diameter_caps{origin_host = OH,
['DWR', {'Origin-Host', OH},
{'Origin-Realm', OR},
{'Origin-State-Id', OSI}].
+
+%% restrict_nodes/1
+
+restrict_nodes(false) ->
+ [];
+
+restrict_nodes(nodes) ->
+ [node() | nodes()];
+
+restrict_nodes(node) ->
+ [node()];
+
+restrict_nodes(Nodes)
+ when [] == Nodes;
+ is_atom(hd(Nodes)) ->
+ Nodes;
+
+restrict_nodes(F) ->
+ diameter_lib:eval(F).
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-08 15:58:24 +0000