From 380a3f0aece61c649efaed45041c8679038891f1 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Fri, 12 Oct 2012 11:19:49 +0200 Subject: Implement transport_opt() disconnect_cb --- lib/diameter/src/base/diameter.erl | 1 + lib/diameter/src/base/diameter_peer_fsm.erl | 187 +++++++++++++++++++++------- lib/diameter/src/base/diameter_service.erl | 64 +++++----- 3 files changed, 179 insertions(+), 73 deletions(-) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl index 3e3a6be0ef..95702f03d4 100644 --- a/lib/diameter/src/base/diameter.erl +++ b/lib/diameter/src/base/diameter.erl @@ -330,6 +330,7 @@ call(SvcName, App, Message) -> | {applications, [app_alias()]} | {capabilities, [capability()]} | {capabilities_cb, evaluable()} + | {disconnect_cb, evaluable()} | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}} | {reconnect_timer, 'Unsigned32'()} | {private, any()}. diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index 3f4945f7a6..ecdb09d34a 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -48,15 +48,19 @@ -include("diameter_internal.hrl"). -include("diameter_gen_base_rfc3588.hrl"). +%% Values of Disconnect-Cause in DPR. -define(GOAWAY, ?'DIAMETER_BASE_DISCONNECT-CAUSE_DO_NOT_WANT_TO_TALK_TO_YOU'). -define(REBOOT, ?'DIAMETER_BASE_DISCONNECT-CAUSE_REBOOTING'). +-define(BUSY, ?'DIAMETER_BASE_DISCONNECT-CAUSE_BUSY'). -define(NO_INBAND_SECURITY, 0). -define(TLS, 1). %% Keys in process dictionary. -define(CB_KEY, cb). %% capabilities callback +-define(DPR_KEY, dpr). %% disconnect callback -define(DWA_KEY, dwa). %% outgoing DWA +-define(REF_KEY, ref). %% transport_ref() -define(Q_KEY, q). %% transport start queue -define(START_KEY, start). %% start of connected transport -define(SEQUENCE_KEY, mask). %% mask for sequence numbers @@ -68,6 +72,13 @@ %% A 2xxx series Result-Code. Not necessarily 2001. -define(IS_SUCCESS(N), 2 == (N) div 1000). +%% Guards. +-define(IS_UINT32(N), (0 =< N andalso 0 == N bsr 32)). +-define(IS_TIMEOUT(N), ?IS_UINT32(N)). +-define(IS_CAUSE(N), N == ?REBOOT; N == rebooting; + N == ?GOAWAY; N == goaway; + N == ?BUSY; N == busy). + %% RFC 3588: %% %% Timeout An application-defined timer has expired while waiting @@ -75,18 +86,16 @@ %% -define(EVENT_TIMEOUT, 10000). -%% How long to wait for a DPA in response to DPR before simply -%% aborting. Used to distinguish between shutdown and not but there's -%% not really any need. Stopping a service will require a timeout if -%% the peer doesn't answer DPR so the value should be short-ish. +%% Default timeout for DPA in response to DPR. A bit short but the +%% timeout used to be hardcoded. (So it could be worse.) -define(DPA_TIMEOUT, 1000). -record(state, {state = 'Wait-Conn-Ack' %% state of RFC 3588 Peer State Machine :: 'Wait-Conn-Ack' | recv_CER | 'Wait-CEA' | 'Open', mode :: accept | connect | {connect, reference()}, - parent :: pid(), - transport :: pid(), + parent :: pid(), %% watchdog process + transport :: pid(), %% transport process service :: #diameter_service{}, dpr = false :: false | {diameter:'Unsigned32'(), diameter:'Unsigned32'()}}). @@ -163,14 +172,16 @@ i({WPid, Type, Opts, #diameter_service{} = Svc}) -> %% from old code i({WPid, Type, Opts, {?NOMASK, [node() | nodes()], Svc}}); i({WPid, T, Opts, {Mask, Nodes, #diameter_service{applications = Apps, - capabilities = Caps} + capabilities = LCaps} = Svc}}) -> [] /= Apps orelse ?ERROR({no_apps, T, Opts}), - putr(?DWA_KEY, dwa(Caps)), + putr(?DWA_KEY, dwa(LCaps)), {M, Ref} = T, diameter_stats:reg(Ref), - {[Ts], Rest} = proplists:split(Opts, [capabilities_cb]), - putr(?CB_KEY, {Ref, [F || {_,F} <- Ts]}), + {[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), erlang:monitor(process, WPid), @@ -188,8 +199,8 @@ i({WPid, T, Opts, {Mask, Nodes, #diameter_service{applications = Apps, %% watchdog start (start/2) succeeds regardless so as not to crash the %% service. -start_transport(T, Opts, #diameter_service{capabilities = Caps} = Svc) -> - Addrs0 = Caps#diameter_caps.host_ip_address, +start_transport(T, Opts, #diameter_service{capabilities = LCaps} = Svc) -> + Addrs0 = LCaps#diameter_caps.host_ip_address, start_transport(Addrs0, {T, Opts, Svc}). start_transport(Addrs0, T) -> @@ -212,9 +223,9 @@ svc(Svc, []) -> svc(Svc, Addrs) -> readdr(Svc, Addrs). -readdr(#diameter_service{capabilities = Caps0} = Svc, Addrs) -> - Caps = Caps0#diameter_caps{host_ip_address = Addrs}, - Svc#diameter_service{capabilities = Caps}. +readdr(#diameter_service{capabilities = LCaps0} = Svc, Addrs) -> + LCaps = LCaps0#diameter_caps{host_ip_address = Addrs}, + Svc#diameter_service{capabilities = LCaps}. %% The 4-tuple Data returned from diameter_peer:start/1 identifies the %% transport module/config use to start the transport process in @@ -375,17 +386,17 @@ transition({send, Msg}, #state{transport = TPid}) -> send(TPid, Msg), ok; -%% Request for graceful shutdown. -transition({shutdown, Pid}, #state{parent = Pid, dpr = false} = S) -> - dpr(?GOAWAY, S); -transition({shutdown, Pid}, #state{parent = Pid}) -> - ok; - -%% Application shutdown. -transition(shutdown, #state{dpr = false} = S) -> - dpr(?REBOOT, S); -transition(shutdown, _) -> %% DPR already send: ensure expected timeout - dpa_timer(), +%% Messages from old (diameter_service) code. +transition(shutdown = T, #state{parent = Pid} = S) -> + transition({T, Pid, service}, S); %% Reason irrelevant: old code has no cb + +%% Request for graceful shutdown at remove_transport, stop_service of +%% application shutdown. +transition({shutdown = T, Pid}, S) -> + transition({T, Pid, transport}, S); +transition({shutdown, Pid, Reason}, #state{parent = Pid, dpr = false} = S) -> + dpr(Reason, S); +transition({shutdown, Pid, _}, #state{parent = Pid}) -> ok; %% Request to close the transport connection. @@ -441,13 +452,13 @@ start_next(#state{service = Svc0} = S) -> %% send_CER/1 send_CER(#state{mode = {connect, Remote}, - service = #diameter_service{capabilities = Caps}, + service = #diameter_service{capabilities = LCaps}, transport = TPid} = S) -> - OH = Caps#diameter_caps.origin_host, + OH = LCaps#diameter_caps.origin_host, req_send_CER(OH, Remote) orelse - close({already_connected, Remote, Caps}, S), + close({already_connected, Remote, LCaps}, S), CER = build_CER(S), ?LOG(send, 'CER'), send(TPid, encode(CER)), @@ -471,8 +482,8 @@ start_timer(#state{state = PS} = S) -> %% build_CER/1 -build_CER(#state{service = #diameter_service{capabilities = Caps}}) -> - {ok, CER} = diameter_capx:build_CER(Caps), +build_CER(#state{service = #diameter_service{capabilities = LCaps}}) -> + {ok, CER} = diameter_capx:build_CER(LCaps), CER. %% encode/1 @@ -800,8 +811,8 @@ a('CER', #diameter_caps{vendor_id = Vid, {'Product-Name', Name}, {'Origin-State-Id', OSI}]; -a('DPR', #diameter_caps{origin_host = Host, - origin_realm = Realm}) -> +a('DPR', #diameter_caps{origin_host = {Host, _}, + origin_realm = {Realm, _}}) -> ['DPA', {'Origin-Host', Host}, {'Origin-Realm', Realm}]. @@ -909,7 +920,9 @@ rejected(N) %% open/5 -open(Pkt, SupportedApps, Caps, {Type, IS}, #state{parent = Pid} = S) -> +open(Pkt, SupportedApps, Caps, {Type, IS}, #state{parent = Pid, + service = Svc} + = S) -> #diameter_caps{origin_host = {_,_} = H, inband_security_id = {LS,_}} = Caps, @@ -917,7 +930,9 @@ open(Pkt, SupportedApps, Caps, {Type, IS}, #state{parent = Pid} = S) -> tls_ack(lists:member(?TLS, LS), Caps, Type, IS, S), Pid ! {open, self(), H, {Caps, SupportedApps, Pkt}}, - S#state{state = 'Open'}. + %% Replace capabilities record with local/remote pairs. + S#state{state = 'Open', + service = Svc#diameter_service{capabilities = Caps}}. %% We've advertised TLS support: tell the transport the result %% and expect a reply when the handshake is complete. @@ -970,24 +985,110 @@ dwa(#diameter_caps{origin_host = OH, {'Origin-State-Id', OSI}]. %% dpr/2 +%% +%% The RFC isn't clear on whether DPR should be send 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 +%% yet reached Open. + +%% Connection is open, DPR has not been sent. +dpr(Reason, #state{state = 'Open', + dpr = false, + service = #diameter_service{capabilities = Caps}} + = S) -> + case getr(?DPR_KEY) of + CBs when is_list(CBs) -> + Ref = getr(?REF_KEY), + Peer = {self(), Caps}, + dpr(CBs, [Reason, Ref, Peer], S); + undefined -> %% started in old code + send_dpr(Reason, [], S) + end; -dpr(Cause, #state{transport = TPid, - service = #diameter_service{capabilities = Caps}} - = S) -> - #diameter_caps{origin_host = OH, - origin_realm = OR} +%% Connection is open, DPR already sent. +dpr(_, #state{state = 'Open'}) -> + ok; + +%% Connection not open. +dpr(_Reason, _S) -> + stop. + +%% dpr/3 +%% +%% Note that an implementation that wants to do something +%% transport_module-specific can lookup the pid of the transport +%% process and contact it. (eg. diameter:service_info/2) + +dpr([CB|Rest], [Reason | _] = Args, S) -> + try diameter_lib:eval([CB | Args]) of + {dpr, Opts} when is_list(Opts) -> + send_dpr(Reason, Opts, S); + dpr -> + send_dpr(Reason, [], S); + close = T -> + {stop, {disconnect_cb, T}}; + ignore -> + dpr(Rest, Args, S); + T -> + No = {disconnect_cb, T}, + diameter_lib:error_report(invalid, No), + {stop, No} + catch + E:R -> + No = {disconnect_cb, E, R, ?STACK}, + diameter_lib:error_report(failure, No), + {stop, No} + end; + +dpr([], [Reason | _], S) -> + send_dpr(Reason, [], S). + +-record(opts, {cause, timeout = ?DPA_TIMEOUT}). + +send_dpr(Reason, Opts, #state{transport = TPid, + service = #diameter_service{capabilities = Caps}} + = S) -> + #opts{cause = Cause, timeout = Tmo} + = lists:foldl(fun opt/2, + #opts{cause = case Reason of + transport -> ?GOAWAY; + _ -> ?REBOOT + end, + timeout = ?DPA_TIMEOUT}, + Opts), + #diameter_caps{origin_host = {OH, _}, + origin_realm = {OR, _}} = Caps, Bin = encode(['DPR', {'Origin-Host', OH}, {'Origin-Realm', OR}, {'Disconnect-Cause', Cause}]), send(TPid, Bin), - dpa_timer(), + dpa_timer(Tmo), ?LOG(send, 'DPR'), S#state{dpr = diameter_codec:sequence_numbers(Bin)}. -dpa_timer() -> - erlang:send_after(?DPA_TIMEOUT, self(), dpa_timeout). +opt({timeout, Tmo}, Rec) + when ?IS_TIMEOUT(Tmo) -> + Rec#opts{timeout = Tmo}; +opt({cause, Cause}, Rec) + when ?IS_CAUSE(Cause) -> + Rec#opts{cause = cause(Cause)}; +opt(T, _) -> + ?ERROR({invalid_option, T}). + +cause(rebooting) -> ?REBOOT; +cause(goaway) -> ?GOAWAY; +cause(busy) -> ?BUSY; +cause(N) + when ?IS_CAUSE(N) -> + N; +cause(N) -> + ?ERROR({invalid_cause, N}). + +dpa_timer(Tmo) -> + erlang:send_after(Tmo, self(), dpa_timeout). %% register_everywhere/1 %% diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index cffba4fc94..0a0fd6ded1 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -494,7 +494,7 @@ handle_call({info, Item}, _From, S) -> {reply, service_info(Item, S), S}; handle_call(stop, _From, S) -> - shutdown(S), + shutdown(service, S), {stop, normal, ok, S}; %% The server currently isn't guaranteed to be dead when the caller %% gets the reply. We deal with this in the call to the server, @@ -683,7 +683,7 @@ upgrade_insert(#state{service = #diameter_service{pid = Pid}} = S) -> terminate(Reason, #state{service_name = Name} = S) -> ets:delete(?STATE_TABLE, Name), shutdown == Reason %% application shutdown - andalso shutdown(S). + andalso shutdown(application, S). %%% --------------------------------------------------------------------------- %%% # code_change(FromVsn, State, Extra) @@ -766,44 +766,48 @@ mod_state(Alias, ModS) -> %%% # shutdown/2 %%% --------------------------------------------------------------------------- -shutdown(Refs, #state{peerT = PeerT}) -> - ets:foldl(fun(P,ok) -> s(P, Refs), ok end, ok, PeerT). +%% remove_transport: ask watchdogs to terminate their transport. +shutdown(Refs, #state{peerT = PeerT}) + when is_list(Refs) -> + ets:foldl(fun(P,ok) -> sp(P, Refs), ok end, ok, PeerT); -s(#peer{ref = Ref, pid = Pid}, Refs) -> - s(lists:member(Ref, Refs), Pid); - -s(true, Pid) -> - Pid ! {shutdown, self()}; %% 'DOWN' will cleanup as usual -s(false, _) -> - ok. - -%%% --------------------------------------------------------------------------- -%%% # shutdown/1 -%%% --------------------------------------------------------------------------- - -shutdown(#state{peerT = PeerT}) -> +%% application/service shutdown: ask transports to terminate themselves. +shutdown(Reason, #state{peerT = PeerT}) -> %% A transport might not be alive to receive the shutdown request %% but give those that are a chance to shutdown gracefully. - wait(fun st/2, PeerT), + shutdown(conn, Reason, PeerT), %% Kill the watchdogs explicitly in case there was no transport. - wait(fun sw/2, PeerT). + shutdown(peer, Reason, PeerT). -wait(Fun, T) -> - diameter_lib:wait(ets:foldl(Fun, [], T)). +%% sp/2 -st(#peer{op_state = {OS,_}} = P, Acc) -> - st(P#peer{op_state = OS}, Acc); -st(#peer{op_state = ?STATE_UP, conn = Pid}, Acc) -> - Pid ! shutdown, - [Pid | Acc]; -st(#peer{}, Acc) -> - Acc. +sp(#peer{ref = Ref, pid = Pid}, Refs) -> + lists:member(Ref, Refs) + andalso (Pid ! {shutdown, self()}). %% 'DOWN' cleans up + +%% shutdown/3 + +shutdown(Who, Reason, T) -> + diameter_lib:wait(ets:foldl(fun(X,A) -> shutdown(Who, X, Reason, A) end, + [], + T)). -sw(#peer{pid = Pid}, Acc) +shutdown(conn = Who, #peer{op_state = {OS,_}} = P, Reason, Acc) -> + shutdown(Who, P#peer{op_state = OS}, Reason, Acc); + +shutdown(conn, + #peer{pid = Pid, op_state = ?STATE_UP, conn = TPid}, + Reason, + Acc) -> + TPid ! {shutdown, Pid, Reason}, + [TPid | Acc]; + +shutdown(peer, #peer{pid = Pid}, _Reason, Acc) when is_pid(Pid) -> exit(Pid, shutdown), [Pid | Acc]; -sw(#peer{}, Acc) -> + +shutdown(_, #peer{}, _, Acc) -> Acc. %%% --------------------------------------------------------------------------- -- cgit v1.2.3 From 3ce7a87d6a7e096fa1850dbaa0942a919a81955b Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Tue, 13 Nov 2012 10:30:02 +0100 Subject: Remove dead clause Long dead. --- lib/diameter/src/base/diameter_peer_fsm.erl | 6 ------ 1 file changed, 6 deletions(-) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index ecdb09d34a..4acfd8313b 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -399,12 +399,6 @@ transition({shutdown, Pid, Reason}, #state{parent = Pid, dpr = false} = S) -> transition({shutdown, Pid, _}, #state{parent = Pid}) -> ok; -%% Request to close the transport connection. -transition({close = T, Pid}, #state{parent = Pid, - transport = TPid}) -> - diameter_peer:close(TPid), - {stop, T}; - %% DPA reception has timed out. transition(dpa_timeout, _) -> stop; -- cgit v1.2.3 From 5903d6db03084f4279bb4834f8542f726c9d3cdb Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Thu, 8 Nov 2012 12:54:20 +0100 Subject: Ensure watchdog dies with transport if DPR was sent A watchdog timeout after DPR but before DPA would previously result in the watchdog restarting the transport. --- lib/diameter/src/base/diameter_watchdog.erl | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl index d814f1afe2..d6a2d2833b 100644 --- a/lib/diameter/src/base/diameter_watchdog.erl +++ b/lib/diameter/src/base/diameter_watchdog.erl @@ -48,18 +48,19 @@ -record(watchdog, {%% PCB - Peer Control Block; see RFC 3539, Appendix A status = initial :: initial | okay | suspect | down | reopen, - pending = false :: boolean(), + pending = false :: boolean(), %% DWA tw :: 6000..16#FFFFFFFF | {module(), atom(), list()}, %% {M,F,A} -> integer() >= 0 num_dwa = 0 :: -1 | non_neg_integer(), %% number of DWAs received during reopen %% end PCB - parent = self() :: pid(), - transport :: pid() | undefined, + parent = self() :: pid(), %% service process + transport :: pid() | undefined, %% peer_fsm process tref :: reference(), %% reference for current watchdog timer message_data, %% term passed into diameter_service with message sequence :: diameter:sequence(), %% mask - restrict :: {diameter:restriction(), boolean()}}). + restrict :: {diameter:restriction(), boolean()}, + shutdown = false :: boolean()}). %% start/2 %% @@ -168,7 +169,8 @@ handle_info(T, S) -> handle_info(T, upgrade(S)). upgrade(S) -> - #watchdog{} = list_to_tuple(tuple_to_list(S) ++ [?NOMASK, {nodes, true}]). + #watchdog{} = list_to_tuple(tuple_to_list(S) + ++ [?NOMASK, {nodes, true}, false]). event(#watchdog{status = T}, #watchdog{status = T}) -> ok; @@ -225,9 +227,10 @@ transition({shutdown, Pid}, #watchdog{parent = Pid, down = S, %% sanity check stop; transition({shutdown = T, Pid}, #watchdog{parent = Pid, - transport = TPid}) -> + transport = TPid} + = S) -> TPid ! {T, self()}, - ok; + S#watchdog{shutdown = true}; %% Parent process has died, transition({'DOWN', _, process, Pid, _Reason}, @@ -301,7 +304,10 @@ transition({open = P, TPid, _Hosts, T}, transition({'DOWN', _, process, TPid, _}, #watchdog{transport = TPid, - status = initial}) -> + status = S, + shutdown = D}) + when S == initial; + D -> stop; transition({'DOWN', _, process, TPid, _}, -- cgit v1.2.3 From 91a223d37d6b57f53d26135a4cbaf4ac22854ba2 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Thu, 15 Nov 2012 10:52:58 +0100 Subject: Allow a handle_request callback to return a #diameter_packet{} This allows it to set transport_data and header, inappropriately so even. --- lib/diameter/src/base/diameter_service.erl | 49 ++++++++++++++++++------------ 1 file changed, 30 insertions(+), 19 deletions(-) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index 0a0fd6ded1..467bb585cd 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -2175,15 +2175,13 @@ reply([Msg], Dict, TPid, Fs, Pkt) reply(Msg, Dict, TPid, Fs, Pkt#diameter_packet{errors = []}); %% No errors or a diameter_header/avp list. -reply(Msg, Dict, TPid, Fs, #diameter_packet{errors = Es, - transport_data = TD} - = ReqPkt) +reply(Msg, Dict, TPid, Fs, #diameter_packet{errors = Es} = ReqPkt) when [] == Es; is_record(hd(Msg), diameter_header) -> Pkt = diameter_codec:encode(Dict, make_answer_packet(Msg, ReqPkt)), eval_packet(Pkt, Fs), incr(send, Pkt, Dict, TPid), %% count result codes in sent answers - send(TPid, Pkt#diameter_packet{transport_data = TD}); + send(TPid, Pkt); %% Or not: set Result-Code and Failed-AVP AVP's. reply(Msg, Dict, TPid, Fs, #diameter_packet{errors = [H|_] = Es} = Pkt) -> @@ -2198,23 +2196,36 @@ eval_packet(Pkt, Fs) -> %% make_answer_packet/2 +%% A reply message clears the R and T flags and retains the P flag. +%% The E flag will be set at encode. 6.2 of 3588 requires the same P +%% flag on an answer as on the request. A #diameter_packet{} returned +%% from a handle_request callback can circumvent this by setting its +%% own header values. +make_answer_packet(#diameter_packet{header = Hdr, + msg = Msg, + transport_data = TD}, + #diameter_packet{header = ReqHdr}) -> + Hdr0 = ReqHdr#diameter_header{version = ?DIAMETER_VERSION, + is_request = false, + is_error = undefined, + is_retransmitted = false}, + #diameter_packet{header = fold_record(Hdr0, Hdr), + msg = Msg, + transport_data = TD}; + %% Binaries and header/avp lists are sent as-is. -make_answer_packet(Bin, _) +make_answer_packet(Bin, #diameter_packet{transport_data = TD}) when is_binary(Bin) -> - #diameter_packet{bin = Bin}; -make_answer_packet([#diameter_header{} | _] = Msg, _) -> - #diameter_packet{msg = Msg}; - -%% Otherwise a reply message clears the R and T flags and retains the -%% P flag. The E flag will be set at encode. 6.2 of 3588 requires the -%% same P flag on an answer as on the request. -make_answer_packet(Msg, #diameter_packet{header = ReqHdr}) -> - Hdr = ReqHdr#diameter_header{version = ?DIAMETER_VERSION, - is_request = false, - is_error = undefined, - is_retransmitted = false}, - #diameter_packet{header = Hdr, - msg = Msg}. + #diameter_packet{bin = Bin, + transport_data = TD}; +make_answer_packet([#diameter_header{} | _] = Msg, + #diameter_packet{transport_data = TD}) -> + #diameter_packet{msg = Msg, + transport_data = TD}; + +%% Otherwise, preserve transport_data. +make_answer_packet(Msg, #diameter_packet{transport_data = TD} = Pkt) -> + make_answer_packet(#diameter_packet{msg = Msg, transport_data = TD}, Pkt). %% rc/1 -- cgit v1.2.3 From 4ddc8f40399d7c4856d0c8f9463832012c25002f Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Thu, 15 Nov 2012 14:57:18 +0100 Subject: Add check of End-to-End and Hop-by-Hop identfiers in received CEA --- lib/diameter/src/base/diameter_peer_fsm.erl | 43 ++++++++++++++++++++--------- 1 file changed, 30 insertions(+), 13 deletions(-) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index 4acfd8313b..3cf1129d59 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -90,15 +90,20 @@ %% timeout used to be hardcoded. (So it could be worse.) -define(DPA_TIMEOUT, 1000). +-type uint32() :: diameter:'Unsigned32'(). + -record(state, {state = 'Wait-Conn-Ack' %% state of RFC 3588 Peer State Machine - :: 'Wait-Conn-Ack' | recv_CER | 'Wait-CEA' | 'Open', + :: 'Wait-Conn-Ack' + | recv_CER + | 'Wait-CEA' %% old code + | {'Wait-CEA', uint32(), uint32()} + | 'Open', mode :: accept | connect | {connect, reference()}, parent :: pid(), %% watchdog process transport :: pid(), %% transport process service :: #diameter_service{}, - dpr = false :: false | {diameter:'Unsigned32'(), - diameter:'Unsigned32'()}}). + dpr = false :: false | {uint32(), uint32()}}). %% | hop by hop and end to end identifiers %% There are non-3588 states possible as a consequence of 5.6.1 of the @@ -455,8 +460,12 @@ send_CER(#state{mode = {connect, Remote}, close({already_connected, Remote, LCaps}, S), CER = build_CER(S), ?LOG(send, 'CER'), - send(TPid, encode(CER)), - start_timer(S#state{state = 'Wait-CEA'}). + #diameter_packet{header = #diameter_header{end_to_end_id = Eid, + hop_by_hop_id = Hid}} + = Pkt + = encode(CER), + send(TPid, Pkt), + start_timer(S#state{state = {'Wait-CEA', Hid, Eid}}). %% Register ourselves as connecting to the remote endpoint in %% question. This isn't strictly necessary since a peer implementing @@ -487,10 +496,8 @@ encode(Rec) -> Hdr = #diameter_header{version = ?DIAMETER_VERSION, end_to_end_id = Seq, hop_by_hop_id = Seq}, - Pkt = #diameter_packet{header = Hdr, - msg = Rec}, - #diameter_packet{bin = Bin} = diameter_codec:encode(?BASE, Pkt), - Bin. + diameter_codec:encode(?BASE, #diameter_packet{header = Hdr, + msg = Rec}). sequence() -> case getr(?SEQUENCE_KEY) of @@ -558,7 +565,14 @@ discard(Reason, F, A) -> %% rcv/3 %% Incoming CEA. -rcv('CEA', Pkt, #state{state = 'Wait-CEA'} = S) -> +rcv('CEA', + #diameter_packet{header = #diameter_header{end_to_end_id = Eid, + hop_by_hop_id = Hid}} + = Pkt, + #state{state = {'Wait-CEA' = T, Hid, Eid}} + = S) -> + handle_CEA(Pkt, S#state{state = T}); +rcv('CEA', Pkt, #state{state = 'Wait-CEA'} = S) -> %% old code handle_CEA(Pkt, S); %% Incoming CER @@ -1055,13 +1069,16 @@ send_dpr(Reason, Opts, #state{transport = TPid, origin_realm = {OR, _}} = Caps, - Bin = encode(['DPR', {'Origin-Host', OH}, + #diameter_packet{header = #diameter_header{end_to_end_id = Eid, + hop_by_hop_id = Hid}} + = Pkt + = encode(['DPR', {'Origin-Host', OH}, {'Origin-Realm', OR}, {'Disconnect-Cause', Cause}]), - send(TPid, Bin), + send(TPid, Pkt), dpa_timer(Tmo), ?LOG(send, 'DPR'), - S#state{dpr = diameter_codec:sequence_numbers(Bin)}. + S#state{dpr = {Hid, Eid}}. opt({timeout, Tmo}, Rec) when ?IS_TIMEOUT(Tmo) -> -- cgit v1.2.3 From 62403f98c887a1b2e99e41da576c171c29d59193 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Thu, 15 Nov 2012 15:05:26 +0100 Subject: Add check of End-to-End and Hop-by-Hop identfiers in received DPA --- lib/diameter/src/base/diameter_peer_fsm.erl | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index 3cf1129d59..e06dc136ce 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -592,16 +592,16 @@ rcv(N, Pkt, S) N == 'DPR' -> handle_request(N, Pkt, S); -%% DPA even though we haven't sent DPR: ignore. -rcv('DPA', _Pkt, #state{dpr = false}) -> - ok; - -%% DPA in response to DPR. We could check the sequence numbers but -%% don't bother, just close. -rcv('DPA' = N, _Pkt, #state{transport = TPid}) -> +%% DPA in response to DPR and with the expected identifiers. +rcv('DPA' = N, + #diameter_packet{header = #diameter_header{end_to_end_id = Eid, + hop_by_hop_id = Hid}}, + #state{transport = TPid, + dpr = {Hid, Eid}}) -> diameter_peer:close(TPid), {stop, N}; +%% Ignore anything else, an unsolicited DPA in particular. rcv(_, _, _) -> ok. -- cgit v1.2.3 From 81e2c30e1dd83974e6a41f0988124501da1e8ad6 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Thu, 15 Nov 2012 15:47:02 +0100 Subject: Add comment about lack of identifier checks on DWA --- lib/diameter/src/base/diameter_watchdog.erl | 8 ++++++++ 1 file changed, 8 insertions(+) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl index d6a2d2833b..243ad0a986 100644 --- a/lib/diameter/src/base/diameter_watchdog.erl +++ b/lib/diameter/src/base/diameter_watchdog.erl @@ -487,6 +487,14 @@ throwaway(S) -> throw({?MODULE, throwaway, S}). %% rcv/2 +%% +%% The lack of Hop-by-Hop and End-to-End Identifiers checks in a +%% received DWA is intentional. The purpose of the message is to +%% demonstrate life but a peer that consistently bungles it by sending +%% the wrong identifiers causes the connection to toggle between OPEN +%% and SUSPECT, with failover and failback as result, despite there +%% being no real problem with connectivity. Thus, relax and accept any +%% incoming DWA as being in response to an outgoing DWR. %% INITIAL Receive DWA Pending = FALSE %% Throwaway() INITIAL -- cgit v1.2.3 From 23925103bfb66268113e533e701eb3865d6c1085 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Thu, 15 Nov 2012 18:39:54 +0100 Subject: Implement transport_opt() capx_timeout --- lib/diameter/src/base/diameter.erl | 1 + lib/diameter/src/base/diameter_peer_fsm.erl | 43 +++++++++++++++++++---------- 2 files changed, 30 insertions(+), 14 deletions(-) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl index 95702f03d4..8f9901907a 100644 --- a/lib/diameter/src/base/diameter.erl +++ b/lib/diameter/src/base/diameter.erl @@ -330,6 +330,7 @@ call(SvcName, App, Message) -> | {applications, [app_alias()]} | {capabilities, [capability()]} | {capabilities_cb, evaluable()} + | {capx_timeout, 'Unsigned32'()} | {disconnect_cb, evaluable()} | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}} | {reconnect_timer, 'Unsigned32'()} diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index e06dc136ce..c4320fcb99 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -73,7 +73,7 @@ -define(IS_SUCCESS(N), 2 == (N) div 1000). %% Guards. --define(IS_UINT32(N), (0 =< N andalso 0 == N bsr 32)). +-define(IS_UINT32(N), (is_integer(N) andalso 0 =< N andalso 0 == N bsr 32)). -define(IS_TIMEOUT(N), ?IS_UINT32(N)). -define(IS_CAUSE(N), N == ?REBOOT; N == rebooting; N == ?GOAWAY; N == goaway; @@ -85,6 +85,7 @@ %% for some event. %% -define(EVENT_TIMEOUT, 10000). +%% Default timeout for reception of CER/CEA. %% Default timeout for DPA in response to DPR. A bit short but the %% timeout used to be hardcoded. (So it could be worse.) @@ -93,8 +94,9 @@ -type uint32() :: diameter:'Unsigned32'(). -record(state, - {state = 'Wait-Conn-Ack' %% state of RFC 3588 Peer State Machine - :: 'Wait-Conn-Ack' + {state %% of RFC 3588 Peer State Machine + :: 'Wait-Conn-Ack' %% old code + | {'Wait-Conn-Ack', uint32()} | recv_CER | 'Wait-CEA' %% old code | {'Wait-CEA', uint32(), uint32()} @@ -191,7 +193,10 @@ i({WPid, T, Opts, {Mask, Nodes, #diameter_service{applications = Apps, putr(?RESTRICT_KEY, Nodes), erlang:monitor(process, WPid), {TPid, Addrs} = start_transport(T, Rest, Svc), - #state{parent = WPid, + Tmo = proplists:get_value(capx_timeout, Opts, ?EVENT_TIMEOUT), + ?IS_TIMEOUT(Tmo) orelse ?ERROR({invalid, {capx_timeout, Tmo}}), + #state{state = {'Wait-Conn-Ack', Tmo}, + parent = WPid, transport = TPid, mode = M, service = svc(Svc, Addrs)}. @@ -329,13 +334,17 @@ eraser(Key) -> %% transition/2 +%% Started in old code. +transition(T, #state{state = 'Wait-Conn-Ack' = PS} = S) -> + transition(T, S#state{state = {PS, ?EVENT_TIMEOUT}}); + %% Connection to peer. transition({diameter, {TPid, connected, Remote}}, #state{transport = TPid, state = PS, mode = M} = S) -> - 'Wait-Conn-Ack' = PS, %% assert + {'Wait-Conn-Ack', _} = PS, %% assert connect = M, %% keep_transport(TPid), send_CER(S#state{mode = {M, Remote}}); @@ -347,11 +356,11 @@ transition({diameter, {TPid, connected}}, mode = M, parent = Pid} = S) -> - 'Wait-Conn-Ack' = PS, %% assert + {'Wait-Conn-Ack', Tmo} = PS, %% assert accept = M, %% keep_transport(TPid), Pid ! {accepted, self()}, - start_timer(S#state{state = recv_CER}); + start_timer(Tmo, S#state{state = recv_CER}); %% Connection established after receiving a connection_timeout %% message. This may be followed by an incoming message which arrived @@ -365,7 +374,7 @@ transition({diameter, {_, connected, _}}, _) -> %% Connection has timed out: start an alternate. transition({connection_timeout = T, TPid}, #state{transport = TPid, - state = 'Wait-Conn-Ack'} + state = {'Wait-Conn-Ack', _}} = S) -> exit(TPid, {shutdown, T}), start_next(S); @@ -380,7 +389,7 @@ transition({diameter, {recv, Pkt}}, S) -> %% Timeout when still in the same state ... transition({timeout, PS}, #state{state = PS}) -> - stop; + {stop, {capx(PS), timeout}}; %% ... or not. transition({timeout, _}, _) -> @@ -435,6 +444,11 @@ transition({state, Pid}, #state{state = S, transport = TPid}) -> %% Crash on anything unexpected. +capx(recv_CER) -> + 'CER'; +capx({'Wait-CEA', _, _}) -> + 'CEA'. + %% start_next/1 start_next(#state{service = Svc0} = S) -> @@ -450,7 +464,8 @@ start_next(#state{service = Svc0} = S) -> %% send_CER/1 -send_CER(#state{mode = {connect, Remote}, +send_CER(#state{state = {'Wait-Conn-Ack', Tmo}, + mode = {connect, Remote}, service = #diameter_service{capabilities = LCaps}, transport = TPid} = S) -> @@ -465,7 +480,7 @@ send_CER(#state{mode = {connect, Remote}, = Pkt = encode(CER), send(TPid, Pkt), - start_timer(S#state{state = {'Wait-CEA', Hid, Eid}}). + start_timer(Tmo, S#state{state = {'Wait-CEA', Hid, Eid}}). %% Register ourselves as connecting to the remote endpoint in %% question. This isn't strictly necessary since a peer implementing @@ -477,10 +492,10 @@ send_CER(#state{mode = {connect, Remote}, req_send_CER(OriginHost, Remote) -> register_everywhere({?MODULE, connection, OriginHost, {remote, Remote}}). -%% start_timer/1 +%% start_timer/2 -start_timer(#state{state = PS} = S) -> - erlang:send_after(?EVENT_TIMEOUT, self(), {timeout, PS}), +start_timer(Tmo, #state{state = PS} = S) -> + erlang:send_after(Tmo, self(), {timeout, PS}), S. %% build_CER/1 -- cgit v1.2.3 From 476db5308088f0a907aa72726a75e31c695a92d3 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Thu, 11 Oct 2012 15:44:24 +0200 Subject: Add start/stop service_event() --- lib/diameter/src/base/diameter_service.erl | 2 ++ 1 file changed, 2 insertions(+) (limited to 'lib/diameter/src') diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index cffba4fc94..36907a8d1c 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -681,6 +681,7 @@ upgrade_insert(#state{service = #diameter_service{pid = Pid}} = S) -> %%% --------------------------------------------------------------------------- terminate(Reason, #state{service_name = Name} = S) -> + send_event(Name, stop), ets:delete(?STATE_TABLE, Name), shutdown == Reason %% application shutdown andalso shutdown(S). @@ -857,6 +858,7 @@ i(SvcName) -> lists:foreach(fun(T) -> start_fsm(T,S) end, CL), init_shared(S), + send_event(SvcName, start), S. cfg_acc({SvcName, #diameter_service{applications = Apps} = Rec, Opts}, -- cgit v1.2.3