diff options
Diffstat (limited to 'lib/diameter/src')
| -rw-r--r-- | lib/diameter/src/base/diameter.erl | 53 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_callback.erl | 6 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_config.erl | 297 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_lib.erl | 2 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_peer.erl | 6 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_peer_fsm.erl | 24 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_service.erl | 89 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_traffic.erl | 1 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_watchdog.erl | 35 | ||||
| -rw-r--r-- | lib/diameter/src/compiler/diameter_exprecs.erl | 4 | ||||
| -rw-r--r-- | lib/diameter/src/transport/diameter_sctp.erl | 96 | ||||
| -rw-r--r-- | lib/diameter/src/transport/diameter_tcp.erl | 96 |
12 files changed, 403 insertions, 306 deletions
diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl index 2e18a1d903..3b41feac0d 100644 --- a/lib/diameter/src/base/diameter.erl +++ b/lib/diameter/src/base/diameter.erl @@ -46,7 +46,8 @@ -export([start/0, stop/0]). --export_type([evaluable/0, +-export_type([eval/0, + evaluable/0, %% deprecated decode_format/0, strict_arities/0, restriction/0, @@ -301,7 +302,7 @@ call(SvcName, App, Message) -> | realm | {host, any|'DiameterIdentity'()} | {realm, any|'DiameterIdentity'()} - | {eval, evaluable()} + | {eval, eval()} | {neg, peer_filter()} | {all, [peer_filter()]} | {any, [peer_filter()]}. @@ -309,10 +310,13 @@ call(SvcName, App, Message) -> -opaque peer_ref() :: pid(). --type evaluable() +-type eval() :: {module(), atom(), list()} | fun() - | maybe_improper_list(evaluable(), list()). + | maybe_improper_list(eval(), list()). + +-type evaluable() + :: eval(). -type sequence() :: {'Unsigned32'(), 0..32}. @@ -322,12 +326,12 @@ call(SvcName, App, Message) -> | node | nodes | [node()] - | evaluable(). + | eval(). -type remotes() :: boolean() | [node()] - | evaluable(). + | eval(). -type message_length() :: 0..16#FFFFFF. @@ -344,22 +348,38 @@ call(SvcName, App, Message) -> | encode | decode. +%% Options common to both start_service/2 and add_transport/2. + +-type common_opt() + :: {pool_size, pos_integer()} + | {capabilities_cb, eval()} + | {capx_timeout, 'Unsigned32'()} + | {strict_capx, boolean()} + | {strict_mbit, boolean()} + | {disconnect_cb, eval()} + | {dpr_timeout, 'Unsigned32'()} + | {dpa_timeout, 'Unsigned32'()} + | {incoming_maxlen, message_length()} + | {length_errors, exit | handle | discard} + | {connect_timer, 'Unsigned32'()} + | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}} + | {watchdog_config, [{okay|suspect, non_neg_integer()}]} + | {spawn_opt, list()}. + %% Options passed to start_service/2 -type service_opt() :: capability() | {application, [application_opt()]} | {restrict_connections, restriction()} - | {sequence, sequence() | evaluable()} + | {sequence, sequence() | eval()} | {share_peers, remotes()} | {decode_format, decode_format()} | {traffic_counters, boolean()} | {string_decode, boolean()} | {strict_arities, true | strict_arities()} - | {strict_mbit, boolean()} - | {incoming_maxlen, message_length()} | {use_shared_peers, remotes()} - | {spawn_opt, list()}. + | common_opt(). -type application_opt() :: {alias, app_alias()} @@ -389,20 +409,9 @@ 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'()} - | {capx_strictness, boolean()} - | {disconnect_cb, evaluable()} - | {dpr_timeout, 'Unsigned32'()} - | {dpa_timeout, 'Unsigned32'()} - | {length_errors, exit | handle | discard} - | {connect_timer, 'Unsigned32'()} - | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}} - | {watchdog_config, [{okay|suspect, non_neg_integer()}]} - | {spawn_opt, list()} + | common_opt() | {private, any()}. %% Predicate passed to remove_transport/2 diff --git a/lib/diameter/src/base/diameter_callback.erl b/lib/diameter/src/base/diameter_callback.erl index f9cdc66c70..d04a416bef 100644 --- a/lib/diameter/src/base/diameter_callback.erl +++ b/lib/diameter/src/base/diameter_callback.erl @@ -26,16 +26,16 @@ %% as the Diameter application callback in question. The record has %% one field for each callback function as well as 'default' and %% 'extra' fields. A function-specific field can be set to a -%% diameter:evaluable() in order to redirect the callback +%% diameter:eval() in order to redirect the callback %% corresponding to that field, or to 'false' to request the default %% callback implemented in this module. If neither of these fields are %% set then the 'default' field determines the form of the callback: a %% module name results in the usual callback as if the module had been -%% configured directly as the callback module, a diameter_evaluable() +%% configured directly as the callback module, a diameter_eval() %% in a callback applied to the atom-valued callback name and argument %% list. For all callbacks not to this module, the 'extra' field is a %% list of additional arguments, following arguments supplied by -%% diameter but preceding those of the diameter:evaluable() being +%% diameter but preceding those of the diameter:eval() being %% applied. %% %% For example, the following config to diameter:start_service/2, in diff --git a/lib/diameter/src/base/diameter_config.erl b/lib/diameter/src/base/diameter_config.erl index f1b6e56782..4721be1ca0 100644 --- a/lib/diameter/src/base/diameter_config.erl +++ b/lib/diameter/src/base/diameter_config.erl @@ -102,9 +102,6 @@ -record(monitor, {mref = make_ref() :: reference(), service}). %% name -%% The default sequence mask. --define(NOMASK, {0,32}). - %% Time to lay low before restarting a dead service. -define(RESTART_SLEEP, 2000). @@ -560,87 +557,184 @@ add(SvcName, Type, Opts0) -> end. transport_opts(Opts) -> - lists:map(fun topt/1, Opts). + [setopt(transport, T) || T <- Opts]. + +%% setopt/2 -topt(T) -> - case opt(T) of +setopt(K, T) -> + case opt(K, T) of {value, X} -> X; true -> T; false -> - ?THROW({invalid, T}) + ?THROW({invalid, T}); + {error, Reason} -> + ?THROW({invalid, T, Reason}) end. -opt({transport_module, M}) -> +%% opt/2 + +opt(_, {incoming_maxlen, N}) -> + is_integer(N) andalso 0 =< N andalso N < 1 bsl 24; + +opt(service, {K, B}) + when K == string_decode; + K == traffic_counters -> + is_boolean(B); + +opt(service, {K, false}) + when K == share_peers; + K == use_shared_peers; + K == monitor; + K == restrict_connections; + K == strict_arities; + K == decode_format -> + true; + +opt(service, {K, true}) + when K == share_peers; + K == use_shared_peers; + K == strict_arities -> + true; + +opt(service, {decode_format, T}) + when T == record; + T == list; + T == map; + T == record_from_map -> + true; + +opt(service, {strict_arities, T}) + when T == encode; + T == decode -> + true; + +opt(service, {restrict_connections, T}) + when T == node; + T == nodes -> + true; + +opt(service, {K, T}) + when (K == share_peers + orelse K == use_shared_peers + orelse K == restrict_connections), ([] == T + orelse is_atom(hd(T))) -> + true; + +opt(service, {monitor, P}) -> + is_pid(P); + +opt(service, {K, F}) + when K == restrict_connections; + K == share_peers; + K == use_shared_peers -> + try diameter_lib:eval(F) of %% but no guarantee that it won't fail later + Nodes -> + is_list(Nodes) orelse {error, Nodes} + catch + E:R -> + {error, {E, R, ?STACK}} + end; + +opt(service, {sequence, {H,N}}) -> + 0 =< N andalso N =< 32 + andalso is_integer(H) + andalso 0 =< H + andalso 0 == H bsr (32-N); + +opt(service = S, {sequence = K, F}) -> + try diameter_lib:eval(F) of + {_,_} = T -> + KT = {K,T}, + opt(S, KT) andalso {value, KT}; + V -> + {error, V} + catch + E:R -> + {error, {E, R, ?STACK}} + end; + +opt(transport, {transport_module, M}) -> is_atom(M); -opt({transport_config, _, Tmo}) -> +opt(transport, {transport_config, _, Tmo}) -> ?IS_UINT32(Tmo) orelse Tmo == infinity; -opt({applications, As}) -> +opt(transport, {applications, As}) -> is_list(As); -opt({capabilities, Os}) -> - is_list(Os) andalso ok == encode_CER(Os); +opt(transport, {capabilities, Os}) -> + is_list(Os) andalso case encode_CER(Os) of + ok -> true; + No -> {error, No} + end; -opt({K, Tmo}) +opt(_, {K, Tmo}) when K == capx_timeout; K == dpr_timeout; K == dpa_timeout -> ?IS_UINT32(Tmo); -opt({capx_strictness, B}) -> +opt(_, {capx_strictness, B}) -> + is_boolean(B) andalso {value, {strict_capx, B}}; +opt(_, {K, B}) + when K == strict_capx; + K == strict_mbit -> is_boolean(B); -opt({length_errors, T}) -> +opt(_, {length_errors, T}) -> lists:member(T, [exit, handle, discard]); -opt({K, Tmo}) - when K == reconnect_timer; %% deprecated - K == connect_timer -> +opt(transport, {reconnect_timer, Tmo}) -> %% deprecated + ?IS_UINT32(Tmo) andalso {value, {connect_timer, Tmo}}; +opt(_, {connect_timer, Tmo}) -> ?IS_UINT32(Tmo); -opt({watchdog_timer, {M,F,A}}) +opt(_, {watchdog_timer, {M,F,A}}) when is_atom(M), is_atom(F), is_list(A) -> true; -opt({watchdog_timer, Tmo}) -> +opt(_, {watchdog_timer, Tmo}) -> ?IS_UINT32(Tmo); -opt({watchdog_config, L}) -> - is_list(L) andalso lists:all(fun wdopt/1, L); +opt(_, {watchdog_config, L}) -> + is_list(L) andalso lists:all(fun wd/1, L); -opt({spawn_opt, {M,F,A}}) +opt(_, {spawn_opt, {M,F,A}}) when is_atom(M), is_atom(F), is_list(A) -> true; -opt({spawn_opt = K, Opts}) -> +opt(_, {spawn_opt = K, Opts}) -> if is_list(Opts) -> {value, {K, spawn_opts(Opts)}}; true -> false end; -opt({pool_size, N}) -> +opt(_, {pool_size, N}) -> is_integer(N) andalso 0 < N; -%% Options that we can't validate. -opt({K, _}) +%% Options we can't validate. +opt(_, {K, _}) + when K == disconnect_cb; + K == capabilities_cb -> + true; +opt(transport, {K, _}) when K == transport_config; - K == capabilities_cb; - K == disconnect_cb; K == private -> true; -%% Anything else, which is ignored by us. This makes options sensitive -%% to spelling mistakes but arbitrary options are passed by some users -%% as a way to identify transports. (That is, can't just do away with -%% it.) -opt(_) -> - true. +%% Anything else, which is ignored in transport config. This makes +%% options sensitive to spelling mistakes, but arbitrary options are +%% passed by some users as a way to identify transports so can't just +%% do away with it. +opt(K, _) -> + K == transport. + +%% wd/1 -wdopt({K,N}) -> +wd({K,N}) -> (K == okay orelse K == suspect) andalso is_integer(N) andalso 0 =< N; -wdopt(_) -> +wd(_) -> false. %% start_transport/2 @@ -705,19 +799,7 @@ make_config(SvcName, Opts) -> ok = encode_CER(CapOpts), - SvcOpts = make_opts((Opts -- AppOpts) -- CapOpts, - [{false, share_peers}, - {false, use_shared_peers}, - {false, monitor}, - {?NOMASK, sequence}, - {nodes, restrict_connections}, - {16#FFFFFF, incoming_maxlen}, - {true, strict_arities}, - {true, strict_mbit}, - {record, decode_format}, - {true, traffic_counters}, - {true, string_decode}, - {[], spawn_opt}]), + SvcOpts = service_opts((Opts -- AppOpts) -- CapOpts), D = proplists:get_value(string_decode, SvcOpts, true), @@ -731,115 +813,22 @@ binary_caps(Caps, true) -> 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, +%% service_opts/1 - [] == Unknown orelse ?THROW({invalid, hd(Unknown)}), - - [{K, opt(K,V)} || {K,V} <- Known]. - -opt(incoming_maxlen, N) - when 0 =< N, N < 1 bsl 24 -> - N; - -opt(spawn_opt, {M,F,A} = T) - when is_atom(M), is_atom(F), is_list(A) -> - T; - -opt(spawn_opt, L) - when is_list(L) -> - spawn_opts(L); - -opt(K, false = B) - when K == share_peers; - K == use_shared_peers; - K == monitor; - K == restrict_connections; - K == strict_arities; - K == strict_mbit; - K == decode_format; - K == traffic_counters; - K == string_decode -> - B; - -opt(K, true = B) - when K == share_peers; - K == use_shared_peers; - K == strict_arities; - K == strict_mbit; - K == traffic_counters; - K == string_decode -> - B; - -opt(decode_format, T) - when T == record; - T == list; - T == map; - T == record_from_map -> - T; - -opt(strict_arities, T) - when T == encode; - T == decode -> - T; - -opt(restrict_connections, T) - when T == node; - T == nodes -> - T; - -opt(K, T) - when (K == share_peers - orelse K == use_shared_peers - orelse K == restrict_connections), ([] == T - orelse is_atom(hd(T))) -> - T; - -opt(monitor, P) - when is_pid(P) -> - P; - -opt(K, F) - when K == restrict_connections; - K == share_peers; - K == use_shared_peers -> - try diameter_lib:eval(F) of %% but no guarantee that it won't fail later - Nodes when is_list(Nodes) -> - F; - V -> - ?THROW({value, {K,V}}) - catch - E:R -> - ?THROW({value, {K, E, R, ?STACK}}) - end; - -opt(sequence, {_,_} = T) -> - sequence(T); - -opt(sequence = K, F) -> - try diameter_lib:eval(F) of - T -> sequence(T) - catch - E:R -> - ?THROW({value, {K, E, R, ?STACK}}) - end; - -opt(K, _) -> - ?THROW({value, K}). +service_opts(Opts) -> + Res = [setopt(service, T) || T <- Opts], + Keys = sets:to_list(sets:from_list([K || {K,_} <- Res])), %% unique + Dups = lists:foldl(fun(K,A) -> lists:keydelete(K, 1, A) end, Res, Keys), + [] == Dups orelse ?THROW({duplicate, Dups}), + Res. +%% Reject duplicates on a service, but not on a transport. There's no +%% particular reason for the inconsistency, but the historic behaviour +%% ignores all but the first of a transport_opt(), and there's no real +%% reason to change it. spawn_opts(L) -> [T || T <- L, T /= link, T /= monitor]. -sequence({H,N} = T) - when 0 =< N, N =< 32, 0 =< H, 0 == H bsr (32-N) -> - T; - -sequence(_) -> - ?THROW({value, sequence}). - make_caps(Caps, Opts) -> case diameter_capx:make_caps(Caps, Opts) of {ok, T} -> diff --git a/lib/diameter/src/base/diameter_lib.erl b/lib/diameter/src/base/diameter_lib.erl index 8792e97621..1c1ea42cb5 100644 --- a/lib/diameter/src/base/diameter_lib.erl +++ b/lib/diameter/src/base/diameter_lib.erl @@ -283,7 +283,7 @@ ip(T) %% Or not: convert from '.'/':'-separated decimal/hex. ip(Addr) -> - {ok, A} = inet_parse:address(Addr), %% documented in inet(3) + {ok, A} = inet:parse_address(Addr), A. %% --------------------------------------------------------------------------- diff --git a/lib/diameter/src/base/diameter_peer.erl b/lib/diameter/src/base/diameter_peer.erl index 2759f17e64..4cb5a57a54 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-2015. All Rights Reserved. +%% Copyright Ericsson AB 2010-2017. 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. @@ -202,10 +202,10 @@ match1(Addr, Match) -> match(Addr, {ok, A}, _) -> Addr == A; match(Addr, {error, _}, RE) -> - match == re:run(inet_parse:ntoa(Addr), RE, [{capture, none}]). + match == re:run(inet:ntoa(Addr), RE, [{capture, none}, caseless]). addr([_|_] = A) -> - inet_parse:address(A); + inet:parse_address(A); addr(A) -> {ok, A}. diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index 6c47d8804e..d99f11a697 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -238,7 +238,7 @@ i({Ack, WPid, {M, Ref} = T, Opts, {SvcOpts, Nodes, Dict0, Svc}}) -> proplists:get_value(dpa_timeout, Opts, ?DPA_TIMEOUT)}), Tmo = proplists:get_value(capx_timeout, Opts, ?CAPX_TIMEOUT), - Strictness = proplists:get_value(capx_strictness, Opts, true), + Strict = proplists:get_value(strict_capx, Opts, true), LengthErr = proplists:get_value(length_errors, Opts, exit), {TPid, Addrs} = start_transport(T, Rest, Svc), @@ -252,7 +252,7 @@ i({Ack, WPid, {M, Ref} = T, Opts, {SvcOpts, Nodes, Dict0, Svc}}) -> mode = M, service = svc(Svc, Addrs), length_errors = LengthErr, - strict = Strictness, + strict = Strict, incoming_maxlen = Maxlen, codec = maps:with([decode_format, string_decode, @@ -544,11 +544,11 @@ put_route(Pid) -> MRef = monitor(process, Pid), put(Pid, MRef). -%% get_route/2 +%% get_route/3 -%% incoming answer -get_route(_, #diameter_packet{header = #diameter_header{is_request = false}} - = Pkt) -> +%% Incoming answer. +get_route(_, _, #diameter_packet{header = #diameter_header{is_request = false}} + = Pkt) -> Seqs = diameter_codec:sequence_numbers(Pkt), case erase(Seqs) of {Pid, Ref, MRef} -> @@ -559,8 +559,14 @@ get_route(_, #diameter_packet{header = #diameter_header{is_request = false}} false end; -%% incoming request -get_route(Ack, _) -> +%% Requests answered here ... +get_route(_, N, _) + when N == 'CER'; + N == 'DPR' -> + false; + +%% ... or not. +get_route(Ack, _, _) -> Ack. %% erase_route/1 @@ -745,7 +751,7 @@ recv1('DPA' = Name, %% Any other message with a header and no length errors. recv1(Name, H, Msg, #state{parent = Pid, ack = Ack} = S) -> Pkt = pkt(H, Msg), - Pid ! {recv, self(), get_route(Ack, Pkt), Name, Pkt}, + Pid ! {recv, self(), get_route(Ack, Name, Pkt), Name, Pkt}, handle(Name, Pkt, S). %% pkt/2 diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index 07f1fd3a4a..802e6e7243 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -117,6 +117,18 @@ decode_format := diameter:decode_format(), traffic_counters := boolean(), string_decode := boolean(), + capabilities_cb => diameter:evaluable(), + pool_size => pos_integer(), + capx_timeout => diameter:'Unsigned32'(), + strict_capx => boolean(), + disconnect_cb => diameter:evaluable(), + dpr_timeout => diameter:'Unsigned32'(), + dpa_timeout => diameter:'Unsigned32'(), + length_errors => exit | handle | discard, + connect_timer => diameter:'Unsigned32'(), + watchdog_timer => diameter:'Unsigned32'() + | {module(), atom(), list()}, + watchdog_config => [{okay|suspect, non_neg_integer()}], spawn_opt := list() | {module(), atom(), list()}}}). %% Record representing an RFC 3539 watchdog process implemented by @@ -517,6 +529,13 @@ transition({tc_timeout, T}, S) -> tc_timeout(T, S), ok; +transition({nodeup, Node, _}, S) -> + nodeup(Node, S), + ok; + +transition({nodedown, _Node, _}, _) -> + ok; + transition(Req, S) -> unexpected(handle_info, [Req], S), ok. @@ -682,12 +701,15 @@ i(SvcName) -> cfg_acc({SvcName, #diameter_service{applications = Apps} = Rec, Opts}, {false, Acc}) -> lists:foreach(fun init_mod/1, Apps), + #{monitor := M} + = SvcOpts + = service_opts(Opts), 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(lists:keydelete(monitor, 1, Opts))}, + monitor = mref(M), + options = maps:remove(monitor, SvcOpts)}, {S, Acc}; cfg_acc({_Ref, Type, _Opts} = T, {S, Acc}) @@ -702,8 +724,23 @@ init_peers() -> %% Alias, %% TPid} -service_options(Opts) -> - maps:from_list(lists:delete({strict_arities, true}, Opts)). +service_opts(Opts) -> + T = {strict_arities, true}, + maps:merge(maps:from_list([{monitor, false} | def_opts() -- [T]]), + maps:from_list(Opts -- [T])). + +def_opts() -> %% defaults on the service map + [{share_peers, false}, + {use_shared_peers, false}, + {sequence, {0,32}}, + {restrict_connections, nodes}, + {incoming_maxlen, 16#FFFFFF}, + {strict_arities, true}, + {strict_mbit, true}, + {decode_format, record}, + {traffic_counters, true}, + {string_decode, true}, + {spawn_opt, []}]. mref(false = No) -> No; @@ -712,6 +749,8 @@ mref(P) -> init_shared(#state{options = #{use_shared_peers := T}, service_name = Svc}) -> + T == false orelse net_kernel:monitor_nodes(true, [{node_type, visible}, + nodedown_reason]), notify(T, Svc, {service, self()}). init_mod(#diameter_app{alias = Alias, @@ -721,16 +760,17 @@ init_mod(#diameter_app{alias = Alias, start_fsm({Ref, Type, Opts}, S) -> start(Ref, {Type, Opts}, S). -get_value(Key, Vs) -> - {_, V} = lists:keyfind(Key, 1, Vs), - V. - notify(Share, SvcName, T) -> Nodes = remotes(Share), [] /= Nodes andalso diameter_peer:notify(Nodes, SvcName, T). %% Test for the empty list for upgrade reasons: there's no %% diameter_peer:notify/3 in old code. +nodeup(Node, #state{options = #{share_peers := SP}, + service_name = SvcName}) -> + lists:member(Node, remotes(SP)) + andalso diameter_peer:notify([Node], SvcName, {service, self()}). + remotes(false) -> []; @@ -809,7 +849,7 @@ start(Ref, Type, Opts, State) -> start(Ref, Type, Opts, N, #state{watchdogT = WatchdogT, local = {PeerT, _, _}, options = #{string_decode := SD} - = SvcOpts0, + = SvcOpts, service_name = SvcName, service = Svc0}) when Type == connect; @@ -818,12 +858,12 @@ start(Ref, Type, Opts, N, #state{watchdogT = WatchdogT, = Svc1 = merge_service(Opts, Svc0), Svc = binary_caps(Svc1, SD), - SvcOpts = merge_options(Opts, SvcOpts0), - RecvData = diameter_traffic:make_recvdata([SvcName, PeerT, Apps, SvcOpts]), - T = {Opts, SvcOpts, RecvData, Svc}, + {SOpts, TOpts} = merge_opts(SvcOpts, Opts), + RecvData = diameter_traffic:make_recvdata([SvcName, PeerT, Apps, SOpts]), + T = {TOpts, SOpts, RecvData, Svc}, Rec = #watchdog{type = Type, ref = Ref, - options = Opts}, + options = TOpts}, diameter_lib:fold_n(fun(_,A) -> [wd(Type, Ref, T, WatchdogT, Rec) | A] @@ -831,10 +871,14 @@ start(Ref, Type, Opts, N, #state{watchdogT = WatchdogT, [], N). -merge_options(Opts, SvcOpts) -> - Keys = maps:keys(SvcOpts), - Map = maps:from_list([KV || {K,_} = KV <- Opts, lists:member(K, Keys)]), - maps:merge(SvcOpts, Map). +merge_opts(SvcOpts, Opts) -> + Keys = [K || {K,_} <- def_opts()], + SO = [T || {K,_} = T <- Opts, lists:member(K, Keys)], + TO = Opts -- SO, + {maps:merge(maps:with(Keys, SvcOpts), maps:from_list(SO)), + TO ++ [T || {K,_} = T <- maps:to_list(SvcOpts), + not lists:member(K, Keys), + not lists:keymember(K, 1, Opts)]}. binary_caps(Svc, true) -> Svc; @@ -1403,9 +1447,15 @@ is_remote(Pid, T) -> %% # remote_peer_up/4 %% --------------------------------------------------------------------------- -remote_peer_up(TPid, Aliases, Caps, #state{options = #{use_shared_peers := T}} +remote_peer_up(TPid, Aliases, Caps, #state{options = #{use_shared_peers := T}, + remote = {PeerT, _, _}} = S) -> - is_remote(TPid, T) andalso rpu(TPid, Aliases, Caps, S). + is_remote(TPid, T) + andalso not ets:member(PeerT, TPid) + andalso rpu(TPid, Aliases, Caps, S). + +%% Notification can be duplicate since remote nodes push and the local +%% node pulls. rpu(TPid, Aliases, Caps, #state{service = Svc, remote = RT}) -> #diameter_service{applications = Apps} = Svc, @@ -1415,6 +1465,7 @@ rpu(TPid, Aliases, Caps, #state{service = Svc, remote = RT}) -> rpu(_, [] = No, _, _) -> No; + rpu(TPid, Aliases, Caps, {PeerT, _, _} = RT) -> monitor(process, TPid), ets:insert(PeerT, #peer{pid = TPid, diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl index 4388175c3a..9a1c28c42d 100644 --- a/lib/diameter/src/base/diameter_traffic.erl +++ b/lib/diameter/src/base/diameter_traffic.erl @@ -92,6 +92,7 @@ caller :: pid() | undefined, %% calling process handler :: pid(), %% request process peer :: undefined | {pid(), #diameter_caps{}}, + caps :: undefined, %% no longer used packet :: #diameter_packet{} | undefined}). %% of request %% --------------------------------------------------------------------------- diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl index b2172356ee..bb671e9860 100644 --- a/lib/diameter/src/base/diameter_watchdog.erl +++ b/lib/diameter/src/base/diameter_watchdog.erl @@ -76,10 +76,8 @@ string_decode := false, strict_arities => diameter:strict_arities(), strict_mbit := boolean(), - failed_avp := false, rfc := 3588 | 6733, - ordered_encode := false, - incoming_maxlen := diameter:message_length()}, + ordered_encode := false}, shutdown = false :: boolean()}). %% --------------------------------------------------------------------------- @@ -137,15 +135,6 @@ i({Ack, T, Pid, {Opts, putr(restart, {T, Opts, Svc, SvcOpts}), %% save seeing it in trace putr(dwr, dwr(Caps)), %% Nodes = restrict_nodes(Restrict), - CodecKeys = [decode_format, - string_decode, - strict_arities, - strict_mbit, - incoming_maxlen, - spawn_opt, - rfc, - ordered_encode], - #watchdog{parent = Pid, transport = start(T, Opts, SvcOpts, Nodes, Dict0, Svc), tw = proplists:get_value(watchdog_timer, @@ -153,13 +142,21 @@ i({Ack, T, Pid, {Opts, ?DEFAULT_TW_INIT), receive_data = RecvData, dictionary = Dict0, - config = - maps:without([traffic_counters | CodecKeys], - config(SvcOpts#{restrict => restrict(Nodes), - suspect => 1, - okay => 3}, - Opts)), - codec = maps:with(CodecKeys -- [strict_arities], + config = maps:with([sequence, + restrict_connections, + restrict, + suspect, + okay], + config(SvcOpts#{restrict => restrict(Nodes), + suspect => 1, + okay => 3}, + Opts)), + codec = maps:with([decode_format, + strict_arities, + strict_mbit, + string_decode, + rfc, + ordered_encode], SvcOpts#{decode_format := false, string_decode := false, ordered_encode => false})}. diff --git a/lib/diameter/src/compiler/diameter_exprecs.erl b/lib/diameter/src/compiler/diameter_exprecs.erl index 9a0cb6baf2..143dede037 100644 --- a/lib/diameter/src/compiler/diameter_exprecs.erl +++ b/lib/diameter/src/compiler/diameter_exprecs.erl @@ -110,9 +110,9 @@ %% parse_transform/2 parse_transform(Forms, _Options) -> - Rs = [R || {attribute, _, record, R} <- Forms], - Es = lists:append([E || {attribute, _, export_records, E} <- Forms]), {H,T} = lists:splitwith(fun is_head/1, Forms), + Rs = [R || {attribute, _, record, R} <- H], + Es = lists:append([E || {attribute, _, export_records, E} <- H]), H ++ [a_export(Es) | f_accessors(Es, Rs)] ++ T. is_head(T) -> diff --git a/lib/diameter/src/transport/diameter_sctp.erl b/lib/diameter/src/transport/diameter_sctp.erl index 6a9f1f940b..4eb3379d59 100644 --- a/lib/diameter/src/transport/diameter_sctp.erl +++ b/lib/diameter/src/transport/diameter_sctp.erl @@ -79,7 +79,7 @@ -type option() :: {sender, boolean()} | sender | {packet, boolean() | raw} - | {message_cb, false | diameter:evaluable()}. + | {message_cb, false | diameter:eval()}. -type uint() :: non_neg_integer(). @@ -102,9 +102,10 @@ streams :: {uint(), uint()} %% {InStream, OutStream} counts | undefined, os = 0 :: uint(), %% next output stream + rotate = 1 :: boolean() | 0 | 1, %% rotate os? packet = true :: boolean() %% legacy transport_data? | raw, - message_cb = false :: false | diameter:evaluable(), + message_cb = false :: false | diameter:eval(), send = false :: pid() | boolean()}). %% sending process %% Monitor process state. @@ -112,7 +113,7 @@ {transport :: pid(), ack = false :: boolean(), socket :: gen_sctp:sctp_socket(), - assoc_id :: gen_sctp:assoc_id()}). %% next output stream + assoc_id :: gen_sctp:assoc_id()}). %% Listener process state. -record(listener, @@ -120,7 +121,7 @@ socket :: gen_sctp:sctp_socket(), service :: pid(), %% service process pending = {0, queue:new()}, - opts :: [[match()] | boolean() | diameter:evaluable()]}). + opts :: [[match()] | boolean() | diameter:eval()]}). %% Field pending implements two queues: the first of transport-to-be %% processes to which an association has been assigned but for which %% diameter hasn't yet spawned a transport process, a short-lived @@ -156,12 +157,7 @@ start(T, Svc, Opts) = Svc, diameter_sctp_sup:start(), %% start supervisors on demand Addrs = Caps#diameter_caps.host_ip_address, - s(T, Addrs, Pid, lists:map(fun ip/1, Opts)). - -ip({ifaddr, A}) -> - {ip, A}; -ip(T) -> - T. + s(T, Addrs, Pid, Opts). %% A listener spawns transports either as a consequence of this call %% when there is not yet an association to assign it, or at comm_up on @@ -354,23 +350,35 @@ l([], Ref, T) -> %% open/3 open(Addrs, Opts, PortNr) -> - {LAs, Os} = addrs(Addrs, Opts), - {LAs, case gen_sctp:open(gen_opts(portnr(Os, PortNr))) of - {ok, Sock} -> - Sock; - {error, Reason} -> - x({open, Reason}) - end}. + case gen_sctp:open(gen_opts(portnr(addrs(Addrs, Opts), PortNr))) of + {ok, Sock} -> + {addrs(Sock), Sock}; + {error, Reason} -> + x({open, Reason}) + end. addrs(Addrs, Opts) -> - case proplists:split(Opts, [ip]) of - {[[]], _} -> - {Addrs, Opts ++ [{ip, A} || A <- Addrs]}; - {[As], Os} -> - LAs = [diameter_lib:ipaddr(A) || {ip, A} <- As], - {LAs, Os ++ [{ip, A} || A <- LAs]} + case lists:mapfoldl(fun ipaddr/2, false, Opts) of + {Os, true} -> + Os; + {_, false} -> + Opts ++ [{ip, A} || A <- Addrs] end. +ipaddr({K,A}, _) + when K == ifaddr; + K == ip -> + {{ip, ipaddr(A)}, true}; +ipaddr(T, B) -> + {T, B}. + +ipaddr(A) + when A == loopback; + A == any -> + A; +ipaddr(A) -> + diameter_lib:ipaddr(A). + portnr(Opts, PortNr) -> case proplists:get_value(port, Opts) of undefined -> @@ -379,6 +387,14 @@ portnr(Opts, PortNr) -> Opts end. +addrs(Sock) -> + case inet:socknames(Sock) of + {ok, As} -> + [A || {A,_} <- As]; + {error, Reason} -> + x({socknames, Reason}) + end. + %% x/1 x(Reason) -> @@ -565,7 +581,7 @@ transition(Msg, S) %% Deferred actions from a message_cb. transition({actions, Dir, Acts}, S) -> - actions(Acts, Dir, S); + setopts(ok, actions(Acts, Dir, S)); %% Request to close the transport connection. transition({diameter, {close, Pid}}, #transport{parent = Pid}) -> @@ -677,11 +693,16 @@ send(#diameter_packet{transport_data = {outstream, SId}} = S) -> send(SId rem OS, Msg, S); -%% ... or not: rotate through all streams. -send(Msg, #transport{streams = {_, OS}, +%% ... or not: rotate when sending on multiple streams ... +send(Msg, #transport{rotate = true, + streams = {_, OS}, os = N} = S) -> - send(N, Msg, S#transport{os = (N + 1) rem OS}). + send(N, Msg, S#transport{os = (N + 1) rem OS}); + +%% ... or send on the only stream available. +send(Msg, S) -> + send(0, Msg, S). %% send/3 @@ -749,7 +770,7 @@ recv({[#sctp_sndrcvinfo{assoc_id = Id}], _Bin} %% Inbound Diameter message. recv({[#sctp_sndrcvinfo{}], Bin} = Msg, S) when is_binary(Bin) -> - message(recv, Msg, S); + message(recv, Msg, recv(S)); recv({_, #sctp_shutdown_event{}}, _) -> stop; @@ -769,6 +790,25 @@ recv({_, #sctp_paddr_change{}}, _) -> recv({_, #sctp_pdapi_event{}}, _) -> ok. +%% recv/1 +%% +%% Start sending unordered after the second reception, so that an +%% outgoing CER/CEA will arrive at the peer before another request. + +recv(#transport{rotate = B} = S) + when is_boolean(B) -> + S; + +recv(#transport{rotate = 0, streams = {_,N}, socket = Sock} = S) -> + ok = inet:setopts(Sock, [{sctp_default_send_param, + #sctp_sndrcvinfo{flags = [unordered]}}]), + S#transport{rotate = 1 < N}; + +recv(#transport{rotate = N} = S) -> + S#transport{rotate = N-1}. + +%% publish/4 + publish(T, Ref, Id, Sock) -> true = diameter_reg:add_new({?MODULE, T, {Ref, {Id, Sock}}}), putr(?INFO_KEY, {gen_sctp, Sock}). %% for info/1 diff --git a/lib/diameter/src/transport/diameter_tcp.erl b/lib/diameter/src/transport/diameter_tcp.erl index 6252dbddfb..a8639baa11 100644 --- a/lib/diameter/src/transport/diameter_tcp.erl +++ b/lib/diameter/src/transport/diameter_tcp.erl @@ -110,7 +110,7 @@ -type option() :: {port, non_neg_integer()} | {sender, boolean()} | sender - | {message_cb, false | diameter:evaluable()} + | {message_cb, false | diameter:eval()} | {fragment_timer, 0..16#FFFFFFFF}. %% Accepting/connecting transport process state. @@ -125,7 +125,7 @@ timeout :: infinity | 0..16#FFFFFFFF, %% fragment timeout tref = false :: false | reference(), %% fragment timer reference flush = false :: boolean(), %% flush fragment at timeout? - message_cb :: false | diameter:evaluable(), + message_cb :: false | diameter:eval(), send :: pid() | false}). %% sending process %% The usual transport using gen_tcp can be replaced by anything @@ -142,8 +142,7 @@ -> {ok, pid(), [inet:ip_address()]} when Ref :: diameter:transport_ref(); ({connect, Ref}, #diameter_service{}, [connect_option()]) - -> {ok, pid(), [inet:ip_address()]} - | {ok, pid()} + -> {ok, pid()} when Ref :: diameter:transport_ref(). start({T, Ref}, Svc, Opts) -> @@ -258,22 +257,14 @@ i(#monitor{parent = Pid, transport = TPid} = S) -> 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), + {[LP], Rest} = proplists:split(Opts, [port]), + {ok, LSock} = Mod:listen(get_port(LP), gen_opts(Addrs, Rest)), + {ok, {LAddr, _}} = sockname(Mod, LSock), true = diameter_reg:add_new({?MODULE, listener, {Ref, {LAddr, LSock}}}), proc_lib:init_ack({ok, self(), {LAddr, LSock}}), #listener{socket = LSock, module = Mod}. -laddr([], Mod, Sock) -> - {ok, {Addr, _Port}} = sockname(Mod, Sock), - Addr; -laddr([{ip, Addr}], _, _) -> - Addr. - ssl_opts([]) -> false; ssl_opts([{ssl_options, true}]) -> @@ -308,24 +299,16 @@ init(accept = T, Ref, Mod, Pid, Opts, Addrs, SvcPid) -> Sock; init(connect = T, Ref, Mod, Pid, Opts, Addrs, _SvcPid) -> - {[LA, RA, RP], Rest} = proplists:split(Opts, [ip, raddr, rport]), - LAddrOpt = get_addr(LA, Addrs), + {[RA, RP], Rest} = proplists:split(Opts, [raddr, rport]), RAddr = get_addr(RA), RPort = get_port(RP), - proc_lib:init_ack(init_rc(LAddrOpt)), - Sock = ok(connect(Mod, RAddr, RPort, gen_opts(LAddrOpt, Rest))), + proc_lib:init_ack({ok, self()}), + Sock = ok(connect(Mod, RAddr, RPort, gen_opts(Addrs, Rest))), publish(Mod, T, Ref, Sock), - up(Pid, {RAddr, RPort}, LAddrOpt, Mod, Sock), + up(Pid, {RAddr, RPort}, Mod, Sock), Sock. -init_rc([{ip, Addr}]) -> - {ok, self(), [Addr]}; -init_rc([]) -> - {ok, self()}. - -up(Pid, Remote, [{ip, _Addr}], _, _) -> - diameter_peer:up(Pid, Remote); -up(Pid, Remote, [], Mod, Sock) -> +up(Pid, Remote, Mod, Sock) -> {Addr, _Port} = ok(sockname(Mod, Sock)), diameter_peer:up(Pid, Remote, [Addr]). @@ -382,25 +365,41 @@ l([{{?MODULE, listener, {_, AS}}, LPid}], _, _) -> l([], Ref, T) -> diameter_tcp_sup:start_child({listen, Ref, T}). -%% get_addr/1 +%% addrs/2 +%% +%% Take the first address from the service if several are specified +%% and not address is configured. + +addrs(Addrs, Opts) -> + case lists:mapfoldr(fun ipaddr/2, [], Opts) of + {Os, [_]} -> + Os; + {_, []} -> + Opts ++ [{ip, A} || [A|_] <- [Addrs]]; + {_, As} -> + ?ERROR({invalid_addrs, As, Addrs}) + end. -get_addr(As) -> - diameter_lib:ipaddr(addr(As, [])). +ipaddr({K,A}, As) + when K == ifaddr; + K == ip -> + {{ip, ipaddr(A)}, [A | As]}; +ipaddr(T, B) -> + {T, B}. -%% get_addr/2 +ipaddr(A) + when A == loopback; + A == any -> + A; +ipaddr(A) -> + diameter_lib:ipaddr(A). -get_addr([], []) -> - []; -get_addr(As, Def) -> - [{ip, diameter_lib:ipaddr(addr(As, Def))}]. +%% get_addr/1 -%% Take the first address from the service if several are unspecified. -addr([], [Addr | _]) -> - Addr; -addr([{_, Addr}], _) -> - Addr; -addr(As, Addrs) -> - ?ERROR({invalid_addrs, As, Addrs}). +get_addr([{_, Addr}]) -> + diameter_lib:ipaddr(Addr); +get_addr(Addrs) -> + ?ERROR({invalid_addrs, Addrs}). %% get_port/1 @@ -413,10 +412,15 @@ get_port(Ps) -> %% gen_opts/2 -gen_opts(LAddrOpt, Opts) -> +gen_opts(Addrs, Opts) -> + gen_opts(addrs(Addrs, Opts)). + +%% gen_opts/1 + +gen_opts(Opts) -> {L,_} = proplists:split(Opts, [binary, packet, active]), [[],[],[]] == L orelse ?ERROR({reserved_options, Opts}), - [binary, {packet, 0}, {active, false}] ++ LAddrOpt ++ Opts. + [binary, {packet, 0}, {active, false} | Opts]. %% --------------------------------------------------------------------------- %% # ports/1 @@ -640,7 +644,7 @@ transition(Msg, S) %% Deferred actions from a message_cb. transition({actions, Dir, Acts}, S) -> - actions(Acts, Dir, S); + setopts(actions(Acts, Dir, S)); %% Request to close the transport connection. transition({diameter, {close, Pid}}, #transport{parent = Pid, |