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.

1 files changed, 46 insertions, 30 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
 %%