aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorAnders Svensson <[email protected]>2016-08-06 00:55:20 +0200
committerAnders Svensson <[email protected]>2016-08-26 01:20:41 +0200
commit2afd1fe5fca3082193ff555fce8c62a31a5ea83f (patch)
tree23c2d16ea6d631e096d19a6434114ccc769abfe5
parent010925dd108b560ed86d9e0d8f48beefd677728e (diff)
downloadotp-2afd1fe5fca3082193ff555fce8c62a31a5ea83f.tar.gz
otp-2afd1fe5fca3082193ff555fce8c62a31a5ea83f.tar.bz2
otp-2afd1fe5fca3082193ff555fce8c62a31a5ea83f.zip
Close listening sockets at service death
Commit 5ca5fb71 ensured that they were closed immediately at transport removal, but in so doing broke their closing at stop service completely, by removing the timer that caused sockets to be closed even belatedly. Monitor on the service process to make it happen. This could still be improved, since stop_service listening ports aren't closed until after the service process has died. They could be closed earlier in the case of stop_service.
-rw-r--r--lib/diameter/src/transport/diameter_sctp.erl76
-rw-r--r--lib/diameter/src/transport/diameter_tcp.erl104
2 files changed, 108 insertions, 72 deletions
diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl
index 4a005b853d..f48e4347ee 100644
--- a/lib/diameter/src/transport/diameter_sctp.erl
+++ b/lib/diameter/src/transport/diameter_sctp.erl
@@ -98,7 +98,7 @@
-record(listener,
{ref :: reference(),
socket :: gen_sctp:sctp_socket(),
- count = 0 :: uint(), %% attached transport processes
+ service = false :: false | pid(), %% service process
pending = {0, queue:new()},
accept :: [match()]}).
%% Field pending implements two queues: the first of transport-to-be
@@ -129,11 +129,14 @@
-> {ok, pid(), [inet:ip_address()]}
when Ref :: diameter:transport_ref().
-start(T, #diameter_service{capabilities = Caps}, Opts)
+start(T, Svc, Opts)
when is_list(Opts) ->
+ #diameter_service{capabilities = Caps,
+ pid = SPid}
+ = Svc,
diameter_sctp_sup:start(), %% start supervisors on demand
Addrs = Caps#diameter_caps.host_ip_address,
- s(T, Addrs, lists:map(fun ip/1, Opts)).
+ s(T, Addrs, SPid, lists:map(fun ip/1, Opts)).
ip({ifaddr, A}) ->
{ip, A};
@@ -144,18 +147,22 @@ ip(T) ->
%% when there is not yet an association to assign it, or at comm_up on
%% a new association in which case the call retrieves a transport from
%% the pending queue.
-s({accept, Ref} = A, Addrs, Opts) ->
- {LPid, LAs} = listener(Ref, {Opts, Addrs}),
- try gen_server:call(LPid, {A, self()}, infinity) of
- {ok, TPid} -> {ok, TPid, LAs}
+s({accept, Ref} = A, Addrs, SPid, Opts) ->
+ {ok, LPid, LAs} = listener(Ref, {Opts, Addrs}),
+ try gen_server:call(LPid, {A, self(), SPid}, infinity) of
+ {ok, TPid} ->
+ {ok, TPid, LAs};
+ No ->
+ {error, No}
catch
- exit: Reason -> {error, Reason}
+ exit: Reason ->
+ {error, Reason}
end;
%% This implementation is due to there being no accept call in
%% gen_sctp in order to be able to accept a new association only
%% *after* an accepting transport has been spawned.
-s({connect = C, Ref}, Addrs, Opts) ->
+s({connect = C, Ref}, Addrs, _SPid, Opts) ->
diameter_sctp_sup:start_child({C, self(), Opts, Addrs, Ref}).
%% start_link/1
@@ -281,24 +288,23 @@ i({K, Ref}, #transport{mode = {accept, _}} = S) ->
%% Accepting processes can be started concurrently: ensure only one
%% listener is started.
-listener(LRef, T) ->
- diameter_sync:call({?MODULE, listener, LRef},
- {?MODULE, listener, [{LRef, T}]},
+listener(Ref, T) ->
+ diameter_sync:call({?MODULE, listener, Ref},
+ {?MODULE, listener, [{Ref, T}]},
infinity,
infinity).
-listener({LRef, T}) ->
- l(diameter_reg:match({?MODULE, listener, {LRef, '_'}}), LRef, T).
+listener({Ref, T}) ->
+ l(diameter_reg:match({?MODULE, listener, {Ref, '_'}}), Ref, T).
%% Existing listening process ...
l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) ->
- {LAs, _Sock} = AS,
- {LPid, LAs};
+ {LAs, _Sock} = AS,
+ {ok, LPid, LAs};
%% ... or not.
-l([], LRef, T) ->
- {ok, LPid, LAs} = diameter_sctp_sup:start_child({listen, LRef, T}),
- {LPid, LAs}.
+l([], Ref, T) ->
+ diameter_sctp_sup:start_child({listen, Ref, T}).
%% open/3
@@ -364,11 +370,17 @@ type(T) ->
%% # handle_call/3
%% ---------------------------------------------------------------------------
-handle_call({{accept, Ref}, Pid}, _, #listener{ref = Ref,
- count = K}
- = S) ->
+handle_call({{accept, Ref}, Pid}, _, #listener{ref = Ref} = S) ->
{TPid, NewS} = accept(Ref, Pid, S),
- {reply, {ok, TPid}, NewS#listener{count = K+1}};
+ {reply, {ok, TPid}, NewS};
+
+handle_call({{accept, _} = T, Pid, SPid}, From, #listener{service = P} = S) ->
+ handle_call({T, Pid}, From, if not is_pid(P), is_pid(SPid) ->
+ monitor(process, SPid),
+ S#listener{service = SPid};
+ true ->
+ S
+ end);
handle_call(_, _, State) ->
{reply, nok, State}.
@@ -441,6 +453,13 @@ l({sctp, Sock, _RA, _RP, Data} = T, #listener{socket = Sock,
setopts(Sock),
NewS;
+%% Service process has died.
+l({'DOWN', _, process, Pid, _} = T, #listener{service = Pid,
+ socket = Sock}) ->
+ gen_sctp:close(Sock),
+ x(T);
+
+%% Accepting process has died.
l({'DOWN', _MRef, process, TPid, _}, #listener{pending = {_,Q}} = S) ->
down(queue:member(TPid, Q), TPid, S);
@@ -454,20 +473,17 @@ l({transport, remove, _} = T, #listener{socket = Sock}) ->
%% Accepting transport has died.
%% One that's waiting for transport start in the pending queue ...
-down(true, TPid, #listener{pending = {N,Q},
- count = K}
- = S) ->
+down(true, TPid, #listener{pending = {N,Q}} = S) ->
NQ = queue:filter(fun(P) -> P /= TPid end, Q),
if N < 0 -> %% awaiting an association ...
- S#listener{count = K-1,
- pending = {N+1, NQ}};
+ S#listener{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) ->
- S#listener{count = K-1}.
+down(false, _TPid, S) ->
+ S.
%% t/2
%%
diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl
index 546c2cfa5e..44abc5c3b4 100644
--- a/lib/diameter/src/transport/diameter_tcp.erl
+++ b/lib/diameter/src/transport/diameter_tcp.erl
@@ -71,11 +71,8 @@
%% a process owning the listening port.
%% Listener process state.
--record(listener, {socket :: inet:socket(),
- 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.
+-record(listener, {socket :: inet:socket(),
+ service = false :: false | pid()}). %% service process
%% Monitor process state.
-record(monitor,
@@ -138,11 +135,15 @@
| {ok, pid()}
when Ref :: diameter:transport_ref().
-start({T, Ref}, #diameter_service{capabilities = Caps}, Opts) ->
+start({T, Ref}, Svc, Opts) ->
+ #diameter_service{capabilities = Caps,
+ pid = SPid}
+ = Svc,
+
diameter_tcp_sup:start(), %% start tcp supervisors on demand
{Mod, Rest} = split(Opts),
Addrs = Caps#diameter_caps.host_ip_address,
- Arg = {T, Ref, Mod, self(), Rest, Addrs},
+ Arg = {T, Ref, Mod, self(), Rest, Addrs, SPid},
diameter_tcp_sup:start_child(Arg).
split([{module, M} | Opts]) ->
@@ -196,7 +197,7 @@ init(T) ->
%% i/1
%% A transport process.
-i({T, Ref, Mod, Pid, Opts, Addrs})
+i({T, Ref, Mod, Pid, Opts, Addrs, SPid})
when T == accept;
T == connect ->
monitor(process, Pid),
@@ -214,7 +215,7 @@ i({T, Ref, Mod, Pid, Opts, Addrs})
?DEFAULT_FRAGMENT_TIMEOUT),
?IS_TIMEOUT(Tmo) orelse ?ERROR({fragment_timer, Tmo}),
Throttle = proplists:get_value(throttle_cb, OwnOpts, false),
- Sock = init(T, Ref, Mod, Pid, SslOpts, Rest, Addrs),
+ Sock = init(T, Ref, Mod, Pid, SslOpts, Rest, Addrs, SPid),
MPid ! {stop, self()}, %% tell the monitor to die
M = if SslOpts -> ssl; true -> Mod end,
putr(?REF_KEY, Ref),
@@ -228,6 +229,11 @@ i({T, Ref, Mod, Pid, Opts, Addrs})
%% Put the reference in the process dictionary since we now use it
%% advertise the ssl socket after TLS upgrade.
+i({T, _Ref, _Mod, _Pid, _Opts, _Addrs} = Arg) %% from old code
+ when T == accept;
+ T == connect ->
+ i(erlang:append_element(Arg, _SPid = false));
+
%% A monitor process to kill the transport if the parent dies.
i(#monitor{parent = Pid, transport = TPid} = S) ->
proc_lib:init_ack({ok, self()}),
@@ -240,16 +246,18 @@ i(#monitor{parent = Pid, transport = TPid} = S) ->
%% death. However, a link can be unlinked and this is exactly what
%% 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
+i({listen = L, Ref, _APid, T}) -> %% from old code
+ i({L, Ref, T});
+
+i({listen, Ref, {Mod, Opts, Addrs}}) ->
+ [_] = diameter_config:subscribe(Ref, transport), %% assert existence
{[LA, LP], Rest} = proplists:split(Opts, [ip, port]),
LAddrOpt = get_addr(LA, Addrs),
LPort = get_port(LP),
{ok, LSock} = Mod:listen(LPort, gen_opts(LAddrOpt, Rest)),
LAddr = laddr(LAddrOpt, Mod, LSock),
- true = diameter_reg:add_new({?MODULE, listener, {LRef, {LAddr, LSock}}}),
+ true = diameter_reg:add_new({?MODULE, listener, {Ref, {LAddr, LSock}}}),
proc_lib:init_ack({ok, self(), {LAddr, LSock}}),
- monitor(process, APid),
#listener{socket = LSock}.
laddr([], Mod, Sock) ->
@@ -268,21 +276,22 @@ ssl_opts([{ssl_options, Opts}])
ssl_opts(T) ->
?ERROR({ssl_options, T}).
-%% init/7
+%% init/8
%% Establish a TLS connection before capabilities exchange ...
-init(Type, Ref, Mod, Pid, true, Opts, Addrs) ->
- init(Type, Ref, ssl, Pid, [{cb_info, ?TCP_CB(Mod)} | Opts], Addrs);
+init(Type, Ref, Mod, Pid, true, Opts, Addrs, SPid) ->
+ init(Type, Ref, ssl, Pid, [{cb_info, ?TCP_CB(Mod)} | Opts], Addrs, SPid);
%% ... or not.
-init(Type, Ref, Mod, Pid, _, Opts, Addrs) ->
- init(Type, Ref, Mod, Pid, Opts, Addrs).
+init(Type, Ref, Mod, Pid, _, Opts, Addrs, SPid) ->
+ init(Type, Ref, Mod, Pid, Opts, Addrs, SPid).
-%% init/6
+%% init/7
-init(accept = T, Ref, Mod, Pid, Opts, Addrs) ->
+init(accept = T, Ref, Mod, Pid, Opts, Addrs, SPid) ->
{[Matches], Rest} = proplists:split(Opts, [accept]),
- {LAddr, LSock} = listener(Ref, {Mod, Rest, Addrs}),
+ {ok, LPid, {LAddr, LSock}} = listener(Ref, {Mod, Rest, Addrs}),
+ ok = gen_server:call(LPid, {accept, SPid}, infinity),
proc_lib:init_ack({ok, self(), [LAddr]}),
Sock = ok(accept(Mod, LSock)),
ok = accept_peer(Mod, Sock, accept(Matches)),
@@ -290,7 +299,7 @@ init(accept = T, Ref, Mod, Pid, Opts, Addrs) ->
diameter_peer:up(Pid),
Sock;
-init(connect = T, Ref, Mod, Pid, Opts, Addrs) ->
+init(connect = T, Ref, Mod, Pid, Opts, Addrs, _SPid) ->
{[LA, RA, RP], Rest} = proplists:split(Opts, [ip, raddr, rport]),
LAddrOpt = get_addr(LA, Addrs),
RAddr = get_addr(RA),
@@ -344,24 +353,26 @@ accept(Opts) ->
%% Accepting processes can be started concurrently: ensure only one
%% listener is started.
-listener(LRef, T) ->
- diameter_sync:call({?MODULE, listener, LRef},
- {?MODULE, listener, [{LRef, T, self()}]},
+listener(Ref, T) ->
+ diameter_sync:call({?MODULE, listener, Ref},
+ {?MODULE, listener, [{Ref, T, self()}]},
infinity,
infinity).
-listener({LRef, T, TPid}) ->
- l(diameter_reg:match({?MODULE, listener, {LRef, '_'}}), LRef, T, TPid).
+%% listener/1
+
+listener({Ref, T, _TPid}) ->
+ l(diameter_reg:match({?MODULE, listener, {Ref, '_'}}), Ref, T).
+
+%% l/3
%% Existing listening process ...
-l([{{?MODULE, listener, {_, AS}}, LPid}], _, _, TPid) ->
- LPid ! {accept, TPid},
- AS;
+l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) ->
+ {ok, LPid, AS};
%% ... or not.
-l([], LRef, T, TPid) ->
- {ok, _, AS} = diameter_tcp_sup:start_child({listen, LRef, TPid, T}),
- AS.
+l([], Ref, T) ->
+ diameter_tcp_sup:start_child({listen, Ref, T}).
%% get_addr/1
@@ -440,6 +451,14 @@ portnr(Sock) ->
%% # handle_call/3
%% ---------------------------------------------------------------------------
+handle_call({accept, SPid}, _From, #listener{service = P} = S) ->
+ {reply, ok, if not is_pid(P), is_pid(SPid) ->
+ monitor(process, SPid),
+ S#listener{service = SPid};
+ true ->
+ S
+ end};
+
handle_call(_, _, State) ->
{reply, nok, State}.
@@ -507,19 +526,20 @@ m({'DOWN', _, process, Pid, _}, #monitor{parent = Pid,
%%
%% Transition listener state.
-%% 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) ->
- S#listener{count = N-1};
+%% Service process has died.
+l({'DOWN', _, process, Pid, _} = T, #listener{service = Pid,
+ socket = Sock}) ->
+ gen_tcp:close(Sock),
+ x(T);
%% Transport has been removed.
l({transport, remove, _} = T, #listener{socket = Sock}) ->
gen_tcp:close(Sock),
- x(T).
+ x(T);
+
+%% Possibly death of an accepting process monitored in old code.
+l(_, S) ->
+ S.
%% t/2
%%