Add transport_opt() pool_size

Transport processes are started by diameter one at a time. In the
listening case, a transport process accepts a connection, tells the
peer_fsm process, which tells its watchdog process, which tells its
service process, which then starts a new watchdog, which starts a new
peer_fsm, which starts a new transport process, which (finally) goes
about accepting another connection. In other words, not particularly
aggressive in accepting new connections. This behaviour doesn't do
particularly well with a large number of concurrent connections: with
TCP and 250 connecting peers we see connections being refused.

This commit adds the possibilty of configuring a pool of accepting
processes, by way of a new transport option, pool_size. Instead of
diameter:add_transport/2 starting just a single process, it now starts
the configured number, so that instead of a single process waiting for a
connection there's now a pool.

The option is even available for connecting processes, which provides an
alternate to adding multiple transports when multiple connections to the
same peer are required. In practice this also means configuring
{restrict_connections, false}: this is not implicit.

For backwards compatibility, the form of
diameter:service_info(_,transport) differs in the connecting case,
depending on whether or not pool_size is configured.

Note that transport processes for the same transport_ref() can be
started concurrently when pool_size > 1. This places additional
requirements on diameter_{tcp,sctp}, that will be dealt with in a
subsequent commit.

4 files changed, 92 insertions, 34 deletions

diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl
index d74e091e11..1bbdf6e34d 100644
--- a/lib/diameter/src/base/diameter.erl
+++ b/lib/diameter/src/base/diameter.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2013. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2014. All Rights Reserved.
 %%
 %% The contents of this file are subject to the Erlang Public License,
 %% Version 1.1, (the "License"); you may not use this file except in
@@ -337,6 +337,7 @@ call(SvcName, App, Message) ->
    :: {transport_module, atom()}
     | {transport_config, any()}
     | {transport_config, any(), 'Unsigned32'() | infinity}
+    | {pool_size, pos_integer()}
     | {applications, [app_alias()]}
     | {capabilities, [capability()]}
     | {capabilities_cb, evaluable()}
diff --git a/lib/diameter/src/base/diameter_config.erl b/lib/diameter/src/base/diameter_config.erl
index dd1c9b73bb..270faa9542 100644
--- a/lib/diameter/src/base/diameter_config.erl
+++ b/lib/diameter/src/base/diameter_config.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2014. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
 %%
 %% The contents of this file are subject to the Erlang Public License,
 %% Version 1.1, (the "License"); you may not use this file except in
@@ -554,6 +554,9 @@ opt({watchdog_config, L}) ->
 opt({spawn_opt, Opts}) ->
     is_list(Opts);
 
+opt({pool_size, N}) ->
+    is_integer(N) andalso 0 < N;
+
 %% Options that we can't validate.
 opt({K, _})
   when K == transport_config;
diff --git a/lib/diameter/src/base/diameter_lib.erl b/lib/diameter/src/base/diameter_lib.erl
index 5b3a2063f8..23eba7e80c 100644
--- a/lib/diameter/src/base/diameter_lib.erl
+++ b/lib/diameter/src/base/diameter_lib.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2014. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
 %%
 %% The contents of this file are subject to the Erlang Public License,
 %% Version 1.1, (the "License"); you may not use this file except in
@@ -31,6 +31,8 @@
          spawn_opts/2,
          wait/1,
          fold_tuple/3,
+         fold_n/3,
+         for_n/2,
          log/4]).
 
 %% ---------------------------------------------------------------------------
@@ -290,6 +292,35 @@ ft(Value, {Idx, T}) ->
     setelement(Idx, T, Value).
 
 %% ---------------------------------------------------------------------------
+%% # fold_n/3
+%% ---------------------------------------------------------------------------
+
+-spec fold_n(F, Acc0, N)
+   -> term()
+ when F    :: fun((non_neg_integer(), term()) -> term()),
+      Acc0 :: term(),
+      N    :: non_neg_integer().
+
+fold_n(F, Acc, N)
+  when is_integer(N), 0 < N ->
+    fold_n(F, F(N, Acc), N-1);
+
+fold_n(_, Acc, _) ->
+    Acc.
+
+%% ---------------------------------------------------------------------------
+%% # for_n/2
+%% ---------------------------------------------------------------------------
+
+-spec for_n(F, N)
+   -> non_neg_integer()
+ when F :: fun((non_neg_integer()) -> term()),
+      N :: non_neg_integer().
+
+for_n(F, N) ->
+    fold_n(fun(M,A) -> F(M), A+1 end, 0, N).
+
+%% ---------------------------------------------------------------------------
 %% # log/4
 %%
 %% Called to have something to trace on for happenings of interest.
diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl
index 76b05a2ad4..96897c71eb 100644
--- a/lib/diameter/src/base/diameter_service.erl
+++ b/lib/diameter/src/base/diameter_service.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2014. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
 %%
 %% The contents of this file are subject to the Erlang Public License,
 %% Version 1.1, (the "License"); you may not use this file except in
@@ -765,8 +765,9 @@ reason(failure) ->
 start(Ref, {T, Opts}, S)
   when T == connect;
        T == listen ->
