Merge branch 'anders/diameter/overload/OTP-13330'

* anders/diameter/overload/OTP-13330:
  Suppress dialyzer warning
  Remove dead case clause
  Let throttling callback send a throttle message
  Acknowledge answers to notification pids when throttling
  Throttle properly with TLS
  Don't ask throttling callback to receive more unless needed
  Let a throttling callback answer a received message
  Let a throttling callback discard a received message
  Let throttling callback return a notification pid
  Make throttling callbacks on message reception
  Add diameter_tcp option throttle_cb

3 files changed, 86 insertions, 41 deletions

diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index fb874013a3..deb9404216 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.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.
@@ -416,8 +416,8 @@ transition({connection_timeout, _}, _) ->
     ok;
 
 %% Incoming message from the transport.
-transition({diameter, {recv, Pkt}}, S) ->
-    recv(Pkt, S);
+transition({diameter, {recv, MsgT}}, S) ->
+    incoming(MsgT, S);
 
 %% Timeout when still in the same state ...
 transition({timeout = T, PS}, #state{state = PS}) ->
@@ -543,6 +543,28 @@ encode(Rec, Dict) ->
     diameter_codec:encode(Dict, #diameter_packet{header = Hdr,
                                                  msg = Rec}).
 
+%% incoming/2
+
+incoming({Msg, NPid}, S) ->
+    try recv(Msg, S) of
+        T ->
+            NPid ! {diameter, discard},
+            T
+    catch
+        {?MODULE, Name, Pkt} ->
+            S#state.parent ! {recv, self(), Name, {Pkt, NPid}},
+            rcv(Name, Pkt, S)
+    end;
+
+incoming(Msg, S) ->
+    try
+        recv(Msg, S)
+    catch
+        {?MODULE, Name, Pkt} ->
+            S#state.parent ! {recv, self(), Name, Pkt},
+            rcv(Name, Pkt, S)
+    end.
+
 %% recv/2
 
 recv(#diameter_packet{header = #diameter_header{} = Hdr}
@@ -597,9 +619,8 @@ recv1('DPA' = N,
 
 %% Any other message with a header and no length errors: send to the
 %% parent.
-recv1(Name, Pkt, #state{parent = Pid} = S) ->
-    Pid ! {recv, self(), Name, Pkt},
-    rcv(Name, Pkt, S).
+recv1(Name, Pkt, #state{}) ->
+    throw({?MODULE, Name, Pkt}).
 
 %% recv/3
 
diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl
index c169d3fc2c..2112941d5e 100644
--- a/lib/diameter/src/base/diameter_traffic.erl
+++ b/lib/diameter/src/base/diameter_traffic.erl
@@ -230,7 +230,15 @@ pending(TPids) ->
 %% used to come through the service process but this avoids that
 %% becoming a bottleneck.
 
-receive_message(TPid, Pkt, Dict0, RecvData)
+receive_message(TPid, {Pkt, NPid}, Dict0, RecvData) ->
+    NPid ! {diameter, incoming(TPid, Pkt, Dict0, RecvData)};
+
+receive_message(TPid, Pkt, Dict0, RecvData) ->
+    incoming(TPid, Pkt, Dict0, RecvData).
+
+%% incoming/4
+
+incoming(TPid, Pkt, Dict0, RecvData)
   when is_pid(TPid) ->
     #diameter_packet{header = #diameter_header{is_request = R}} = Pkt,
     recv(R,
@@ -244,11 +252,18 @@ receive_message(TPid, Pkt, Dict0, RecvData)
 
 %% Incoming request ...
 recv(true, false, TPid, Pkt, Dict0, T) ->
-    spawn_request(TPid, Pkt, Dict0, T);
+    try
+        {request, spawn_request(TPid, Pkt, Dict0, T)}
+    catch
+        error: system_limit = E ->  %% discard
+            ?LOG(error, E),
+            discard
+    end;
 
 %% ... answer to known request ...
 recv(false, #request{ref = Ref, handler = Pid} = Req, _, Pkt, Dict0, _) ->
-    Pid ! {answer, Ref, Req, Dict0, Pkt};
+    Pid ! {answer, Ref, Req, Dict0, Pkt},
+    {answer, Pid};
 
 %% Note that failover could have happened prior to this message being
 %% received and triggering failback. That is, both a failover message
@@ -263,7 +278,7 @@ recv(false, #request{ref = Ref, handler = Pid} = Req, _, Pkt, Dict0, _) ->
 recv(false, false, TPid, Pkt, _, _) ->
     ?LOG(discarded, Pkt#diameter_packet.header),
     incr(TPid, {{unknown, 0}, recv, discarded}),
-    ok.
+    discard.
 
 %% spawn_request/4
 
@@ -273,12 +288,7 @@ spawn_request(TPid, Pkt, Dict0, RecvData) ->
     spawn_request(TPid, Pkt, Dict0, ?DEFAULT_SPAWN_OPTS, RecvData).
 
 spawn_request(TPid, Pkt, Dict0, Opts, RecvData) ->
-    try
-        spawn_opt(fun() -> recv_request(TPid, Pkt, Dict0, RecvData) end, Opts)
-    catch
-        error: system_limit = E ->  %% discard
-            ?LOG(error, E)
-    end.
+    spawn_opt(fun() -> recv_request(TPid, Pkt, Dict0, RecvData) end, Opts).
 
 %% ---------------------------------------------------------------------------
 %% recv_request/4
diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl
index ea8b2fdb0e..3fd87b223e 100644
--- a/lib/diameter/src/base/diameter_watchdog.erl
+++ b/lib/diameter/src/base/diameter_watchdog.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.
@@ -449,8 +449,14 @@ transition({'DOWN', _, process, TPid, _Reason} = D,
     end;
 
 %% Incoming message.
-transition({recv, TPid, Name, Pkt}, #watchdog{transport = TPid} = S) ->
-    recv(Name, Pkt, S);
+transition({recv, TPid, Name, PktT}, #watchdog{transport = TPid} = S) ->
+    try
+        incoming(Name, PktT, S)
+    catch
+        #watchdog{dictionary = Dict0, receive_data = T} = NS ->
+            diameter_traffic:receive_message(TPid, PktT, Dict0, T),
+            NS
+    end;
 
 %% Current watchdog has timed out.
 transition({timeout, TRef, tw}, #watchdog{tref = TRef} = S) ->
@@ -578,22 +584,32 @@ send_watchdog(#watchdog{pending = false,
 
 %% Don't count encode errors since we don't expect any on DWR/DWA.
 
+%% incoming/3
+
+incoming(Name, {Pkt, NPid}, S) ->
+    NS = recv(Name, Pkt, S),
+    NPid ! {diameter, discard},
+    NS;
+
+incoming(Name, Pkt, S) ->
+    recv(Name, Pkt, S).
+
 %% recv/3
 
 recv(Name, Pkt, S) ->
-    try rcv(Name, S) of
+    try rcv(Name, Pkt, rcv(Name, S)) of
         #watchdog{} = NS ->
-            rcv(Name, Pkt, S),
-            NS
+            throw(NS)
     catch
-        {?MODULE, throwaway, #watchdog{} = NS} ->
+        #watchdog{} = NS ->  %% throwaway
             NS
     end.
 
 %% rcv/3
 
 rcv('DWR', Pkt, #watchdog{transport = TPid,
-                          dictionary = Dict0}) ->
+                          dictionary = Dict0}
+                = S) ->
     ?LOG(recv, 'DWR'),
     DPkt = diameter_codec:decode(Dict0, Pkt),
     diameter_traffic:incr(recv, DPkt, TPid, Dict0),
@@ -610,32 +626,30 @@ rcv('DWR', Pkt, #watchdog{transport = TPid,
     send(TPid, {send, #diameter_packet{header = H,
                                        transport_data = T,
                                        bin = Bin}}),
-    ?LOG(send, 'DWA');
+    ?LOG(send, 'DWA'),
+    throw(S);
 
 rcv('DWA', Pkt, #watchdog{transport = TPid,
-                          dictionary = Dict0}) ->
+                          dictionary = Dict0}
+                = S) ->
     ?LOG(recv, 'DWA'),
     diameter_traffic:incr(recv, Pkt, TPid, Dict0),
     diameter_traffic:incr_rc(recv,
                              diameter_codec:decode(Dict0, Pkt),
                              TPid,
-                             Dict0);
+                             Dict0),
+    throw(S);
 
-rcv(N, _, _)
+rcv(N, _, S)
   when N == 'CER';
        N == 'CEA';
        N == 'DPR' ->
-    false;
+    throw(S);
 %% DPR can be sent explicitly with diameter:call/4. Only the
 %% corresponding DPAs arrive here.
 
-rcv(_, Pkt, #watchdog{transport = TPid,
-                      dictionary = Dict0,
-                      receive_data = T}) ->
-    diameter_traffic:receive_message(TPid, Pkt, Dict0, T).
-
-throwaway(S) ->
-    throw({?MODULE, throwaway, S}).
+rcv(_, _, S)->
+    S.
 
 %% rcv/2
 %%
@@ -652,20 +666,20 @@ throwaway(S) ->
 %%   INITIAL       Receive non-DWA      Throwaway()          INITIAL
 
 rcv('DWA', #watchdog{status = initial} = S) ->
-    throwaway(S#watchdog{pending = false});
+    throw(S#watchdog{pending = false});
 
 rcv(_, #watchdog{status = initial} = S) ->
-    throwaway(S);
+    throw(S);
 
 %%   DOWN          Receive DWA          Pending = FALSE
 %%                                      Throwaway()          DOWN
 %%   DOWN          Receive non-DWA      Throwaway()          DOWN
 
 rcv('DWA', #watchdog{status = down} = S) ->
-    throwaway(S#watchdog{pending = false});
+    throw(S#watchdog{pending = false});
 
 rcv(_, #watchdog{status = down} = S) ->
-    throwaway(S);
+    throw(S);
 
 %%   OKAY          Receive DWA          Pending = FALSE
 %%                                      SetWatchdog()        OKAY
@@ -721,7 +735,7 @@ rcv('DWR', #watchdog{status = reopen} = S) ->
     S;  %% ensure DWA: the RFC isn't explicit about answering
 
 rcv(_, #watchdog{status = reopen} = S) ->
-    throwaway(S).
+    throw(S).
 
 %% timeout/1
 %%