From 4b691d8d8a5eae2b0b34e1e0ec00fa5116384d69 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Fri, 23 Jan 2015 20:02:25 +0100 Subject: 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. --- lib/diameter/doc/src/diameter.xml | 23 +++++++- lib/diameter/src/base/diameter.erl | 3 +- lib/diameter/src/base/diameter_config.erl | 5 +- lib/diameter/src/base/diameter_lib.erl | 33 +++++++++++- lib/diameter/src/base/diameter_service.erl | 85 +++++++++++++++++++----------- 5 files changed, 114 insertions(+), 35 deletions(-) diff --git a/lib/diameter/doc/src/diameter.xml b/lib/diameter/doc/src/diameter.xml index 00b54ffbc4..538011b789 100644 --- a/lib/diameter/doc/src/diameter.xml +++ b/lib/diameter/doc/src/diameter.xml @@ -21,7 +21,7 @@ 2011 -2014 +2015 Ericsson AB. All Rights Reserved. @@ -1188,6 +1188,19 @@ See &man_tcp; for the behaviour of that module.

+{pool_size, pos_integer()} + +

+Number of transport processes to start. +For a listening transport, defines the size of the pool of accepting +transport processes, a larger number being desirable for processing +multiple concurrent peer connection attempts. +For a connecting transport, the &service_opt; +restrict_connections must also be configured on the service in +question to allow multiple connections to the same peer.

+ + + {spawn_opt, [term()]} @@ -1737,6 +1750,14 @@ connection might look as follows.

The information presented here is as in the connect case except that the client connections are grouped under an accept tuple.

+

+Whether or not the &transport_opt; pool_size affects the format +of the listing in the case of a connecting transport, since a value +greater than 1 implies multiple transport processes for the same +&transport_ref;, as in the listening case. +The format in this case is similar to the listening case, with a +pool tuple in place of an accept tuple.

+ connections 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]). %% --------------------------------------------------------------------------- @@ -289,6 +291,35 @@ ft(undefined, {_, T}) -> 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 %% 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. -- cgit v1.2.3