diff options
Diffstat (limited to 'lib/diameter/src')
22 files changed, 1127 insertions, 405 deletions
diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl index d74e091e11..67dfc7bdbf 100644 --- a/lib/diameter/src/base/diameter.erl +++ b/lib/diameter/src/base/diameter.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2013. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -306,6 +306,7 @@ call(SvcName, App, Message) -> | {restrict_connections, restriction()} | {sequence, sequence() | evaluable()} | {share_peers, remotes()} + | {string_decode, boolean()} | {use_shared_peers, remotes()} | {spawn_opt, list()}. @@ -337,11 +338,14 @@ call(SvcName, App, Message) -> :: {transport_module, atom()} | {transport_config, any()} | {transport_config, any(), 'Unsigned32'() | infinity} + | {pool_size, pos_integer()} | {applications, [app_alias()]} | {capabilities, [capability()]} | {capabilities_cb, evaluable()} | {capx_timeout, 'Unsigned32'()} | {disconnect_cb, evaluable()} + | {dpr_timeout, 'Unsigned32'()} + | {dpa_timeout, 'Unsigned32'()} | {length_errors, exit | handle | discard} | {connect_timer, 'Unsigned32'()} | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}} diff --git a/lib/diameter/src/base/diameter_capx.erl b/lib/diameter/src/base/diameter_capx.erl index 93548ecafd..7dc61f229f 100644 --- a/lib/diameter/src/base/diameter_capx.erl +++ b/lib/diameter/src/base/diameter_capx.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2013. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -50,7 +50,8 @@ -export([build_CER/2, recv_CER/3, recv_CEA/3, - make_caps/2]). + make_caps/2, + binary_caps/1]). -include_lib("diameter/include/diameter.hrl"). -include("diameter_internal.hrl"). @@ -115,7 +116,8 @@ mk_caps(Caps0, Opts) -> -define(SC(K,F), set_cap({K, Val}, {Caps, #diameter_caps{F = false} = C}) -> - {Caps#diameter_caps{F = cap(K, Val)}, C#diameter_caps{F = true}}). + {Caps#diameter_caps{F = cap(K, copy(Val))}, + C#diameter_caps{F = true}}). ?SC('Origin-Host', origin_host); ?SC('Origin-Realm', origin_realm); @@ -375,10 +377,10 @@ capx_to_caps(CEX, Dict) -> 'Firmware-Revision', 'AVP'], CEX), - #diameter_caps{origin_host = OH, - origin_realm = OR, + #diameter_caps{origin_host = copy(OH), + origin_realm = copy(OR), vendor_id = VId, - product_name = PN, + product_name = copy(PN), origin_state_id = OSI, host_ip_address = IP, supported_vendor_id = SV, @@ -389,6 +391,32 @@ capx_to_caps(CEX, Dict) -> firmware_revision = FR, avp = X}. +%% Copy binaries to avoid retaining a reference to a large binary +%% containing AVPs we aren't interested in. +copy(B) + when is_binary(B) -> + binary:copy(B); + +copy(T) -> + T. + +%% binary_caps/1 +%% +%% Encode stringish capabilities with {string_decode, false}. + +binary_caps(Caps) -> + lists:foldl(fun bcaps/2, Caps, [#diameter_caps.origin_host, + #diameter_caps.origin_realm, + #diameter_caps.product_name]). + +bcaps(N, Caps) -> + case element(N, Caps) of + undefined -> + Caps; + V -> + setelement(N, Caps, iolist_to_binary(V)) + end. + %% --------------------------------------------------------------------------- %% --------------------------------------------------------------------------- diff --git a/lib/diameter/src/base/diameter_codec.erl b/lib/diameter/src/base/diameter_codec.erl index a2b04bfd63..15a4c5e86f 100644 --- a/lib/diameter/src/base/diameter_codec.erl +++ b/lib/diameter/src/base/diameter_codec.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -22,6 +22,8 @@ -export([encode/2, decode/2, decode/3, + setopts/1, + getopt/1, collect_avps/1, decode_header/1, sequence_numbers/1, @@ -59,6 +61,50 @@ %% +-+-+-+-+-+-+-+-+-+-+-+-+- %%% --------------------------------------------------------------------------- +%%% # setopts/1 +%%% # getopt/1 +%%% --------------------------------------------------------------------------- + +%% These functions are a compromise in the same vein as the use of the +%% process dictionary in diameter_gen.hrl in generated codec modules. +%% Instead of rewriting the entire dictionary generation to pass +%% encode/decode options around, the calling process sets them by +%% calling setopts/1. At current, the only option is whether or not to +%% decode binaries as strings, which is used by diameter_types. + +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}) -> + setopt(K, B); + +%% Regard anything but the generated RFC 3588 dictionary as modern. +%% This affects the interpretation of defaults during the decode +%% of values of type DiameterURI, this having changed from RFC 3588. +%% (So much for backwards compatibility.) +setopt({common_dictionary, diameter_gen_base_rfc3588}) -> + setopt(rfc, 3588); + +setopt(_) -> + ok. + +setopt(Key, Value) -> + put({diameter, Key}, Value). + +getopt(Key) -> + case get({diameter, Key}) of + undefined when Key == string_decode -> + true; + undefined when Key == rfc -> + 6733; + V -> + V + end. + +%%% --------------------------------------------------------------------------- %%% # encode/2 %%% --------------------------------------------------------------------------- @@ -90,7 +136,7 @@ encode(Mod, Msg) -> msg = Msg}). e(_, #diameter_packet{msg = [#diameter_header{} = Hdr | As]} = Pkt) -> - try encode_avps(As) of + try encode_avps(reorder(As)) of Avps -> Length = size(Avps) + 20, @@ -183,26 +229,50 @@ values(Avps) -> %% Message as a list of #diameter_avp{} ... encode_avps(_, _, [#diameter_avp{} | _] = Avps) -> - encode_avps(reorder(Avps, [], Avps)); + encode_avps(reorder(Avps)); %% ... or as a tuple list or record. encode_avps(Mod, MsgName, Values) -> Mod:encode_avps(MsgName, Values). %% reorder/1 +%% +%% Reorder AVPs for the relay case using the index field of +%% diameter_avp records. Decode populates this field in collect_avps +%% and presents AVPs in reverse order. A relay then sends the reversed +%% list with a Route-Record AVP prepended. The goal here is just to do +%% lists:reverse/1 in Grouped AVPs and the outer list, but only in the +%% case there are indexed AVPs at all, so as not to reverse lists that +%% have been explicilty sent (unindexed, in the desired order) as a +%% diameter_avp list. The effect is the same as lists:keysort/2, but +%% only on the cases we expect, not a general sort. + +reorder(Avps) -> + case reorder(Avps, []) of + false -> + Avps; + Sorted -> + Sorted + end. + +%% reorder/3 -reorder([#diameter_avp{index = 0} | _] = Avps, Acc, _) -> +%% In case someone has reversed the list already. (Not likely.) +reorder([#diameter_avp{index = 0} | _] = Avps, Acc) -> Avps ++ Acc; -reorder([#diameter_avp{index = N} = A | Avps], Acc, _) +%% Assume indexed AVPs are in reverse order. +reorder([#diameter_avp{index = N} = A | Avps], Acc) when is_integer(N) -> lists:reverse(Avps, [A | Acc]); -reorder([H | T], Acc, Avps) -> - reorder(T, [H | Acc], Avps); +%% An unindexed AVP. +reorder([H | T], Acc) -> + reorder(T, [H | Acc]); -reorder([], Acc, _) -> - Acc. +%% No indexed members. +reorder([], _) -> + false. %% encode_avps/1 @@ -390,6 +460,9 @@ sequence_numbers(#diameter_packet{bin = Bin}) sequence_numbers(#diameter_packet{header = #diameter_header{} = H}) -> sequence_numbers(H); +sequence_numbers(#diameter_packet{msg = [#diameter_header{} = H | _]}) -> + sequence_numbers(H); + sequence_numbers(#diameter_header{hop_by_hop_id = H, end_to_end_id = E}) -> {H,E}; @@ -561,14 +634,14 @@ split_data(Bin, Len) -> <<Data:Len/binary, _:Pad/binary, Rest/binary>> -> {Data, Rest}; _ -> - %% Header length points past the end of the message. As - %% stated in the 6733 text above, it's sufficient to - %% return a zero-filled minimal payload if this is a - %% request. Do this (in cases that we know the type) by - %% inducing a decode failure and letting the dictionary's - %% decode (in diameter_gen) deal with it. Here we don't - %% know type. If the type isn't known, then the decode - %% just strips the extra bit. + %% Header length points past the end of the message, or + %% doesn't span the header. As stated in the 6733 text + %% above, it's sufficient to return a zero-filled minimal + %% payload if this is a request. Do this (in cases that we + %% know the type) by inducing a decode failure and letting + %% the dictionary's decode (in diameter_gen) deal with it. + %% Here we don't know type. If the type isn't known, then + %% the decode just strips the extra bit. {<<0:1, Bin/binary>>, <<>>} end. @@ -582,6 +655,8 @@ split_data(Bin, Len) -> %% 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)}); diff --git a/lib/diameter/src/base/diameter_config.erl b/lib/diameter/src/base/diameter_config.erl index dd1c9b73bb..0d0304bf33 100644 --- a/lib/diameter/src/base/diameter_config.erl +++ b/lib/diameter/src/base/diameter_config.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -35,10 +35,11 @@ %% -module(diameter_config). --compile({no_auto_import, [monitor/2]}). - -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, @@ -531,7 +532,10 @@ opt({applications, As}) -> opt({capabilities, Os}) -> is_list(Os) andalso ok == encode_CER(Os); -opt({capx_timeout, Tmo}) -> +opt({K, Tmo}) + when K == capx_timeout; + K == dpr_timeout; + K == dpa_timeout -> ?IS_UINT32(Tmo); opt({length_errors, T}) -> @@ -554,6 +558,9 @@ opt({watchdog_config, L}) -> opt({spawn_opt, Opts}) -> is_list(Opts); +opt({pool_size, N}) -> + is_integer(N) andalso 0 < N; + %% Options that we can't validate. opt({K, _}) when K == transport_config; @@ -638,13 +645,23 @@ make_config(SvcName, Opts) -> {false, monitor}, {?NOMASK, sequence}, {nodes, restrict_connections}, + {true, string_decode}, {[], spawn_opt}]), + D = proplists:get_value(string_decode, SvcOpts, true), + #service{name = SvcName, rec = #diameter_service{applications = Apps, - capabilities = Caps}, + capabilities = binary_caps(Caps, D)}, options = SvcOpts}. +binary_caps(Caps, true) -> + Caps; +binary_caps(Caps, false) -> + diameter_capx:binary_caps(Caps). + +%% make_opts/2 + make_opts(Opts, Defs) -> Known = [{K, get_opt(K, Opts, D)} || {D,K} <- Defs], Unknown = Opts -- Known, @@ -663,7 +680,8 @@ opt(K, false = B) opt(K, true = B) when K == share_peers; - K == use_shared_peers -> + K == use_shared_peers; + K == string_decode -> B; opt(restrict_connections, T) diff --git a/lib/diameter/src/base/diameter_lib.erl b/lib/diameter/src/base/diameter_lib.erl index 549d7daca0..51d203d722 100644 --- a/lib/diameter/src/base/diameter_lib.erl +++ b/lib/diameter/src/base/diameter_lib.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -18,13 +18,19 @@ %% -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/1, now_diff/1, + micro_diff/1, + micro_diff/2, time/1, + seed/0, eval/1, eval_name/1, get_stacktrace/0, @@ -32,6 +38,8 @@ spawn_opts/2, wait/1, fold_tuple/3, + fold_n/3, + for_n/2, log/4]). %% --------------------------------------------------------------------------- @@ -91,13 +99,51 @@ fmt(T) -> end. %% --------------------------------------------------------------------------- +%% # 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. + +now() -> + try + erlang:monotonic_time() + catch + error: undef -> erlang:now() + end. + +%% --------------------------------------------------------------------------- +%% # timestamp/1 +%% --------------------------------------------------------------------------- + +-spec timestamp(NowT :: now()) + -> timestamp(). + +timestamp({_,_,_} = T) -> %% erlang:now() + T; + +timestamp(MonoT) -> %% monotonic time + MicroSecs = monotonic_to_microseconds(MonoT + erlang:time_offset()), + Secs = MicroSecs div 1000000, + {Secs div 1000000, Secs rem 1000000, MicroSecs rem 1000000}. + +monotonic_to_microseconds(MonoT) -> + erlang:convert_time_unit(MonoT, native, micro_seconds). + +%% --------------------------------------------------------------------------- %% # now_diff/1 %% --------------------------------------------------------------------------- --spec now_diff(NowT) +-spec now_diff(NowT :: now()) -> {Hours, Mins, Secs, MicroSecs} - when NowT :: {non_neg_integer(), 0..999999, 0..999999}, - Hours :: non_neg_integer(), + when Hours :: non_neg_integer(), Mins :: 0..59, Secs :: 0..59, MicroSecs :: 0..999999. @@ -105,8 +151,37 @@ fmt(T) -> %% Return timer:now_diff(now(), NowT) as an {H, M, S, MicroS} tuple %% instead of as integer microseconds. -now_diff({_,_,_} = Time) -> - time(timer:now_diff(now(), Time)). +now_diff(Time) -> + time(micro_diff(Time)). + +%% --------------------------------------------------------------------------- +%% # micro_diff/1 +%% --------------------------------------------------------------------------- + +-spec micro_diff(NowT :: now()) + -> 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). + +%% --------------------------------------------------------------------------- +%% # micro_diff/2 +%% --------------------------------------------------------------------------- + +-spec micro_diff(T1 :: now(), T0 :: now()) + -> 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)). %% --------------------------------------------------------------------------- %% # time/1 @@ -116,7 +191,7 @@ now_diff({_,_,_} = Time) -> -spec time(NowT | Diff) -> {Hours, Mins, Secs, MicroSecs} - when NowT :: {non_neg_integer(), 0..999999, 0..999999}, + when NowT :: timestamp(), Diff :: non_neg_integer(), Hours :: non_neg_integer(), Mins :: 0..59, @@ -135,6 +210,27 @@ 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. @@ -248,17 +344,19 @@ opts(HeapSize, Opts) -> %% # wait/1 %% --------------------------------------------------------------------------- --spec wait([pid()]) +-spec wait([pid() | reference()]) -> ok. wait(L) -> - down([erlang:monitor(process, P) || P <- L]). + lists:foreach(fun down/1, L). -down([]) -> - ok; -down([MRef|T]) -> - receive {'DOWN', MRef, process, _, _} -> ok end, - down(T). +down(Pid) + when is_pid(Pid) -> + down(monitor(process, Pid)); + +down(MRef) + when is_reference(MRef) -> + receive {'DOWN', MRef, process, _, _} = T -> T end. %% --------------------------------------------------------------------------- %% # fold_tuple/3 @@ -291,6 +389,35 @@ ft(Value, {Idx, T}) -> setelement(Idx, T, Value). %% --------------------------------------------------------------------------- +%% # fold_n/3 +%% --------------------------------------------------------------------------- + +-spec fold_n(F, Acc0, N) + -> term() + when F :: fun((non_neg_integer(), term()) -> term()), + Acc0 :: term(), + N :: non_neg_integer(). + +fold_n(F, Acc, N) + when is_integer(N), 0 < N -> + fold_n(F, F(N, Acc), N-1); + +fold_n(_, Acc, _) -> + Acc. + +%% --------------------------------------------------------------------------- +%% # for_n/2 +%% --------------------------------------------------------------------------- + +-spec for_n(F, N) + -> non_neg_integer() + when F :: fun((non_neg_integer()) -> term()), + N :: non_neg_integer(). + +for_n(F, N) -> + fold_n(fun(M,A) -> F(M), A+1 end, 0, N). + +%% --------------------------------------------------------------------------- %% # log/4 %% %% Called to have something to trace on for happenings of interest. diff --git a/lib/diameter/src/base/diameter_peer.erl b/lib/diameter/src/base/diameter_peer.erl index e5d4b28766..89b63c8a92 100644 --- a/lib/diameter/src/base/diameter_peer.erl +++ b/lib/diameter/src/base/diameter_peer.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2013. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -18,9 +18,11 @@ %% -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, @@ -230,12 +232,22 @@ recv(Pid, Pkt) -> %% # send/2 %% --------------------------------------------------------------------------- -send(Pid, #diameter_packet{transport_data = undefined, - bin = Bin}) -> - send(Pid, Bin); +send(Pid, Msg) -> + ifc_send(Pid, {send, strip(Msg)}). + +%% Send only binary when possible. +strip(#diameter_packet{transport_data = undefined, + bin = Bin}) -> + Bin; + +%% Strip potentially large message terms. +strip(#diameter_packet{transport_data = T, + bin = Bin}) -> + #diameter_packet{transport_data = T, + bin = Bin}; -send(Pid, Pkt) -> - ifc_send(Pid, {send, Pkt}). +strip(Msg) -> + Msg. %% --------------------------------------------------------------------------- %% # close/1 @@ -324,7 +336,6 @@ code_change(_OldVsn, State, _Extra) -> {ok, State}. %% --------------------------------------------------------- -%% INTERNAL FUNCTIONS %% --------------------------------------------------------- %% ifc_send/2 diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index ee6e7dd89e..aac2685514 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-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -63,6 +63,8 @@ %% Keys in process dictionary. -define(CB_KEY, cb). %% capabilities callback -define(DPR_KEY, dpr). %% disconnect callback +-define(DPA_KEY, dpa). %% timeout for incoming DPA, or shutdown after + %% outgoing DPA -define(REF_KEY, ref). %% transport_ref() -define(Q_KEY, q). %% transport start queue -define(START_KEY, start). %% start of connected transport @@ -82,18 +84,26 @@ N == ?GOAWAY; N == goaway; N == ?BUSY; N == busy). -%% RFC 3588: +%% RFC 6733: %% %% Timeout An application-defined timer has expired while waiting %% for some event. %% --define(EVENT_TIMEOUT, 10000). + %% Default timeout for reception of CER/CEA. +-define(CAPX_TIMEOUT, 10000). -%% Default timeout for DPA in response to DPR. A bit short but the -%% timeout used to be hardcoded. (So it could be worse.) +%% Default timeout for DPA to be received in response to an outgoing +%% DPR. A bit short but the timeout used to be hardcoded. (So it could +%% be worse.) -define(DPA_TIMEOUT, 1000). +%% Default timeout for the connection to be closed by the peer +%% following an outgoing DPA in response to an incoming DPR. It's the +%% recipient of DPA that should close the connection according to the +%% RFC. +-define(DPR_TIMEOUT, 5000). + -type uint32() :: diameter:'Unsigned32'(). -record(state, @@ -107,8 +117,14 @@ transport :: pid(), %% transport process dictionary :: module(), %% common dictionary service :: #diameter_service{}, - dpr = false :: false | {uint32(), uint32()}, - %% | hop by hop and end to end identifiers + dpr = false :: false + | true %% DPR received, DPA sent + | {uint32(), uint32()} %% set in old code + | {boolean(), uint32(), uint32()}, + %% hop by hop and end to end identifiers in + %% outgoing DPR; boolean says whether or not + %% the request was sent explicitly with + %% diameter:call/4. length_errors :: exit | handle | discard}). %% There are non-3588 states possible as a consequence of 5.6.1 of the @@ -138,7 +154,8 @@ %% # start/3 %% --------------------------------------------------------------------------- --spec start(T, [Opt], {diameter:sequence(), +-spec start(T, [Opt], {[diameter:service_opt()] + | diameter:sequence(), %% from old code [node()], module(), #diameter_service{}}) @@ -177,18 +194,25 @@ init(T) -> proc_lib:init_ack({ok, self()}), gen_server:enter_loop(?MODULE, [], i(T)). -i({Ack, WPid, {M, Ref} = T, Opts, {Mask, Nodes, Dict0, Svc}}) -> +i({Ack, WPid, T, Opts, {{_,_} = Mask, Nodes, Dict0, Svc}}) -> %% from old code + i({Ack, WPid, T, Opts, {[{sequence, Mask}], Nodes, Dict0, Svc}}); + +i({Ack, WPid, {M, Ref} = T, Opts, {SvcOpts, Nodes, Dict0, Svc}}) -> erlang:monitor(process, WPid), wait(Ack, WPid), diameter_stats:reg(Ref), + diameter_codec:setopts([{common_dictionary, Dict0} | SvcOpts]), + {_,_} = Mask = proplists:get_value(sequence, SvcOpts), {[Cs,Ds], Rest} = proplists:split(Opts, [capabilities_cb, disconnect_cb]), putr(?CB_KEY, {Ref, [F || {_,F} <- Cs]}), putr(?DPR_KEY, [F || {_, F} <- Ds]), putr(?REF_KEY, Ref), putr(?SEQUENCE_KEY, Mask), putr(?RESTRICT_KEY, Nodes), + putr(?DPA_KEY, {proplists:get_value(dpr_timeout, Opts, ?DPR_TIMEOUT), + proplists:get_value(dpa_timeout, Opts, ?DPA_TIMEOUT)}), - Tmo = proplists:get_value(capx_timeout, Opts, ?EVENT_TIMEOUT), + Tmo = proplists:get_value(capx_timeout, Opts, ?CAPX_TIMEOUT), OnLengthErr = proplists:get_value(length_errors, Opts, exit), {TPid, Addrs} = start_transport(T, Rest, Svc), @@ -212,9 +236,12 @@ wait(Ref, Pid) -> Ref -> ok; {'DOWN', _, process, Pid, _} = D -> - exit({shutdown, D}) + x(D) end. +x(T) -> + exit({shutdown, T}). + start_transport(T, Opts, #diameter_service{capabilities = LCaps} = Svc) -> Addrs0 = LCaps#diameter_caps.host_ip_address, start_transport(Addrs0, {T, Opts, Svc}). @@ -226,7 +253,7 @@ start_transport(Addrs0, T) -> q_next(TPid, Addrs0, Tmo, Data), {TPid, Addrs}; {error, No} -> - exit({shutdown, {no_connection, No}}) + x({no_connection, No}) end. svc(#diameter_service{capabilities = LCaps0} = Svc, Addrs) -> @@ -333,6 +360,9 @@ eraser(Key) -> %% transition/2 +transition(T, #state{dpr = {Hid, Eid}} = S) -> %% DPR sent from old code + transition(T, S#state{dpr = {false, Hid, Eid}}); + %% Connection to peer. transition({diameter, {TPid, connected, Remote}}, #state{transport = TPid, @@ -397,9 +427,8 @@ transition({timeout, _}, _) -> ok; %% Outgoing message. -transition({send, Msg}, #state{transport = TPid}) -> - send(TPid, Msg), - ok; +transition({send, Msg}, S) -> + outgoing(Msg, S); %% Request for graceful shutdown at remove_transport, stop_service of %% application shutdown. @@ -408,7 +437,8 @@ transition({shutdown, Pid, Reason}, #state{parent = Pid, dpr = false} = S) -> transition({shutdown, Pid, _}, #state{parent = Pid}) -> ok; -%% DPA reception has timed out. +%% DPA reception has timed out, or peer has not closed the connection +%% as a result of outgoing DPA. transition(dpa_timeout, _) -> stop; @@ -516,12 +546,9 @@ encode(Rec, Dict) -> recv(#diameter_packet{header = #diameter_header{} = Hdr} = Pkt, - #state{parent = Pid, - dictionary = Dict0} + #state{dictionary = Dict0} = S) -> - Name = diameter_codec:msg_name(Dict0, Hdr), - Pid ! {recv, self(), Name, Pkt}, - rcv(Name, Pkt, S); + recv1(diameter_codec:msg_name(Dict0, Hdr), Pkt, S); recv(#diameter_packet{header = undefined, bin = Bin} @@ -532,6 +559,41 @@ recv(#diameter_packet{header = undefined, recv(Bin, S) -> recv(#diameter_packet{bin = Bin}, S). +%% recv1/3 + +%% Incoming request after outgoing DPR: discard. Don't discard DPR, so +%% both ends don't do so when sending simultaneously. +recv1(Name, + #diameter_packet{header = #diameter_header{is_request = true} = H}, + #state{dpr = {_,_,_}}) + when Name /= 'DPR' -> + invalid(false, recv_after_outgoing_dpr, H); + +%% Incoming request after incoming DPR: discard. +recv1(_, + #diameter_packet{header = #diameter_header{is_request = true} = H}, + #state{dpr = true}) -> + invalid(false, recv_after_incoming_dpr, H); + +%% DPA with identifier mismatch, or in response to a DPR initiated by +%% the service. +recv1('DPA' = N, + #diameter_packet{header = #diameter_header{hop_by_hop_id = Hid, + end_to_end_id = Eid}} + = Pkt, + #state{dpr = {X,H,E}} + = S) + when H /= Hid; + E /= Eid; + not X -> + rcv(N, Pkt, S); + +%% 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). + %% recv/3 recv(#diameter_header{length = Len} @@ -588,7 +650,7 @@ rcv(Name, _, #state{state = PS}) rcv('DPR' = N, Pkt, S) -> handle_request(N, Pkt, S); -%% DPA in response to DPR and with the expected identifiers. +%% DPA in response to DPR, with the expected identifiers. rcv('DPA' = N, #diameter_packet{header = #diameter_header{end_to_end_id = Eid, hop_by_hop_id = Hid} @@ -596,14 +658,21 @@ rcv('DPA' = N, = Pkt, #state{dictionary = Dict0, transport = TPid, - dpr = {Hid, Eid}}) -> + dpr = {X, Hid, Eid}}) -> ?LOG(recv, N), - incr(recv, H, Dict0), - incr_rc(recv, diameter_codec:decode(Dict0, Pkt), Dict0), + X orelse begin + %% Only count DPA in response to a DPR sent by the + %% service: explicit DPR is counted in the same way + %% as other explicitly sent requests. + incr(recv, H, Dict0), + incr_rc(recv, diameter_codec:decode(Dict0, Pkt), Dict0) + end, diameter_peer:close(TPid), {stop, N}; -%% Ignore anything else, an unsolicited DPA in particular. +%% Ignore anything else, an unsolicited DPA in particular. Note that +%% dpa_timeout deals with the case in which the peer sends the wrong +%% identifiers in DPA. rcv(N, #diameter_packet{header = H}, _) when N == 'CER'; N == 'CEA'; @@ -637,9 +706,61 @@ incr_error(Dir, Pkt, Dict0) -> %% Msg here could be a #diameter_packet or a binary depending on who's %% sending. In particular, the watchdog will send DWR as a binary %% while messages coming from clients will be in a #diameter_packet. + send(Pid, Msg) -> diameter_peer:send(Pid, Msg). +%% outgoing/2 + +%% Explicit DPR. +outgoing(#diameter_packet{header = #diameter_header{application_id = 0, + cmd_code = 282, + is_request = true} + = H} + = Pkt, + #state{dpr = T, + parent = Pid} + = S) -> + if T == false -> + inform_dpr(Pid), + send_dpr(true, Pkt, dpa_timeout(), S); + T == true -> + invalid(false, dpr_after_dpa, H); %% DPA sent: discard + true -> + invalid(false, dpr_after_dpr, H) %% DPR sent: discard + end; + +%% Explict CER or DWR: discard. These are sent by us. +outgoing(#diameter_packet{header = #diameter_header{application_id = 0, + cmd_code = C, + is_request = true} + = H}, + _) + when 257 == C; %% CER + 280 == C -> %% DWR + invalid(false, invalid_request, H); + +%% DPR not sent: send. +outgoing(Msg, #state{transport = TPid, dpr = false}) -> + send(TPid, Msg), + ok; + +%% Outgoing answer: send. +outgoing(#diameter_packet{header = #diameter_header{is_request = false}} + = Pkt, + #state{transport = TPid}) -> + send(TPid, Pkt), + ok; + +%% Outgoing request: discard. +outgoing(Msg, #state{dpr = {_,_,_}}) -> + invalid(false, send_after_dpr, header(Msg)). + +header(#diameter_packet{header = H}) -> + H; +header(Bin) -> %% DWR + diameter_codec:decode_header(Bin). + %% handle_request/3 %% %% Incoming CER or DPR. @@ -699,6 +820,8 @@ build_answer('CER', = Pkt, #state{dictionary = Dict0} = S) -> + diameter_codec:setopts([{string_decode, false}]), + {SupportedApps, RCaps, CEA} = recv_CER(CER, S), [RC, IS] = Dict0:'#get-'(['Result-Code', 'Inband-Security-Id'], CEA), @@ -731,7 +854,7 @@ build_answer(Type, errors = Es} = Pkt, S) -> - {RC, FailedAVP} = result_code(H, Es), + {RC, FailedAVP} = result_code(Type, H, Es), {answer(Type, RC, FailedAVP, S), post(Type, RC, Pkt, S)}. inband_security([]) -> @@ -748,8 +871,16 @@ cea(CEA, RC, Dict0) -> post('CER' = T, RC, Pkt, S) -> {T, caps(S), {RC, Pkt}}; -post('DPR' = T, _, _, #state{parent = Pid}) -> - [fun(S) -> Pid ! {T, self()}, S end]. +post('DPR', _, _, #state{parent = Pid}) -> + [fun(S) -> dpr_timer(), inform_dpr(Pid), dpr(S) end]. + +dpr(#state{dpr = false} = S) -> %% not awaiting DPA + S#state{dpr = true}; %% DPR received +dpr(S) -> %% DPR already sent or received + S. + +inform_dpr(Pid) -> + Pid ! {'DPR', self()}. %% tell watchdog to die with us rejected({capabilities_cb, _F, Reason}, T, S) -> rejected(Reason, T, S); @@ -798,6 +929,19 @@ set(['answer-message' | _] = Ans, FailedAvp) -> set([_|_] = Ans, FailedAvp) -> Ans ++ FailedAvp. +%% result_code/3 + +%% Be lenient with errors in DPR since there's no reason to be +%% otherwise. Rejecting may cause the peer to missinterpret the error +%% as meaning that the connection should not be closed, which may well +%% lead to more problems than any errors in the DPR. + +result_code('DPR', _, _) -> + {2001, []}; + +result_code('CER', H, Es) -> + result_code(H, Es). + %% result_code/2 result_code(#diameter_header{is_error = true}, _) -> @@ -886,6 +1030,8 @@ handle_CEA(#diameter_packet{header = H} = DPkt = diameter_codec:decode(Dict0, Pkt), + diameter_codec:setopts([{string_decode, false}]), + RC = result_code(incr_rc(recv, DPkt, Dict0)), {SApps, IS, RCaps} = recv_CEA(DPkt, S), @@ -1026,7 +1172,7 @@ close(Reason) -> %% dpr/2 %% -%% The RFC isn't clear on whether DPR should be send in a non-Open +%% The RFC isn't clear on whether DPR should be sent in a non-Open %% state. The Peer State Machine transitions it documents aren't %% exhaustive (no Stop in Wait-I-CEA for example) so assume it's up to %% the implementation and transition to Closed (ie. die) if we haven't @@ -1042,7 +1188,7 @@ dpr(Reason, #state{state = 'Open', Peer = {self(), Caps}, dpr(CBs, [Reason, Ref, Peer], S); -%% Connection is open, DPR already sent. +%% Connection is open, DPR already sent or received. dpr(_, #state{state = 'Open'}) -> ok; @@ -1073,10 +1219,9 @@ dpr([CB|Rest], [Reason | _] = Args, S) -> dpr([], [Reason | _], S) -> send_dpr(Reason, [], S). --record(opts, {cause, timeout = ?DPA_TIMEOUT}). +-record(opts, {cause, timeout}). -send_dpr(Reason, Opts, #state{transport = TPid, - dictionary = Dict, +send_dpr(Reason, Opts, #state{dictionary = Dict, service = #diameter_service{capabilities = Caps}} = S) -> #opts{cause = Cause, timeout = Tmo} @@ -1085,24 +1230,37 @@ send_dpr(Reason, Opts, #state{transport = TPid, transport -> ?GOAWAY; _ -> ?REBOOT end, - timeout = ?DPA_TIMEOUT}, + timeout = dpa_timeout()}, Opts), #diameter_caps{origin_host = {OH, _}, origin_realm = {OR, _}} = Caps, - #diameter_packet{header = #diameter_header{end_to_end_id = Eid, - hop_by_hop_id = Hid}} - = Pkt - = encode(['DPR', {'Origin-Host', OH}, + Pkt = encode(['DPR', {'Origin-Host', OH}, {'Origin-Realm', OR}, {'Disconnect-Cause', Cause}], Dict), - incr(send, Pkt, Dict), + send_dpr(false, Pkt, Tmo, S). + +%% send_dpr/4 + +send_dpr(X, + #diameter_packet{header = #diameter_header{end_to_end_id = Eid, + hop_by_hop_id = Hid}} + = Pkt, + Tmo, + #state{transport = TPid, + dictionary = Dict} + = S) -> + %% Only count DPR sent by the service: explicit DPR is counted in + %% the same way as other explicitly sent requests. + X orelse incr(send, Pkt, Dict), send(TPid, Pkt), dpa_timer(Tmo), ?LOG(send, 'DPR'), - S#state{dpr = {Hid, Eid}}. + S#state{dpr = {X, Hid, Eid}}. + +%% opt/2 opt({timeout, Tmo}, Rec) when ?IS_TIMEOUT(Tmo) -> @@ -1125,6 +1283,27 @@ cause(N) -> dpa_timer(Tmo) -> erlang:send_after(Tmo, self(), dpa_timeout). +dpa_timeout() -> + dpa_timeout(getr(?DPA_KEY)). + +dpa_timeout({_, Tmo}) -> + Tmo; +dpa_timeout(undefined) -> %% set in old code + ?DPA_TIMEOUT; +dpa_timeout(Tmo) -> %% ditto + Tmo. + +dpr_timer() -> + dpa_timer(dpr_timeout()). + +dpr_timeout() -> + dpr_timeout(getr(?DPA_KEY)). + +dpr_timeout({Tmo, _}) -> + Tmo; +dpr_timeout(_) -> %% set in old code + ?DPR_TIMEOUT. + %% register_everywhere/1 %% %% Register a term and ensure it's not registered elsewhere. Note that diff --git a/lib/diameter/src/base/diameter_reg.erl b/lib/diameter/src/base/diameter_reg.erl index 3197c1aee1..f785777874 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-2013. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -22,10 +22,11 @@ %% -module(diameter_reg). --compile({no_auto_import, [monitor/2]}). - -behaviour(gen_server). +-compile({no_auto_import, [monitor/2, now/0]}). +-import(diameter_lib, [now/0]). + -export([add/1, add_new/1, del/1, diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index 76b05a2ad4..a01bcdd4e7 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -24,6 +24,9 @@ -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]). @@ -127,7 +130,8 @@ :: [{sequence, diameter:sequence()} %% sequence mask | {share_peers, diameter:remotes()} %% broadcast to | {use_shared_peers, diameter:remotes()} %% use from - | {restrict_connections, diameter:restriction()}]}). + | {restrict_connections, diameter:restriction()} + | {string_decode, boolean()}]}). %% shared_peers reflects the peers broadcast from remote nodes. %% Record representing an RFC 3539 watchdog process implemented by @@ -258,16 +262,22 @@ whois(SvcName) -> %% --------------------------------------------------------------------------- -spec pick_peer(SvcName, AppOrAlias, Opts) - -> {{TPid, Caps, App}, Mask} - | false - | {error, term()} + -> {{TPid, Caps, App}, Mask, SvcOpts} + | false %% no selection + | {error, no_service} when SvcName :: diameter:service_name(), - AppOrAlias :: {alias, diameter:app_alias()} | #diameter_app{}, - Opts :: tuple(), + AppOrAlias :: #diameter_app{} + | {alias, diameter:app_alias()}, + Opts :: {fun((Dict :: module()) -> [term()]), + diameter:peer_filter(), + Xtra :: list()}, TPid :: pid(), Caps :: #diameter_caps{}, App :: #diameter_app{}, - Mask :: diameter:sequence(). + Mask :: diameter:sequence(), + SvcOpts :: [diameter:service_opt()]. +%% Extract Mask in the returned tuple so that diameter_traffic doesn't +%% need to know about the ordering of SvcOpts used here. pick_peer(SvcName, App, Opts) -> pick(lookup_state(SvcName), App, Opts). @@ -284,10 +294,10 @@ pick(#state{service = #diameter_service{applications = Apps}} Opts) -> %% initial call from diameter:call/4 pick(S, find_outgoing_app(Alias, Apps), Opts); -pick(_, false, _) -> - false; +pick(_, false = No, _) -> + No; -pick(#state{options = [{_, Mask} | _]} +pick(#state{options = [{_, Mask} | SvcOpts]} = S, #diameter_app{module = ModX, dictionary = Dict} = App0, @@ -296,7 +306,7 @@ pick(#state{options = [{_, Mask} | _]} [_,_] = RealmAndHost = diameter_lib:eval([DestF, Dict]), case pick_peer(App, RealmAndHost, Filter, S) of {TPid, Caps} -> - {{TPid, Caps, App}, Mask}; + {{TPid, Caps, App}, Mask, SvcOpts}; false = No -> No end. @@ -610,8 +620,9 @@ st(#watchdog{ref = Ref, pid = Pid}, Refs) -> %% st/3 st(#watchdog{pid = Pid}, Reason, Acc) -> + MRef = monitor(process, Pid), Pid ! {shutdown, self(), Reason}, - [Pid | Acc]. + [MRef | Acc]. %% --------------------------------------------------------------------------- %% # call_service/2 @@ -686,7 +697,8 @@ service_options(Opts) -> {restrict_connections, proplists:get_value(restrict_connections, Opts, ?RESTRICT)}, - {spawn_opt, proplists:get_value(spawn_opt, Opts, [])}]. + {spawn_opt, proplists:get_value(spawn_opt, Opts, [])}, + {string_decode, proplists:get_value(string_decode, Opts, true)}]. %% The order of options is significant since we match against the list. mref(false = No) -> @@ -765,8 +777,9 @@ reason(failure) -> start(Ref, {T, Opts}, S) when T == connect; T == listen -> + N = proplists:get_value(pool_size, Opts, 1), try - {ok, start(Ref, type(T), Opts, S)} + {ok, start(Ref, type(T), Opts, N, S)} catch ?FAILURE(Reason) -> {error, Reason} @@ -784,26 +797,44 @@ type(connect = T) -> T. %% start/4 -start(Ref, Type, Opts, #state{watchdogT = WatchdogT, - peerT = PeerT, - options = SvcOpts, - service_name = SvcName, - service = Svc0}) +start(Ref, Type, Opts, State) -> + start(Ref, Type, Opts, 1, State). + +%% start/5 + +start(Ref, Type, Opts, N, #state{watchdogT = WatchdogT, + peerT = PeerT, + options = SvcOpts, + service_name = SvcName, + service = Svc0}) when Type == connect; Type == accept -> #diameter_service{applications = Apps} - = Svc + = Svc1 = merge_service(Opts, Svc0), - {_,_} = Mask = proplists:get_value(sequence, SvcOpts), - RecvData = diameter_traffic:make_recvdata([SvcName, PeerT, Apps, Mask]), - Pid = s(Type, Ref, {{spawn_opts([Opts, SvcOpts]), RecvData}, - Opts, - SvcOpts, - Svc}), - insert(WatchdogT, #watchdog{pid = Pid, - type = Type, - ref = Ref, - options = Opts}), + Svc = binary_caps(Svc1, proplists:get_value(string_decode, SvcOpts, true)), + RecvData = diameter_traffic:make_recvdata([SvcName, + PeerT, + Apps, + SvcOpts]), + T = {{spawn_opts([Opts, SvcOpts]), RecvData}, Opts, SvcOpts, Svc}, + Rec = #watchdog{type = Type, + ref = Ref, + options = Opts}, + diameter_lib:fold_n(fun(_,A) -> + [wd(Type, Ref, T, WatchdogT, Rec) | A] + end, + [], + N). + +binary_caps(Svc, true) -> + Svc; +binary_caps(#diameter_service{capabilities = Caps} = Svc, false) -> + Svc#diameter_service{capabilities = diameter_capx:binary_caps(Caps)}. + +wd(Type, Ref, T, WatchdogT, Rec) -> + Pid = start_watchdog(Type, Ref, T), + insert(WatchdogT, Rec#watchdog{pid = Pid}), Pid. %% Note that the service record passed into the watchdog is the merged @@ -816,7 +847,7 @@ spawn_opts(Optss) -> T /= link, T /= monitor]. -s(Type, Ref, T) -> +start_watchdog(Type, Ref, T) -> {_MRef, Pid} = diameter_watchdog:start({Type, Ref}, T), Pid. @@ -837,7 +868,7 @@ ms({applications, As}, #diameter_service{applications = Apps} = S) %% The fact that all capabilities can be configured on the transports %% means that the service doesn't necessarily represent a single -%% locally implemented Diameter peer as identified by Origin-Host: a +%% locally implemented Diameter node as identified by Origin-Host: a %% transport can configure its own Origin-Host. This means that the %% service little more than a placeholder for default capabilities %% plus a list of applications that individual transports can choose @@ -1185,7 +1216,7 @@ connect_timer(Opts, Def0) -> %% continuous restarted in case of faulty config or other problems. tc(Time, Tc) -> choose(Tc > ?RESTART_TC - orelse timer:now_diff(now(), Time) > 1000*?RESTART_TC, + orelse diameter_lib:micro_diff(Time) > 1000*?RESTART_TC, Tc, ?RESTART_TC). @@ -1718,31 +1749,43 @@ info_transport(S) -> [], PeerD). -%% Only a config entry for a listening transport: use it. -transport([[{type, listen}, _] = L]) -> - L ++ [{accept, []}]; - -%% Only one config or peer entry for a connecting transport: use it. -transport([[{type, connect} | _] = L]) -> - L; +%% Single config entry. Distinguish between pool_size config or not on +%% a connecting transport for backwards compatibility: with the option +%% the form is similar to the listening case, with connections grouped +%% in a pool tuple (for lack of a better name), without as before. +transport([[{type, Type}, {options, Opts}] = L]) + when Type == listen; + Type == connect -> + L ++ [{K, []} || [{_,K}] <- [keys(Type, Opts)]]; %% Peer entries: discard config. Note that the peer entries have %% length at least 3. transport([[_,_] | L]) -> transport(L); -%% Possibly many peer entries for a listening transport. Note that all -%% have the same options by construction, which is not terribly space -%% efficient. -transport([[{type, accept}, {options, Opts} | _] | _] = Ls) -> - [{type, listen}, +%% Multiple tranports. Note that all have the same options by +%% construction, which is not terribly space efficient. +transport([[{type, Type}, {options, Opts} | _] | _] = Ls) -> + transport(keys(Type, Opts), Ls). + +%% Group transports in an accept or pool tuple ... +transport([{Type, Key}], [[{type, _}, {options, Opts} | _] | _] = Ls) -> + [{type, Type}, {options, Opts}, - {accept, [lists:nthtail(2,L) || L <- Ls]}]. + {Key, [tl(tl(L)) || L <- Ls]}]; + +%% ... or not: there can only be one. +transport([], [L]) -> + L. + +keys(connect = T, Opts) -> + [{T, pool} || lists:keymember(pool_size, 1, Opts)]; +keys(_, _) -> + [{listen, accept}]. peer_dict(#state{watchdogT = WatchdogT, peerT = PeerT}, Dict0) -> try ets:tab2list(WatchdogT) of - L -> - lists:foldl(fun(T,A) -> peer_acc(PeerT, A, T) end, Dict0, L) + L -> lists:foldl(fun(T,A) -> peer_acc(PeerT, A, T) end, Dict0, L) catch error: badarg -> Dict0 %% service has gone down end. diff --git a/lib/diameter/src/base/diameter_service_sup.erl b/lib/diameter/src/base/diameter_service_sup.erl index 153fff902f..e3177f0083 100644 --- a/lib/diameter/src/base/diameter_service_sup.erl +++ b/lib/diameter/src/base/diameter_service_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -58,7 +58,7 @@ init([]) -> ChildSpec = {Mod, {Mod, start_link, []}, temporary, - 1000, + 5000, worker, [Mod]}, {ok, {Flags, [ChildSpec]}}. diff --git a/lib/diameter/src/base/diameter_session.erl b/lib/diameter/src/base/diameter_session.erl index 3b236f109a..c5ea0428b5 100644 --- a/lib/diameter/src/base/diameter_session.erl +++ b/lib/diameter/src/base/diameter_session.erl @@ -157,8 +157,8 @@ session_id(Host) -> %% --------------------------------------------------------------------------- init() -> - Now = now(), - random:seed(Now), + {Now, Seed} = diameter_lib:seed(), + random:seed(Seed), Time = time32(Now), Seq = (?INT32 band (Time bsl 20)) bor (random:uniform(1 bsl 20) - 1), ets:insert(diameter_sequence, [{origin_state_id, Time}, diff --git a/lib/diameter/src/base/diameter_stats.erl b/lib/diameter/src/base/diameter_stats.erl index 8353613d32..64ea082be0 100644 --- a/lib/diameter/src/base/diameter_stats.erl +++ b/lib/diameter/src/base/diameter_stats.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -22,9 +22,11 @@ %% -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, diff --git a/lib/diameter/src/base/diameter_sup.erl b/lib/diameter/src/base/diameter_sup.erl index e5afd23dcd..4ede4086d8 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-2011. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -64,7 +64,7 @@ spec(Mod) -> {Mod, {Mod, start_link, []}, permanent, - 1000, + infinity, supervisor, [Mod]}. diff --git a/lib/diameter/src/base/diameter_sync.erl b/lib/diameter/src/base/diameter_sync.erl index ce2db4b3a2..90eabece3d 100644 --- a/lib/diameter/src/base/diameter_sync.erl +++ b/lib/diameter/src/base/diameter_sync.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -27,6 +27,9 @@ -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]). diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl index 3b62afca47..784f9ca08f 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-2014. All Rights Reserved. +%% Copyright Ericsson AB 2013-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -77,7 +77,8 @@ {peerT :: ets:tid(), service_name :: diameter:service_name(), apps :: [#diameter_app{}], - sequence :: diameter:sequence()}). + sequence :: diameter:sequence(), + codec :: list()}). %% Record stored in diameter_request for each outgoing request. -record(request, @@ -92,11 +93,16 @@ %% # make_recvdata/1 %% --------------------------------------------------------------------------- -make_recvdata([SvcName, PeerT, Apps, Mask | _]) -> +make_recvdata([SvcName, PeerT, Apps, {_,_} = Mask | _]) -> %% from old code + make_recvdata([SvcName, PeerT, Apps, [{sequence, Mask}]]); + +make_recvdata([SvcName, PeerT, Apps, SvcOpts | _]) -> + {_,_} = Mask = proplists:get_value(sequence, SvcOpts), #recvdata{service_name = SvcName, peerT = PeerT, apps = Apps, - sequence = Mask}. + sequence = Mask, + codec = [T || {K,_} = T <- SvcOpts, K == string_decode]}. %% --------------------------------------------------------------------------- %% peer_up/1 @@ -162,24 +168,28 @@ incr_error(Dir, Id, TPid) -> %% incr_rc/4 %% --------------------------------------------------------------------------- --spec incr_rc(send|recv, Pkt, TPid, Dict0) +-spec incr_rc(send|recv, Pkt, TPid, DictT) -> {Counter, non_neg_integer()} | Reason when Pkt :: #diameter_packet{}, TPid :: pid(), - Dict0 :: module(), + DictT :: module() | {module(), module(), module()}, Counter :: {'Result-Code', integer()} | {'Experimental-Result', integer(), integer()}, Reason :: atom(). -incr_rc(Dir, Pkt, TPid, Dict0) -> +incr_rc(Dir, Pkt, TPid, {Dict, _, _} = DictT) -> try - incr_result(Dir, Pkt, TPid, {Dict0, Dict0, Dict0}) + incr_result(Dir, Pkt, TPid, DictT) catch exit: {E,_} when E == no_result_code; E == invalid_error_bit -> + incr(TPid, {msg_id(Pkt#diameter_packet.header, Dict), Dir, E}), E - end. + end; + +incr_rc(Dir, Pkt, TPid, Dict0) -> + incr_rc(Dir, Pkt, TPid, {Dict0, Dict0, Dict0}). %% --------------------------------------------------------------------------- %% pending/1 @@ -266,8 +276,11 @@ recv_request(TPid, #diameter_packet{header = #diameter_header{application_id = Id}} = Pkt, Dict0, - #recvdata{peerT = PeerT, apps = Apps} + #recvdata{peerT = PeerT, + apps = Apps, + codec = Opts} = RecvData) -> + diameter_codec:setopts([{common_dictionary, Dict0} | Opts]), send_A(recv_R(diameter_service:find_incoming_app(PeerT, TPid, Id, Apps), TPid, Pkt, @@ -275,7 +288,13 @@ recv_request(TPid, RecvData), TPid, Dict0, - RecvData). + RecvData); + +recv_request(TPid, Pkt, Dict0, RecvData) -> %% from old code + recv_request(TPid, + Pkt, + Dict0, + #recvdata{} = erlang:append_element(RecvData, [])). %% recv_R/5 @@ -592,7 +611,7 @@ resend(false, Route = #diameter_avp{data = {Dict0, 'Route-Record', OH}}, Seq = diameter_session:sequence(Mask), Hdr = Hdr0#diameter_header{hop_by_hop_id = Seq}, - Msg = [Hdr, Route | Avps], + Msg = [Hdr, Route | Avps], %% reordered at encode resend(send_request(SvcName, App, Msg, Opts), Caps, Dict0, Pkt). %% The incoming request is relayed with the addition of a %% Route-Record. Note the requirement on the return from call/4 below, @@ -678,7 +697,7 @@ local(Msg, TPid, {Dict, AppDict, Dict0} = DictT, Fs, ReqPkt) -> reset(make_answer_packet(Msg, ReqPkt), Dict, Dict0), Fs), incr(send, Pkt, TPid, AppDict), - incr_result(send, Pkt, TPid, DictT), %% count outgoing + incr_rc(send, Pkt, TPid, DictT), %% count outgoing send(TPid, Pkt). %% reset/3 @@ -1221,10 +1240,9 @@ answer_rc(_, _, Sent) -> send_R(SvcName, AppOrAlias, Msg, Opts, Caller) -> case pick_peer(SvcName, AppOrAlias, Msg, Opts) of - {{_,_,_} = Transport, Mask} -> + {Transport, Mask, SvcOpts} -> + diameter_codec:setopts(SvcOpts), send_request(Transport, Mask, Msg, Opts, Caller, SvcName); - false -> - {error, no_connection}; {error, _} = No -> No end. @@ -1286,6 +1304,8 @@ send_request({TPid, Caps, App} SvcName, []). +%% send_R/7 + send_R({send, Msg}, Pkt, Transport, Opts, Caller, SvcName, Fs) -> send_R(make_request_packet(Msg, Pkt), Transport, @@ -1388,6 +1408,21 @@ 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(#diameter_packet{msg = [#diameter_header{} = Hdr + | Avps]} + = Pkt) -> + Pkt#diameter_packet{msg = [make_retransmit_header(Hdr) | Avps]}; + +make_retransmit_packet(#diameter_packet{header = Hdr} = Pkt) -> + Pkt#diameter_packet{header = make_retransmit_header(Hdr)}. + +%% make_retransmit_header/1 + +make_retransmit_header(Hdr) -> + Hdr#diameter_header{is_retransmitted = true}. + %% fold_record/2 fold_record(undefined, R) -> @@ -1531,7 +1566,9 @@ a(Hdr, SvcName, discard) -> %% timer value is ignored. This means that an answer could be accepted %% from a peer after timeout in the case of failover. -retransmit({{_,_,App} = Transport, _Mask}, Req, Opts, SvcName, Timeout) -> +%% retransmit/5 + +retransmit({{_,_,App} = Transport, _, _}, Req, Opts, SvcName, Timeout) -> try retransmit(Transport, Req, SvcName, Timeout) of T -> recv_A(Timeout, SvcName, App, Opts, T) catch @@ -1552,17 +1589,26 @@ pick_peer(SvcName, pick_peer(SvcName, App, Msg, Opts#options{extra = []}); pick_peer(_, _, undefined, _) -> - false; + {error, no_connection}; pick_peer(SvcName, AppOrAlias, Msg, #options{filter = Filter, extra = Xtra}) -> - diameter_service:pick_peer(SvcName, - AppOrAlias, - {fun(D) -> get_destination(D, Msg) end, - Filter, - Xtra}). + pick(diameter_service:pick_peer(SvcName, + AppOrAlias, + {fun(D) -> get_destination(D, Msg) end, + Filter, + Xtra})). + +pick({{_,_,_} = Transport, Mask}) -> %% from old code; dialyzer complains + {Transport, Mask, []}; %% about this + +pick(false) -> + {error, no_connection}; + +pick(T) -> + T. %% handle_error/4 @@ -1660,8 +1706,14 @@ recv(TPid, Pid, TRef, Ref) -> %% send/2 -send(Pid, Pkt) -> - Pid ! {send, Pkt}. +send(Pid, Pkt) -> %% Strip potentially large message terms. + #diameter_packet{header = H, + bin = Bin, + transport_data = T} + = Pkt, + Pid ! {send, #diameter_packet{header = H, + bin = Bin, + transport_data = T}}. %% retransmit/4 @@ -1674,9 +1726,7 @@ retransmit({TPid, Caps, App} have_request(Pkt0, TPid) %% Don't failover to a peer we've andalso ?THROW(timeout), %% already sent to. - #diameter_packet{header = Hdr0} = Pkt0, - Hdr = Hdr0#diameter_header{is_retransmitted = true}, - Pkt = Pkt0#diameter_packet{header = Hdr}, + Pkt = make_retransmit_packet(Pkt0), retransmit(cb(App, prepare_retransmit, [Pkt, SvcName, {TPid, Caps}]), Transport, diff --git a/lib/diameter/src/base/diameter_types.erl b/lib/diameter/src/base/diameter_types.erl index ca3338be5f..fe7613541c 100644 --- a/lib/diameter/src/base/diameter_types.erl +++ b/lib/diameter/src/base/diameter_types.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2013. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -75,7 +75,7 @@ %% message indicating this error MUST include the offending AVPs %% within a Failed-AVP AVP. %% --define(INVALID_LENGTH(Bin), erlang:error({'DIAMETER', 5014, Bin})). +-define(INVALID_LENGTH(Bitstr), erlang:error({'DIAMETER', 5014, Bitstr})). %% ------------------------------------------------------------------------- %% 3588, 4.2. Basic AVP Data Formats @@ -90,7 +90,12 @@ 'OctetString'(decode, Bin) when is_binary(Bin) -> - binary_to_list(Bin); + case diameter_codec:getopt(string_decode) of + true -> + binary_to_list(Bin); + _ -> + Bin + end; 'OctetString'(decode, B) -> ?INVALID_LENGTH(B); @@ -298,21 +303,29 @@ 'OctetString'(M, lists:duplicate(0,7)); 'DiameterURI'(encode, #diameter_uri{type = Type, - fqdn = D, - port = P, + fqdn = DN, + port = PN, transport = T, - protocol = Prot} - = U) -> - S = lists:append([atom_to_list(Type), "://", D, - ":", integer_to_list(P), + protocol = P}) + when (Type == 'aaa' orelse Type == 'aaas'), + is_integer(PN), + 0 =< PN, + (T == tcp orelse T == sctp orelse T == udp), + (P == diameter orelse P == radius orelse P == 'tacacs+'), + (P /= diameter orelse T /= udp) -> + iolist_to_binary([atom_to_list(Type), "://", DN, + ":", integer_to_list(PN), ";transport=", atom_to_list(T), - ";protocol=", atom_to_list(Prot)]), - U = scan_uri(S), %% assert - list_to_binary(S); + ";protocol=", atom_to_list(P)]); +%% Don't omit defaults since they're dependent on whether RFC 3588 or +%% 6733 is being followed. For one, we don't know this at encode; for +%% two (more importantly), we don't know how the peer will interpret +%% defaults, so it's best to be explicit. Interpret defaults on decode +%% since there's no choice. 'DiameterURI'(encode, Str) -> Bin = iolist_to_binary(Str), - #diameter_uri{} = scan_uri(Bin), %% type check + #diameter_uri{} = scan_uri(Bin), %% assert Bin. %% -------------------- @@ -321,7 +334,6 @@ 'IPFilterRule'(encode = M, zero) -> 'OctetString'(M, lists:duplicate(0,33)); -%% TODO: parse grammar. 'IPFilterRule'(M, X) -> 'OctetString'(M, X). @@ -331,7 +343,6 @@ 'QoSFilterRule'(encode = M, zero = X) -> 'IPFilterRule'(M, X); -%% TODO: parse grammar. 'QoSFilterRule'(M, X) -> 'OctetString'(M, X). @@ -339,7 +350,13 @@ 'UTF8String'(decode, Bin) when is_binary(Bin) -> - tl([0|_] = unicode:characters_to_list([0, Bin])); %% assert list return + case diameter_codec:getopt(string_decode) of + true -> + %% assert list return + tl([0|_] = unicode:characters_to_list([0, Bin])); + false -> + <<_/binary>> = unicode:characters_to_binary(Bin) + end; 'UTF8String'(decode, B) -> ?INVALID_LENGTH(B); @@ -507,55 +524,87 @@ msb(false) -> ?TIME_2036. %% %% aaa-protocol = ( "diameter" / "radius" / "tacacs+" ) -scan_uri(Bin) - when is_binary(Bin) -> - scan_uri(binary_to_list(Bin)); -scan_uri("aaa://" ++ Rest) -> - scan_fqdn(Rest, #diameter_uri{type = aaa}); -scan_uri("aaas://" ++ Rest) -> - scan_fqdn(Rest, #diameter_uri{type = aaas}). - -scan_fqdn(S, U) -> - {[_|_] = F, Rest} = lists:splitwith(fun is_fqdn/1, S), - scan_opt_port(Rest, U#diameter_uri{fqdn = F}). - -scan_opt_port(":" ++ S, U) -> - {[_|_] = P, Rest} = lists:splitwith(fun is_digit/1, S), - scan_opt_transport(Rest, U#diameter_uri{port = list_to_integer(P)}); -scan_opt_port(S, U) -> - scan_opt_transport(S, U). - -scan_opt_transport(";transport=" ++ S, U) -> - {P, Rest} = transport(S), - scan_opt_protocol(Rest, U#diameter_uri{transport = P}); -scan_opt_transport(S, U) -> - scan_opt_protocol(S, U). - -scan_opt_protocol(";protocol=" ++ S, U) -> - {P, ""} = protocol(S), - U#diameter_uri{protocol = P}; -scan_opt_protocol("", U) -> - U. - -transport("tcp" ++ S) -> - {tcp, S}; -transport("sctp" ++ S) -> - {sctp, S}; -transport("udp" ++ S) -> - {udp, S}. - -protocol("diameter" ++ S) -> - {diameter, S}; -protocol("radius" ++ S) -> - {radius, S}; -protocol("tacacs+" ++ S) -> - {'tacacs+', S}. - -is_fqdn(C) -> - is_digit(C) orelse is_alpha(C) orelse C == $. orelse C == $-. - -is_alpha(C) -> - ($a =< C andalso C =< $z) orelse ($A =< C andalso C =< $Z). - -is_digit(C) -> - $0 =< C andalso C =< $9. +%% RFC 6733, 4.3.1, changes the defaults: +%% +%% "aaa://" FQDN [ port ] [ transport ] [ protocol ] +%% +%% ; No transport security +%% +%% "aaas://" FQDN [ port ] [ transport ] [ protocol ] +%% +%% ; Transport security used +%% +%% FQDN = < Fully Qualified Domain Name > +%% +%% port = ":" 1*DIGIT +%% +%% ; One of the ports used to listen for +%% ; incoming connections. +%% ; If absent, the default Diameter port +%% ; (3868) is assumed if no transport +%% ; security is used and port 5658 when +%% ; transport security (TLS/TCP and DTLS/SCTP) +%% ; is used. +%% +%% transport = ";transport=" transport-protocol +%% +%% ; One of the transports used to listen +%% ; for incoming connections. If absent, +%% ; the default protocol is assumed to be TCP. +%% ; UDP MUST NOT be used when the aaa-protocol +%% ; field is set to diameter. +%% +%% transport-protocol = ( "tcp" / "sctp" / "udp" ) +%% +%% protocol = ";protocol=" aaa-protocol +%% +%% ; If absent, the default AAA protocol +%% ; is Diameter. +%% +%% aaa-protocol = ( "diameter" / "radius" / "tacacs+" ) + +scan_uri(Bin) -> + RE = "^(aaas?)://" + "([-a-zA-Z0-9.]+)" + "(:([0-9]+))?" + "(;transport=(tcp|sctp|udp))?" + "(;protocol=(diameter|radius|tacacs\\+))?$", + {match, [A, DN, PN, T, P]} = re:run(Bin, + RE, + [{capture, [1,2,4,6,8], binary}]), + Type = to_atom(A), + {PN0, T0} = defaults(diameter_codec:getopt(rfc), Type), + #diameter_uri{type = Type, + fqdn = from_bin(DN), + port = to_int(PN, PN0), + transport = to_atom(T, T0), + protocol = to_atom(P, diameter)}. + +%% Choose defaults based on the RFC, since 6733 has changed them. +defaults(3588, _) -> + {3868, sctp}; +defaults(6733, aaa) -> + {3868, tcp}; +defaults(6733, aaas) -> + {5658, tcp}. + +from_bin(B) -> + case diameter_codec:getopt(string_decode) of + true -> + binary_to_list(B); + false -> + B + end. + +to_int(<<>>, N) -> + N; +to_int(B, _) -> + binary_to_integer(B). + +to_atom(<<>>, A) -> + A; +to_atom(B, _) -> + to_atom(B). + +to_atom(B) -> + binary_to_atom(B, latin1). diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl index b7f2d24941..de9c4bca33 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-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -122,15 +122,18 @@ i({Ack, T, Pid, {RecvData, = Svc}}) -> erlang:monitor(process, Pid), wait(Ack, Pid), - random:seed(now()), - putr(restart, {T, Opts, Svc}), %% save seeing it in trace - putr(dwr, dwr(Caps)), %% + {_, 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), Restrict = proplists:get_value(restrict_connections, SvcOpts), Nodes = restrict_nodes(Restrict), Dict0 = common_dictionary(Apps), + diameter_codec:setopts([{common_dictionary, Dict0}, + {string_decode, false}]), #watchdog{parent = Pid, - transport = start(T, Opts, Mask, Nodes, Dict0, Svc), + transport = start(T, Opts, SvcOpts, Nodes, Dict0, Svc), tw = proplists:get_value(watchdog_timer, Opts, ?DEFAULT_TW_INIT), @@ -165,11 +168,11 @@ config({okay, N}, Rec) when ?IS_NATURAL(N) -> Rec#config{okay = N}. -%% start/5 +%% start/6 -start(T, Opts, Mask, Nodes, Dict0, Svc) -> +start(T, Opts, SvcOpts, Nodes, Dict0, Svc) -> {_MRef, Pid} - = diameter_peer_fsm:start(T, Opts, {Mask, Nodes, Dict0, Svc}), + = diameter_peer_fsm:start(T, Opts, {SvcOpts, Nodes, Dict0, Svc}), Pid. %% common_dictionary/1 @@ -319,7 +322,7 @@ code_change(_, State, _) -> %% expiry; or another watchdog is saying the same after reestablishing %% a connection previously had by this one. transition(close, #watchdog{}) -> - {{accept, _}, _, _} = getr(restart), %% assert + {accept, _} = role(), %% assert stop; %% Service is asking for the peer to be taken down gracefully. @@ -332,8 +335,9 @@ transition({shutdown = T, Pid, Reason}, #watchdog{parent = Pid, send(TPid, {T, self(), Reason}), S#watchdog{shutdown = true}; -%% Transport is telling us that DPA has been sent in response to DPR: -%% its death should lead to ours. +%% Transport is telling us that DPA has been sent in response to DPR, +%% or that DPR has been explicitly sent: transport death should lead +%% to ours. transition({'DPR', TPid}, #watchdog{transport = TPid} = S) -> S#watchdog{shutdown = true}; @@ -368,7 +372,7 @@ transition({open, TPid, Hosts, _} = Open, restrict = {_,R}, config = #config{suspect = OS}} = S) -> - case okay(getr(restart), Hosts, R) of + case okay(role(), Hosts, R) of okay -> set_watchdog(S#watchdog{status = okay, num_dwa = OS}); @@ -422,7 +426,7 @@ transition({'DOWN', _, process, TPid, _Reason} = D, = S0) -> S = S0#watchdog{pending = false, transport = undefined}, - {{M,_}, _, _} = getr(restart), + {M,_} = role(), %% Close an accepting watchdog immediately if there's no %% restriction on the number of connections to the same peer: the @@ -489,7 +493,7 @@ encode(dwa, Dict0, #diameter_packet{header = H, transport_data = TD} %% okay/3 -okay({{accept, Ref}, _, _}, Hosts, Restrict) -> +okay({accept, Ref}, Hosts, Restrict) -> T = {?MODULE, connection, Ref, Hosts}, diameter_reg:add(T), if Restrict -> @@ -500,7 +504,7 @@ okay({{accept, Ref}, _, _}, Hosts, Restrict) -> %% Register before matching so that at least one of two registering %% processes will match the other. -okay({{connect, _}, _, _}, _, _) -> +okay({connect, _}, _, _) -> okay. %% okay/2 @@ -515,6 +519,11 @@ okay(C) -> [_|_] = [send(P, close) || {_,P} <- C, self() /= P], reopen. +%% role/0 + +role() -> + element(1, getr(restart)). + %% set_watchdog/1 set_watchdog(#watchdog{tw = TwInit, @@ -549,7 +558,7 @@ send_watchdog(#watchdog{pending = false, ?LOG(send, 'DWR'), S#watchdog{pending = true}. -%% Dont' count encode errors since we don't expect any on DWR/DWA. +%% Don't count encode errors since we don't expect any on DWR/DWA. %% recv/3 @@ -571,11 +580,18 @@ rcv('DWR', Pkt, #watchdog{transport = TPid, DPkt = diameter_codec:decode(Dict0, Pkt), diameter_traffic:incr(recv, DPkt, TPid, Dict0), diameter_traffic:incr_error(recv, DPkt, TPid, Dict0), - EPkt = encode(dwa, Dict0, Pkt), + #diameter_packet{header = H, + transport_data = T, + bin = Bin} + = EPkt + = encode(dwa, Dict0, Pkt), diameter_traffic:incr(send, EPkt, TPid, Dict0), diameter_traffic:incr_rc(send, EPkt, TPid, Dict0), - send(TPid, {send, EPkt}), + %% Strip potentially large message terms. + send(TPid, {send, #diameter_packet{header = H, + transport_data = T, + bin = Bin}}), ?LOG(send, 'DWA'); rcv('DWA', Pkt, #watchdog{transport = TPid, @@ -590,9 +606,10 @@ rcv('DWA', Pkt, #watchdog{transport = TPid, rcv(N, _, _) when N == 'CER'; N == 'CEA'; - N == 'DPR'; - N == 'DPA' -> + N == 'DPR' -> false; +%% DPR can be sent explicitly with diameter:call/4. Only the +%% corresponding DPAs arrive here. rcv(_, Pkt, #watchdog{transport = TPid, dictionary = Dict0, @@ -793,26 +810,28 @@ restart(S) -> %% reconnect has won race with timeout %% state down rather then initial when receiving notification of an %% open connection. -restart({{connect, _} = T, Opts, Svc}, +restart({T, Opts, Svc}, S) -> %% put in old code + restart({T, Opts, Svc, []}, S); + +restart({{connect, _} = T, Opts, Svc, SvcOpts}, #watchdog{parent = Pid, - sequence = Mask, restrict = {R,_}, dictionary = Dict0} = S) -> send(Pid, {reconnect, self()}), Nodes = restrict_nodes(R), - S#watchdog{transport = start(T, Opts, Mask, Nodes, Dict0, Svc), + S#watchdog{transport = start(T, Opts, SvcOpts, Nodes, Dict0, Svc), restrict = {R, lists:member(node(), Nodes)}}; %% No restriction on the number of connections to the same peer: just %% die. Note that a state machine never enters state REOPEN in this %% case. -restart({{accept, _}, _, _}, #watchdog{restrict = {_, false}}) -> +restart({{accept, _}, _, _, _}, #watchdog{restrict = {_, false}}) -> stop; %% 'DOWN' was in old code: 'close' was not sent %% Otherwise hang around until told to die, either by the service or %% by another watchdog. -restart({{accept, _}, _, _}, S) -> +restart({{accept, _}, _, _, _}, S) -> S. %% Don't currently use Opts/Svc in the accept case. diff --git a/lib/diameter/src/diameter.appup.src b/lib/diameter/src/diameter.appup.src index 881d25b5fb..a54eb24031 100644 --- a/lib/diameter/src/diameter.appup.src +++ b/lib/diameter/src/diameter.appup.src @@ -2,7 +2,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -59,7 +59,32 @@ {"1.7.1", [{load_module, diameter_traffic}, %% 17.3 {load_module, diameter_watchdog}, {load_module, diameter_peer_fsm}, - {load_module, diameter_service}]} + {load_module, diameter_service}]}, + {"1.8", [{load_module, diameter_lib}, %% 17.4 + {load_module, diameter_peer}, + {load_module, diameter_reg}, + {load_module, diameter_session}, + {load_module, diameter_stats}, + {load_module, diameter_sync}, + {load_module, diameter_capx}, + {load_module, diameter_codec}, + {load_module, diameter_types}, + {load_module, diameter_traffic}, + {load_module, diameter_service}, + {load_module, diameter_peer_fsm}, + {load_module, diameter_watchdog}, + {load_module, diameter_tcp}, + {load_module, diameter_sctp}, + {load_module, diameter_config}, + {load_module, diameter}, + {load_module, diameter_gen_base_rfc6733}, + {load_module, diameter_gen_acct_rfc6733}, + {load_module, diameter_gen_base_rfc3588}, + {load_module, diameter_gen_base_accounting}, + {load_module, diameter_gen_relay}, + {update, diameter_transport_sup, supervisor}, + {update, diameter_service_sup, supervisor}, + {update, diameter_sup, supervisor}]} ], [ {"0.9", [{restart_application, diameter}]}, @@ -101,6 +126,31 @@ {"1.7.1", [{load_module, diameter_service}, {load_module, diameter_peer_fsm}, {load_module, diameter_watchdog}, - {load_module, diameter_traffic}]} + {load_module, diameter_traffic}]}, + {"1.8", [{update, diameter_sup, supervisor}, + {update, diameter_service_sup, supervisor}, + {update, diameter_transport_sup, supervisor}, + {load_module, diameter_gen_relay}, + {load_module, diameter_gen_base_accounting}, + {load_module, diameter_gen_base_rfc3588}, + {load_module, diameter_gen_acct_rfc6733}, + {load_module, diameter_gen_base_rfc6733}, + {load_module, diameter}, + {load_module, diameter_config}, + {load_module, diameter_sctp}, + {load_module, diameter_tcp}, + {load_module, diameter_watchdog}, + {load_module, diameter_peer_fsm}, + {load_module, diameter_service}, + {load_module, diameter_traffic}, + {load_module, diameter_types}, + {load_module, diameter_codec}, + {load_module, diameter_capx}, + {load_module, diameter_sync}, + {load_module, diameter_stats}, + {load_module, diameter_session}, + {load_module, diameter_reg}, + {load_module, diameter_peer}, + {load_module, diameter_lib}]} ] }. diff --git a/lib/diameter/src/modules.mk b/lib/diameter/src/modules.mk index a2a7a51892..c9dd4e683a 100644 --- a/lib/diameter/src/modules.mk +++ b/lib/diameter/src/modules.mk @@ -1,7 +1,7 @@ # %CopyrightBegin% # -# Copyright Ericsson AB 2010-2014. All Rights Reserved. +# Copyright Ericsson AB 2010-2015. All Rights Reserved. # # The contents of this file are subject to the Erlang Public License, # Version 1.1, (the "License"); you may not use this file except in @@ -94,7 +94,7 @@ BINS = \ # Released files relative to ../examples. EXAMPLES = \ code/GNUmakefile \ - code/peer.erl \ + code/node.erl \ code/client.erl \ code/client_cb.erl \ code/server.erl \ diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl index 32e7aaca39..2c8d6f0a14 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-2014. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -18,9 +18,11 @@ %% -module(diameter_sctp). - -behaviour(gen_server). +-compile({no_auto_import, [now/0]}). +-import(diameter_lib, [now/0]). + %% interface -export([start/3]). @@ -37,7 +39,8 @@ code_change/3, terminate/2]). --export([info/1]). %% service_info callback +-export([listener/1,%% diameter_sync callback + info/1]). %% service_info callback -export([ports/0, ports/1]). @@ -99,22 +102,31 @@ -record(listener, {ref :: reference(), socket :: gen_sctp:sctp_socket(), - count = 0 :: uint(), + count = 0 :: uint(), %% attached transport processes tmap = ets:new(?MODULE, []) :: ets:tid(), %% {MRef, Pid|AssocId}, {AssocId, Pid} pending = {0, ets:new(?MODULE, [ordered_set])}, tref :: reference(), accept :: [match()]}). %% Field tmap is used to map an incoming message or event to the -%% relevent transport process. Field pending implements a queue of -%% transport processes to which an association has been assigned (at -%% comm_up and written into tmap) but for which diameter hasn't yet -%% spawned a transport process: a short-lived state of affairs as a -%% new transport is spawned as a consequence of a peer being taken up, -%% transport processes being spawned by the listener on demand. In -%% case diameter starts a transport before comm_up on a new -%% association, pending is set to an improper list with the spawned -%% transport as head and the queue as tail. +%% relevant transport process. Field pending implements two queues: +%% the first of transport-to-be processes to which an association has +%% been assigned (at comm_up and written into tmap) but for which +%% diameter hasn't yet spawned a transport process, a short-lived +%% state of affairs as a new transport is spawned as a consequence of +%% a peer being taken up, transport processes being spawned by the +%% listener on demand; the second of started transport processes that +%% have not yet been assigned an association. +%% +%% When diameter calls start/3, the transport process is either taken +%% from the first queue or spawned and placed in the second queue +%% until an association is established. When an association is +%% established, a controlling process is either taken from the second +%% queue or spawned and placed in the first queue. Thus, there are +%% only elements in one queue at a time, so share an ets table queue +%% and tag it with a positive length if it contains the first queue, a +%% negative length if it contains the second queue. The case -1 is +%% handled differently for backwards compatibility reasons. %% --------------------------------------------------------------------------- %% # start/3 @@ -139,9 +151,9 @@ ip(T) -> T. %% A listener spawns transports either as a consequence of this call -%% when there is not yet an association to associate with it, or at -%% comm_up on a new association in which case the call retrieves a -%% transport from the pending queue. +%% 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 @@ -226,7 +238,7 @@ i({connect, Pid, Opts, Addrs, Ref}) -> {LAs, Sock} = open(Addrs, Rest, 0), putr(?REF_KEY, Ref), proc_lib:init_ack({ok, self(), LAs}), - erlang:monitor(process, Pid), + monitor(process, Pid), #transport{parent = Pid, mode = {connect, connect(Sock, RAs, RP, [])}, socket = Sock}; @@ -236,8 +248,8 @@ i({accept, Pid, LPid, Sock, Ref}) when is_pid(Pid) -> putr(?REF_KEY, Ref), proc_lib:init_ack({ok, self()}), - erlang:monitor(process, Pid), - erlang:monitor(process, LPid), + monitor(process, Pid), + monitor(process, LPid), #transport{parent = Pid, mode = {accept, LPid}, socket = Sock}; @@ -246,7 +258,7 @@ i({accept, Pid, LPid, Sock, Ref}) i({accept, Ref, LPid, Sock, Id}) -> putr(?REF_KEY, Ref), proc_lib:init_ack({ok, self()}), - MRef = erlang:monitor(process, LPid), + MRef = monitor(process, LPid), %% Wait for a signal that the transport has been started before %% processing other messages. receive @@ -270,15 +282,23 @@ close(Sock, Id) -> %% listener/2 +%% 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}]}, + infinity, + infinity). + +listener({LRef, T}) -> l(diameter_reg:match({?MODULE, listener, {LRef, '_'}}), LRef, T). -%% Existing process with the listening socket ... +%% Existing listening process ... l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) -> - {LAs, _Sock} = AS, - {LPid, LAs}; - -%% ... or not: start one. + {LAs, _Sock} = AS, + {LPid, LAs}; + +%% ... or not. l([], LRef, T) -> {ok, LPid, LAs} = diameter_sctp_sup:start_child({listen, LRef, T}), {LPid, LAs}. @@ -347,11 +367,17 @@ type(T) -> %% # handle_call/3 %% --------------------------------------------------------------------------- +handle_call(T, From, #listener{pending = L} = S) + when is_list(L) -> + handle_call(T, From, upgrade(S)); + handle_call({{accept, Ref}, Pid}, _, #listener{ref = Ref, - count = N} + pending = {N,Q}, + count = K} = S) -> - {TPid, NewS} = accept(Ref, Pid, S), - {reply, {ok, TPid}, NewS#listener{count = N+1}}; + TPid = accept(Ref, Pid, S), + {reply, {ok, TPid}, downgrade(S#listener{pending = {N-1,Q}, + count = K+1})}; handle_call(_, _, State) -> {reply, nok, State}. @@ -370,8 +396,46 @@ handle_cast(_, State) -> handle_info(T, #transport{} = S) -> {noreply, #transport{} = t(T,S)}; +handle_info(T, #listener{pending = L} = S) + when is_list(L) -> + handle_info(T, upgrade(S)); + handle_info(T, #listener{} = S) -> - {noreply, #listener{} = l(T,S)}. + {noreply, downgrade(#listener{} = l(T,S))}. + +%% upgrade/1 + +upgrade(#listener{pending = [TPid | {0,Q}]} = S) -> + ets:insert(Q, {TPid, now()}), + S#listener{pending = {-1,Q}}. +%% Prior to the possiblity of setting pool_size on in transport +%% configuration, a new accepting transport was only started following +%% the death of a predecessor, so that there was only at most one +%% previously started transport process waiting for an association. +%% This assumption no longer holds with pool_size > 1, in which case +%% several accepting transports are started concurrently. Deal with +%% this by placing the started transports in a new queue of transport +%% processes waiting for an association. +%% +%% Since only one of this queue and the existing queue of controlling +%% processes waiting for a transport to be started can be non-empty at +%% any given time, implement both queues in the same ets table. The +%% absolute value of the first element of the 2-tuple is the queue +%% length, the sign says which queue it is. + +%% downgrade/1 +%% +%% Revert to the pre-pool_size representation when possible, for +%% backwards compatibility in the case that the pool_size option +%% hasn't been used. + +downgrade(#listener{pending = {-1,Q}} = S) -> + TPid = ets:first(Q), + ets:delete(Q, TPid), + S#listener{pending = [TPid | {0,Q}]}; + +downgrade(S) -> + S. %% --------------------------------------------------------------------------- %% # code_change/3 @@ -436,54 +500,46 @@ l({sctp, Sock, _RA, _RP, Data} = Msg, #listener{socket = Sock} = S) -> setopts(Sock) end; -%% Transport is asking message to be sent. See send/3 for why the send -%% isn't directly from the transport. -l({send, AssocId, StreamId, Bin}, #listener{socket = Sock} = S) -> - send(Sock, AssocId, StreamId, Bin), - S; +l({'DOWN', MRef, process, TPid, _}, #listener{pending = {_,Q}} = S) -> + down(ets:member(Q, TPid), MRef, 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. + +%% down/4 %% Accepting transport has died. One that's awaiting an association ... -l({'DOWN', MRef, process, TPid, _}, #listener{pending = [TPid | Q], - tmap = T, - count = N} - = S) -> +down(true, MRef, TPid, #listener{pending = {N,Q}, + tmap = T, + count = K} + = S) + when N < 0 -> + ets:delete(Q, TPid), ets:delete(T, MRef), ets:delete(T, TPid), - start_timer(S#listener{count = N-1, - pending = Q}); - -%% ... ditto and a new transport has already been started ... -l({'DOWN', _, process, _, _} = T, #listener{pending = [TPid | Q]} - = S) -> - #listener{pending = NQ} - = NewS - = l(T, S#listener{pending = Q}), - NewS#listener{pending = [TPid | NQ]}; - -%% ... or not. -l({'DOWN', MRef, process, TPid, _}, #listener{socket = Sock, - tmap = T, - count = N, - pending = {P,Q}} - = S) -> + start_timer(S#listener{count = K-1, + pending = {N+1,Q}}); + +%% ... or one that already has one. +down(B, MRef, TPid, #listener{socket = Sock, + tmap = T, + count = K, + pending = {N,Q}} + = S) -> [{MRef, Id}] = ets:lookup(T, MRef), %% Id = TPid | AssocId ets:delete(T, MRef), ets:delete(T, Id), Id == TPid orelse close(Sock, Id), - case ets:lookup(Q, TPid) of - [{TPid, _}] -> %% transport in the pending queue ... + if B -> %% Waiting for attachment in the pending queue ... ets:delete(Q, TPid), - S#listener{pending = {P-1, Q}}; - [] -> %% ... or not - start_timer(S#listener{count = N-1}) - end; - -%% 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. + S#listener{pending = {N-1,Q}}; + true -> %% ... or already attached + start_timer(S#listener{count = K-1}) + end. %% t/2 %% @@ -582,29 +638,24 @@ accept(Opts) -> %% No pending associations: spawn a new transport. accept(Ref, Pid, #listener{socket = Sock, tmap = T, - pending = {0,_} = Q} - = S) -> + pending = {N,Q}}) + when N =< 0 -> Arg = {accept, Pid, self(), Sock, Ref}, {ok, TPid} = diameter_sctp_sup:start_child(Arg), - MRef = erlang:monitor(process, TPid), + MRef = monitor(process, TPid), ets:insert(T, [{MRef, TPid}, {TPid, MRef}]), - {TPid, S#listener{pending = [TPid | Q]}}; -%% Placing the transport in the pending field makes it available to -%% the next association. The stack starts a new accepting transport -%% only after this one brings the connection up (or dies). - -%% Accepting transport has died. This can happen if a new transport is -%% started before the DOWN has arrived. -accept(Ref, Pid, #listener{pending = [TPid | {0,_} = Q]} = S) -> - false = is_process_alive(TPid), %% assert - accept(Ref, Pid, S#listener{pending = Q}); + ets:insert(Q, {TPid, now()}), + TPid; +%% Placing the transport in the second pending table makes it +%% available to the next association. %% Pending associations: attach to the first in the queue. -accept(_, Pid, #listener{ref = Ref, pending = {N,Q}} = S) -> +accept(_, Pid, #listener{ref = Ref, + pending = {_,Q}}) -> TPid = ets:first(Q), TPid ! {Ref, Pid}, ets:delete(Q, TPid), - {TPid, S#listener{pending = {N-1, Q}}}. + TPid. %% send/2 @@ -718,34 +769,12 @@ up(#transport{parent = Pid, find(Id, Data, #listener{tmap = T} = S) -> f(ets:lookup(T, Id), Data, S). -%% New association and a transport waiting for one: use it. -f([], - {_, #sctp_assoc_change{state = comm_up, - assoc_id = Id}}, - #listener{tmap = T, - pending = [TPid | {_,_} = Q]} - = S) -> - [{TPid, MRef}] = ets:lookup(T, TPid), - ets:insert(T, [{MRef, Id}, {Id, TPid}]), - ets:delete(T, TPid), - {TPid, S#listener{pending = Q}}; - -%% New association and no transport start yet: spawn one and place it -%% in the queue. +%% New association ... f([], - {_, #sctp_assoc_change{state = comm_up, - assoc_id = Id}}, - #listener{ref = Ref, - socket = Sock, - tmap = T, - pending = {N,Q}} + {_, #sctp_assoc_change{state = comm_up, assoc_id = Id}}, + #listener{pending = {N,Q}} = S) -> - Arg = {accept, Ref, self(), Sock, Id}, - {ok, TPid} = diameter_sctp_sup:start_child(Arg), - MRef = erlang:monitor(process, TPid), - ets:insert(T, [{MRef, Id}, {Id, TPid}]), - ets:insert(Q, {TPid, now()}), - {TPid, S#listener{pending = {N+1, Q}}}; + {find(Id, S), S#listener{pending = {N+1,Q}}}; %% Known association ... f([{_, TPid}], _, S) -> @@ -755,6 +784,31 @@ f([{_, TPid}], _, S) -> f([], _, _) -> false. +%% find/2 + +%% Transport waiting for an association: use it. +find(Id, #listener{tmap = T, + pending = {N,Q}}) + when N < 0 -> + TPid = ets:first(Q), + [{TPid, MRef}] = ets:lookup(T, TPid), + ets:insert(T, [{MRef, Id}, {Id, TPid}]), + ets:delete(T, TPid), + ets:delete(Q, TPid), + TPid; + +%% No transport start yet: spawn one and queue. +find(Id, #listener{ref = Ref, + socket = Sock, + tmap = T, + pending = {_,Q}}) -> + Arg = {accept, Ref, self(), Sock, Id}, + {ok, TPid} = diameter_sctp_sup:start_child(Arg), + MRef = monitor(process, TPid), + ets:insert(T, [{MRef, Id}, {Id, TPid}]), + ets:insert(Q, {TPid, now()}), + TPid. + %% assoc_id/1 assoc_id({[#sctp_sndrcvinfo{assoc_id = Id}], _}) -> diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl index 4d1b8bec51..0b26f429fb 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-2013. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -37,7 +37,8 @@ code_change/3, terminate/2]). --export([info/1]). %% service_info callback +-export([listener/1,%% diameter_sync callback + info/1]). %% service_info callback -export([ports/0, ports/1]). @@ -191,7 +192,7 @@ init(T) -> i({T, Ref, Mod, Pid, Opts, Addrs}) when T == accept; T == connect -> - erlang:monitor(process, Pid), + monitor(process, Pid), %% Since accept/connect might block indefinitely, spawn a process %% that does nothing but kill us with the parent until call %% returns. @@ -218,8 +219,8 @@ i({T, Ref, Mod, Pid, Opts, Addrs}) %% A monitor process to kill the transport if the parent dies. i(#monitor{parent = Pid, transport = TPid} = S) -> proc_lib:init_ack({ok, self()}), - erlang:monitor(process, Pid), - erlang:monitor(process, TPid), + monitor(process, Pid), + monitor(process, TPid), S; %% In principle a link between the transport and killer processes %% could do the same thing: have the accepting/connecting process be @@ -235,7 +236,7 @@ i({listen, LRef, APid, {Mod, Opts, Addrs}}) -> LAddr = laddr(LAddrOpt, Mod, LSock), true = diameter_reg:add_new({?MODULE, listener, {LRef, {LAddr, LSock}}}), proc_lib:init_ack({ok, self(), {LAddr, LSock}}), - erlang:monitor(process, APid), + monitor(process, APid), start_timer(#listener{socket = LSock}). laddr([], Mod, Sock) -> @@ -336,17 +337,25 @@ accept(Opts) -> %% listener/2 +%% Accepting processes can be started concurrently: ensure only one +%% listener is started. listener(LRef, T) -> - l(diameter_reg:match({?MODULE, listener, {LRef, '_'}}), LRef, T). + diameter_sync:call({?MODULE, listener, LRef}, + {?MODULE, listener, [{LRef, T, self()}]}, + infinity, + infinity). -%% Existing process with the listening socket ... -l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) -> - LPid ! {accept, self()}, +listener({LRef, T, TPid}) -> + l(diameter_reg:match({?MODULE, listener, {LRef, '_'}}), LRef, T, TPid). + +%% Existing listening process ... +l([{{?MODULE, listener, {_, AS}}, LPid}], _, _, TPid) -> + LPid ! {accept, TPid}, AS; -%% ... or not: start one. -l([], LRef, T) -> - {ok, _, AS} = diameter_tcp_sup:start_child({listen, LRef, self(), T}), +%% ... or not. +l([], LRef, T, TPid) -> + {ok, _, AS} = diameter_tcp_sup:start_child({listen, LRef, TPid, T}), AS. %% get_addr/1 @@ -502,7 +511,7 @@ m({'DOWN', _, process, Pid, _}, #monitor{parent = Pid, %% Another accept transport is attaching. l({accept, TPid}, #listener{count = N} = S) -> - erlang:monitor(process, TPid), + monitor(process, TPid), S#listener{count = N+1}; %% Accepting process has died. diff --git a/lib/diameter/src/transport/diameter_transport_sup.erl b/lib/diameter/src/transport/diameter_transport_sup.erl index 6457ab78b0..284a41a752 100644 --- a/lib/diameter/src/transport/diameter_transport_sup.erl +++ b/lib/diameter/src/transport/diameter_transport_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. All Rights Reserved. +%% Copyright Ericsson AB 2010-2015. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -54,7 +54,7 @@ start_child(Name, Module) -> Spec = {Name, {Module, start_link, [Name]}, permanent, - 1000, + infinity, supervisor, [Module]}, supervisor:start_child(?MODULE, Spec). |