+    N = proplists:get_value(pool_size, Opts, 1),
     try
-        {ok, start(Ref, type(T), Opts, S)}
+        {ok, start(Ref, type(T), Opts, N, S)}
     catch
         ?FAILURE(Reason) ->
             {error, Reason}
@@ -784,11 +785,16 @@ type(connect = T) -> T.
 
 %% start/4
 
-start(Ref, Type, Opts, #state{watchdogT = WatchdogT,
-                              peerT = PeerT,
-                              options = SvcOpts,
-                              service_name = SvcName,
-                              service = Svc0})
+start(Ref, Type, Opts, State) ->
+    start(Ref, Type, Opts, 1, State).
+
+%% start/5
+
+start(Ref, Type, Opts, N, #state{watchdogT = WatchdogT,
+                                 peerT = PeerT,
+                                 options = SvcOpts,
+                                 service_name = SvcName,
+                                 service = Svc0})
   when Type == connect;
        Type == accept ->
     #diameter_service{applications = Apps}
@@ -796,14 +802,19 @@ start(Ref, Type, Opts, #state{watchdogT = WatchdogT,
         = merge_service(Opts, Svc0),
     {_,_} = Mask = proplists:get_value(sequence, SvcOpts),
     RecvData = diameter_traffic:make_recvdata([SvcName, PeerT, Apps, Mask]),
-    Pid = s(Type, Ref, {{spawn_opts([Opts, SvcOpts]), RecvData},
-                        Opts,
-                        SvcOpts,
-                        Svc}),
-    insert(WatchdogT, #watchdog{pid = Pid,
-                                type = Type,
-                                ref = Ref,
-                                options = Opts}),
+    T = {{spawn_opts([Opts, SvcOpts]), RecvData}, Opts, SvcOpts, Svc},
+    Rec = #watchdog{type = Type,
+                    ref = Ref,
+                    options = Opts},
+    diameter_lib:fold_n(fun(_,A) ->
+                                [wd(Type, Ref, T, WatchdogT, Rec) | A]
+                        end,
+                        [],
+                        N).
+
+wd(Type, Ref, T, WatchdogT, Rec) ->
+    Pid = wd(Type, Ref, T),
+    insert(WatchdogT, Rec#watchdog{pid = Pid}),
     Pid.
 
 %% Note that the service record passed into the watchdog is the merged
@@ -816,7 +827,7 @@ spawn_opts(Optss) ->
           T /= link,
           T /= monitor].
 
-s(Type, Ref, T) ->
+wd(Type, Ref, T) ->
     {_MRef, Pid} = diameter_watchdog:start({Type, Ref}, T),
     Pid.
 
@@ -1718,31 +1729,43 @@ info_transport(S) ->
               [],
               PeerD).
 
-%% Only a config entry for a listening transport: use it.
-transport([[{type, listen}, _] = L]) ->
-    L ++ [{accept, []}];
-
-%% Only one config or peer entry for a connecting transport: use it.
-transport([[{type, connect} | _] = L]) ->
-    L;
+%% Single config entry. Distinguish between pool_size config or not on
+%% a connecting transport for backwards compatibility: with the option
+%% the form is similar to the listening case, with connections grouped
+%% in a pool tuple (for lack of a better name), without as before.
+transport([[{type, Type}, {options, Opts}] = L])
+  when Type == listen;
+       Type == connect ->
+    L ++ [{K, []} || [{_,K}] <- [keys(Type, Opts)]];
 
 %% Peer entries: discard config. Note that the peer entries have
 %% length at least 3.
 transport([[_,_] | L]) ->
     transport(L);
 
-%% Possibly many peer entries for a listening transport. Note that all
-%% have the same options by construction, which is not terribly space
-%% efficient.
-transport([[{type, accept}, {options, Opts} | _] | _] = Ls) ->
-    [{type, listen},
+%% Multiple tranports. Note that all have the same options by
+%% construction, which is not terribly space efficient.
+transport([[{type, Type}, {options, Opts} | _] | _] = Ls) ->
+    transport(keys(Type, Opts), Ls).
+
+%% Group transports in an accept or pool tuple ...
+transport([{Type, Key}], [[{type, _}, {options, Opts} | _] | _] = Ls) ->
+    [{type, Type},
      {options, Opts},
-     {accept, [lists:nthtail(2,L) || L <- Ls]}].
+     {Key, [tl(tl(L)) || L <- Ls]}];
+
+%% ... or not: there can only be one.
+transport([], [L]) ->
+    L.
+
+keys(connect = T, Opts) ->
+    [{T, pool} || lists:keymember(pool_size, 1, Opts)];
+keys(_, _) ->
+    [{listen, accept}].
 
 peer_dict(#state{watchdogT = WatchdogT, peerT = PeerT}, Dict0) ->
     try ets:tab2list(WatchdogT) of
-        L ->
-            lists:foldl(fun(T,A) -> peer_acc(PeerT, A, T) end, Dict0, L)
+        L -> lists:foldl(fun(T,A) -> peer_acc(PeerT, A, T) end, Dict0, L)
     catch
         error: badarg -> Dict0  %% service has gone down
     end.