Move failover out of service process

In order to be able to remove fields from the request process that don't
need to be there and do less in the service process. The pick_peer
callback now takes place in the request process in the case of immutable
state, just as in the case of the initial send.

1 files changed, 123 insertions, 93 deletions

diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl
index 6e7adb1be2..d5a09338cd 100644
--- a/lib/diameter/src/base/diameter_service.erl
+++ b/lib/diameter/src/base/diameter_service.erl
@@ -334,14 +334,14 @@ call_rc(_, _, Sent) ->
 call(SvcName, App, Msg, Opts, Caller) ->
     c(find_state(SvcName), App, Msg, Opts, Caller).
 
-c(#state{service_name = Svc, options = [{_, Mask} | _]} = S,
+c(#state{service_name = SvcName, options = [{_, Mask} | _]} = S,
   App,
   Msg,
   Opts,
   Caller) ->
     case find_transport(App, Msg, Opts, S) of
         {_,_,_} = T ->
-            send_request(T, Mask, Msg, Opts, Caller, Svc);
+            send_request(T, Mask, Msg, Opts, Caller, SvcName);
         false ->
             {error, no_connection};
         {error, _} = No ->
@@ -587,8 +587,8 @@ transition({tc_timeout, T}, S) ->
 %% Request process is telling us it may have missed a failover message
 %% after a transport went down and the service process looked up
 %% outstanding requests.
-transition({failover, TRef, Seqs}, S) ->
-    failover(TRef, Seqs, S),
+transition({failover, TRef, Seqs}, _) ->
+    failover(TRef, Seqs),
     ok;
 
 transition(Req, S) ->
@@ -1064,7 +1064,7 @@ connection_down(#watchdog{state = ?WD_OKAY,
                 = S) ->
     report_status(down, Wd, Pr, S, []),
     remove_local_peer(SApps, {{TPid, Caps}, {SvcName, Apps}}, LDict),
-    request_peer_down(TPid, S);
+    request_peer_down(TPid);
 
 connection_down(#watchdog{}, #peer{}, _) ->
     ok;
@@ -1302,7 +1302,13 @@ cm([_,_|_], _, _, _) ->
 %% The mod field of the #diameter_app{} here includes any extra
 %% arguments passed to diameter:call/2.
 
-send_request({TPid, Caps, App} = T, Mask, Msg, Opts, Caller, SvcName) ->
+send_request({TPid, Caps, App}
+             = Transport,
+             Mask,
+             Msg,
+             Opts,
+             Caller,
+             SvcName) ->
     #diameter_app{module = ModX}
         = App,
 
@@ -1310,14 +1316,19 @@ send_request({TPid, Caps, App} = T, Mask, Msg, Opts, Caller, SvcName) ->
 
     send_req(cb(ModX, prepare_request, [Pkt, SvcName, {TPid, Caps}]),
              Pkt,
-             T,
+             Transport,
              Opts,
              Caller,
              SvcName,
              []).
 
-send_req({send, P}, Pkt, T, Opts, Caller, SvcName, Fs) ->
-    send_req(make_request_packet(P, Pkt), T, Opts, Caller, SvcName, Fs);
+send_req({send, Msg}, Pkt, Transport, Opts, Caller, SvcName, Fs) ->
+    send_req(make_request_packet(Msg, Pkt),
+             Transport,
+             Opts,
+             Caller,
+             SvcName,
+             Fs);
 
 send_req({discard, Reason} , _, _, _, _, _, _) ->
     {error, Reason};
@@ -1446,11 +1457,11 @@ send_req(Pkt0, {TPid, Caps, App}, Opts, Caller, SvcName, Fs) ->
                    module = ModX},
 
     try
-        TRef = send_request(TPid, Pkt, Req, Timeout),
+        TRef = send_request(TPid, Pkt, Req, SvcName, Timeout),
         ack(Caller),
         handle_answer(SvcName, AE, recv_answer(Timeout, SvcName, {TRef, Req}))
     after
-        erase_request(Pkt)
+        erase_requests(Pkt)
     end.
 
 %% Tell caller a send has been attempted.
@@ -1459,41 +1470,39 @@ ack({Pid, Ref}) ->
 
 %% recv_answer/3
 
-recv_answer(Timeout,
-            SvcName,
-            {TRef, #request{from = {_, Ref}, packet = RPkt} = Req}
-            = T) ->
-
+recv_answer(Timeout, SvcName, {TRef, #request{from = {_, Ref}} = Req}) ->
     %% Matching on TRef below ensures we ignore messages that pertain
     %% to a previous transport prior to failover. The answer message
     %% includes the #request{} since it's not necessarily Req; that
     %% is, from the last peer to which we've transmitted.
-
     receive
         {answer = A, Ref, Rq, Dict0, Pkt} ->  %% Answer from peer
             {A, Rq, Dict0, Pkt};
         {timeout = Reason, TRef, _} ->        %% No timely reply
             {error, Req, Reason};
-        {failover = Reason, TRef, false} ->   %% No alternate peer
-            {error, Req, Reason};
-        {failover, TRef, Transport} ->        %% Resend to alternate peer
-            try_retransmit(Timeout, SvcName, Req, Transport);
-        {failover, TRef} ->  %% May have missed failover notification
-            Seqs = diameter_codec:sequence_numbers(RPkt),
-            Pid  = whois(SvcName),
-            is_pid(Pid) andalso (Pid ! {failover, TRef, Seqs}),
-            recv_answer(Timeout, SvcName, T)
+        {failover, TRef} ->       %% Service says peer has gone down
+            retransmit(Req, find_state(SvcName), Timeout)
     end.
+
 %% Note that failover starts a new timer and that expiry of an old
 %% timer value is ignored. This means that an answer could be accepted
 %% from a peer after timeout in the case of failover.
 
-try_retransmit(Timeout, SvcName, Req, Transport) ->
+retransmit(Req, #state{service_name = SvcName} = S, Timeout) ->
+    rt(find_transport(Req, S), Req, SvcName, Timeout);
+            
+retransmit(Req, false, _) ->  %% service has gone down
+    {error, Req, failover}.
+
+rt({_,_,_} = Transport, Req, SvcName, Timeout) ->
     try retransmit(Transport, Req, SvcName, Timeout) of
         T -> recv_answer(Timeout, SvcName, T)
     catch
         ?FAILURE(Reason) -> {error, Req, Reason}
-    end.
+    end;
+
+rt(_, Req, _, _) ->  %% no alternate peer
+    {error, Req, failover}.
 
 %% handle_error/3
 
@@ -1532,28 +1541,32 @@ encode(Dict, #diameter_packet{bin = undefined} = Pkt) ->
 encode(_, #diameter_packet{} = Pkt) ->
     Pkt.
 
-%% send_request/4
+%% send_request/5
 
-send_request(TPid, #diameter_packet{bin = Bin} = Pkt, Req, Timeout)
+send_request(TPid, #diameter_packet{bin = Bin} = Pkt, Req, SvcName, Timeout)
   when node() == node(TPid) ->
     %% Store the outgoing request before sending to avoid a race with
     %% reply reception.
-    TRef = store_request(TPid, Bin, Req, Timeout),
+    TRef = store_request(TPid, Bin, Req, Timeout, SvcName),
     send(TPid, Pkt),
     TRef;
 
 %% Send using a remote transport: spawn a process on the remote node
 %% to relay the answer.
-send_request(TPid, #diameter_packet{} = Pkt, Req, Timeout) ->
+send_request(TPid, #diameter_packet{} = Pkt, Req, SvcName, Timeout) ->
     TRef = erlang:start_timer(Timeout, self(), timeout),
-    T = {TPid, Pkt, Req, Timeout, TRef},
+    T = {TPid, Pkt, Req, SvcName, Timeout, TRef},
     spawn(node(TPid), ?MODULE, send, [T]),
     TRef.
 
 %% send/1
 
-send({TPid, Pkt, #request{handler = Pid} = Req, Timeout, TRef}) ->
-    Ref = send_request(TPid, Pkt, Req#request{handler = self()}, Timeout),
+send({TPid, Pkt, #request{handler = Pid} = Req, SvcName, Timeout, TRef}) ->
+    Ref = send_request(TPid,
+                       Pkt,
+                       Req#request{handler = self()},
+                       SvcName,
+                       Timeout),
     Pid ! reref(receive T -> T end, Ref, TRef).
 
 reref({T, Ref, R}, Ref, TRef) ->
@@ -1568,62 +1581,84 @@ send(Pid, Pkt) ->
 
 %% retransmit/4
 
-retransmit({TPid, Caps, #diameter_app{alias = Alias} = App} = T,
+retransmit({TPid, Caps, #diameter_app{alias = Alias} = App}
+           = Transport,
            #request{app = Alias, packet = Pkt0}
            = Req,
            SvcName,
            Timeout) ->
-    have_request(Pkt0, TPid)      %% Don't failover to a peer we've
+    have_request(Pkt0, TPid)     %% Don't failover to a peer we've
         andalso ?THROW(timeout), %% already sent to.
 
     #diameter_packet{header = Hdr0} = Pkt0,
     Hdr = Hdr0#diameter_header{is_retransmitted = true},
     Pkt = Pkt0#diameter_packet{header = Hdr},
 
-    resend_req(cb(App, prepare_retransmit, [Pkt, SvcName, {TPid, Caps}]),
-               T,
+    retransmit(cb(App, prepare_retransmit, [Pkt, SvcName, {TPid, Caps}]),
+               Transport,
                Req#request{packet = Pkt},
+               SvcName,
                Timeout,
                []).
 
-resend_req({send, P}, T, #request{packet = Pkt} = Req, Timeout, Fs) ->
-    retransmit(make_request_packet(P, Pkt), T, Req, Timeout, Fs);
-
-resend_req({discard, Reason}, _, _, _, _) ->
+retransmit({send, Msg},
+           Transport,
+           #request{packet = Pkt}
+           = Req,
+           SvcName,
+           Timeout,
+           Fs) ->
+    resend_request(make_request_packet(Msg, Pkt),
+                   Transport,
+                   Req,
+                   SvcName,
+                   Timeout,
+                   Fs);
+
+retransmit({discard, Reason}, _, _, _, _, _) ->
     ?THROW(Reason);
 
-resend_req(discard, _, _, _, _) ->
+retransmit(discard, _, _, _, _, _) ->
     ?THROW(discarded);
 
-resend_req({eval_packet, RC, F}, T, Req, Timeout, Fs) ->
-    resend_req(RC, T, Req, Timeout, [F|Fs]);
+retransmit({eval_packet, RC, F}, Transport, Req, SvcName, Timeout, Fs) ->
+    retransmit(RC, Transport, Req, SvcName, Timeout, [F|Fs]);
     
-resend_req(T, {_, _, App}, _, _, _) ->
+retransmit(T, {_, _, App}, _, _, _, _) ->
     ?ERROR({invalid_return, prepare_retransmit, App, T}).
 
-%% retransmit/6
-
-retransmit(Pkt0, {TPid, Caps, _}, #request{dictionary = D} = Req0, Tmo, Fs) ->
-    Pkt = encode(D, Pkt0, Fs),
+resend_request(Pkt0,
+               {TPid, Caps, _},
+               #request{dictionary = Dict}
+               = Req0,
+               SvcName,
+               Tmo,
+               Fs) ->
+    Pkt = encode(Dict, Pkt0, Fs),
 
     Req = Req0#request{transport = TPid,
                        packet = Pkt0,
                        caps = Caps},
 
     ?LOG(retransmission, Req),
-    TRef = send_request(TPid, Pkt, Req, Tmo),
+    TRef = send_request(TPid, Pkt, Req, SvcName, Tmo),
     {TRef, Req}.
 
-%% store_request/4
+%% store_request/5
 
-store_request(TPid, Bin, Req, Timeout) ->
+store_request(TPid, Bin, Req, Timeout, SvcName) ->
     Seqs = diameter_codec:sequence_numbers(Bin),
     TRef = erlang:start_timer(Timeout, self(), timeout),
     ets:insert(?REQUEST_TABLE, {Seqs, Req, TRef}),
     ets:member(?REQUEST_TABLE, TPid)
-        orelse (self() ! {failover, TRef}),  %% possibly missed failover
+        orelse failover(whois(SvcName), TRef, Seqs),
     TRef.
 
+%% Induce failover for a request that was stored after a transport
+%% went down and which the service may have missed.
+failover(Pid, TRef, Seqs) ->
+    is_pid(Pid) andalso (Pid ! {failover, TRef, Seqs}).
+
 %% lookup_request/2
 
 lookup_request(Msg, TPid)
@@ -1646,9 +1681,9 @@ lookup(Msg, TPid, TRef) ->
             false
     end.
 
-%% erase_request/1
+%% erase_requests/1
 
-erase_request(Pkt) ->
+erase_requests(Pkt) ->
     ets:delete(?REQUEST_TABLE, diameter_codec:sequence_numbers(Pkt)).
 
 %% match_requests/1
@@ -1671,16 +1706,14 @@ have_request(Pkt, TPid) ->
 request_peer_up(TPid) ->
     ets:insert(?REQUEST_TABLE, {TPid}).
 
-%% request_peer_down/2
+%% request_peer_down/1
 
-request_peer_down(TPid, S) ->
+request_peer_down(TPid) ->
     ets:delete(?REQUEST_TABLE, TPid),
-    lists:foreach(fun(T) -> failover(T,S) end, match_requests(TPid)).
+    lists:foreach(fun failover/1, match_requests(TPid)).
 %% Note that a request process can store its request after failover
 %% notifications are sent here: store_request/4 sends the notification
-%% in that case. Note also that we'll send as many notifications to a
-%% given handler as there are peers its sent to. All but one of these
-%% will be ignored.
+%% in that case.
 
 %% ---------------------------------------------------------------------------
 %% recv_request/4
@@ -2558,34 +2591,30 @@ x(T) ->
     exit(T).
 
 %% ---------------------------------------------------------------------------
-%% # failover/[23]
+%% # failover/1-2
 %% ---------------------------------------------------------------------------
 
-%% Failover as a consequence of request_peer_down/2.
-failover({_, #request{handler = Pid} = Req, TRef}, S) ->
-    Pid ! {failover, TRef, rt(Req, S)}.
+%% Failover as a consequence of request_peer_down/1: inform the
+%% request process.
+failover({_, Req, TRef}) ->
+    #request{handler = Pid,
+             packet = #diameter_packet{msg = M}}
+        = Req,
+    M /= undefined andalso (Pid ! {failover, TRef}).
+%% Failover is not performed when msg = binary() since sending
+%% pre-encoded binaries is only partially supported. (Mostly for
+%% test.)
 
 %% Failover as a consequence of store_request/4.
-failover(TRef, Seqs, S)
+failover(TRef, Seqs)
   when is_reference(TRef) ->
     case lookup_request(Seqs, TRef) of
         #request{} = Req ->
-            failover({Seqs, Req, TRef}, S);
+            failover({Seqs, Req, TRef});
         false ->
             ok
     end.
 
-%% prepare_request returned a binary ...
-rt(#request{packet = #diameter_packet{msg = undefined}}, _) ->
-    false; %% Not what we should do but binaries are only parially supported
-
-%% ... or not.
-rt(#request{packet = #diameter_packet{msg = Msg},
-            dictionary = Dict}
-   = Req,
-   S) ->
-    find_transport(get_destination(Dict, Msg), Req, S).
-
 %% ---------------------------------------------------------------------------
 %% # report_status/5
 %% ---------------------------------------------------------------------------
@@ -2672,7 +2701,7 @@ rpd(Pid, Alias, PDict) ->
     ?Dict:update(Alias, fun(Ps) -> lists:keydelete(Pid, 1, Ps) end, PDict).
 
 %% ---------------------------------------------------------------------------
-%% find_transport/[34]
+%% find_transport/2-3
 %%
 %% Return: {TransportPid, #diameter_caps{}, #diameter_app{}}
 %%         | false
@@ -2709,26 +2738,27 @@ find_send_app(Alias, Apps) ->
             T
     end.
 
-%% Retransmission, in the service process.
-find_transport([_,_] = RH,
-               Req,
-               #state{service = #diameter_service{pid = Pid,
-                                                  applications = Apps}}
+%% Retransmission after failover.
+find_transport(#request{app = Alias,
+                        filter = Filter,
+                        module = ModX,
+                        packet = #diameter_packet{msg = Msg},
+                        dictionary = Dict},
+               #state{service = #diameter_service{applications = Apps}}
                = S)
-  when self() == Pid ->
-    #request{app = Alias,
-             filter = Filter,
-             module = ModX}
-        = Req,
+  when Msg /= undefined ->  %% retransmission of binaries is unsupported
     #diameter_app{}
         = App
         = lists:keyfind(Alias, #diameter_app.alias, Apps),
 
     pick_peer(App#diameter_app{module = ModX},
-              RH,
+              get_destination(Dict, Msg),
               Filter,
-              S).
+              S);
 
+find_transport(_, _) -> 
+    false.
+    
 %% get_destination/2
 
 get_destination(Dict, Msg) ->