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.

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
 %%