Merge branch 'anders/diameter/sctp/OTP-10889' into maint

* anders/diameter/sctp/OTP-10889:
  Send unordered on all outbound diameter_sctp streams
  Delay rotation of diameter_sctp outbound streams
  Exercise unordered delivery in traffic suite
  Use unordered delivery on a lone outbound stream in diameter_sctp

2 files changed, 40 insertions, 5 deletions

diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl
index 8de19e904e..4eb3379d59 100644
--- a/lib/diameter/src/transport/diameter_sctp.erl
+++ b/lib/diameter/src/transport/diameter_sctp.erl
@@ -102,6 +102,7 @@
          streams  :: {uint(), uint()}      %% {InStream, OutStream} counts
                    | undefined,
          os = 0   :: uint(),               %% next output stream
+         rotate = 1 :: boolean() | 0 | 1,  %% rotate os?
          packet = true :: boolean()        %% legacy transport_data?
                         | raw,
          message_cb = false :: false | diameter:eval(),
@@ -692,11 +693,16 @@ send(#diameter_packet{transport_data = {outstream, SId}}
      = S) ->
     send(SId rem OS, Msg, S);
 
-%% ... or not: rotate through all streams.
-send(Msg, #transport{streams = {_, OS},
+%% ... or not: rotate when sending on multiple streams ...
+send(Msg, #transport{rotate = true,
+                     streams = {_, OS},
                      os = N}
           = S) ->
-    send(N, Msg, S#transport{os = (N + 1) rem OS}).
+    send(N, Msg, S#transport{os = (N + 1) rem OS});
+
+%% ... or send on the only stream available.
+send(Msg, S) ->
+    send(0, Msg, S).
 
 %% send/3
 
@@ -764,7 +770,7 @@ recv({[#sctp_sndrcvinfo{assoc_id = Id}], _Bin}
 %% Inbound Diameter message.
 recv({[#sctp_sndrcvinfo{}], Bin} = Msg, S)
   when is_binary(Bin) ->
-    message(recv, Msg, S);
+    message(recv, Msg, recv(S));
 
 recv({_, #sctp_shutdown_event{}}, _) ->
     stop;
@@ -784,6 +790,25 @@ recv({_, #sctp_paddr_change{}}, _) ->
 recv({_, #sctp_pdapi_event{}}, _) ->
     ok.
 
+%% recv/1
+%%
+%% Start sending unordered after the second reception, so that an
+%% outgoing CER/CEA will arrive at the peer before another request.
+
+recv(#transport{rotate = B} = S)
+  when is_boolean(B) ->
+    S;
+
+recv(#transport{rotate = 0, streams = {_,N}, socket = Sock} = S) ->
+    ok = inet:setopts(Sock, [{sctp_default_send_param,
+                              #sctp_sndrcvinfo{flags = [unordered]}}]),
+    S#transport{rotate = 1 < N};
+
+recv(#transport{rotate = N} = S) ->
+    S#transport{rotate = N-1}.
+
+%% publish/4
+
 publish(T, Ref, Id, Sock) ->
     true = diameter_reg:add_new({?MODULE, T, {Ref, {Id, Sock}}}),
     putr(?INFO_KEY, {gen_sctp, Sock}).  %% for info/1
diff --git a/lib/diameter/test/diameter_traffic_SUITE.erl b/lib/diameter/test/diameter_traffic_SUITE.erl
index 6f30055a0a..f2c7ac0ff9 100644
--- a/lib/diameter/test/diameter_traffic_SUITE.erl
+++ b/lib/diameter/test/diameter_traffic_SUITE.erl
@@ -114,6 +114,8 @@
 %% diameter_{tcp,sctp} callbacks
 -export([message/3]).
 
+-include_lib("kernel/include/inet_sctp.hrl").
+
 -include("diameter.hrl").
 -include("diameter_gen_base_rfc3588.hrl").
 -include("diameter_gen_base_accounting.hrl").
@@ -533,7 +535,7 @@ add_transports(Config) ->
                          | server_apps()]
                         ++ [{spawn_opt, {erlang, spawn, []}} || CS]),
     Cs = [?util:connect(CN,
-                        [T, {sender, CS}],
+                        [T, {sender, CS} | client_opts(T)],
                         LRef,
                         [{id, Id}
                          | client_apps(R, [{'Origin-State-Id', origin(Id)}])])
@@ -543,6 +545,14 @@ add_transports(Config) ->
              Id <- [{D,E}]],
     ?util:write_priv(Config, "transport", [LRef | Cs]).
 
+client_opts(tcp) ->
+    [];
+client_opts(sctp) ->
+    [{sctp_initmsg, #sctp_initmsg{num_ostreams = N,
+                                  max_instreams = 5}}
+     || N <- [rand:uniform(8)],
+        N =< 6].
+
 server_apps() ->
     B = have_nas(),
     [{applications, [diameter_gen_base_rfc3588,