diff options
Diffstat (limited to 'lib/diameter/src')
50 files changed, 1564 insertions, 821 deletions
diff --git a/lib/diameter/src/Makefile b/lib/diameter/src/Makefile index e0bbbdfe63..6bf748a727 100644 --- a/lib/diameter/src/Makefile +++ b/lib/diameter/src/Makefile @@ -1,7 +1,7 @@ # # %CopyrightBegin% # -# Copyright Ericsson AB 2010-2014. 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. @@ -123,7 +123,7 @@ ERL_COMPILE_FLAGS += \ # erl/hrl from dictionary file. gen/diameter_gen_%.erl gen/diameter_gen_%.hrl: dict/%.dia $(dia_verbose) \ - ../bin/diameterc -o gen -i $(EBIN) $< + escript ../bin/diameterc -o gen -i $(EBIN) $< opt: $(TARGET_FILES) diff --git a/lib/diameter/src/app.sed b/lib/diameter/src/app.sed index 78e5bd2bad..dd3806f5f1 100644 --- a/lib/diameter/src/app.sed +++ b/lib/diameter/src/app.sed @@ -1,7 +1,7 @@ # # %CopyrightBegin% # -# Copyright Ericsson AB 2014. All Rights Reserved. +# Copyright Ericsson AB 2014-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. diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl index e82c2c168c..e8f2f63f86 100644 --- a/lib/diameter/src/base/diameter.erl +++ b/lib/diameter/src/base/diameter.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. @@ -36,6 +36,8 @@ %% Information. -export([services/0, + peer_info/1, + peer_find/1, service_info/2]). %% Start/stop the application. In a "real" application this should @@ -53,6 +55,7 @@ service_name/0, capability/0, peer_filter/0, + peer_ref/0, service_opt/0, application_opt/0, app_module/0, @@ -147,6 +150,27 @@ service_info(SvcName, Option) -> diameter_service:info(SvcName, Option). %% --------------------------------------------------------------------------- +%% peer_info/2 +%% --------------------------------------------------------------------------- + +-spec peer_info(peer_ref()) + -> [tuple()]. + +peer_info(PeerRef) -> + diameter_service:peer_info(PeerRef). + +%% --------------------------------------------------------------------------- +%% peer_find/1 +%% --------------------------------------------------------------------------- + +-spec peer_find(peer_ref() | pid()) + -> {peer_ref(), pid()} + | false. + +peer_find(Pid) -> + diameter_peer_fsm:find(Pid). + +%% --------------------------------------------------------------------------- %% add_transport/3 %% --------------------------------------------------------------------------- @@ -280,6 +304,9 @@ call(SvcName, App, Message) -> | {all, [peer_filter()]} | {any, [peer_filter()]}. +-opaque peer_ref() + :: pid(). + -type evaluable() :: {module(), atom(), list()} | fun() @@ -312,6 +339,7 @@ call(SvcName, App, Message) -> | {sequence, sequence() | evaluable()} | {share_peers, remotes()} | {string_decode, boolean()} + | {strict_mbit, boolean()} | {incoming_maxlen, message_length()} | {use_shared_peers, remotes()} | {spawn_opt, list()}. diff --git a/lib/diameter/src/base/diameter_app.erl b/lib/diameter/src/base/diameter_app.erl index 6f0c78094a..122f60dd88 100644 --- a/lib/diameter/src/base/diameter_app.erl +++ b/lib/diameter/src/base/diameter_app.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/base/diameter_callback.erl b/lib/diameter/src/base/diameter_callback.erl index 70c70fb5bd..f479cb6612 100644 --- a/lib/diameter/src/base/diameter_callback.erl +++ b/lib/diameter/src/base/diameter_callback.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/base/diameter_codec.erl b/lib/diameter/src/base/diameter_codec.erl index 34276a1674..1ea5357924 100644 --- a/lib/diameter/src/base/diameter_codec.erl +++ b/lib/diameter/src/base/diameter_codec.erl @@ -77,9 +77,10 @@ setopts(Opts) when is_list(Opts) -> lists:foreach(fun setopt/1, Opts). -%% Decode stringish types to string()? The default true is for -%% backwards compatibility. -setopt({string_decode = K, false = B}) -> +%% The default string_decode true is for backwards compatibility. +setopt({K, false = B}) + when K == string_decode; + K == strict_mbit -> setopt(K, B); %% Regard anything but the generated RFC 3588 dictionary as modern. @@ -97,7 +98,8 @@ setopt(Key, Value) -> getopt(Key) -> case get({diameter, Key}) of - undefined when Key == string_decode -> + undefined when Key == string_decode; + Key == strict_mbit -> true; undefined when Key == rfc -> 6733; @@ -657,16 +659,23 @@ split_data(Bin, Len) -> %% The normal case here is data as an #diameter_avp{} list or an %% iolist, which are the cases that generated codec modules use. The -%% other case is as a convenience in the relay case in which the +%% other cases are a convenience in the relay case in which the %% dictionary doesn't know about specific AVP's. -%% Grouped AVP whose components need packing ... -pack_avp([#diameter_avp{} = A | Avps]) -> - pack_avp(A#diameter_avp{data = Avps}); -pack_avp(#diameter_avp{data = [#diameter_avp{} | _] = Avps} = A) -> - pack_avp(A#diameter_avp{data = encode_avps(Avps)}); +%% Decoded Grouped AVP with decoded components: ignore components +%% since they're already encoded in the Grouped AVP. +pack_avp([#diameter_avp{} = Grouped | _Components]) -> + pack_avp(Grouped); -%% ... data as a type/value tuple ... +%% Grouped AVP whose components need packing. It's intentional that +%% this isn't equivalent to [Grouped | Components]: here the +%% components need to be encoded before wrapping with the Grouped AVP, +%% and the list is flat, nesting being accomplished in the data +%% fields. +pack_avp(#diameter_avp{data = [#diameter_avp{} | _] = Components} = Grouped) -> + pack_avp(Grouped#diameter_avp{data = encode_avps(Components)}); + +%% Data as a type/value tuple ... pack_avp(#diameter_avp{data = {Type, Value}} = A) when is_atom(Type) -> pack_avp(A#diameter_avp{data = diameter_types:Type(encode, Value)}); diff --git a/lib/diameter/src/base/diameter_config.erl b/lib/diameter/src/base/diameter_config.erl index 5ae4ff1f46..fdbbd412a1 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-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. @@ -38,18 +38,17 @@ -module(diameter_config). -behaviour(gen_server). --compile({no_auto_import, [monitor/2, now/0]}). --import(diameter_lib, [now/0]). - -export([start_service/2, stop_service/1, add_transport/2, remove_transport/2, have_transport/2, - lookup/1]). + lookup/1, + subscribe/2]). -%% child server start --export([start_link/0]). +%% server start +-export([start_link/0, + start_link/1]). %% gen_server callbacks -export([init/1, @@ -59,8 +58,8 @@ handle_info/2, code_change/3]). -%% diameter_sync requests. --export([sync/1]). +%% callbacks +-export([sync/1]). %% diameter_sync requests %% debug -export([state/0, @@ -70,7 +69,8 @@ -include("diameter_internal.hrl"). %% Server state. --record(state, {id = now()}). +-record(state, {id = diameter_lib:now(), + role :: server | transport}). %% Registered name of the server. -define(SERVER, ?MODULE). @@ -78,6 +78,9 @@ %% Table config is written to. -define(TABLE, ?MODULE). +%% Key on which a transport-specific child registers itself. +-define(TRANSPORT_KEY(Ref), {?MODULE, transport, Ref}). + %% Workaround for dialyzer's lack of understanding of match specs. -type match(T) :: T | '_' | '$1' | '$2' | '$3' | '$4'. @@ -226,6 +229,13 @@ pred(_) -> ?THROW(pred). %% -------------------------------------------------------------------------- +%% # subscribe/2 +%% -------------------------------------------------------------------------- + +subscribe(Ref, T) -> + diameter_reg:subscribe(?TRANSPORT_KEY(Ref), T). + +%% -------------------------------------------------------------------------- %% # have_transport/2 %% %% Output: true | false @@ -265,6 +275,9 @@ start_link() -> Options = [{spawn_opt, diameter_lib:spawn_opts(server, [])}], gen_server:start_link(ServerName, Module, Args, Options). +start_link(T) -> + proc_lib:start_link(?MODULE, init, [T], infinity, []). + state() -> call(state). @@ -275,8 +288,27 @@ uptime() -> %%% # init/1 %%% ---------------------------------------------------------- +%% ?SERVER start. init([]) -> - {ok, #state{}}. + {ok, #state{role = server}}; + +%% Child start as a consequence of add_transport. +init({SvcName, Type, Opts}) -> + Res = try + add(SvcName, Type, Opts) + catch + ?FAILURE(Reason) -> {error, Reason} + end, + proc_lib:init_ack({ok, self(), Res}), + loop(Res). + +%% loop/1 + +loop({ok, _}) -> + gen_server:enter_loop(?MODULE, [], #state{role = transport}); + +loop({error, _}) -> + ok. %% die %%% ---------------------------------------------------------- %%% # handle_call/2 @@ -285,8 +317,8 @@ init([]) -> handle_call(state, _, State) -> {reply, State, State}; -handle_call(uptime, _, #state{id = Time} = State) -> - {reply, diameter_lib:now_diff(Time), State}; +handle_call(uptime, _, #state{id = Time} = S) -> + {reply, diameter_lib:now_diff(Time), S}; handle_call(Req, From, State) -> ?UNEXPECTED([Req, From]), @@ -305,30 +337,34 @@ handle_cast(Msg, State) -> %%% # handle_info/2 %%% ---------------------------------------------------------- +%% remove_transport is telling published child to die. +handle_info(stop, #state{role = transport} = S) -> + {stop, normal, S}; + %% A service process has died. This is most likely a consequence of %% stop_service, in which case the restart will find no config for the %% service and do nothing. The entry keyed on the monitor ref is only %% removed as a result of the 'DOWN' notification however. -handle_info({'DOWN', MRef, process, _, Reason}, State) -> +handle_info({'DOWN', MRef, process, _, Reason}, #state{role = server} = S) -> [#monitor{service = SvcName} = T] = select([{#monitor{mref = MRef, _ = '_'}, [], ['$_']}]), queue_restart(Reason, SvcName), delete_object(T), - {noreply, State}; + {noreply, S}; -handle_info({monitor, SvcName, Pid}, State) -> - monitor(Pid, SvcName), - {noreply, State}; +handle_info({monitor, SvcName, Pid}, #state{role = server} = S) -> + insert_monitor(Pid, SvcName), + {noreply, S}; -handle_info({restart, SvcName}, State) -> +handle_info({restart, SvcName}, #state{role = server} = S) -> restart(SvcName), - {noreply, State}; + {noreply, S}; -handle_info(restart, State) -> +handle_info(restart, #state{role = server} = S) -> restart(), - {noreply, State}; + {noreply, S}; handle_info(Info, State) -> ?UNEXPECTED([Info]), @@ -405,19 +441,22 @@ sync({start_service, SvcName, Opts}) -> sync({stop_service, SvcName}) -> stop(SvcName); +%% Start a child whose only purpose is to be alive for the lifetime of +%% the transport configuration and publish itself in diameter_reg. +%% This is to provide a way for processes to to be notified when the +%% configuration is removed (diameter_reg:subscribe/2). sync({add, SvcName, Type, Opts}) -> - try - add(SvcName, Type, Opts) - catch - ?FAILURE(Reason) -> {error, Reason} - end; + {ok, _Pid, Res} = diameter_config_sup:start_child({SvcName, Type, Opts}), + Res; sync({remove, SvcName, Pred}) -> - remove(select([{#transport{service = '$1', _ = '_'}, + Recs = select([{#transport{service = '$1', _ = '_'}, [{'=:=', '$1', {const, SvcName}}], ['$_']}]), - SvcName, - Pred). + F = fun(#transport{ref = R, type = T, options = O}) -> + Pred(R,T,O) + end, + remove(SvcName, lists:filter(F, Recs)). %% start/3 @@ -439,8 +478,8 @@ startmon(SvcName, {ok, Pid}) -> startmon(_, {error, _}) -> ok. -monitor(Pid, SvcName) -> - MRef = erlang:monitor(process, Pid), +insert_monitor(Pid, SvcName) -> + MRef = monitor(process, Pid), insert(#monitor{mref = MRef, service = SvcName}). %% queue_restart/2 @@ -504,6 +543,7 @@ add(SvcName, Type, Opts) -> ok = transport_opts(Opts), Ref = make_ref(), + true = diameter_reg:add_new(?TRANSPORT_KEY(Ref)), T = {Ref, Type, Opts}, %% The call to the service returns error if the service isn't %% started yet, which is harmless. The transport will be started @@ -595,26 +635,30 @@ start_transport(SvcName, T) -> No end. -%% remove/3 +%% remove/2 -remove(L, SvcName, Pred) -> - rm(SvcName, lists:filter(fun(#transport{ref = R, type = T, options = O}) -> - Pred(R,T,O) - end, - L)). - -rm(_, []) -> +remove(_, []) -> ok; -rm(SvcName, L) -> + +remove(SvcName, L) -> Refs = lists:map(fun(#transport{ref = R}) -> R end, L), case stop_transport(SvcName, Refs) of ok -> + lists:foreach(fun stop_child/1, Refs), diameter_stats:flush(Refs), lists:foreach(fun delete_object/1, L); {error, _} = No -> No end. +stop_child(Ref) -> + case diameter_reg:match(?TRANSPORT_KEY(Ref)) of + [{_, Pid}] -> %% tell the transport-specific child to die + Pid ! stop; + [] -> %% already removed/dead + ok + end. + stop_transport(SvcName, Refs) -> case diameter_service:stop_transport(SvcName, Refs) of ok -> @@ -648,6 +692,7 @@ make_config(SvcName, Opts) -> {?NOMASK, sequence}, {nodes, restrict_connections}, {16#FFFFFF, incoming_maxlen}, + {true, strict_mbit}, {true, string_decode}, {[], spawn_opt}]), @@ -686,12 +731,14 @@ opt(K, false = B) K == use_shared_peers; K == monitor; K == restrict_connections; + K == strict_mbit; K == string_decode -> B; opt(K, true = B) when K == share_peers; K == use_shared_peers; + K == strict_mbit; K == string_decode -> B; diff --git a/lib/diameter/src/base/diameter_config_sup.erl b/lib/diameter/src/base/diameter_config_sup.erl new file mode 100644 index 0000000000..9524573378 --- /dev/null +++ b/lib/diameter/src/base/diameter_config_sup.erl @@ -0,0 +1,58 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2016. 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 +%% compliance with the License. You should have received a copy of the +%% Erlang Public License along with this software. If not, it can be +%% retrieved online at http://www.erlang.org/. +%% +%% Software distributed under the License is distributed on an "AS IS" +%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See +%% the License for the specific language governing rights and limitations +%% under the License. +%% +%% %CopyrightEnd% +%% + +%% +%% Supervisor for config processes. +%% + +-module(diameter_config_sup). + +-behaviour(supervisor). + +%% interface +-export([start_link/0, %% supervisor start + start_child/1]). %% config start + +-export([init/1]). + +-define(NAME, ?MODULE). %% supervisor name + +%% start_link/0 + +start_link() -> + SupName = {local, ?NAME}, + supervisor:start_link(SupName, ?MODULE, []). + +%% start_child/1 + +start_child(T) -> + supervisor:start_child(?NAME, [T]). + +%% init/1 + +init([]) -> + Mod = diameter_config, + Flags = {simple_one_for_one, 0, 1}, + ChildSpec = {Mod, + {Mod, start_link, []}, + temporary, + 1000, + worker, + [Mod]}, + {ok, {Flags, [ChildSpec]}}. diff --git a/lib/diameter/src/base/diameter_dict.erl b/lib/diameter/src/base/diameter_dict.erl index 1013690a5b..7db294a1b1 100644 --- a/lib/diameter/src/base/diameter_dict.erl +++ b/lib/diameter/src/base/diameter_dict.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/base/diameter_internal.hrl b/lib/diameter/src/base/diameter_internal.hrl index 518c0b9b1f..a0f4a8567d 100644 --- a/lib/diameter/src/base/diameter_internal.hrl +++ b/lib/diameter/src/base/diameter_internal.hrl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2013. 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. diff --git a/lib/diameter/src/base/diameter_lib.erl b/lib/diameter/src/base/diameter_lib.erl index b5b37dcf62..3928769b5e 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-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. @@ -20,18 +20,17 @@ -module(diameter_lib). -compile({no_auto_import, [now/0]}). --compile({nowarn_deprecated_function, [{erlang, now, 0}]}). -export([info_report/2, error_report/2, warning_report/2, now/0, + timestamp/0, timestamp/1, now_diff/1, micro_diff/1, micro_diff/2, time/1, - seed/0, eval/1, eval_name/1, get_stacktrace/0, @@ -103,32 +102,28 @@ fmt(T) -> %% # now/0 %% --------------------------------------------------------------------------- --type timestamp() :: {non_neg_integer(), 0..999999, 0..999999}. --type now() :: integer() %% monotonic time - | timestamp(). - -spec now() - -> now(). - -%% Use monotonic time if it exists, fall back to erlang:now() -%% otherwise. + -> integer(). now() -> - try - erlang:monotonic_time() - catch - error: undef -> erlang:now() - end. + erlang:monotonic_time(). %% --------------------------------------------------------------------------- -%% # timestamp/1 +%% # timestamp/0 %% --------------------------------------------------------------------------- --spec timestamp(NowT :: now()) - -> timestamp(). +-spec timestamp() + -> erlang:timestamp(). -timestamp({_,_,_} = T) -> %% erlang:now() - T; +timestamp() -> + timestamp(now()). + +%% --------------------------------------------------------------------------- +%% # timestamp/1 +%% --------------------------------------------------------------------------- + +-spec timestamp(integer()) + -> erlang:timestamp(). timestamp(MonoT) -> %% monotonic time MicroSecs = monotonic_to_microseconds(MonoT + erlang:time_offset()), @@ -142,30 +137,27 @@ monotonic_to_microseconds(MonoT) -> %% # now_diff/1 %% --------------------------------------------------------------------------- --spec now_diff(NowT :: now()) +-spec now_diff(T0 :: integer()) -> {Hours, Mins, Secs, MicroSecs} when Hours :: non_neg_integer(), Mins :: 0..59, Secs :: 0..59, MicroSecs :: 0..999999. -%% Return timer:now_diff(now(), NowT) as an {H, M, S, MicroS} tuple -%% instead of as integer microseconds. +%% Return time difference as an {H, M, S, MicroS} tuple instead of as +%% integer microseconds. -now_diff(Time) -> - time(micro_diff(Time)). +now_diff(T0) -> + time(micro_diff(T0)). %% --------------------------------------------------------------------------- %% # micro_diff/1 %% --------------------------------------------------------------------------- --spec micro_diff(NowT :: now()) +-spec micro_diff(T0 :: integer()) -> MicroSecs when MicroSecs :: non_neg_integer(). -micro_diff({_,_,_} = T0) -> - timer:now_diff(erlang:now(), T0); - micro_diff(T0) -> %% monotonic time monotonic_to_microseconds(erlang:monotonic_time() - T0). @@ -173,16 +165,12 @@ micro_diff(T0) -> %% monotonic time %% # micro_diff/2 %% --------------------------------------------------------------------------- --spec micro_diff(T1 :: now(), T0 :: now()) +-spec micro_diff(T1 :: integer(), T0 :: integer()) -> MicroSecs when MicroSecs :: non_neg_integer(). -micro_diff(T1, T0) - when is_integer(T1), is_integer(T0) -> %% monotonic time - monotonic_to_microseconds(T1 - T0); - -micro_diff(T1, T0) -> %% at least one erlang:now() - timer:now_diff(timestamp(T1), timestamp(T0)). +micro_diff(T1, T0) -> %% monotonic time + monotonic_to_microseconds(T1 - T0). %% --------------------------------------------------------------------------- %% # time/1 @@ -190,19 +178,13 @@ micro_diff(T1, T0) -> %% at least one erlang:now() %% Return an elapsed time as an {H, M, S, MicroS} tuple. %% --------------------------------------------------------------------------- --spec time(NowT | Diff) +-spec time(Diff :: non_neg_integer()) -> {Hours, Mins, Secs, MicroSecs} - when NowT :: timestamp(), - Diff :: non_neg_integer(), - Hours :: non_neg_integer(), + when Hours :: non_neg_integer(), Mins :: 0..59, Secs :: 0..59, MicroSecs :: 0..999999. -time({_,_,_} = NowT) -> %% time of day - %% 24 hours = 24*60*60*1000000 = 86400000000 microsec - time(timer:now_diff(NowT, {0,0,0}) rem 86400000000); - time(Micro) -> %% elapsed time Seconds = Micro div 1000000, H = Seconds div 3600, @@ -211,27 +193,6 @@ time(Micro) -> %% elapsed time {H, M, S, Micro rem 1000000}. %% --------------------------------------------------------------------------- -%% # seed/0 -%% --------------------------------------------------------------------------- - --spec seed() - -> {timestamp(), {integer(), integer(), integer()}}. - -%% Return an argument for random:seed/1. - -seed() -> - T = now(), - {timestamp(T), seed(T)}. - -%% seed/1 - -seed({_,_,_} = T) -> - T; - -seed(T) -> %% monotonic time - {erlang:phash2(node()), T, erlang:unique_integer()}. - -%% --------------------------------------------------------------------------- %% # eval/1 %% %% Evaluate a function in various forms. @@ -338,8 +299,28 @@ spawn_opts(server, Opts) -> spawn_opts(worker, Opts) -> opts(5000, Opts). -opts(HeapSize, Opts) -> - [{min_heap_size, HeapSize} | lists:keydelete(min_heap_size, 1, Opts)]. +%% These setting are historical rather than useful. In particular, the +%% server setting can bloat many processes unnecessarily. Let them be +%% disabled with -diameter min_heap_size false. + +opts(Def, Opts) -> + Key = min_heap_size, + case getenv(Key, Def) of + N when is_integer(N), 0 =< N -> + [{Key, N} | lists:keydelete(Key, 1, Opts)]; + _ -> + Opts + end. + +%% getenv/1 + +getenv(Key, Def) -> + case application:get_env(Key) of + {ok, T} -> + T; + undefined -> + Def + end. %% --------------------------------------------------------------------------- %% # wait/1 diff --git a/lib/diameter/src/base/diameter_misc_sup.erl b/lib/diameter/src/base/diameter_misc_sup.erl index 2054ea7831..343688be23 100644 --- a/lib/diameter/src/base/diameter_misc_sup.erl +++ b/lib/diameter/src/base/diameter_misc_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/base/diameter_peer.erl b/lib/diameter/src/base/diameter_peer.erl index acec91c43f..2759f17e64 100644 --- a/lib/diameter/src/base/diameter_peer.erl +++ b/lib/diameter/src/base/diameter_peer.erl @@ -21,9 +21,6 @@ -module(diameter_peer). -behaviour(gen_server). --compile({no_auto_import, [now/0]}). --import(diameter_lib, [now/0]). - %% Interface towards transport modules ... -export([recv/2, up/1, @@ -60,7 +57,7 @@ -define(SERVER, ?MODULE). %% Server state. --record(state, {id = now()}). +-record(state, {id = diameter_lib:now()}). %% Default transport_module/config. -define(DEFAULT_TMOD, diameter_tcp). diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index f5e04d3eae..996e75a8d3 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. @@ -32,6 +32,9 @@ -export([start/3, result_code/2]). +%% Interface towards diameter. +-export([find/1]). + %% gen_server callbacks -export([init/1, handle_call/3, @@ -117,7 +120,7 @@ parent :: pid(), %% watchdog process transport :: pid(), %% transport process dictionary :: module(), %% common dictionary - service :: #diameter_service{}, + service :: #diameter_service{} | undefined, dpr = false :: false | true %% DPR received, DPA sent | {boolean(), uint32(), uint32()}, @@ -185,6 +188,25 @@ start_link(T) -> infinity, diameter_lib:spawn_opts(server, [])). +%% find/1 +%% +%% Identify both pids of a peer_fsm/transport pair. + +find(Pid) -> + findl([{?MODULE, '_', Pid}, {?MODULE, Pid, '_'}]). + +findl([]) -> + false; + +findl([Pat | Rest]) -> + try + [{{_, Pid, TPid}, Pid}] = diameter_reg:match(Pat), + {Pid, TPid} + catch + error:_ -> + findl(Rest) + end. + %% --------------------------------------------------------------------------- %% --------------------------------------------------------------------------- @@ -215,6 +237,8 @@ i({Ack, WPid, {M, Ref} = T, Opts, {SvcOpts, Nodes, Dict0, Svc}}) -> {TPid, Addrs} = start_transport(T, Rest, Svc), + diameter_reg:add({?MODULE, self(), TPid}), %% lets pairs be discovered + #state{state = {'Wait-Conn-Ack', Tmo}, parent = WPid, transport = TPid, @@ -315,7 +339,7 @@ handle_info(T, #state{} = State) -> ?LOG(stop, Reason), {stop, {shutdown, Reason}, State}; stop -> - ?LOG(stop, T), + ?LOG(stop, truncate(T)), {stop, {shutdown, T}, State} catch exit: {diameter_codec, encode, T} = Reason -> @@ -348,6 +372,11 @@ code_change(_, State, _) -> %% --------------------------------------------------------------------------- %% --------------------------------------------------------------------------- +truncate({'DOWN' = T, _, process, Pid, _}) -> + {T, Pid}; +truncate(T) -> + T. + putr(Key, Val) -> put({?MODULE, Key}, Val). @@ -411,8 +440,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}) -> @@ -538,6 +567,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} @@ -592,9 +643,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_peer_fsm_sup.erl b/lib/diameter/src/base/diameter_peer_fsm_sup.erl index 54bd06929d..cf3c205e3f 100644 --- a/lib/diameter/src/base/diameter_peer_fsm_sup.erl +++ b/lib/diameter/src/base/diameter_peer_fsm_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/base/diameter_reg.erl b/lib/diameter/src/base/diameter_reg.erl index ce29baae45..9027130063 100644 --- a/lib/diameter/src/base/diameter_reg.erl +++ b/lib/diameter/src/base/diameter_reg.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. @@ -25,15 +25,12 @@ -module(diameter_reg). -behaviour(gen_server). --compile({no_auto_import, [monitor/2, now/0]}). --import(diameter_lib, [now/0]). - -export([add/1, add_new/1, - del/1, - repl/2, + remove/1, match/1, - wait/1]). + wait/1, + subscribe/2]). -export([start_link/0]). @@ -47,29 +44,32 @@ %% test -export([pids/0, - terms/0]). + terms/0, + subs/0, + waits/0]). %% debug -export([state/0, uptime/0]). --include("diameter_internal.hrl"). - -define(SERVER, ?MODULE). -define(TABLE, ?MODULE). -%% Table entry used to keep from starting more than one monitor on the -%% same process. This isn't a problem but there's no point in starting -%% multiple monitors if we can avoid it. Note that we can't have a 2-tuple -%% keyed on Pid since a registered term can be anything. Want the entry -%% keyed on Pid so that lookup is fast. --define(MONITOR(Pid, MRef), {Pid, monitor, MRef}). +-type key() :: term(). +-type from() :: {pid(), term()}. +-type pattern() :: term(). -%% Table entry containing the Term -> Pid mapping. --define(MAPPING(Term, Pid), {Term, Pid}). +-record(state, {id = diameter_lib:now(), + receivers = dict:new() + :: dict:dict(pattern(), [[pid() | term()]%% subscribe + | from()]), %% wait + monitors = sets:new() :: sets:set(pid())}). --record(state, {id = now(), - q = []}). %% [{From, Pat}] +%% The ?TABLE bag contains the Key -> Pid mapping, as {Key, Pid} +%% tuples. Each pid is stored in the monitors set to ensure only one +%% monitor for each pid: more are harmless, but unnecessary. A pattern +%% is added to receivers a result of calls to wait/1 or subscribe/2: +%% changes to ?TABLE causes processes to be notified as required. %% =========================================================================== %% # add(T) @@ -78,18 +78,18 @@ %% this or other assocations can be retrieved using match/1. %% %% An association is removed when the calling process dies or as a -%% result of calling del/1. Adding the same term more than once is -%% equivalent to adding it exactly once. +%% result of calling remove/1. Adding the same term more than once is +%% equivalent to adding it once. %% %% Note that since match/1 takes a pattern as argument, specifying a %% term that contains match variables is probably not a good idea %% =========================================================================== --spec add(any()) +-spec add(key()) -> true. add(T) -> - call({add, fun ets:insert/2, T, self()}). + call({add, false, T}). %% =========================================================================== %% # add_new(T) @@ -98,36 +98,23 @@ add(T) -> %% association, false being returned if an association already exists. %% =========================================================================== --spec add_new(any()) +-spec add_new(key()) -> boolean(). add_new(T) -> - call({add, fun insert_new/2, T, self()}). - -%% =========================================================================== -%% # repl(T, NewT) -%% -%% Like add/1 but only replace an existing association on T, false -%% being returned if it doesn't exist. -%% =========================================================================== - --spec repl(any(), any()) - -> boolean(). - -repl(T, U) -> - call({repl, T, U, self()}). + call({add, true, T}). %% =========================================================================== -%% # del(Term) +%% # remove(Term) %% %% Remove any existing association of Term with self(). %% =========================================================================== --spec del(any()) +-spec remove(key()) -> true. -del(T) -> - call({del, T, self()}). +remove(T) -> + call({remove, T}). %% =========================================================================== %% # match(Pat) @@ -140,12 +127,17 @@ del(T) -> %% associations removed.) %% =========================================================================== --spec match(any()) - -> [{term(), pid()}]. +-spec match(pattern()) + -> [{key(), pid()}]. match(Pat) -> - ets:match_object(?TABLE, ?MAPPING(Pat, '_')). + match(Pat, '_'). +%% match/2 + +match(Pat, Pid) -> + ets:match_object(?TABLE, {Pat, Pid}). + %% =========================================================================== %% # wait(Pat) %% @@ -153,10 +145,29 @@ match(Pat) -> %% It's up to the caller to ensure that the wait won't be forever. %% =========================================================================== +-spec wait(pattern()) + -> [{key(), pid()}]. + wait(Pat) -> + _ = match(Pat), %% ensure match can succeed call({wait, Pat}). %% =========================================================================== +%% # subscribe(Pat, T) +%% +%% Like match/1, but additionally receive messages of the form +%% {T, add|remove, {term(), pid()} when associations are added +%% or removed. +%% =========================================================================== + +-spec subscribe(Pat :: any(), T :: term()) + -> [{term(), pid()}]. + +subscribe(Pat, T) -> + _ = match(Pat), %% ensure match can succeed + call({subscribe, Pat, T}). + +%% =========================================================================== start_link() -> ServerName = {local, ?SERVER}, @@ -170,19 +181,15 @@ uptime() -> call(uptime). %% pids/0 -%% -%% Return: list of {Pid, [Term, ...]} + +-spec pids() + -> [{pid(), [key()]}]. pids() -> to_list(fun swap/1). to_list(Fun) -> - ets:foldl(fun(T,A) -> acc(Fun, T, A) end, orddict:new(), ?TABLE). - -acc(Fun, ?MAPPING(Term, Pid), Dict) -> - append(Fun({Term, Pid}), Dict); -acc(_, _, Dict) -> - Dict. + ets:foldl(fun(T,D) -> append(Fun(T), D) end, orddict:new(), ?TABLE). append({K,V}, Dict) -> orddict:append(K, V, Dict). @@ -190,14 +197,47 @@ append({K,V}, Dict) -> id(T) -> T. %% terms/0 -%% -%% Return: list of {Term, [Pid, ...]} + +-spec terms() + -> [{key(), [pid()]}]. terms() -> to_list(fun id/1). swap({X,Y}) -> {Y,X}. +%% subs/0 + +-spec subs() + -> [{pattern(), [{pid(), term()}]}]. + +subs() -> + #state{receivers = RD} = state(), + dict:fold(fun sub/3, orddict:new(), RD). + +sub(Pat, Ps, Dict) -> + lists:foldl(fun([P|T], D) -> orddict:append(Pat, {P,T}, D); + (_, D) -> D + end, + Dict, + Ps). + +%% waits/0 + +-spec waits() + -> [{pattern(), [{from(), term()}]}]. + +waits() -> + #state{receivers = RD} = state(), + dict:fold(fun wait/3, orddict:new(), RD). + +wait(Pat, Ps, Dict) -> + lists:foldl(fun({_,_} = F, D) -> orddict:append(Pat, F, D); + (_, D) -> D + end, + Dict, + Ps). + %% ---------------------------------------------------------- %% # init/1 %% ---------------------------------------------------------- @@ -210,57 +250,58 @@ init(_) -> %% # handle_call/3 %% ---------------------------------------------------------- -handle_call({add, Fun, Key, Pid}, _, S) -> - B = Fun(?TABLE, {Key, Pid}), - monitor(B andalso no_monitor(Pid), Pid), - {reply, B, pending(B, S)}; - -handle_call({del, Key, Pid}, _, S) -> - {reply, ets:delete_object(?TABLE, ?MAPPING(Key, Pid)), S}; - -handle_call({repl, T, U, Pid}, _, S) -> - MatchSpec = [{?MAPPING('$1', Pid), - [{'=:=', '$1', {const, T}}], - ['$_']}], - {reply, repl(ets:select(?TABLE, MatchSpec), U, Pid), S}; - -handle_call({wait, Pat}, From, #state{q = Q} = S) -> - case find(Pat) of - {ok, L} -> - {reply, L, S}; - false -> - {noreply, S#state{q = [{From, Pat} | Q]}} +handle_call({add, Uniq, Key}, {Pid, _}, S0) -> + Rec = {Key, Pid}, + S1 = flush(Uniq, Rec, S0), + {Res, New} = insert(Uniq, Rec), + {Recvs, S} = add(New, Rec, S1), + notify(Recvs, Rec), + {reply, Res, S}; + +handle_call({remove, Key}, {Pid, _}, S) -> + Rec = {Key, Pid}, + Recvs = delete([Rec], S), + ets:delete_object(?TABLE, Rec), + notify(Recvs, remove), + {reply, true, S}; + +handle_call({wait, Pat}, {Pid, _} = From, #state{receivers = RD} = S) -> + NS = add_monitor(Pid, S), + case match(Pat) of + [_|_] = L -> + {reply, L, NS}; + [] -> + {noreply, NS#state{receivers = dict:append(Pat, From, RD)}} end; +handle_call({subscribe, Pat, T}, {Pid, _}, #state{receivers = RD} = S) -> + NS = add_monitor(Pid, S), + {reply, match(Pat), NS#state{receivers = dict:append(Pat, [Pid | T], RD)}}; + handle_call(state, _, S) -> {reply, S, S}; handle_call(uptime, _, #state{id = Time} = S) -> {reply, diameter_lib:now_diff(Time), S}; -handle_call(Req, From, S) -> - ?UNEXPECTED([Req, From]), +handle_call(_Req, _From, S) -> {reply, nok, S}. %% ---------------------------------------------------------- %% # handle_cast/2 %% ---------------------------------------------------------- -handle_cast(Msg, S)-> - ?UNEXPECTED([Msg]), +handle_cast(_Msg, S)-> {noreply, S}. %% ---------------------------------------------------------- %% # handle_info/2 %% ---------------------------------------------------------- -handle_info({'DOWN', MRef, process, Pid, _}, S) -> - ets:delete_object(?TABLE, ?MONITOR(Pid, MRef)), - ets:match_delete(?TABLE, ?MAPPING('_', Pid)), - {noreply, S}; +handle_info({'DOWN', _MRef, process, Pid, _}, S) -> + {noreply, down(Pid, S)}; -handle_info(Info, S) -> - ?UNEXPECTED([Info]), +handle_info(_Info, S) -> {noreply, S}. %% ---------------------------------------------------------- @@ -279,71 +320,166 @@ code_change(_OldVsn, State, _Extra) -> %% =========================================================================== -monitor(true, Pid) -> - ets:insert(?TABLE, ?MONITOR(Pid, erlang:monitor(process, Pid))); -monitor(false, _) -> - ok. +%% insert/2 + +insert(false, Rec) -> + Spec = [{'$1', [{'==', '$1', {const, Rec}}], ['$_']}], + X = '$end_of_table' /= ets:select(?TABLE, Spec, 1), %% entry exists? + X orelse ets:insert(?TABLE, Rec), + {true, not X}; + +insert(true, Rec) -> + B = ets:insert_new(?TABLE, Rec), %% entry inserted? + {B, B}. -%% Do we need a monitor for the specified Pid? -no_monitor(Pid) -> - [] == ets:match_object(?TABLE, ?MONITOR(Pid, '_')). +%% add/3 -%% insert_new/2 +%% Only add a single monitor for any given process, since there's no +%% use to more. +add(true, {_Key, Pid} = Rec, S) -> + NS = add_monitor(Pid, S), + {Recvs, RD} = add(Rec, NS), + {Recvs, S#state{receivers = RD}}; -insert_new(?TABLE, {Key, _} = T) -> - flush(ets:lookup(?TABLE, Key)), - ets:insert_new(?TABLE, T). +add(false = No, _, S) -> + {No, S}. + +%% add/2 + +%% Notify processes whose patterns match the inserted key. +add({_Key, Pid} = Rec, #state{receivers = RD}) -> + dict:fold(fun(Pt, Ps, A) -> + add(lists:member(Rec, match(Pt, Pid)), Pt, Ps, Rec, A) + end, + {sets:new(), RD}, + RD). + +%% add/5 + +add(true, Pat, Recvs, {_,_} = Rec, {Set, Dict}) -> + {lists:foldl(fun sets:add_element/2, Set, Recvs), + remove(fun erlang:is_list/1, Pat, Recvs, Dict)}; + +add(false, _, _, _, Acc) -> + Acc. + +%% add_monitor/2 + +add_monitor(Pid, #state{monitors = MS} = S) -> + add_monitor(sets:is_element(Pid, MS), Pid, S). + +%% add_monitor/3 + +add_monitor(false, Pid, #state{monitors = MS} = S) -> + monitor(process, Pid), + S#state{monitors = sets:add_element(Pid, MS)}; + +add_monitor(true, _, S) -> + S. + +%% delete/2 + +delete(Recs, #state{receivers = RD}) -> + lists:foldl(fun(R,S) -> delete(R, RD, S) end, sets:new(), Recs). + +%% delete/3 + +delete({_Key, Pid} = Rec, RD, Set) -> + dict:fold(fun(Pt, Ps, S) -> + delete(lists:member(Rec, match(Pt, Pid)), Rec, Ps, S) + end, + Set, + RD). + +%% delete/4 + +%% Entry matches a pattern ... +delete(true, Rec, Recvs, Set) -> + lists:foldl(fun(R,S) -> sets:add_element({R, Rec}, S) end, + Set, + Recvs); + +%% ... or not. +delete(false, _, _, Set) -> + Set. + +%% notify/2 + +notify(false = No, _) -> + No; + +notify(Recvs, remove = Op) -> + sets:fold(fun({P,R}, N) -> send(P, R, Op), N+1 end, 0, Recvs); + +notify(Recvs, {_,_} = Rec) -> + sets:fold(fun(P,N) -> send(P, Rec, add), N+1 end, 0, Recvs). + +%% send/3 + +%% No processes waiting on remove, by construction: they've either +%% received notification at add or aren't waiting. +send([Pid | T], Rec, Op) -> + Pid ! {T, Op, Rec}; + +send({_,_} = From, Rec, add) -> + gen_server:reply(From, [Rec]). + +%% down/2 + +down(Pid, #state{monitors = MS} = S) -> + NS = flush(Pid, S), + Recvs = delete(match('_', Pid), NS), + ets:match_delete(?TABLE, {'_', Pid}), + notify(Recvs, remove), + NS#state{monitors = sets:del_element(Pid, MS)}. + +%% flush/3 %% Remove any processes that are dead but for which we may not have -%% received 'DOWN' yet. This is to ensure that add_new can be used -%% to register a unique name each time a process restarts. -flush(List) -> - lists:foreach(fun({_,P} = T) -> - del(erlang:is_process_alive(P), T) - end, - List). - -del(Alive, T) -> - Alive orelse ets:delete_object(?TABLE, T). - -%% repl/3 - -repl([?MAPPING(_, Pid) = M], Key, Pid) -> - ets:delete_object(?TABLE, M), - true = ets:insert(?TABLE, ?MAPPING(Key, Pid)); -repl([], _, _) -> - false. - -%% pending/1 - -pending(true, #state{q = [_|_] = Q} = S) -> - S#state{q = q(lists:reverse(Q), [])}; %% retain reply order -pending(_, S) -> +%% received 'DOWN' yet, to ensure that add_new can be used to register +%% a unique name each time a registering process restarts. +flush(true, {Key, Pid}, S) -> + Spec = [{{'$1', '$2'}, + [{'andalso', {'==', '$1', {const, Key}}, + {'/=', '$2', Pid}}], + ['$2']}], + lists:foldl(fun down/2, S, [P || P <- ets:select(?TABLE, Spec), + not is_process_alive(P)]); + +flush(false, _, S) -> S. -q([], Q) -> - Q; -q([{From, Pat} = T | Rest], Q) -> - case find(Pat) of - {ok, L} -> - gen_server:reply(From, L), - q(Rest, Q); - false -> - q(Rest, [T|Q]) - end. - -%% find/1 - -find(Pat) -> - try match(Pat) of - [] -> - false; - L -> - {ok, L} - catch - _:_ -> - {ok, []} - end. +%% flush/2 + +%% Process has died and should no longer receive messages/replies. +flush(Pid, #state{receivers = RD} = S) + when is_pid(Pid) -> + S#state{receivers = dict:fold(fun(Pt,Ps,D) -> flush(Pid, Pt, Ps, D) end, + RD, + RD)}. + +%% flush/4 + +flush(Pid, Pat, Recvs, Dict) -> + remove(fun(T) -> Pid /= head(T) end, Pat, Recvs, Dict). + +%% head/1 + +head([P|_]) -> + P; + +head({P,_}) -> + P. + +%% remove/4 + +remove(Pred, Key, Values, Dict) -> + case lists:filter(Pred, Values) of + [] -> + dict:erase(Key, Dict); + Rest -> + dict:store(Key, Rest, Dict) + end. %% call/1 diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index 5cb7fa5abe..ccf68f4d93 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-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. @@ -25,9 +25,6 @@ -module(diameter_service). -behaviour(gen_server). --compile({no_auto_import, [now/0]}). --import(diameter_lib, [now/0]). - %% towards diameter_service_sup -export([start_link/1]). @@ -35,6 +32,7 @@ -export([subscribe/1, unsubscribe/1, services/0, + peer_info/1, info/2]). %% towards diameter_config @@ -86,16 +84,8 @@ -define(DEFAULT_TC, 30000). %% RFC 3588 ch 2.1 -define(RESTART_TC, 1000). %% if restart was this recent -%% Used to be able to swap this with anything else dict-like but now -%% rely on the fact that a service's #state{} record does not change -%% in storing in it ?STATE table and not always going through the -%% service process. In particular, rely on the fact that operations on -%% a ?Dict don't change the handle to it. --define(Dict, diameter_dict). - -%% Maintains state in a table. In contrast to previously, a service's -%% stat is not constant and is accessed outside of the service -%% process. +%% Maintain state in a table since a service's state is accessed +%% outside of the service process. -define(STATE_TABLE, ?MODULE). %% The default sequence mask. @@ -115,23 +105,23 @@ %% to determine whether or not we need to call the process for a %% pick_peer callback in the statefull case. -record(state, - {id = now(), + {id = diameter_lib:now(), service_name :: diameter:service_name(), %% key in ?STATE_TABLE service :: #diameter_service{}, watchdogT = ets_new(watchdogs) %% #watchdog{} at start :: ets:tid(), - peerT = ets_new(peers) %% #peer{pid = TPid} at okay/reopen - :: ets:tid(), - shared_peers = ?Dict:new() %% Alias -> [{TPid, Caps}, ...] - :: ets:tid(), - local_peers = ?Dict:new() %% Alias -> [{TPid, Caps}, ...] - :: ets:tid(), + peerT, %% undefined in new code, but remain for upgrade + shared_peers, %% reasons. Replaced by local/remote. + local_peers, %% + local :: {ets:tid(), ets:tid(), ets:tid()}, + remote :: {ets:tid(), ets:tid(), ets:tid()}, monitor = false :: false | pid(), %% process to die with options :: [{sequence, diameter:sequence()} %% sequence mask | {share_peers, diameter:remotes()} %% broadcast to | {use_shared_peers, diameter:remotes()} %% use from | {restrict_connections, diameter:restriction()} + | {strict_mbit, boolean()} | {string_decode, boolean()} | {incoming_maxlen, diameter:message_length()}]}). %% shared_peers reflects the peers broadcast from remote nodes. @@ -139,23 +129,25 @@ %% Record representing an RFC 3539 watchdog process implemented by %% diameter_watchdog. -record(watchdog, - {pid :: match(pid()), + {pid :: match(pid()) | undefined, type :: match(connect | accept), ref :: match(reference()), %% key into diameter_config options :: match([diameter:transport_opt()]),%% from start_transport state = ?WD_INITIAL :: match(wd_state()), - started = now(), %% at process start + started = diameter_lib:now(),%% at process start peer = false :: match(boolean() | pid())}). - %% true at accepted, pid() at okay/reopen + %% true at accepted/remove, pid() at okay/reopen -%% Record representing an Peer State Machine processes implemented by +%% Record representing a Peer State Machine processes implemented by %% diameter_peer_fsm. -record(peer, - {pid :: pid(), - apps :: [{0..16#FFFFFFFF, diameter:app_alias()}], %% {Id, Alias} - caps :: #diameter_caps{}, - started = now(), %% at process start - watchdog :: pid()}). %% key into watchdogT + {pid :: pid(), + apps :: match([{0..16#FFFFFFFF, diameter:app_alias()}] %% {Id, Alias} + | [diameter:app_alias()]), %% remote + caps :: match(#diameter_caps{}), + started = diameter_lib:now(), %% at process start or sharing + watchdog :: match(pid() %% key into watchdogT + | undefined)}). %% undefined if remote %% --------------------------------------------------------------------------- %% # start/1 @@ -183,7 +175,7 @@ stop(SvcName) -> end. stop(ok, Pid) -> - MRef = erlang:monitor(process, Pid), + MRef = monitor(process, Pid), receive {'DOWN', MRef, process, _, _} -> ok end; stop(No, _) -> No. @@ -210,7 +202,7 @@ stop_transport(SvcName, [_|_] = Refs) -> info(SvcName, Item) -> case lookup_state(SvcName) of - [#state{} = S] -> + [S] -> service_info(Item, S); [] -> undefined @@ -219,7 +211,35 @@ info(SvcName, Item) -> %% lookup_state/1 lookup_state(SvcName) -> - ets:lookup(?STATE_TABLE, SvcName). + case ets:lookup(?STATE_TABLE, SvcName) of + [#state{}] = L -> + L; + _ -> + [] + end. + +%% --------------------------------------------------------------------------- +%% # peer_info/2 +%% --------------------------------------------------------------------------- + +%% An extended version of info_peer/1 for peer_info/1. +peer_info(Pid) -> + try + {_, PD} = process_info(Pid, dictionary), + {_, T} = lists:keyfind({diameter_peer_fsm, start}, 1, PD), + {TPid, {{Type, Ref}, TMod, Cfg}} = T, + {_, TD} = process_info(TPid, dictionary), + {_, Data} = lists:keyfind({TMod, info}, 1, TD), + [{ref, Ref}, + {type, Type}, + {owner, TPid}, + {module, TMod}, + {config, Cfg} + | try TMod:info(Data) catch _:_ -> [] end] + catch + error:_ -> + [] + end. %% --------------------------------------------------------------------------- %% # subscribe/1 @@ -230,7 +250,7 @@ subscribe(SvcName) -> diameter_reg:add({?MODULE, subscriber, SvcName}). unsubscribe(SvcName) -> - diameter_reg:del({?MODULE, subscriber, SvcName}). + diameter_reg:remove({?MODULE, subscriber, SvcName}). subscriptions(Pat) -> pmap(diameter_reg:match({?MODULE, subscriber, Pat})). @@ -492,6 +512,9 @@ transition({service, Pid}, S) -> transition({peer, TPid, Aliases, Caps}, S) -> remote_peer_up(TPid, Aliases, Caps, S), ok; +transition({peer, TPid}, S) -> + remote_peer_down(TPid, S), + ok; %% Remote peer process has died. transition({'DOWN', _, process, TPid, _}, S) -> @@ -511,7 +534,7 @@ transition(Req, S) -> %% # terminate/2 %% --------------------------------------------------------------------------- -terminate(Reason, #state{service_name = Name, peerT = PeerT} = S) -> +terminate(Reason, #state{service_name = Name, local = {PeerT, _, _}} = S) -> send_event(Name, stop), ets:delete(?STATE_TABLE, Name), @@ -533,23 +556,80 @@ terminate(Reason, #state{service_name = Name, peerT = PeerT} = S) -> %% # code_change/3 %% --------------------------------------------------------------------------- -code_change(FromVsn, - #state{service_name = SvcName, - service = #diameter_service{applications = Apps}} - = S, - Extra) -> - lists:foreach(fun(A) -> - code_change(FromVsn, SvcName, Extra, A) - end, - Apps), +code_change(_FromVsn, #state{} = S, _Extra) -> + {ok, S}; + +%% Don't support downgrade since we won't in appup. +code_change({down = T, _}, _, _Extra) -> + {error, T}; + +%% Upgrade local/shared peers dicts populated in old code. Don't +code_change(_FromVsn, S0, _Extra) -> + {state, Id, SvcName, Svc, WT, PeerT, SDict, LDict, Monitor, Opts} + = S0, + + init_peers(LT = setelement(1, {PT, _, _} = init_peers(), PeerT), + fun({_,A}) -> A end), + init_peers(init_peers(RT = init_peers(), SDict), + fun(A) -> A end), + + S = #state{id = Id, + service_name = SvcName, + service = Svc, + watchdogT = WT, + peerT = PT, %% empty + shared_peers = SDict, + local_peers = LDict, + local = LT, + remote = RT, + monitor = Monitor, + options = Opts}, + + %% Replacing the table entry and deleting the old shared tables + %% can make outgoing requests return {error, no_connection} until + %% everyone is running new code. Don't delete the tables to avoid + %% crashing request processes. + ets:delete_all_objects(SDict), + ets:delete_all_objects(LDict), + ets:insert(?STATE_TABLE, S), {ok, S}. -code_change(FromVsn, SvcName, Extra, #diameter_app{alias = Alias} = A) -> - {ok, S} = cb(A, code_change, [FromVsn, - mod_state(Alias), - Extra, - SvcName]), - mod_state(Alias, S). +%% init_peers/2 + +%% Populate app and identity bags from a new-style #peer{} sets. +init_peers({PeerT, _, _} = T, F) + when is_function(F) -> + ets:foldl(fun(#peer{pid = P, apps = As, caps = C}, N) -> + insert_peer(P, lists:map(F, As), C, T), + N+1 + end, + 0, + PeerT); + +%% Populate #peer{} table given a shared peers dict. +init_peers({PeerT, _, _}, SDict) -> + dict:fold(fun(P, As, N) -> + ets:update_element(PeerT, P, {#peer.apps, As}), + N+1 + end, + 0, + diameter_dict:fold(fun(A, Ps, D) -> + init_peers(A, Ps, PeerT, D) + end, + dict:new(), + SDict)). + +%% init_peers/4 + +init_peers(App, TCs, PeerT, Dict) -> + lists:foldl(fun({P,C}, D) -> + ets:insert(PeerT, #peer{pid = P, + apps = [], + caps = C}), + dict:append(P, App, D) + end, + Dict, + TCs). %% =========================================================================== %% =========================================================================== @@ -557,9 +637,6 @@ code_change(FromVsn, SvcName, Extra, #diameter_app{alias = Alias} = A) -> unexpected(F, A, #state{service_name = Name}) -> ?UNEXPECTED(F, A ++ [Name]). -cb(#diameter_app{module = [_|_] = M}, F, A) -> - eval(M, F, A). - eval([M|X], F, A) -> apply(M, F, A ++ X). @@ -579,10 +656,6 @@ choose(false, _, X) -> X. ets_new(Tbl) -> ets:new(Tbl, [{keypos, 2}]). -insert(Tbl, Rec) -> - ets:insert(Tbl, Rec), - Rec. - %% Using the process dictionary for the callback state was initially %% just a way to make what was horrendous trace (big state record and %% much else everywhere) somewhat more readable. There's not as much @@ -603,25 +676,34 @@ mod_state(Alias, ModS) -> %% remove_transport shutdown(Refs, #state{watchdogT = WatchdogT}) when is_list(Refs) -> - ets:foldl(fun(P,ok) -> st(P, Refs), ok end, ok, WatchdogT); + ets:insert(WatchdogT, ets:foldl(fun(R,A) -> st(R, Refs, A) end, + [], + WatchdogT)); %% application/service shutdown shutdown(Reason, #state{watchdogT = WatchdogT}) when Reason == application; Reason == service -> - diameter_lib:wait(ets:foldl(fun(P,A) -> st(P, Reason, A) end, + diameter_lib:wait(ets:foldl(fun(P,A) -> ss(P, Reason, A) end, [], WatchdogT)). -%% st/2 +%% st/3 -st(#watchdog{ref = Ref, pid = Pid}, Refs) -> - lists:member(Ref, Refs) - andalso (Pid ! {shutdown, self(), transport}). %% 'DOWN' cleans up +%% Mark replacement as started so that a subsequent accept doesn't +%% result in a new process that isn't terminated. +st(#watchdog{ref = Ref, pid = Pid, peer = P} = Rec, Refs, Acc) -> + case lists:member(Ref, Refs) of + true -> + Pid ! {shutdown, self(), transport}, %% 'DOWN' cleans up + [Rec#watchdog{peer = true} || P == false] ++ Acc; + false -> + Acc + end. -%% st/3 +%% ss/3 -st(#watchdog{pid = Pid}, Reason, Acc) -> +ss(#watchdog{pid = Pid}, Reason, Acc) -> MRef = monitor(process, Pid), Pid ! {shutdown, self(), Reason}, [MRef | Acc]. @@ -683,6 +765,8 @@ cfg_acc({SvcName, #diameter_service{applications = Apps} = Rec, Opts}, lists:foreach(fun init_mod/1, Apps), S = #state{service_name = SvcName, service = Rec#diameter_service{pid = self()}, + local = init_peers(), + remote = init_peers(), monitor = mref(get_value(monitor, Opts)), options = service_options(Opts)}, {S, Acc}; @@ -692,6 +776,13 @@ cfg_acc({_Ref, Type, _Opts} = T, {S, Acc}) Type == listen -> {S, [T | Acc]}. +init_peers() -> + {ets_new(caps), %% #peer{} + ets:new(apps, [bag]), %% {Alias, TPid} + ets:new(idents, [bag])}. %% {{host, OH} | {realm, OR} | {OR, OH}, + %% Alias, + %% TPid} + service_options(Opts) -> [{sequence, proplists:get_value(sequence, Opts, ?NOMASK)}, {share_peers, get_value(share_peers, Opts)}, @@ -701,13 +792,14 @@ service_options(Opts) -> ?RESTRICT)}, {spawn_opt, proplists:get_value(spawn_opt, Opts, [])}, {string_decode, proplists:get_value(string_decode, Opts, true)}, - {incoming_maxlen, proplists:get_value(incoming_maxlen, Opts, 16#FFFFFF)}]. + {incoming_maxlen, proplists:get_value(incoming_maxlen, Opts, 16#FFFFFF)}, + {strict_mbit, proplists:get_value(strict_mbit, Opts, true)}]. %% The order of options is significant since we match against the list. mref(false = No) -> No; mref(P) -> - erlang:monitor(process, P). + monitor(process, P). init_shared(#state{options = [_, _, {_,T} | _], service_name = Svc}) -> @@ -806,7 +898,7 @@ start(Ref, Type, Opts, State) -> %% start/5 start(Ref, Type, Opts, N, #state{watchdogT = WatchdogT, - peerT = PeerT, + local = {PeerT, _, _}, options = SvcOpts, service_name = SvcName, service = Svc0}) @@ -837,7 +929,7 @@ binary_caps(#diameter_service{capabilities = Caps} = Svc, false) -> wd(Type, Ref, T, WatchdogT, Rec) -> Pid = start_watchdog(Type, Ref, T), - insert(WatchdogT, Rec#watchdog{pid = Pid}), + ets:insert(WatchdogT, Rec#watchdog{pid = Pid}), Pid. %% Note that the service record passed into the watchdog is the merged @@ -891,11 +983,22 @@ ms(_, Svc) -> %% --------------------------------------------------------------------------- accepted(Pid, _TPid, #state{watchdogT = WatchdogT} = S) -> - #watchdog{ref = Ref, type = accept = T, peer = false, options = Opts} + #watchdog{type = accept = T, peer = P} = Wd = fetch(WatchdogT, Pid), - insert(WatchdogT, Wd#watchdog{peer = true}),%% mark replacement as started - start(Ref, T, Opts, S). %% start new watchdog + if not P -> + #watchdog{ref = Ref, options = Opts} = Wd, + %% Mark replacement started, and start new watchdog. + ets:insert(WatchdogT, Wd#watchdog{peer = true}), + start(Ref, T, Opts, S); + P -> + %% Transport removal in progress: true has been set in + %% shutdown/2, and the transport will die as a + %% consequence. + ok + end. + +%% fetch/2 fetch(Tid, Key) -> [T] = ets:lookup(Tid, Key), @@ -921,13 +1024,14 @@ watchdog(TPid, [], ?WD_SUSPECT, ?WD_OKAY, Wd, State) -> #watchdog{peer = TPid} = Wd, %% assert connection_up(Wd, State); -%% Watchdog has an unresponsive connection. +%% Watchdog has an unresponsive connection. Note that the peer table +%% entry isn't removed until DOWN. watchdog(TPid, [], ?WD_OKAY, ?WD_SUSPECT = To, Wd, State) -> #watchdog{peer = TPid} = Wd, %% assert watchdog_down(Wd, To, State); %% Watchdog has lost its connection. -watchdog(TPid, [], _, ?WD_DOWN = To, Wd, #state{peerT = PeerT} = S) -> +watchdog(TPid, [], _, ?WD_DOWN = To, Wd, #state{local = {PeerT, _, _}} = S) -> close(Wd), watchdog_down(Wd, To, S), ets:delete(PeerT, TPid); @@ -936,7 +1040,7 @@ watchdog(_, [], _, _, _, _) -> ok. watchdog_down(Wd, To, #state{watchdogT = WatchdogT} = S) -> - insert(WatchdogT, Wd#watchdog{state = To}), + ets:insert(WatchdogT, Wd#watchdog{state = To}), connection_down(Wd, To, S). %% --------------------------------------------------------------------------- @@ -948,14 +1052,14 @@ watchdog_down(Wd, To, #state{watchdogT = WatchdogT} = S) -> connection_up({TPid, {Caps, SupportedApps, Pkt}}, #watchdog{pid = Pid} = Wd, - #state{peerT = PeerT} + #state{local = {PeerT, _, _}} = S) -> - Pr = #peer{pid = TPid, - apps = SupportedApps, - caps = Caps, - watchdog = Pid}, - insert(PeerT, Pr), - connection_up([Pkt], Wd#watchdog{peer = TPid}, Pr, S). + Rec = #peer{pid = TPid, + apps = SupportedApps, + caps = Caps, + watchdog = Pid}, + ets:insert(PeerT, Rec), + connection_up([Pkt], Wd#watchdog{peer = TPid}, Rec, S). %% --------------------------------------------------------------------------- %% # reopen/3 @@ -965,22 +1069,23 @@ reopen({TPid, {Caps, SupportedApps, _Pkt}}, #watchdog{pid = Pid} = Wd, #state{watchdogT = WatchdogT, - peerT = PeerT}) -> - insert(PeerT, #peer{pid = TPid, - apps = SupportedApps, - caps = Caps, - watchdog = Pid}), - insert(WatchdogT, Wd#watchdog{state = ?WD_REOPEN, - peer = TPid}). + local = {PeerT, _, _}}) -> + ets:insert(PeerT, #peer{pid = TPid, + apps = SupportedApps, + caps = Caps, + watchdog = Pid}), + ets:insert(WatchdogT, Wd#watchdog{state = ?WD_REOPEN, + peer = TPid}). %% --------------------------------------------------------------------------- %% # connection_up/2 %% --------------------------------------------------------------------------- -%% Watchdog has recovered as suspect connection. Note that there has +%% Watchdog has recovered a suspect connection. Note that there has %% been no new capabilties exchange in this case. -connection_up(#watchdog{peer = TPid} = Wd, #state{peerT = PeerT} = S) -> +connection_up(#watchdog{peer = TPid} = Wd, #state{local = {PeerT, _, _}} + = S) -> connection_up([], Wd, fetch(PeerT, TPid), S). %% connection_up/4 @@ -991,23 +1096,23 @@ connection_up(Extra, #peer{apps = SApps, caps = Caps} = Pr, #state{watchdogT = WatchdogT, - local_peers = LDict, + local = LT, service_name = SvcName, service = #diameter_service{applications = Apps}} = S) -> - insert(WatchdogT, Wd#watchdog{state = ?WD_OKAY}), + ets:insert(WatchdogT, Wd#watchdog{state = ?WD_OKAY}), diameter_traffic:peer_up(TPid), - insert_local_peer(SApps, {{TPid, Caps}, {SvcName, Apps}}, LDict), + local_peer_up(SApps, {TPid, Caps}, {SvcName, Apps}, LT), report_status(up, Wd, Pr, S, Extra). -insert_local_peer(SApps, T, LDict) -> - lists:foldl(fun(A,D) -> ilp(A, T, D) end, LDict, SApps). +local_peer_up(SApps, {TPid, Caps} = TC, SA, LT) -> + insert_peer(TPid, [A || {_,A} <- SApps], Caps, LT), + lists:foreach(fun(A) -> peer_up(A, TC, SA) end, SApps). -ilp({Id, Alias}, {TC, SA}, LDict) -> - init_conn(Id, Alias, TC, SA), - ?Dict:append(Alias, TC, LDict). +peer_up({Id, Alias}, TC, SA) -> + peer_up(Id, Alias, TC, SA). -init_conn(Id, Alias, {TPid, _} = TC, {SvcName, Apps}) -> +peer_up(Id, Alias, {TPid, _} = TC, {SvcName, Apps}) -> #diameter_app{id = Id} %% assert = App = find_app(Alias, Apps), @@ -1105,17 +1210,17 @@ connection_down(#watchdog{state = ?WD_OKAY, = Pr, #state{service_name = SvcName, service = #diameter_service{applications = Apps}, - local_peers = LDict} + local = LT} = S) -> report_status(down, Wd, Pr, S, []), - remove_local_peer(SApps, {{TPid, Caps}, {SvcName, Apps}}, LDict), + local_peer_down(SApps, {TPid, Caps}, {SvcName, Apps}, LT), diameter_traffic:peer_down(TPid); connection_down(#watchdog{state = ?WD_OKAY, peer = TPid} = Wd, To, - #state{peerT = PeerT} + #state{local = {PeerT, _, _}} = S) when is_atom(To) -> connection_down(Wd, #peer{} = fetch(PeerT, TPid), S); @@ -1123,15 +1228,14 @@ connection_down(#watchdog{state = ?WD_OKAY, connection_down(#watchdog{}, _, _) -> ok. -remove_local_peer(SApps, T, LDict) -> - lists:foldl(fun(A,D) -> rlp(A, T, D) end, LDict, SApps). +local_peer_down(SApps, {TPid, _Caps} = TC, SA, LT) -> + delete_peer(TPid, LT), + lists:foreach(fun(A) -> peer_down(A, TC, SA) end, SApps). -rlp({Id, Alias}, {TC, SA}, LDict) -> - L = ?Dict:fetch(Alias, LDict), - down_conn(Id, Alias, TC, SA), - ?Dict:store(Alias, lists:delete(TC, L), LDict). +peer_down({Id, Alias}, TC, SA) -> + peer_down(Id, Alias, TC, SA). -down_conn(Id, Alias, TC, {SvcName, Apps}) -> +peer_down(Id, Alias, TC, {SvcName, Apps}) -> #diameter_app{id = Id} %% assert = App = find_app(Alias, Apps), @@ -1156,7 +1260,7 @@ wd_down(#watchdog{peer = B}, _) ok; %% ... or maybe it has. -wd_down(#watchdog{peer = TPid} = Wd, #state{peerT = PeerT} = S) -> +wd_down(#watchdog{peer = TPid} = Wd, #state{local = {PeerT, _, _}} = S) -> connection_down(Wd, ?WD_DOWN, S), ets:delete(PeerT, TPid). @@ -1233,8 +1337,7 @@ start_tc(Tc, T, _) -> tc_timeout({Ref, _Type, _Opts} = T, #state{service_name = SvcName} = S) -> tc(diameter_config:have_transport(SvcName, Ref), T, S). -tc(true, {Ref, Type, Opts}, #state{service_name = SvcName} - = S) -> +tc(true, {Ref, Type, Opts}, #state{service_name = SvcName} = S) -> send_event(SvcName, {reconnect, Ref, Opts}), start(Ref, Type, Opts, S); tc(false = No, _, _) -> %% removed @@ -1364,19 +1467,37 @@ share_peer(up, Caps, Apps, TPid, #state{options = [_, {_,T} | _], service_name = Svc}) -> notify(T, Svc, {peer, TPid, [A || {_,A} <- Apps], Caps}); -share_peer(_, _, _, _, _) -> - ok. +share_peer(down, _Caps, _Apps, TPid, #state{options = [_, {_,T} | _], + service_name = Svc}) -> + notify(T, Svc, {peer, TPid}). %% --------------------------------------------------------------------------- %% # share_peers/2 %% --------------------------------------------------------------------------- -share_peers(Pid, #state{options = [_, {_,T} | _], local_peers = PDict}) -> - is_remote(Pid, T) - andalso ?Dict:fold(fun(A,Ps,ok) -> sp(Pid, A, Ps), ok end, ok, PDict). - -sp(Pid, Alias, Peers) -> - lists:foreach(fun({P,C}) -> Pid ! {peer, P, [Alias], C} end, Peers). +share_peers(Pid, #state{options = [_, {_,SP} | _], + local = {PeerT, AppT, _}}) -> + is_remote(Pid, SP) + andalso ets:foldl(fun(T, N) -> N + sp(Pid, AppT, T) end, + 0, + PeerT). + +%% An entry in the peer table doesn't mean the watchdog state is OKAY, +%% an entry in the app table does. + +sp(Pid, AppT, #peer{pid = TPid, + apps = [{_, Alias} | _] = Apps, + caps = Caps}) -> + Spec = [{{'$1', TPid}, + [{'==', '$1', {const, Alias}}], + ['$_']}], + case ets:select(AppT, Spec, 1) of + '$end_of_table' -> + 0; + _ -> + Pid ! {peer, TPid, [A || {_,A} <- Apps], Caps}, + 1 + end. is_remote(Pid, T) -> Node = node(Pid), @@ -1386,32 +1507,49 @@ is_remote(Pid, T) -> %% # remote_peer_up/4 %% --------------------------------------------------------------------------- -remote_peer_up(Pid, Aliases, Caps, #state{options = [_, _, {_,T} | _]} = S) -> - is_remote(Pid, T) - andalso rpu(Pid, Aliases, Caps, S). +remote_peer_up(TPid, Aliases, Caps, #state{options = [_, _, {_,T} | _]} = S) -> + is_remote(TPid, T) andalso rpu(TPid, Aliases, Caps, S). -rpu(Pid, Aliases, Caps, #state{service = Svc, shared_peers = PDict}) -> +rpu(TPid, Aliases, Caps, #state{service = Svc, remote = RT}) -> #diameter_service{applications = Apps} = Svc, Key = #diameter_app.alias, F = fun(A) -> lists:keymember(A, Key, Apps) end, - rpu(Pid, lists:filter(F, Aliases), Caps, PDict); + rpu(TPid, lists:filter(F, Aliases), Caps, RT); rpu(_, [] = No, _, _) -> No; -rpu(Pid, Aliases, Caps, PDict) -> - erlang:monitor(process, Pid), - T = {Pid, Caps}, - lists:foreach(fun(A) -> ?Dict:append(A, T, PDict) end, Aliases). +rpu(TPid, Aliases, Caps, {PeerT, _, _} = RT) -> + monitor(process, TPid), + ets:insert(PeerT, #peer{pid = TPid, + apps = Aliases, + caps = Caps}), + insert_peer(TPid, Aliases, Caps, RT). + +%% insert_peer/4 + +insert_peer(TPid, Aliases, Caps, {_PeerT, AppT, IdentT}) -> + #diameter_caps{origin_host = {_, OH}, + origin_realm = {_, OR}} + = Caps, + ets:insert(AppT, [{A, TPid} || A <- Aliases]), + H = iolist_to_binary(OH), + R = iolist_to_binary(OR), + ets:insert(IdentT, [{T, A, TPid} || T <- [{host, H}, {realm, R}, {R, H}], + A <- Aliases]). %% --------------------------------------------------------------------------- %% # remote_peer_down/2 %% --------------------------------------------------------------------------- -remote_peer_down(Pid, #state{shared_peers = PDict}) -> - lists:foreach(fun(A) -> rpd(Pid, A, PDict) end, ?Dict:fetch_keys(PDict)). +remote_peer_down(TPid, #state{remote = {PeerT, _, _} = RT}) -> + ets:delete(PeerT, TPid), + delete_peer(TPid, RT). + +%% delete_peer/2 -rpd(Pid, Alias, PDict) -> - ?Dict:update(Alias, fun(Ps) -> lists:keydelete(Pid, 1, Ps) end, PDict). +delete_peer(TPid, {_PeerT, AppT, IdentT}) -> + ets:select_delete(AppT, [{{'_', TPid}, [], [true]}]), + ets:select_delete(IdentT, [{{'_', '_', TPid}, [], [true]}]). %% --------------------------------------------------------------------------- %% pick_peer/4 @@ -1421,12 +1559,12 @@ pick_peer(#diameter_app{alias = Alias} = App, RealmAndHost, Filter, - #state{local_peers = L, - shared_peers = S, + #state{local = LT, + remote = RT, service_name = SvcName, service = #diameter_service{pid = Pid}}) -> - pick_peer(peers(Alias, RealmAndHost, Filter, L), - peers(Alias, RealmAndHost, Filter, S), + pick_peer(peers(Alias, RealmAndHost, Filter, LT), + peers(Alias, RealmAndHost, Filter, RT), Pid, SvcName, App). @@ -1491,54 +1629,196 @@ pick_peer(Local, %% peers/4 -peers(Alias, RH, Filter, Peers) -> - case ?Dict:find(Alias, Peers) of - {ok, L} -> - filter(L, RH, Filter); - error -> +peers(Alias, RH, Filter, T) -> + filter(Alias, RH, Filter, T, true). + +%% filter/5 +%% +%% Try to limit the peers list by starting with a host/realm lookup. + +filter(Alias, RH, {neg, F}, T, B) -> + filter(Alias, RH, F, T, not B); + +filter(_, _, none, _, false) -> + []; + +filter(Alias, _, none, T, true) -> + all_peers(Alias, T); + +filter(Alias, [DR,DH] = RH, K, T, B) + when K == realm, DR == undefined; + K == host, DH == undefined -> + filter(Alias, RH, none, T, B); + +filter(Alias, [DR,_] = RH, realm = K, T, B) -> + filter(Alias, RH, {K, DR}, T, B); + +filter(Alias, [_,DH] = RH, host = K, T, B) -> + filter(Alias, RH, {K, DH}, T, B); + +filter(Alias, _, {K, D}, {PeerT, _AppT, IdentT}, true) + when K == host; + K == realm -> + try iolist_to_binary(D) of + B -> + caps(PeerT, ets:select(IdentT, [{{{K, B}, '$1', '$2'}, + [{'==', '$1', {const, Alias}}], + ['$2']}])) + catch + error:_ -> [] - end. + end; -%% filter/3 +filter(Alias, RH, {all, Filters}, T, B) + when is_list(Filters) -> + fltr_all(Alias, RH, Filters, T, B); + +filter(Alias, RH, {first, Filters}, T, B) + when is_list(Filters) -> + fltr_first(Alias, RH, Filters, T, B); + +filter(Alias, RH, Filter, T, B) -> + {Ts, Fs} = filter(all_peers(Alias, T), RH, Filter), + choose(B, Ts, Fs). + +%% fltr_all/5 + +fltr_all(Alias, RH, [{K, any} | Filters], T, B) + when K == host; + K == realm -> + fltr_all(Alias, RH, Filters, T, B); + +fltr_all(Alias, RH, [{host, _} = H, {realm, _} = R | Filters], T, B) -> + fltr_all(Alias, RH, [R, H | Filters], T, B); + +fltr_all(Alias, RH, [{realm, _} = R, {host, any} | Filters], T, B) -> + fltr_all(Alias, RH, [R | Filters], T, B); + +fltr_all(Alias, RH, [{realm, OR}, {host, OH} | Filters], T, true) -> + {PeerT, _AppT, IdentT} = T, + try {iolist_to_binary(OR), iolist_to_binary(OH)} of + BT -> + Peers = caps(PeerT, + ets:select(IdentT, [{{BT, '$1', '$2'}, + [{'==', '$1', {const, Alias}}], + ['$2']}])), + {Ts, _} = filter(Peers, RH, {all, Filters}), + Ts + catch + error:_ -> + [] + end; + +fltr_all(Alias, [undefined,_] = RH, [realm | Filters], T, B) -> + fltr_all(Alias, RH, Filters, T, B); + +fltr_all(Alias, [DR,_] = RH, [realm | Filters], T, B) -> + fltr_all(Alias, RH, [{realm, DR} | Filters], T, B); + +fltr_all(Alias, [_,undefined] = RH, [host | Filters], T, B) -> + fltr_all(Alias, RH, Filters, T, B); + +fltr_all(Alias, [_,DH] = RH, [host | Filters], T, B) -> + fltr_all(Alias, RH, [{host, DH} | Filters], T, B); + +fltr_all(Alias, RH, [{K, _} = KT, KA | Filters], T, B) + when K == host, KA == realm; + K == realm, KA == host -> + fltr_all(Alias, RH, [KA, KT | Filters], T, B); + +fltr_all(Alias, RH, [F | Filters], T, B) -> + {Ts, Fs} = filter(filter(Alias, RH, F, T, B), RH, {all, Filters}), + choose(B, Ts, Fs); + +fltr_all(Alias, RH, [], T, B) -> + filter(Alias, RH, none, T, B). + +%% fltr_first/5 %% -%% Return peers in match order. +%% Like any, but stop at the first filter with any matches. + +fltr_first(Alias, RH, [F | Filters], T, B) -> + case filter(Alias, RH, F, T, B) of + [] -> + fltr_first(Alias, RH, Filters, T, B); + [_|_] = Ts -> + Ts + end; + +fltr_first(Alias, RH, [], T, B) -> + filter(Alias, RH, none, T, not B). -filter(Peers, RH, Filter) -> - {Ts, _} = fltr(Peers, RH, Filter), - Ts. +%% all_peers/2 -%% fltr/4 +all_peers(Alias, {PeerT, AppT, _}) -> + ets:select(PeerT, [{#peer{pid = P, caps = '$1', _ = '_'}, + [], + [{{P, '$1'}}]} + || {_,P} <- ets:lookup(AppT, Alias)]). + +%% caps/2 + +caps(PeerT, Pids) -> + ets:select(PeerT, [{#peer{pid = P, caps = '$1', _ = '_'}, + [], + [{{P, '$1'}}]} + || P <- Pids]). + +%% filter/3 +%% +%% Return peers in match order. -fltr(Peers, _, none) -> +filter(Peers, _, none) -> {Peers, []}; -fltr(Peers, RH, {neg, F}) -> - {Ts, Fs} = fltr(Peers, RH, F), +filter(Peers, RH, {neg, F}) -> + {Ts, Fs} = filter(Peers, RH, F), {Fs, Ts}; -fltr(Peers, RH, {all, L}) +filter(Peers, RH, {all, L}) when is_list(L) -> lists:foldl(fun(F,A) -> fltr_all(F, A, RH) end, {Peers, []}, L); -fltr(Peers, RH, {any, L}) +filter(Peers, RH, {any, L}) when is_list(L) -> lists:foldl(fun(F,A) -> fltr_any(F, A, RH) end, {[], Peers}, L); -fltr(Peers, RH, F) -> +filter(Peers, RH, {first, L}) + when is_list(L) -> + fltr_first(Peers, RH, L); + +filter(Peers, RH, F) -> lists:partition(fun({_,C}) -> caps_filter(C, RH, F) end, Peers). +%% fltr_all/3 + fltr_all(F, {Ts0, Fs0}, RH) -> - {Ts1, Fs1} = fltr(Ts0, RH, F), + {Ts1, Fs1} = filter(Ts0, RH, F), {Ts1, Fs0 ++ Fs1}. +%% fltr_any/3 + fltr_any(F, {Ts0, Fs0}, RH) -> - {Ts1, Fs1} = fltr(Fs0, RH, F), + {Ts1, Fs1} = filter(Fs0, RH, F), {Ts0 ++ Ts1, Fs1}. +%% fltr_first/3 + +fltr_first(Peers, _, []) -> + {[], Peers}; + +fltr_first(Peers, RH, [F | Filters]) -> + case filter(Peers, RH, F) of + {[], _} -> + fltr_first(Peers, RH, Filters); + {_, _} = T -> + T + end. + %% caps_filter/3 caps_filter(#diameter_caps{origin_host = {_,OH}}, [_,DH], host) -> @@ -1582,8 +1862,8 @@ eq(Any, Id, PeerId) -> transports(#state{watchdogT = WatchdogT}) -> ets:select(WatchdogT, [{#watchdog{peer = '$1', _ = '_'}, - [{'is_pid', '$1'}], - ['$1']}]). + [{'is_pid', '$1'}], + ['$1']}]). %% --------------------------------------------------------------------------- %% # service_info/2 @@ -1636,7 +1916,7 @@ tagged_info(Item, S) undefined end; -tagged_info(TPid, #state{watchdogT = WatchdogT, peerT = PeerT}) +tagged_info(TPid, #state{watchdogT = WatchdogT, local = {PeerT, _, _}}) when is_pid(TPid) -> try [#peer{watchdog = Pid}] = ets:lookup(PeerT, TPid), @@ -1786,7 +2066,7 @@ keys(connect = T, Opts) -> keys(_, _) -> [{listen, accept}]. -peer_dict(#state{watchdogT = WatchdogT, peerT = PeerT}, Dict0) -> +peer_dict(#state{watchdogT = WatchdogT, local = {PeerT, _, _}}, Dict0) -> try ets:tab2list(WatchdogT) of L -> lists:foldl(fun(T,A) -> peer_acc(PeerT, A, T) end, Dict0, L) catch diff --git a/lib/diameter/src/base/diameter_session.erl b/lib/diameter/src/base/diameter_session.erl index 4cd76ed1f1..d854bc36a5 100644 --- a/lib/diameter/src/base/diameter_session.erl +++ b/lib/diameter/src/base/diameter_session.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2012. 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. @@ -158,10 +158,9 @@ session_id(Host) -> %% --------------------------------------------------------------------------- init() -> - {Now, Seed} = diameter_lib:seed(), - random:seed(Seed), + Now = diameter_lib:timestamp(), Time = time32(Now), - Seq = (?INT32 band (Time bsl 20)) bor (random:uniform(1 bsl 20) - 1), + Seq = (?INT32 band (Time bsl 20)) bor (rand:uniform(1 bsl 20) - 1), ets:insert(diameter_sequence, [{origin_state_id, Time}, {session_base, Time bsl 32}, {sequence, Seq}]), diff --git a/lib/diameter/src/base/diameter_stats.erl b/lib/diameter/src/base/diameter_stats.erl index 83e562e7fe..8c10464e98 100644 --- a/lib/diameter/src/base/diameter_stats.erl +++ b/lib/diameter/src/base/diameter_stats.erl @@ -25,9 +25,6 @@ -module(diameter_stats). -behaviour(gen_server). --compile({no_auto_import, [now/0]}). --import(diameter_lib, [now/0]). - -export([reg/2, reg/1, incr/3, incr/1, read/1, @@ -61,7 +58,7 @@ -define(SERVER, ?MODULE). %% Server state. --record(state, {id = now()}). +-record(state, {id = diameter_lib:now()}). -type counter() :: any(). -type ref() :: any(). @@ -143,9 +140,14 @@ read(Refs, B) -> L. to_refdict(L) -> - lists:foldl(fun({{C,R}, N}, D) -> orddict:append(R, {C,N}, D) end, - orddict:new(), - L). + lists:foldl(fun append/2, orddict:new(), L). + +%% Order both references and counters in the returned list. +append({{Ctr, Ref}, N}, Dict) -> + orddict:update(Ref, + fun(D) -> orddict:store(Ctr, N, D) end, + [{Ctr, N}], + Dict). %% --------------------------------------------------------------------------- %% # sum(Refs) @@ -221,7 +223,7 @@ uptime() -> %% ---------------------------------------------------------- init([]) -> - ets:new(?TABLE, [named_table, ordered_set, public]), + ets:new(?TABLE, [named_table, set, public, {write_concurrency, true}]), {ok, #state{}}. %% ---------------------------------------------------------- diff --git a/lib/diameter/src/base/diameter_sup.erl b/lib/diameter/src/base/diameter_sup.erl index e89ede9843..482289cb9a 100644 --- a/lib/diameter/src/base/diameter_sup.erl +++ b/lib/diameter/src/base/diameter_sup.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. @@ -34,6 +34,7 @@ -export([init/1]). -define(CHILDREN, [diameter_misc_sup, + diameter_config_sup, diameter_watchdog_sup, diameter_peer_fsm_sup, diameter_transport_sup, diff --git a/lib/diameter/src/base/diameter_sync.erl b/lib/diameter/src/base/diameter_sync.erl index 3309224399..7fb6888e21 100644 --- a/lib/diameter/src/base/diameter_sync.erl +++ b/lib/diameter/src/base/diameter_sync.erl @@ -28,9 +28,6 @@ -module(diameter_sync). -behaviour(gen_server). --compile({no_auto_import, [now/0]}). --import(diameter_lib, [now/0]). - -export([call/4, call/5, cast/4, cast/5, carp/1, carp/2]). @@ -73,7 +70,7 @@ %% Server state. -record(state, - {time = now(), + {time = diameter_lib:now(), pending = 0 :: non_neg_integer(), %% outstanding requests monitor = new() :: ets:tid(), %% MonitorRef -> {Name, From} queue = new() :: ets:tid()}). %% Name -> queue of {Pid, Ref} diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl index 692a01e651..d93a3e71e3 100644 --- a/lib/diameter/src/base/diameter_traffic.erl +++ b/lib/diameter/src/base/diameter_traffic.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2013-2015. All Rights Reserved. +%% Copyright Ericsson AB 2013-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. @@ -41,7 +41,6 @@ -export([make_recvdata/1, peer_up/1, peer_down/1, - failover/1, pending/1]). %% towards ?MODULE @@ -80,6 +79,7 @@ apps :: [#diameter_app{}], sequence :: diameter:sequence(), codec :: [{string_decode, boolean()} + | {strict_mbit, boolean()} | {incoming_maxlen, diameter:message_length()}]}). %% Note that incoming_maxlen is currently handled in diameter_peer_fsm, %% so that any message exceeding the maximum is discarded. Retain the @@ -106,7 +106,8 @@ make_recvdata([SvcName, PeerT, Apps, SvcOpts | _]) -> sequence = Mask, codec = [T || {K,_} = T <- SvcOpts, lists:member(K, [string_decode, - incoming_maxlen])]}. + incoming_maxlen, + strict_mbit])]}. %% --------------------------------------------------------------------------- %% peer_up/1 @@ -122,8 +123,16 @@ peer_up(TPid) -> %% --------------------------------------------------------------------------- peer_down(TPid) -> - ets:delete(?REQUEST_TABLE, TPid), - failover(TPid). + ets:delete_object(?REQUEST_TABLE, {TPid}), + lists:foreach(fun failover/1, ets:lookup(?REQUEST_TABLE, TPid)). +%% Note that a request process can store its request after failover +%% notifications are sent here: insert_request/2 sends the notification +%% in that case. + +%% failover/1 + +failover({_TPid, {Pid, TRef}}) -> + Pid ! {failover, TRef}. %% --------------------------------------------------------------------------- %% incr/4 @@ -167,7 +176,7 @@ incr_error(Dir, Id, TPid, _) -> incr_error(Dir, Id, TPid) -> incr(TPid, {Id, Dir, error}). - + %% --------------------------------------------------------------------------- %% incr_rc/4 %% --------------------------------------------------------------------------- @@ -229,7 +238,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, @@ -243,11 +260,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 @@ -262,7 +286,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 @@ -272,12 +296,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 @@ -900,7 +919,7 @@ failed(Rec, FailedAvp, Dict) -> {'Failed-AVP', [FailedAvp]} catch error: _ -> - Avps = Dict:'get-'('AVP', Rec), + Avps = Dict:'#get-'('AVP', Rec), A = #diameter_avp{name = 'Failed-AVP', value = FailedAvp}, {'AVP', [A|Avps]} @@ -1441,7 +1460,7 @@ make_request_packet(#diameter_packet{header = Hdr} = Pkt, make_request_packet(Msg, Pkt) -> Pkt#diameter_packet{msg = Msg}. -%% make_retransmit_packet/2 +%% make_retransmit_packet/1 make_retransmit_packet(#diameter_packet{msg = [#diameter_header{} = Hdr | Avps]} @@ -1483,16 +1502,12 @@ send_R(Pkt0, caps = Caps, packet = Pkt0}, - try - incr(send, Pkt, TPid, AppDict), - TRef = send_request(TPid, Pkt, Req, SvcName, Timeout), - Pid ! Ref, %% tell caller a send has been attempted - handle_answer(SvcName, - App, - recv_A(Timeout, SvcName, App, Opts, {TRef, Req})) - after - erase_requests(Pkt) - end. + incr(send, Pkt, TPid, AppDict), + TRef = send_request(TPid, Pkt, Req, SvcName, Timeout), + Pid ! Ref, %% tell caller a send has been attempted + handle_answer(SvcName, + App, + recv_A(Timeout, SvcName, App, Opts, {TRef, Req})). %% recv_A/5 @@ -1568,6 +1583,8 @@ answer(Pkt, Req) -> a(Pkt, SvcName, ModX, AE, Req). +-spec a(_, _, _) -> no_return(). %% silence dialyzer + a(#diameter_packet{errors = Es} = Pkt, SvcName, @@ -1692,9 +1709,15 @@ encode(_, _, #diameter_packet{} = Pkt) -> 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), + Seqs = diameter_codec:sequence_numbers(Bin), + TRef = erlang:start_timer(Timeout, self(), TPid), + Entry = {Seqs, #request{handler = Pid} = Req, TRef}, + + %% Ensure that request table is cleaned even if the process is + %% killed. + spawn(fun() -> diameter_lib:wait([Pid]), delete_request(Entry) end), + + insert_request(Entry), send(TPid, Pkt), TRef; @@ -1709,31 +1732,21 @@ send_request(TPid, #diameter_packet{} = Pkt, Req, SvcName, Timeout) -> %% send/1 send({TPid, Pkt, #request{handler = Pid} = Req0, SvcName, Timeout, TRef}) -> - Seqs = diameter_codec:sequence_numbers(Pkt), Req = Req0#request{handler = self()}, - Ref = send_request(TPid, Pkt, Req, SvcName, Timeout), - - try - recv(TPid, Pid, TRef, Ref) - after - %% Remove only the entry for this specific send since a resend - %% from the originating node can pick another transport on - %% this one. - ets:delete_object(?REQUEST_TABLE, {Seqs, Req, Ref}) - end. + recv(TPid, Pid, TRef, send_request(TPid, Pkt, Req, SvcName, Timeout)). %% recv/4 %% %% Relay an answer from a remote node. -recv(TPid, Pid, TRef, Ref) -> +recv(TPid, Pid, TRef, LocalTRef) -> receive {answer, _, _, _, _} = A -> Pid ! A; - {failover = T, Ref} -> + {failover = T, LocalTRef} -> Pid ! {T, TRef}; T -> - exit({timeout, Ref, TPid} = T) + exit({timeout, LocalTRef, TPid} = T) end. %% send/2 @@ -1766,6 +1779,8 @@ retransmit({TPid, Caps, App} SvcName, Timeout, []). +%% When sending a binary, it's up to prepare_retransmit to modify it +%% accordingly. retransmit({send, Msg}, Transport, @@ -1810,17 +1825,27 @@ resend_request(Pkt0, TRef = send_request(TPid, Pkt, Req, SvcName, Tmo), {TRef, Req}. -%% store_request/4 +%% insert_request/1 -store_request(TPid, Bin, Req, Timeout) -> - Seqs = diameter_codec:sequence_numbers(Bin), - TRef = erlang:start_timer(Timeout, self(), TPid), - ets:insert(?REQUEST_TABLE, {Seqs, Req, TRef}), - ets:member(?REQUEST_TABLE, TPid) - orelse (self() ! {failover, TRef}), %% failover/1 may have missed - TRef. +insert_request({_Seqs, #request{transport = TPid}, TRef} = T) -> + ets:insert(?REQUEST_TABLE, [T, {TPid, {self(), TRef}}]), + is_peer_up(TPid) + orelse (self() ! {failover, TRef}). %% failover/1 may have missed + +%% is_peer_up/1 +%% +%% Is the entry written by peer_up/1 and deleted by peer_down/1 still +%% in the request table? + +is_peer_up(TPid) -> + Spec = [{{TPid}, [], ['$_']}], + '$end_of_table' /= ets:select(?REQUEST_TABLE, Spec, 1). %% lookup_request/2 +%% +%% Note the match on both the key and transport pid. The latter is +%% necessary since the same Hop-by-Hop and End-to-End identifiers are +%% reused in the case of retransmission. lookup_request(Msg, TPid) -> Seqs = diameter_codec:sequence_numbers(Msg), @@ -1834,16 +1859,11 @@ lookup_request(Msg, TPid) -> false end. -%% erase_requests/1 - -erase_requests(Pkt) -> - ets:delete(?REQUEST_TABLE, diameter_codec:sequence_numbers(Pkt)). - -%% match_requests/1 +%% delete_request/1 -match_requests(TPid) -> - Pat = {'_', #request{transport = TPid, _ = '_'}, '_'}, - ets:select(?REQUEST_TABLE, [{Pat, [], ['$_']}]). +delete_request({_Seqs, #request{handler = Pid, transport = TPid}, TRef} = T) -> + Spec = [{R, [], [true]} || R <- [T, {TPid, {Pid, TRef}}]], + ets:select_delete(?REQUEST_TABLE, Spec). %% have_request/2 @@ -1852,28 +1872,6 @@ have_request(Pkt, TPid) -> Pat = {Seqs, #request{transport = TPid, _ = '_'}, '_'}, '$end_of_table' /= ets:select(?REQUEST_TABLE, [{Pat, [], ['$_']}], 1). -%% --------------------------------------------------------------------------- -%% # failover/1-2 -%% --------------------------------------------------------------------------- - -failover(TPid) - when is_pid(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. - -%% 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.) - %% get_destination/2 get_destination(Dict, Msg) -> diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl index c6d0a0b6ed..2ba60a65fb 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. @@ -66,7 +66,9 @@ %% end PCB parent = self() :: pid(), %% service process transport :: pid() | undefined, %% peer_fsm process - tref :: reference(), %% reference for current watchdog timer + tref :: reference() %% reference for current watchdog timer + | integer() %% monotonic time + | undefined, dictionary :: module(), %% common dictionary receive_data :: term(), %% term passed into diameter_service with incoming message @@ -94,7 +96,7 @@ start({_,_} = Type, T) -> Ack = make_ref(), {ok, Pid} = diameter_watchdog_sup:start_child({Ack, Type, self(), T}), try - {erlang:monitor(process, Pid), Pid} + {monitor(process, Pid), Pid} after send(Pid, Ack) end. @@ -121,10 +123,8 @@ i({Ack, T, Pid, {RecvData, #diameter_service{applications = Apps, capabilities = Caps} = Svc}}) -> - erlang:monitor(process, Pid), + monitor(process, Pid), wait(Ack, Pid), - {_, Seed} = diameter_lib:seed(), - random:seed(Seed), putr(restart, {T, Opts, Svc, SvcOpts}), %% save seeing it in trace putr(dwr, dwr(Caps)), %% {_,_} = Mask = proplists:get_value(sequence, SvcOpts), @@ -246,11 +246,16 @@ handle_info(T, #watchdog{} = State) -> event(T, State, S), %% before 'watchdog' {noreply, S}; stop -> - ?LOG(stop, T), + ?LOG(stop, truncate(T)), event(T, State, State#watchdog{status = down}), {stop, {shutdown, T}, State} end. +truncate({'DOWN' = T, _, process, Pid, _}) -> + {T, Pid}; +truncate(T) -> + T. + close({'DOWN', _, process, TPid, {shutdown, Reason}}, #watchdog{transport = TPid, parent = Pid}) -> @@ -442,16 +447,23 @@ 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) -> - set_watchdog(timeout(S)); + set_watchdog(0, timeout(S)); -%% Timer was canceled after message was already sent. -transition({timeout, _, tw}, #watchdog{}) -> - ok; +%% Message has arrived since the timer was started: subtract time +%% already elapsed from new timer. +transition({timeout, _, tw}, #watchdog{tref = T0} = S) -> + set_watchdog(diameter_lib:micro_diff(T0) div 1000, S); %% State query. transition({state, Pid}, #watchdog{status = S}) -> @@ -527,22 +539,31 @@ role() -> %% set_watchdog/1 -set_watchdog(#watchdog{tw = TwInit, - tref = TRef} - = S) -> - cancel(TRef), - S#watchdog{tref = erlang:start_timer(tw(TwInit), self(), tw)}; -set_watchdog(stop = No) -> - No. +%% Timer not yet set. +set_watchdog(#watchdog{tref = undefined} = S) -> + set_watchdog(0, S); -cancel(undefined) -> - ok; -cancel(TRef) -> - erlang:cancel_timer(TRef). +%% Timer already set: start at new one only at expiry. +set_watchdog(#watchdog{} = S) -> + S#watchdog{tref = diameter_lib:now()}. + +%% set_watchdog/2 + +set_watchdog(_, stop = No) -> + No; + +set_watchdog(Ms, #watchdog{tw = TwInit} = S) -> + S#watchdog{tref = erlang:start_timer(tw(TwInit, Ms), self(), tw)}. + +%% A callback could return anything, so ensure the result isn't +%% negative. Don't prevent abuse, even though the smallest valid +%% timeout is 4000. +tw(TwInit, Ms) -> + max(tw(TwInit) - Ms, 0). tw(T) when is_integer(T), T >= 6000 -> - T - 2000 + (random:uniform(4001) - 1); %% RFC3539 jitter of +/- 2 sec. + T - 2000 + (rand:uniform(4001) - 1); %% RFC3539 jitter of +/- 2 sec. tw({M,F,A}) -> apply(M,F,A). @@ -561,22 +582,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), @@ -593,32 +624,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 %% @@ -635,20 +664,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 @@ -704,7 +733,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 %% diff --git a/lib/diameter/src/base/diameter_watchdog_sup.erl b/lib/diameter/src/base/diameter_watchdog_sup.erl index 5d24e12f19..7b6669f381 100644 --- a/lib/diameter/src/base/diameter_watchdog_sup.erl +++ b/lib/diameter/src/base/diameter_watchdog_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/compiler/diameter_codegen.erl b/lib/diameter/src/compiler/diameter_codegen.erl index cdaa9aa7f9..4007d6b7b1 100644 --- a/lib/diameter/src/compiler/diameter_codegen.erl +++ b/lib/diameter/src/compiler/diameter_codegen.erl @@ -790,20 +790,7 @@ header() -> ("%% -------------------------------------------------------------------\n" "%% This is a generated file.\n" "%% -------------------------------------------------------------------\n" - "\n" - "%%\n" - "%% Copyright (c) Ericsson AB. All rights reserved.\n" - "%%\n" - "%% The information in this document is the property of Ericsson.\n" - "%%\n" - "%% Except as specifically authorized in writing by Ericsson, the\n" - "%% receiver of this document shall keep the information contained\n" - "%% herein confidential and shall protect the same in whole or in\n" - "%% part from disclosure and dissemination to third parties.\n" - "%%\n" - "%% Disclosure and disseminations to the receivers employees shall\n" - "%% only be made on a strict need to know basis.\n" - "%%\n\n"). + "\n"). hrl_header(Name) -> header() ++ "-hrl_name('" ++ ?S(Name) ++ ".hrl').\n". diff --git a/lib/diameter/src/compiler/diameter_dict_parser.yrl b/lib/diameter/src/compiler/diameter_dict_parser.yrl index ef8d58d63b..4d2eb09831 100644 --- a/lib/diameter/src/compiler/diameter_dict_parser.yrl +++ b/lib/diameter/src/compiler/diameter_dict_parser.yrl @@ -2,7 +2,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/compiler/diameter_dict_scanner.erl b/lib/diameter/src/compiler/diameter_dict_scanner.erl index aeedc89d83..a9abb67640 100644 --- a/lib/diameter/src/compiler/diameter_dict_scanner.erl +++ b/lib/diameter/src/compiler/diameter_dict_scanner.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/compiler/diameter_dict_util.erl b/lib/diameter/src/compiler/diameter_dict_util.erl index 9525393128..f9f2b02e94 100644 --- a/lib/diameter/src/compiler/diameter_dict_util.erl +++ b/lib/diameter/src/compiler/diameter_dict_util.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. 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. diff --git a/lib/diameter/src/compiler/diameter_exprecs.erl b/lib/diameter/src/compiler/diameter_exprecs.erl index 2a04917995..9a0cb6baf2 100644 --- a/lib/diameter/src/compiler/diameter_exprecs.erl +++ b/lib/diameter/src/compiler/diameter_exprecs.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2013. 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. diff --git a/lib/diameter/src/compiler/diameter_make.erl b/lib/diameter/src/compiler/diameter_make.erl index d9709029ae..eae40dbafd 100644 --- a/lib/diameter/src/compiler/diameter_make.erl +++ b/lib/diameter/src/compiler/diameter_make.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. 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. diff --git a/lib/diameter/src/compiler/diameter_vsn.hrl b/lib/diameter/src/compiler/diameter_vsn.hrl index 2efac98bff..5201e6b84d 100644 --- a/lib/diameter/src/compiler/diameter_vsn.hrl +++ b/lib/diameter/src/compiler/diameter_vsn.hrl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/depend.sed b/lib/diameter/src/depend.sed index 5adf7f05d5..e385d7db4a 100644 --- a/lib/diameter/src/depend.sed +++ b/lib/diameter/src/depend.sed @@ -1,7 +1,7 @@ # # %CopyrightBegin% # -# Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/diameter.app.src b/lib/diameter/src/diameter.app.src index 49bfd803e7..d380ebbd92 100644 --- a/lib/diameter/src/diameter.app.src +++ b/lib/diameter/src/diameter.app.src @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. 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. diff --git a/lib/diameter/src/diameter.appup.src b/lib/diameter/src/diameter.appup.src index 86d231179c..b1b8e38d39 100644 --- a/lib/diameter/src/diameter.appup.src +++ b/lib/diameter/src/diameter.appup.src @@ -2,7 +2,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. @@ -42,7 +42,18 @@ {"1.8", [{restart_application, diameter}]}, %% 17.4 {"1.9", [{restart_application, diameter}]}, %% 17.5 {"1.9.1", [{restart_application, diameter}]}, %% 17.5.3 - {"1.9.2", [{restart_application, diameter}]} %% 17.5.5 + {"1.9.2", [{restart_application, diameter}]}, %% 17.5.5 + {"1.9.2.1", [{restart_application, diameter}]}, %% 17.5.6.3 + {"1.9.2.2", [{restart_application, diameter}]}, %% 17.5.6.7 + {"1.9.2.3", [{restart_application, diameter}]}, %% 17.5.6.8 + {"1.10", [{restart_application, diameter}]}, %% 18.0 + {"1.11", [{restart_application, diameter}]}, %% 18.1 + {"1.11.1", [{restart_application, diameter}]}, %% 18.2 + {"1.11.2", [{restart_application, diameter}]}, %% 18.3 + {"1.12", [{load_module, diameter_lib}, %% 19.0 + {load_module, diameter_traffic}, + {load_module, diameter_tcp}, + {load_module, diameter_sctp}]} ], [ {"0.9", [{restart_application, diameter}]}, @@ -66,6 +77,17 @@ {"1.8", [{restart_application, diameter}]}, {"1.9", [{restart_application, diameter}]}, {"1.9.1", [{restart_application, diameter}]}, - {"1.9.2", [{restart_application, diameter}]} + {"1.9.2", [{restart_application, diameter}]}, + {"1.9.2.1", [{restart_application, diameter}]}, + {"1.9.2.2", [{restart_application, diameter}]}, + {"1.9.2.3", [{restart_application, diameter}]}, + {"1.10", [{restart_application, diameter}]}, + {"1.11", [{restart_application, diameter}]}, + {"1.11.1", [{restart_application, diameter}]}, + {"1.11.2", [{restart_application, diameter}]}, + {"1.12", [{load_module, diameter_sctp}, + {load_module, diameter_tcp}, + {load_module, diameter_traffic}, + {load_module, diameter_lib}]} ] }. diff --git a/lib/diameter/src/dict/acct_rfc6733.dia b/lib/diameter/src/dict/acct_rfc6733.dia index 4eb326ce88..7690958932 100644 --- a/lib/diameter/src/dict/acct_rfc6733.dia +++ b/lib/diameter/src/dict/acct_rfc6733.dia @@ -1,7 +1,7 @@ ;; ;; %CopyrightBegin% ;; -;; Copyright Ericsson AB 2013. All Rights Reserved. +;; Copyright Ericsson AB 2013-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. diff --git a/lib/diameter/src/dict/base_accounting.dia b/lib/diameter/src/dict/base_accounting.dia index 839add8764..b719bcd128 100644 --- a/lib/diameter/src/dict/base_accounting.dia +++ b/lib/diameter/src/dict/base_accounting.dia @@ -1,7 +1,7 @@ ;; ;; %CopyrightBegin% ;; -;; Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/dict/base_rfc3588.dia b/lib/diameter/src/dict/base_rfc3588.dia index f7316208a6..ef4c8388c3 100644 --- a/lib/diameter/src/dict/base_rfc3588.dia +++ b/lib/diameter/src/dict/base_rfc3588.dia @@ -1,7 +1,7 @@ ;; ;; %CopyrightBegin% ;; -;; Copyright Ericsson AB 2010-2013. 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. diff --git a/lib/diameter/src/dict/base_rfc6733.dia b/lib/diameter/src/dict/base_rfc6733.dia index d92760711c..a120b4eece 100644 --- a/lib/diameter/src/dict/base_rfc6733.dia +++ b/lib/diameter/src/dict/base_rfc6733.dia @@ -1,7 +1,7 @@ ;; ;; %CopyrightBegin% ;; -;; Copyright Ericsson AB 2013. All Rights Reserved. +;; Copyright Ericsson AB 2013-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. diff --git a/lib/diameter/src/dict/capup_rfc6737.dia b/lib/diameter/src/dict/capup_rfc6737.dia index 396c7de9ac..a473cc9d22 100644 --- a/lib/diameter/src/dict/capup_rfc6737.dia +++ b/lib/diameter/src/dict/capup_rfc6737.dia @@ -1,7 +1,7 @@ ;; ;; %CopyrightBegin% ;; -;; Copyright Ericsson AB 2013. All Rights Reserved. +;; Copyright Ericsson AB 2013-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. diff --git a/lib/diameter/src/dict/relay.dia b/lib/diameter/src/dict/relay.dia index 23772fedf8..c7320cc737 100644 --- a/lib/diameter/src/dict/relay.dia +++ b/lib/diameter/src/dict/relay.dia @@ -1,7 +1,7 @@ ;; ;; %CopyrightBegin% ;; -;; Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/info/diameter_dbg.erl b/lib/diameter/src/info/diameter_dbg.erl index 4f5c91d24f..e1d2086871 100644 --- a/lib/diameter/src/info/diameter_dbg.erl +++ b/lib/diameter/src/info/diameter_dbg.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. 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. diff --git a/lib/diameter/src/info/diameter_info.erl b/lib/diameter/src/info/diameter_info.erl index 2e08662a9e..59a3b94ee4 100644 --- a/lib/diameter/src/info/diameter_info.erl +++ b/lib/diameter/src/info/diameter_info.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. 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. @@ -52,8 +52,6 @@ p/1, p/3]). --compile({no_auto_import,[max/2]}). - -export([collect/2]). -define(LONG_TIMEOUT, 30000). @@ -684,9 +682,6 @@ pt(T) -> recsplit(SFun, Rec) -> fun(Fs,Vs) -> SFun(element(1, Rec), Fs, Vs) end. -max(A, B) -> - if A > B -> A; true -> B end. - keyfetch(Key, List) -> {Key,V} = lists:keyfind(Key, 1, List), V. diff --git a/lib/diameter/src/modules.mk b/lib/diameter/src/modules.mk index 3b223ea391..4e4ce60ddf 100644 --- a/lib/diameter/src/modules.mk +++ b/lib/diameter/src/modules.mk @@ -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. @@ -37,6 +37,7 @@ RT_MODULES = \ base/diameter_callback \ base/diameter_capx \ base/diameter_config \ + base/diameter_config_sup \ base/diameter_codec \ base/diameter_dict \ base/diameter_lib \ diff --git a/lib/diameter/src/transport/diameter_etcp.erl b/lib/diameter/src/transport/diameter_etcp.erl index 9db198ff86..10a014754f 100644 --- a/lib/diameter/src/transport/diameter_etcp.erl +++ b/lib/diameter/src/transport/diameter_etcp.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2012. 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. diff --git a/lib/diameter/src/transport/diameter_etcp_sup.erl b/lib/diameter/src/transport/diameter_etcp_sup.erl index 48794d4fe1..b230397b59 100644 --- a/lib/diameter/src/transport/diameter_etcp_sup.erl +++ b/lib/diameter/src/transport/diameter_etcp_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl index 3e08b78ea1..f48e4347ee 100644 --- a/lib/diameter/src/transport/diameter_sctp.erl +++ b/lib/diameter/src/transport/diameter_sctp.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. @@ -21,9 +21,6 @@ -module(diameter_sctp). -behaviour(gen_server). --compile({no_auto_import, [now/0]}). --import(diameter_lib, [now/0]). - %% interface -export([start/3]). @@ -64,10 +61,6 @@ %% Remote addresses to accept connections from. -define(DEFAULT_ACCEPT, []). %% any -%% How long a listener with no associations lives before offing -%% itself. --define(LISTENER_TIMEOUT, 30000). - %% How long to wait for a transport process to attach after %% association establishment. -define(ACCEPT_TIMEOUT, 5000). @@ -87,25 +80,26 @@ %% Accepting/connecting transport process state. -record(transport, - {parent :: pid(), + {parent :: pid() | undefined, mode :: {accept, pid()} | accept | {connect, {[inet:ip_address()], uint(), list()}} %% {RAs, RP, Errors} | connect, - socket :: gen_sctp:sctp_socket(), + socket :: gen_sctp:sctp_socket() | undefined, assoc_id :: gen_sctp:assoc_id(), %% association identifier - peer :: {[inet:ip_address()], uint()}, %% {RAs, RP} - streams :: {uint(), uint()}, %% {InStream, OutStream} counts + peer :: {[inet:ip_address()], uint()} %% {RAs, RP} + | undefined, + streams :: {uint(), uint()} %% {InStream, OutStream} counts + | undefined, os = 0 :: uint()}). %% next output stream %% Listener process state. -record(listener, {ref :: reference(), socket :: gen_sctp:sctp_socket(), - count = 0 :: uint(), %% attached transport processes + service = false :: false | pid(), %% service process pending = {0, queue:new()}, - tref :: reference(), accept :: [match()]}). %% Field pending implements two queues: the first of transport-to-be %% processes to which an association has been assigned but for which @@ -135,11 +129,14 @@ -> {ok, pid(), [inet:ip_address()]} when Ref :: diameter:transport_ref(). -start(T, #diameter_service{capabilities = Caps}, Opts) +start(T, Svc, Opts) when is_list(Opts) -> + #diameter_service{capabilities = Caps, + pid = SPid} + = Svc, diameter_sctp_sup:start(), %% start supervisors on demand Addrs = Caps#diameter_caps.host_ip_address, - s(T, Addrs, lists:map(fun ip/1, Opts)). + s(T, Addrs, SPid, lists:map(fun ip/1, Opts)). ip({ifaddr, A}) -> {ip, A}; @@ -150,18 +147,22 @@ ip(T) -> %% when there is not yet an association to assign it, or at comm_up on %% a new association in which case the call retrieves a transport from %% the pending queue. -s({accept, Ref} = A, Addrs, Opts) -> - {LPid, LAs} = listener(Ref, {Opts, Addrs}), - try gen_server:call(LPid, {A, self()}, infinity) of - {ok, TPid} -> {ok, TPid, LAs} +s({accept, Ref} = A, Addrs, SPid, Opts) -> + {ok, LPid, LAs} = listener(Ref, {Opts, Addrs}), + try gen_server:call(LPid, {A, self(), SPid}, infinity) of + {ok, TPid} -> + {ok, TPid, LAs}; + No -> + {error, No} catch - exit: Reason -> {error, Reason} + exit: Reason -> + {error, Reason} end; %% This implementation is due to there being no accept call in %% gen_sctp in order to be able to accept a new association only %% *after* an accepting transport has been spawned. -s({connect = C, Ref}, Addrs, Opts) -> +s({connect = C, Ref}, Addrs, _SPid, Opts) -> diameter_sctp_sup:start_child({C, self(), Opts, Addrs, Ref}). %% start_link/1 @@ -217,14 +218,15 @@ init(T) -> %% A process owning a listening socket. i({listen, Ref, {Opts, Addrs}}) -> + [_] = diameter_config:subscribe(Ref, transport), %% assert existence {[Matches], Rest} = proplists:split(Opts, [accept]), {LAs, Sock} = AS = open(Addrs, Rest, ?DEFAULT_PORT), ok = gen_sctp:listen(Sock, true), true = diameter_reg:add_new({?MODULE, listener, {Ref, AS}}), proc_lib:init_ack({ok, self(), LAs}), - start_timer(#listener{ref = Ref, - socket = Sock, - accept = [[M] || {accept, M} <- Matches]}); + #listener{ref = Ref, + socket = Sock, + accept = [[M] || {accept, M} <- Matches]}; %% A connecting transport. i({connect, Pid, Opts, Addrs, Ref}) -> @@ -286,24 +288,23 @@ i({K, Ref}, #transport{mode = {accept, _}} = S) -> %% Accepting processes can be started concurrently: ensure only one %% listener is started. -listener(LRef, T) -> - diameter_sync:call({?MODULE, listener, LRef}, - {?MODULE, listener, [{LRef, T}]}, +listener(Ref, T) -> + diameter_sync:call({?MODULE, listener, Ref}, + {?MODULE, listener, [{Ref, T}]}, infinity, infinity). -listener({LRef, T}) -> - l(diameter_reg:match({?MODULE, listener, {LRef, '_'}}), LRef, T). +listener({Ref, T}) -> + l(diameter_reg:match({?MODULE, listener, {Ref, '_'}}), Ref, T). %% Existing listening process ... l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) -> - {LAs, _Sock} = AS, - {LPid, LAs}; + {LAs, _Sock} = AS, + {ok, LPid, LAs}; %% ... or not. -l([], LRef, T) -> - {ok, LPid, LAs} = diameter_sctp_sup:start_child({listen, LRef, T}), - {LPid, LAs}. +l([], Ref, T) -> + diameter_sctp_sup:start_child({listen, Ref, T}). %% open/3 @@ -369,11 +370,17 @@ type(T) -> %% # handle_call/3 %% --------------------------------------------------------------------------- -handle_call({{accept, Ref}, Pid}, _, #listener{ref = Ref, - count = K} - = S) -> +handle_call({{accept, Ref}, Pid}, _, #listener{ref = Ref} = S) -> {TPid, NewS} = accept(Ref, Pid, S), - {reply, {ok, TPid}, NewS#listener{count = K+1}}; + {reply, {ok, TPid}, NewS}; + +handle_call({{accept, _} = T, Pid, SPid}, From, #listener{service = P} = S) -> + handle_call({T, Pid}, From, if not is_pid(P), is_pid(SPid) -> + monitor(process, SPid), + S#listener{service = SPid}; + true -> + S + end); handle_call(_, _, State) -> {reply, nok, State}. @@ -432,13 +439,6 @@ putr(Key, Val) -> getr(Key) -> get({?MODULE, Key}). -%% start_timer/1 - -start_timer(#listener{count = 0} = S) -> - S#listener{tref = erlang:start_timer(?LISTENER_TIMEOUT, self(), close)}; -start_timer(S) -> - S. - %% l/2 %% %% Transition listener state. @@ -453,35 +453,37 @@ l({sctp, Sock, _RA, _RP, Data} = T, #listener{socket = Sock, setopts(Sock), NewS; +%% Service process has died. +l({'DOWN', _, process, Pid, _} = T, #listener{service = Pid, + socket = Sock}) -> + gen_sctp:close(Sock), + x(T); + +%% Accepting process has died. l({'DOWN', _MRef, process, TPid, _}, #listener{pending = {_,Q}} = S) -> down(queue:member(TPid, Q), TPid, S); -%% Timeout after the last accepting process has died. -l({timeout, TRef, close = T}, #listener{tref = TRef, - count = 0}) -> - x(T); -l({timeout, _, close}, #listener{} = S) -> - S. +%% Transport has been removed. +l({transport, remove, _} = T, #listener{socket = Sock}) -> + gen_sctp:close(Sock), + x(T). %% down/3 %% %% Accepting transport has died. %% One that's waiting for transport start in the pending queue ... -down(true, TPid, #listener{pending = {N,Q}, - count = K} - = S) -> +down(true, TPid, #listener{pending = {N,Q}} = S) -> NQ = queue:filter(fun(P) -> P /= TPid end, Q), if N < 0 -> %% awaiting an association ... - start_timer(S#listener{count = K-1, - pending = {N+1, NQ}}); + S#listener{pending = {N+1, NQ}}; true -> %% ... or one has been assigned S#listener{pending = {N-1, NQ}} end; %% ... or one that's already attached. -down(false, _TPid, #listener{count = K} = S) -> - start_timer(S#listener{count = K-1}). +down(false, _TPid, S) -> + S. %% t/2 %% diff --git a/lib/diameter/src/transport/diameter_sctp_sup.erl b/lib/diameter/src/transport/diameter_sctp_sup.erl index 48df975ae9..36050aaf28 100644 --- a/lib/diameter/src/transport/diameter_sctp_sup.erl +++ b/lib/diameter/src/transport/diameter_sctp_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl index 005b2442c0..44abc5c3b4 100644 --- a/lib/diameter/src/transport/diameter_tcp.erl +++ b/lib/diameter/src/transport/diameter_tcp.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. @@ -19,6 +19,7 @@ %% -module(diameter_tcp). +-dialyzer({no_fail_call, throttle/2}). -behaviour(gen_server). @@ -56,7 +57,6 @@ -define(ERROR(T), erlang:error({T, ?MODULE, ?LINE})). -define(DEFAULT_PORT, 3868). %% RFC 3588, ch 2.1 --define(LISTENER_TIMEOUT, 30000). -define(DEFAULT_FRAGMENT_TIMEOUT, 1000). -define(IS_UINT32(N), (is_integer(N) andalso 0 =< N andalso 0 == N bsr 32)). @@ -71,9 +71,8 @@ %% a process owning the listening port. %% Listener process state. --record(listener, {socket :: inet:socket(), - count = 1 :: non_neg_integer(), - tref :: reference()}). +-record(listener, {socket :: inet:socket(), + service = false :: false | pid()}). %% service process %% Monitor process state. -record(monitor, @@ -102,7 +101,8 @@ | gen_tcp:listen_option(). -type option() :: {port, non_neg_integer()} - | {fragment_timer, 0..16#FFFFFFFF}. + | {fragment_timer, 0..16#FFFFFFFF} + | {throttle_cb, diameter:evaluable()}. %% Accepting/connecting transport process state. -record(transport, @@ -110,10 +110,13 @@ parent :: pid(), %% of process that started us module :: module(), %% gen_tcp-like module frag = <<>> :: frag(), %% message fragment - ssl :: boolean() | [term()], %% ssl options + ssl :: [term()] | boolean(), %% ssl options, ssl or not timeout :: infinity | 0..16#FFFFFFFF, %% fragment timeout tref = false :: false | reference(), %% fragment timer reference - flush = false :: boolean()}). %% flush fragment at timeout? + flush = false :: boolean(), %% flush fragment at timeout? + throttle_cb :: false | diameter:evaluable(), %% ask to receive + throttled :: boolean() | binary()}). %% stopped receiving? + %% The usual transport using gen_tcp can be replaced by anything %% sufficiently gen_tcp-like by passing a 'module' option as the first %% (for simplicity) transport option. The transport_module diameter_etcp @@ -132,11 +135,15 @@ | {ok, pid()} when Ref :: diameter:transport_ref(). -start({T, Ref}, #diameter_service{capabilities = Caps}, Opts) -> +start({T, Ref}, Svc, Opts) -> + #diameter_service{capabilities = Caps, + pid = SPid} + = Svc, + diameter_tcp_sup:start(), %% start tcp supervisors on demand {Mod, Rest} = split(Opts), Addrs = Caps#diameter_caps.host_ip_address, - Arg = {T, Ref, Mod, self(), Rest, Addrs}, + Arg = {T, Ref, Mod, self(), Rest, Addrs, SPid}, diameter_tcp_sup:start_child(Arg). split([{module, M} | Opts]) -> @@ -190,7 +197,7 @@ init(T) -> %% i/1 %% A transport process. -i({T, Ref, Mod, Pid, Opts, Addrs}) +i({T, Ref, Mod, Pid, Opts, Addrs, SPid}) when T == accept; T == connect -> monitor(process, Pid), @@ -198,25 +205,35 @@ i({T, Ref, Mod, Pid, Opts, Addrs}) %% that does nothing but kill us with the parent until call %% returns. {ok, MPid} = diameter_tcp_sup:start_child(#monitor{parent = Pid}), - {SslOpts, Rest0} = ssl(Opts), - {OwnOpts, Rest} = own(Rest0), + {[SO|TO], Rest} = proplists:split(Opts, [ssl_options, + fragment_timer, + throttle_cb]), + SslOpts = ssl_opts(SO), + OwnOpts = lists:append(TO), Tmo = proplists:get_value(fragment_timer, OwnOpts, ?DEFAULT_FRAGMENT_TIMEOUT), ?IS_TIMEOUT(Tmo) orelse ?ERROR({fragment_timer, Tmo}), - Sock = init(T, Ref, Mod, Pid, SslOpts, Rest, Addrs), + Throttle = proplists:get_value(throttle_cb, OwnOpts, false), + Sock = init(T, Ref, Mod, Pid, SslOpts, Rest, Addrs, SPid), MPid ! {stop, self()}, %% tell the monitor to die M = if SslOpts -> ssl; true -> Mod end, - setopts(M, Sock), putr(?REF_KEY, Ref), - #transport{parent = Pid, - module = M, - socket = Sock, - ssl = SslOpts, - timeout = Tmo}; + throttle(#transport{parent = Pid, + module = M, + socket = Sock, + ssl = SslOpts, + timeout = Tmo, + throttle_cb = Throttle, + throttled = false /= Throttle}); %% Put the reference in the process dictionary since we now use it %% advertise the ssl socket after TLS upgrade. +i({T, _Ref, _Mod, _Pid, _Opts, _Addrs} = Arg) %% from old code + when T == accept; + T == connect -> + i(erlang:append_element(Arg, _SPid = false)); + %% A monitor process to kill the transport if the parent dies. i(#monitor{parent = Pid, transport = TPid} = S) -> proc_lib:init_ack({ok, self()}), @@ -229,16 +246,19 @@ i(#monitor{parent = Pid, transport = TPid} = S) -> %% death. However, a link can be unlinked and this is exactly what %% gen_tcp seems to so. Links should be left to supervisors. -i({listen, LRef, APid, {Mod, Opts, Addrs}}) -> +i({listen = L, Ref, _APid, T}) -> %% from old code + i({L, Ref, T}); + +i({listen, Ref, {Mod, Opts, Addrs}}) -> + [_] = diameter_config:subscribe(Ref, transport), %% assert existence {[LA, LP], Rest} = proplists:split(Opts, [ip, port]), LAddrOpt = get_addr(LA, Addrs), LPort = get_port(LP), {ok, LSock} = Mod:listen(LPort, gen_opts(LAddrOpt, Rest)), LAddr = laddr(LAddrOpt, Mod, LSock), - true = diameter_reg:add_new({?MODULE, listener, {LRef, {LAddr, LSock}}}), + true = diameter_reg:add_new({?MODULE, listener, {Ref, {LAddr, LSock}}}), proc_lib:init_ack({ok, self(), {LAddr, LSock}}), - monitor(process, APid), - start_timer(#listener{socket = LSock}). + #listener{socket = LSock}. laddr([], Mod, Sock) -> {ok, {Addr, _Port}} = sockname(Mod, Sock), @@ -246,14 +266,6 @@ laddr([], Mod, Sock) -> laddr([{ip, Addr}], _, _) -> Addr. -own(Opts) -> - {[Own], Rest} = proplists:split(Opts, [fragment_timer]), - {Own, Rest}. - -ssl(Opts) -> - {[SslOpts], Rest} = proplists:split(Opts, [ssl_options]), - {ssl_opts(SslOpts), Rest}. - ssl_opts([]) -> false; ssl_opts([{ssl_options, true}]) -> @@ -261,24 +273,25 @@ ssl_opts([{ssl_options, true}]) -> ssl_opts([{ssl_options, Opts}]) when is_list(Opts) -> Opts; -ssl_opts(L) -> - ?ERROR({ssl_options, L}). +ssl_opts(T) -> + ?ERROR({ssl_options, T}). -%% init/7 +%% init/8 %% Establish a TLS connection before capabilities exchange ... -init(Type, Ref, Mod, Pid, true, Opts, Addrs) -> - init(Type, Ref, ssl, Pid, [{cb_info, ?TCP_CB(Mod)} | Opts], Addrs); +init(Type, Ref, Mod, Pid, true, Opts, Addrs, SPid) -> + init(Type, Ref, ssl, Pid, [{cb_info, ?TCP_CB(Mod)} | Opts], Addrs, SPid); %% ... or not. -init(Type, Ref, Mod, Pid, _, Opts, Addrs) -> - init(Type, Ref, Mod, Pid, Opts, Addrs). +init(Type, Ref, Mod, Pid, _, Opts, Addrs, SPid) -> + init(Type, Ref, Mod, Pid, Opts, Addrs, SPid). -%% init/6 +%% init/7 -init(accept = T, Ref, Mod, Pid, Opts, Addrs) -> +init(accept = T, Ref, Mod, Pid, Opts, Addrs, SPid) -> {[Matches], Rest} = proplists:split(Opts, [accept]), - {LAddr, LSock} = listener(Ref, {Mod, Rest, Addrs}), + {ok, LPid, {LAddr, LSock}} = listener(Ref, {Mod, Rest, Addrs}), + ok = gen_server:call(LPid, {accept, SPid}, infinity), proc_lib:init_ack({ok, self(), [LAddr]}), Sock = ok(accept(Mod, LSock)), ok = accept_peer(Mod, Sock, accept(Matches)), @@ -286,7 +299,7 @@ init(accept = T, Ref, Mod, Pid, Opts, Addrs) -> diameter_peer:up(Pid), Sock; -init(connect = T, Ref, Mod, Pid, Opts, Addrs) -> +init(connect = T, Ref, Mod, Pid, Opts, Addrs, _SPid) -> {[LA, RA, RP], Rest} = proplists:split(Opts, [ip, raddr, rport]), LAddrOpt = get_addr(LA, Addrs), RAddr = get_addr(RA), @@ -340,24 +353,26 @@ accept(Opts) -> %% Accepting processes can be started concurrently: ensure only one %% listener is started. -listener(LRef, T) -> - diameter_sync:call({?MODULE, listener, LRef}, - {?MODULE, listener, [{LRef, T, self()}]}, +listener(Ref, T) -> + diameter_sync:call({?MODULE, listener, Ref}, + {?MODULE, listener, [{Ref, T, self()}]}, infinity, infinity). -listener({LRef, T, TPid}) -> - l(diameter_reg:match({?MODULE, listener, {LRef, '_'}}), LRef, T, TPid). +%% listener/1 + +listener({Ref, T, _TPid}) -> + l(diameter_reg:match({?MODULE, listener, {Ref, '_'}}), Ref, T). + +%% l/3 %% Existing listening process ... -l([{{?MODULE, listener, {_, AS}}, LPid}], _, _, TPid) -> - LPid ! {accept, TPid}, - AS; +l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) -> + {ok, LPid, AS}; %% ... or not. -l([], LRef, T, TPid) -> - {ok, _, AS} = diameter_tcp_sup:start_child({listen, LRef, TPid, T}), - AS. +l([], Ref, T) -> + diameter_tcp_sup:start_child({listen, Ref, T}). %% get_addr/1 @@ -393,7 +408,7 @@ get_port(Ps) -> gen_opts(LAddrOpt, Opts) -> {L,_} = proplists:split(Opts, [binary, packet, active]), [[],[],[]] == L orelse ?ERROR({reserved_options, Opts}), - [binary, {packet, 0}, {active, once}] ++ LAddrOpt ++ Opts. + [binary, {packet, 0}, {active, false}] ++ LAddrOpt ++ Opts. %% --------------------------------------------------------------------------- %% # ports/1 @@ -436,6 +451,14 @@ portnr(Sock) -> %% # handle_call/3 %% --------------------------------------------------------------------------- +handle_call({accept, SPid}, _From, #listener{service = P} = S) -> + {reply, ok, if not is_pid(P), is_pid(SPid) -> + monitor(process, SPid), + S#listener{service = SPid}; + true -> + S + end}; + handle_call(_, _, State) -> {reply, nok, State}. @@ -482,13 +505,6 @@ putr(Key, Val) -> getr(Key) -> get({?MODULE, Key}). -%% start_timer/1 - -start_timer(#listener{count = 0} = S) -> - S#listener{tref = erlang:start_timer(?LISTENER_TIMEOUT, self(), close)}; -start_timer(S) -> - S. - %% m/2 %% %% Transition monitor state. @@ -510,20 +526,19 @@ m({'DOWN', _, process, Pid, _}, #monitor{parent = Pid, %% %% Transition listener state. -%% Another accept transport is attaching. -l({accept, TPid}, #listener{count = N} = S) -> - monitor(process, TPid), - S#listener{count = N+1}; - -%% Accepting process has died. -l({'DOWN', _, process, _, _}, #listener{count = N} = S) -> - start_timer(S#listener{count = N-1}); +%% Service process has died. +l({'DOWN', _, process, Pid, _} = T, #listener{service = Pid, + socket = Sock}) -> + gen_tcp:close(Sock), + x(T); -%% Timeout after the last accepting process has died. -l({timeout, TRef, close = T}, #listener{tref = TRef, - count = 0}) -> +%% Transport has been removed. +l({transport, remove, _} = T, #listener{socket = Sock}) -> + gen_tcp:close(Sock), x(T); -l({timeout, _, close}, #listener{} = S) -> + +%% Possibly death of an accepting process monitored in old code. +l(_, S) -> S. %% t/2 @@ -536,53 +551,37 @@ t(T,S) -> S; #transport{} = NS -> NS; - {stop, Reason} -> - x(Reason); stop -> x(T) end. %% transition/2 -%% Initial incoming message when we might need to upgrade to TLS: -%% don't request another message until we know. - -transition({tcp, Sock, Bin}, #transport{socket = Sock, - parent = Pid, - frag = Head, - module = M, - ssl = Opts} - = S) - when is_list(Opts) -> - case rcv(Head, Bin) of - {Msg, B} when is_binary(Msg) -> - diameter_peer:recv(Pid, Msg), - S#transport{frag = B}; - Frag -> - setopts(M, Sock), - start_fragment_timer(S#transport{frag = Frag}) - end; - %% Incoming message. transition({P, Sock, Bin}, #transport{socket = Sock, - module = M, - ssl = B} + ssl = B, + throttled = T} = S) - when P == tcp, not B; - P == ssl, B -> - setopts(M, Sock), - start_fragment_timer(recv(Bin, S)); + when P == ssl, true == B; + P == tcp -> + false = T, %% assert + recv(Bin, S); + +%% Make a new throttling callback after a timeout. +transition(throttle, #transport{throttled = false}) -> + ok; +transition(throttle, S) -> + throttle(S); %% Capabilties exchange has decided on whether or not to run over TLS. transition({diameter, {tls, Ref, Type, B}}, #transport{parent = Pid} = S) -> - #transport{socket = Sock, - module = M} + true = is_boolean(B), %% assert + #transport{} = NS = tls_handshake(Type, B, S), Pid ! {diameter, {tls, Ref}}, - setopts(M, Sock), - start_fragment_timer(NS#transport{ssl = B}); + throttle(NS#transport{ssl = B}); transition({C, Sock}, #transport{socket = Sock, ssl = B}) @@ -598,14 +597,8 @@ transition({E, Sock, _Reason} = T, #transport{socket = Sock, ?ERROR({T,S}); %% Outgoing message. -transition({diameter, {send, Bin}}, #transport{socket = Sock, - module = M}) -> - case send(M, Sock, Bin) of - ok -> - ok; - {error, Reason} -> - {stop, {send, Reason}} - end; +transition({diameter, {send, Bin}}, S) -> + send(Bin, S); %% Request to close the transport connection. transition({diameter, {close, Pid}}, #transport{parent = Pid, @@ -672,16 +665,25 @@ tls(accept, Sock, Opts) -> %% Reassemble fragmented messages and extract multiple message sent %% using Nagle. -recv(Bin, #transport{parent = Pid, frag = Head} = S) -> +%% Receive packets until a full message is received, +recv(Bin, #transport{frag = Head, throttled = false} = S) -> case rcv(Head, Bin) of - {Msg, B} when is_binary(Msg) -> - diameter_peer:recv(Pid, Msg), - recv(B, S#transport{frag = <<>>}); + {Msg, B} -> + throttle(S#transport{frag = B, throttled = Msg}); Frag -> - S#transport{frag = Frag, - flush = false} + setopts(S), + start_fragment_timer(S#transport{frag = Frag, + flush = false}) end. +%% recv/1 + +recv(#transport{throttled = false} = S) -> + recv(<<>>, S); + +recv(#transport{} = S) -> + S. + %% rcv/2 %% No previous fragment. @@ -765,8 +767,10 @@ bin(Bin) %% since all messages with length problems are discarded this should %% also eventually lead to watchdog failover. -%% No fragment to flush. -flush(#transport{frag = <<>>} = S) -> +%% No fragment to flush or not receiving messages. +flush(#transport{frag = Frag, throttled = B} = S) + when Frag == <<>>; + B /= false -> S; %% Messages have been received since last timer expiry. @@ -807,6 +811,17 @@ accept(Mod, LSock) -> connect(Mod, Host, Port, Opts) -> Mod:connect(Host, Port, Opts). +%% send/2 + +send(Bin, #transport{socket = Sock, + module = M}) -> + case send(M, Sock, Bin) of + ok -> + ok; + {error, Reason} -> + x({send, Reason}) + end. + %% send/3 send(gen_tcp, Sock, Bin) -> @@ -825,6 +840,11 @@ setopts(ssl, Sock, Opts) -> setopts(M, Sock, Opts) -> M:setopts(Sock, Opts). +%% setopts/1 + +setopts(#transport{socket = Sock, module = M}) -> + setopts(M, Sock). + %% setopts/2 setopts(M, Sock) -> @@ -833,6 +853,110 @@ setopts(M, Sock) -> X -> x({setopts, M, Sock, X}) %% possibly on peer disconnect end. +%% throttle/1 + +%% Still collecting packets for a complete message: keep receiving. +throttle(#transport{throttled = false} = S) -> + recv(S); + +%% Decide whether to receive another, or whether to accept a message +%% that's been received. +throttle(#transport{throttle_cb = F, throttled = T} = S) -> + Res = cb(F, T), + + try throttle(Res, S) of + #transport{ssl = SB} = NS when is_boolean(SB) -> + throttle(defrag(NS)); + #transport{throttled = Msg} = NS when is_binary(Msg) -> + %% Initial incoming message when we might need to upgrade + %% to TLS: wait for reception of a tls tuple. + defrag(NS) + catch + #transport{} = NS -> + recv(NS) + end. + +%% cb/2 + +cb(false, _) -> + ok; + +cb(F, B) -> + diameter_lib:eval([F, true /= B andalso B]). + +%% throttle/2 + +%% Callback says to receive another message. +throttle(ok, #transport{throttled = true} = S) -> + throw(S#transport{throttled = false}); + +%% Callback says to accept a received message. +throttle(ok, #transport{parent = Pid, throttled = Msg} = S) + when is_binary(Msg) -> + diameter_peer:recv(Pid, Msg), + S; + +throttle({ok = T, F}, S) -> + throttle(T, S#transport{throttle_cb = F}); + +%% Callback says to accept a received message and acknowledged the +%% returned pid with a {request, Pid} message if a request pid is +%% spawned, a discard message otherwise. The latter does not mean that +%% the message was necessarily discarded: it could have been an +%% answer. +throttle(NPid, #transport{parent = Pid, throttled = Msg} = S) + when is_pid(NPid), is_binary(Msg) -> + diameter_peer:recv(Pid, {Msg, NPid}), + S; + +throttle({NPid, F}, #transport{throttled = Msg} = S) + when is_pid(NPid), is_binary(Msg) -> + throttle(NPid, S#transport{throttle_cb = F}); + +%% Callback to accept a received message says to discard it. +throttle(discard, #transport{throttled = Msg} = S) + when is_binary(Msg) -> + S; + +throttle({discard = T, F}, #transport{throttled = Msg} = S) + when is_binary(Msg) -> + throttle(T, S#transport{throttle_cb = F}); + +%% Callback to accept a received message says to answer it with the +%% supplied binary. +throttle(Bin, #transport{throttled = Msg} = S) + when is_binary(Bin), is_binary(Msg) -> + send(Bin, S), + S; + +throttle({Bin, F}, #transport{throttled = Msg} = S) + when is_binary(Bin), is_binary(Msg) -> + throttle(Bin, S#transport{throttle_cb = F}); + +%% Callback says to ask again in the specified number of milliseconds. +throttle({timeout, Tmo}, S) -> + erlang:send_after(Tmo, self(), throttle), + throw(S); + +throttle({timeout = T, Tmo, F}, S) -> + throttle({T, Tmo}, S#transport{throttle_cb = F}); + +throttle(T, #transport{throttle_cb = F}) -> + ?ERROR({invalid_return, T, F}). + +%% defrag/1 +%% +%% Try to extract another message from packets already read before +%% another throttling callback. + +defrag(#transport{frag = Head} = S) -> + case rcv(Head, <<>>) of + {Msg, B} -> + S#transport{throttled = Msg, frag = B}; + _ -> + S#transport{throttled = true} + end. + %% portnr/2 portnr(gen_tcp, Sock) -> diff --git a/lib/diameter/src/transport/diameter_tcp_sup.erl b/lib/diameter/src/transport/diameter_tcp_sup.erl index a7bdb49968..6cf758748a 100644 --- a/lib/diameter/src/transport/diameter_tcp_sup.erl +++ b/lib/diameter/src/transport/diameter_tcp_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. 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. diff --git a/lib/diameter/src/transport/diameter_transport.erl b/lib/diameter/src/transport/diameter_transport.erl index 5a7c59b4dc..3ca4d97f50 100644 --- a/lib/diameter/src/transport/diameter_transport.erl +++ b/lib/diameter/src/transport/diameter_transport.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2012. All Rights Reserved. +%% Copyright Ericsson AB 2012-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. |