aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorAnders Svensson <[email protected]>2016-05-18 14:39:32 +0200
committerAnders Svensson <[email protected]>2016-05-30 08:56:15 +0200
commit5ca5fb71695ccdf7267007b8405f7a13497a2f17 (patch)
treedf6fb7c7a1fa9138da443e68c3df80bf789219ca
parent7d7271121b4d2e786bf6fb41508166efb96e5842 (diff)
downloadotp-5ca5fb71695ccdf7267007b8405f7a13497a2f17.tar.gz
otp-5ca5fb71695ccdf7267007b8405f7a13497a2f17.tar.bz2
otp-5ca5fb71695ccdf7267007b8405f7a13497a2f17.zip
Close listening sockets at transport removal
The transport interface documented in diameter_transport(3) is used to start/stop accepting/connecting transport processes: they're started with a function call, and told to die with their parent process. In the accepting case, both diameter_tcp and diameter_sctp start a listening process when the first accepting transport is started. However, there's no way for a listening process to find out that that it should stop listening when transport configuration is removed. Both diameter_tcp and diameter_sctp have used a timer to terminate the listening process after all existing accepting processes have died as a consequence of transport removal. The problem with this is that nothing stops a new client from connecting before this, and also that no new transport can succeed in opening the same listening port (eg. reconfiguration) until the old listener dies. This commit solves the problem by adding diameter_reg:subscribe/2, to allow callers to subscribe to messages about added/removed associations. A call to diameter:add_transport/2 results in a new child process that registers a term that a listening process subscribes to. Transport removal results in the death of the child, and the resulting notification to the listener causes the latter to close its socket and terminate. This is still an internal interface, but the subscription mechanism should probably be made external (eg. a diameter:subscribe/1 that can be used to subscribe to specified messages), so that transport modules other than diameter's own can make use of it. There is no support for soft upgrade.
-rw-r--r--lib/diameter/src/base/diameter_config.erl106
-rw-r--r--lib/diameter/src/base/diameter_config_sup.erl58
-rw-r--r--lib/diameter/src/base/diameter_sup.erl3
-rw-r--r--lib/diameter/src/modules.mk3
-rw-r--r--lib/diameter/src/transport/diameter_sctp.erl37
-rw-r--r--lib/diameter/src/transport/diameter_tcp.erl31
-rw-r--r--lib/diameter/test/diameter_transport_SUITE.erl6
7 files changed, 172 insertions, 72 deletions
diff --git a/lib/diameter/src/base/diameter_config.erl b/lib/diameter/src/base/diameter_config.erl
index 702f11593a..73b828536b 100644
--- a/lib/diameter/src/base/diameter_config.erl
+++ b/lib/diameter/src/base/diameter_config.erl
@@ -1,7 +1,7 @@
%%
%% %CopyrightBegin%
%%
-%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
@@ -45,10 +45,12 @@
add_transport/2,
remove_transport/2,
have_transport/2,
- lookup/1]).
+ lookup/1,
+ subscribe/2]).
-%% child server start
--export([start_link/0]).
+%% server start
+-export([start_link/0,
+ start_link/1]).
%% gen_server callbacks
-export([init/1,
@@ -58,8 +60,9 @@
handle_info/2,
code_change/3]).
-%% diameter_sync requests.
--export([sync/1]).
+%% callbacks
+-export([sync/1, %% diameter_sync requests
+ remove/0]). %% transport server termination
%% debug
-export([state/0,
@@ -77,6 +80,9 @@
%% Table config is written to.
-define(TABLE, ?MODULE).
+%% Key on which a transport-specific child registers itself.
+-define(TRANSPORT_KEY(Ref), {?MODULE, transport, Ref}).
+
%% Workaround for dialyzer's lack of understanding of match specs.
-type match(T)
:: T | '_' | '$1' | '$2' | '$3' | '$4'.
@@ -225,6 +231,13 @@ pred(_) ->
?THROW(pred).
%% --------------------------------------------------------------------------
+%% # subscribe/2
+%% --------------------------------------------------------------------------
+
+subscribe(Ref, T) ->
+ diameter_reg:subscribe(?TRANSPORT_KEY(Ref), T).
+
+%% --------------------------------------------------------------------------
%% # have_transport/2
%%
%% Output: true | false
@@ -264,6 +277,9 @@ start_link() ->
Options = [{spawn_opt, diameter_lib:spawn_opts(server, [])}],
gen_server:start_link(ServerName, Module, Args, Options).
+start_link(T) ->
+ proc_lib:start_link(?MODULE, init, [T], infinity, []).
+
state() ->
call(state).
@@ -274,8 +290,42 @@ uptime() ->
%%% # init/1
%%% ----------------------------------------------------------
+%% ?SERVER start.
init([]) ->
- {ok, #state{}}.
+ {ok, #state{}};
+
+%% Child start as a consequence of add_transport.
+init({SvcName, Type, Opts}) ->
+ Res = try
+ add(SvcName, Type, Opts)
+ catch
+ ?FAILURE(Reason) -> {error, Reason}
+ end,
+ proc_lib:init_ack({ok, self(), Res}),
+ sleep(Res).
+
+%% sleep/1
+
+sleep({ok, _}) ->
+ sleep();
+
+sleep({error, _}) ->
+ ok. %% die
+
+%% sleep/0
+
+sleep() ->
+ proc_lib:hibernate(?MODULE, remove, []).
+
+%% remove/0
+
+remove() ->
+ receive
+ {?MODULE, stop} ->
+ ok;
+ _ ->
+ sleep()
+ end.
%%% ----------------------------------------------------------
%%% # handle_call/2
@@ -404,19 +454,22 @@ sync({start_service, SvcName, Opts}) ->
sync({stop_service, SvcName}) ->
stop(SvcName);
+%% Start a child whose only purpose is to be alive for the lifetime of
+%% the transport configuration and publish itself in diameter_reg.
+%% This is to provide a way for processes to to be notified when the
+%% configuration is removed (diameter_reg:subscribe/2).
sync({add, SvcName, Type, Opts}) ->
- try
- add(SvcName, Type, Opts)
- catch
- ?FAILURE(Reason) -> {error, Reason}
- end;
+ {ok, _Pid, Res} = diameter_config_sup:start_child({SvcName, Type, Opts}),
+ Res;
sync({remove, SvcName, Pred}) ->
- remove(select([{#transport{service = '$1', _ = '_'},
+ Recs = select([{#transport{service = '$1', _ = '_'},
[{'=:=', '$1', {const, SvcName}}],
['$_']}]),
- SvcName,
- Pred).
+ F = fun(#transport{ref = R, type = T, options = O}) ->
+ Pred(R,T,O)
+ end,
+ remove(SvcName, lists:filter(F, Recs)).
%% start/3
@@ -503,6 +556,7 @@ add(SvcName, Type, Opts) ->
ok = transport_opts(Opts),
Ref = make_ref(),
+ true = diameter_reg:add_new(?TRANSPORT_KEY(Ref)),
T = {Ref, Type, Opts},
%% The call to the service returns error if the service isn't
%% started yet, which is harmless. The transport will be started
@@ -594,26 +648,30 @@ start_transport(SvcName, T) ->
No
end.
-%% remove/3
-
-remove(L, SvcName, Pred) ->
- rm(SvcName, lists:filter(fun(#transport{ref = R, type = T, options = O}) ->
- Pred(R,T,O)
- end,
- L)).
+%% remove/2
-rm(_, []) ->
+remove(_, []) ->
ok;
-rm(SvcName, L) ->
+
+remove(SvcName, L) ->
Refs = lists:map(fun(#transport{ref = R}) -> R end, L),
case stop_transport(SvcName, Refs) of
ok ->
+ lists:foreach(fun stop_child/1, Refs),
diameter_stats:flush(Refs),
lists:foreach(fun delete_object/1, L);
{error, _} = No ->
No
end.
+stop_child(Ref) ->
+ case diameter_reg:match(?TRANSPORT_KEY(Ref)) of
+ [{_, Pid}] -> %% tell the transport-specific child to die
+ Pid ! {?MODULE, stop};
+ [] -> %% already removed/dead
+ ok
+ end.
+
stop_transport(SvcName, Refs) ->
case diameter_service:stop_transport(SvcName, Refs) of
ok ->
diff --git a/lib/diameter/src/base/diameter_config_sup.erl b/lib/diameter/src/base/diameter_config_sup.erl
new file mode 100644
index 0000000000..9524573378
--- /dev/null
+++ b/lib/diameter/src/base/diameter_config_sup.erl
@@ -0,0 +1,58 @@
+%%
+%% %CopyrightBegin%
+%%
+%% Copyright Ericsson AB 2016. 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%
+%%
+
+%%
+%% Supervisor for config processes.
+%%
+
+-module(diameter_config_sup).
+
+-behaviour(supervisor).
+
+%% interface
+-export([start_link/0, %% supervisor start
+ start_child/1]). %% config start
+
+-export([init/1]).
+
+-define(NAME, ?MODULE). %% supervisor name
+
+%% start_link/0
+
+start_link() ->
+ SupName = {local, ?NAME},
+ supervisor:start_link(SupName, ?MODULE, []).
+
+%% start_child/1
+
+start_child(T) ->
+ supervisor:start_child(?NAME, [T]).
+
+%% init/1
+
+init([]) ->
+ Mod = diameter_config,
+ Flags = {simple_one_for_one, 0, 1},
+ ChildSpec = {Mod,
+ {Mod, start_link, []},
+ temporary,
+ 1000,
+ worker,
+ [Mod]},
+ {ok, {Flags, [ChildSpec]}}.
diff --git a/lib/diameter/src/base/diameter_sup.erl b/lib/diameter/src/base/diameter_sup.erl
index e89ede9843..482289cb9a 100644
--- a/lib/diameter/src/base/diameter_sup.erl
+++ b/lib/diameter/src/base/diameter_sup.erl
@@ -1,7 +1,7 @@
%%
%% %CopyrightBegin%
%%
-%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
@@ -34,6 +34,7 @@
-export([init/1]).
-define(CHILDREN, [diameter_misc_sup,
+ diameter_config_sup,
diameter_watchdog_sup,
diameter_peer_fsm_sup,
diameter_transport_sup,
diff --git a/lib/diameter/src/modules.mk b/lib/diameter/src/modules.mk
index 3b223ea391..4e4ce60ddf 100644
--- a/lib/diameter/src/modules.mk
+++ b/lib/diameter/src/modules.mk
@@ -1,7 +1,7 @@
# %CopyrightBegin%
#
-# Copyright Ericsson AB 2010-2015. All Rights Reserved.
+# Copyright Ericsson AB 2010-2016. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -37,6 +37,7 @@ RT_MODULES = \
base/diameter_callback \
base/diameter_capx \
base/diameter_config \
+ base/diameter_config_sup \
base/diameter_codec \
base/diameter_dict \
base/diameter_lib \
diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl
index 8a80ce630a..4a005b853d 100644
--- a/lib/diameter/src/transport/diameter_sctp.erl
+++ b/lib/diameter/src/transport/diameter_sctp.erl
@@ -1,7 +1,7 @@
%%
%% %CopyrightBegin%
%%
-%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
@@ -61,10 +61,6 @@
%% Remote addresses to accept connections from.
-define(DEFAULT_ACCEPT, []). %% any
-%% How long a listener with no associations lives before offing
-%% itself.
--define(LISTENER_TIMEOUT, 30000).
-
%% How long to wait for a transport process to attach after
%% association establishment.
-define(ACCEPT_TIMEOUT, 5000).
@@ -104,7 +100,6 @@
socket :: gen_sctp:sctp_socket(),
count = 0 :: uint(), %% attached transport processes
pending = {0, queue:new()},
- tref :: reference() | undefined,
accept :: [match()]}).
%% Field pending implements two queues: the first of transport-to-be
%% processes to which an association has been assigned but for which
@@ -216,14 +211,15 @@ init(T) ->
%% A process owning a listening socket.
i({listen, Ref, {Opts, Addrs}}) ->
+ [_] = diameter_config:subscribe(Ref, transport), %% assert existence
{[Matches], Rest} = proplists:split(Opts, [accept]),
{LAs, Sock} = AS = open(Addrs, Rest, ?DEFAULT_PORT),
ok = gen_sctp:listen(Sock, true),
true = diameter_reg:add_new({?MODULE, listener, {Ref, AS}}),
proc_lib:init_ack({ok, self(), LAs}),
- start_timer(#listener{ref = Ref,
- socket = Sock,
- accept = [[M] || {accept, M} <- Matches]});
+ #listener{ref = Ref,
+ socket = Sock,
+ accept = [[M] || {accept, M} <- Matches]};
%% A connecting transport.
i({connect, Pid, Opts, Addrs, Ref}) ->
@@ -431,13 +427,6 @@ putr(Key, Val) ->
getr(Key) ->
get({?MODULE, Key}).
-%% start_timer/1
-
-start_timer(#listener{count = 0} = S) ->
- S#listener{tref = erlang:start_timer(?LISTENER_TIMEOUT, self(), close)};
-start_timer(S) ->
- S.
-
%% l/2
%%
%% Transition listener state.
@@ -455,12 +444,10 @@ l({sctp, Sock, _RA, _RP, Data} = T, #listener{socket = Sock,
l({'DOWN', _MRef, process, TPid, _}, #listener{pending = {_,Q}} = S) ->
down(queue:member(TPid, Q), TPid, S);
-%% Timeout after the last accepting process has died.
-l({timeout, TRef, close = T}, #listener{tref = TRef,
- count = 0}) ->
- x(T);
-l({timeout, _, close}, #listener{} = S) ->
- S.
+%% Transport has been removed.
+l({transport, remove, _} = T, #listener{socket = Sock}) ->
+ gen_sctp:close(Sock),
+ x(T).
%% down/3
%%
@@ -472,15 +459,15 @@ down(true, TPid, #listener{pending = {N,Q},
= S) ->
NQ = queue:filter(fun(P) -> P /= TPid end, Q),
if N < 0 -> %% awaiting an association ...
- start_timer(S#listener{count = K-1,
- pending = {N+1, NQ}});
+ S#listener{count = K-1,
+ pending = {N+1, NQ}};
true -> %% ... or one has been assigned
S#listener{pending = {N-1, NQ}}
end;
%% ... or one that's already attached.
down(false, _TPid, #listener{count = K} = S) ->
- start_timer(S#listener{count = K-1}).
+ S#listener{count = K-1}.
%% t/2
%%
diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl
index 6a5e5fe89d..546c2cfa5e 100644
--- a/lib/diameter/src/transport/diameter_tcp.erl
+++ b/lib/diameter/src/transport/diameter_tcp.erl
@@ -57,7 +57,6 @@
-define(ERROR(T), erlang:error({T, ?MODULE, ?LINE})).
-define(DEFAULT_PORT, 3868). %% RFC 3588, ch 2.1
--define(LISTENER_TIMEOUT, 30000).
-define(DEFAULT_FRAGMENT_TIMEOUT, 1000).
-define(IS_UINT32(N), (is_integer(N) andalso 0 =< N andalso 0 == N bsr 32)).
@@ -73,8 +72,10 @@
%% Listener process state.
-record(listener, {socket :: inet:socket(),
- count = 1 :: non_neg_integer(),
- tref :: reference() | undefined}).
+ count = 1 :: non_neg_integer()}). %% accepting processes
+%% The count of accepting processes was previously used to terminate
+%% the listening process, but diameter_reg:subscribe/2 is now used for
+%% this. Leave the the count for trace purposes.
%% Monitor process state.
-record(monitor,
@@ -240,6 +241,7 @@ i(#monitor{parent = Pid, transport = TPid} = S) ->
%% gen_tcp seems to so. Links should be left to supervisors.
i({listen, LRef, APid, {Mod, Opts, Addrs}}) ->
+ [_] = diameter_config:subscribe(LRef, transport), %% assert existence
{[LA, LP], Rest} = proplists:split(Opts, [ip, port]),
LAddrOpt = get_addr(LA, Addrs),
LPort = get_port(LP),
@@ -248,7 +250,7 @@ i({listen, LRef, APid, {Mod, Opts, Addrs}}) ->
true = diameter_reg:add_new({?MODULE, listener, {LRef, {LAddr, LSock}}}),
proc_lib:init_ack({ok, self(), {LAddr, LSock}}),
monitor(process, APid),
- start_timer(#listener{socket = LSock}).
+ #listener{socket = LSock}.
laddr([], Mod, Sock) ->
{ok, {Addr, _Port}} = sockname(Mod, Sock),
@@ -484,13 +486,6 @@ putr(Key, Val) ->
getr(Key) ->
get({?MODULE, Key}).
-%% start_timer/1
-
-start_timer(#listener{count = 0} = S) ->
- S#listener{tref = erlang:start_timer(?LISTENER_TIMEOUT, self(), close)};
-start_timer(S) ->
- S.
-
%% m/2
%%
%% Transition monitor state.
@@ -512,21 +507,19 @@ m({'DOWN', _, process, Pid, _}, #monitor{parent = Pid,
%%
%% Transition listener state.
-%% Another accept transport is attaching.
+%% An accepting transport is attaching.
l({accept, TPid}, #listener{count = N} = S) ->
monitor(process, TPid),
S#listener{count = N+1};
%% Accepting process has died.
l({'DOWN', _, process, _, _}, #listener{count = N} = S) ->
- start_timer(S#listener{count = N-1});
+ S#listener{count = N-1};
-%% Timeout after the last accepting process has died.
-l({timeout, TRef, close = T}, #listener{tref = TRef,
- count = 0}) ->
- x(T);
-l({timeout, _, close}, #listener{} = S) ->
- S.
+%% Transport has been removed.
+l({transport, remove, _} = T, #listener{socket = Sock}) ->
+ gen_tcp:close(Sock),
+ x(T).
%% t/2
%%
diff --git a/lib/diameter/test/diameter_transport_SUITE.erl b/lib/diameter/test/diameter_transport_SUITE.erl
index 53d2d6660e..f2884eacb4 100644
--- a/lib/diameter/test/diameter_transport_SUITE.erl
+++ b/lib/diameter/test/diameter_transport_SUITE.erl
@@ -1,7 +1,7 @@
%%
%% %CopyrightBegin%
%%
-%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
@@ -140,7 +140,9 @@ sctp_accept(Config) ->
-define(PEER_COUNT, 8).
accept(Prot) ->
- T = {Prot, make_ref()},
+ Ref = make_ref(),
+ true = diameter_reg:add_new({diameter_config, transport, Ref}), %% fake it
+ T = {Prot, Ref},
[] = ?util:run(?util:scramble(acc(2*?PEER_COUNT, T, []))).
acc(0, _, Acc) ->