diff options
| -rw-r--r-- | lib/diameter/src/base/diameter.erl | 1 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_peer.erl | 113 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_peer_fsm.erl | 149 |
3 files changed, 210 insertions, 53 deletions
diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl index 336f0c1f2d..6703841f80 100644 --- a/lib/diameter/src/base/diameter.erl +++ b/lib/diameter/src/base/diameter.erl @@ -312,6 +312,7 @@ call(SvcName, App, Message) -> -type transport_opt() :: {transport_module, atom()} | {transport_config, any()} + | {transport_timeout, non_neg_integer() | infinity} | {applications, [app_alias()]} | {capabilities, [capability()]} | {capabilities_cb, evaluable()} diff --git a/lib/diameter/src/base/diameter_peer.erl b/lib/diameter/src/base/diameter_peer.erl index 3e78c4caef..bfd59bee8e 100644 --- a/lib/diameter/src/base/diameter_peer.erl +++ b/lib/diameter/src/base/diameter_peer.erl @@ -27,7 +27,7 @@ up/2]). %% ... and the stack. --export([start/3, +-export([start/1, send/2, close/1, abort/1, @@ -57,6 +57,11 @@ %% Server state. -record(state, {id = now()}). +%% Default transport_module/config. +-define(DEFAULT_TMOD, diameter_tcp). +-define(DEFAULT_TCFG, []). +-define(DEFAULT_TTMO, infinity). + %%% --------------------------------------------------------------------------- %%% # notify/2 %%% --------------------------------------------------------------------------- @@ -65,12 +70,95 @@ notify(SvcName, T) -> rpc:abcast(nodes(), ?SERVER, {notify, SvcName, T}). %%% --------------------------------------------------------------------------- -%%% # start/3 +%%% # start/1 %%% --------------------------------------------------------------------------- -start(T, Opts, #diameter_service{} = Svc) -> - {Mod, Cfg} = split_transport(Opts), - apply(Mod, start, [T, Svc, Cfg]). +%% Initial start. +start({T, Opts, #diameter_service{} = Svc}) -> + start(T, Svc, pair(Opts, [], []), []); + +%% Subsequent start. +start({#diameter_service{} = Svc, Tmo, {{T, _, Cfg}, Ms, Rest, Errs}}) -> + start(T, Ms, Cfg, Svc, Tmo, Rest, Errs). + +%% pair/3 +%% +%% Pair transport modules with config. + +%% Another transport_module: accumulate it. +pair([{transport_module, M} | Rest], Mods, Acc) -> + pair(Rest, [M|Mods], Acc); + +%% Another transport_config: accumulate another tuple. +pair([{transport_config = T, C} | Rest], Mods, Acc) -> + pair([{T, C, ?DEFAULT_TTMO} | Rest], Mods, Acc); +pair([{transport_config, C, Tmo} | Rest], Mods, Acc) -> + pair(Rest, [], acc({Mods, C, Tmo}, Acc)); + +pair([_ | Rest], Mods, Acc) -> + pair(Rest, Mods, Acc); + +%% No transport_module or transport_config: defaults. +pair([], [], []) -> + [{[?DEFAULT_TMOD], ?DEFAULT_TCFG, ?DEFAULT_TTMO}]; + +%% One transport_module, one transport_config. +pair([], [M], [{[], Cfg, Tmo}]) -> + [{[M], Cfg, Tmo}]; + +%% Trailing transport_module: default transport_config. +pair([], [_|_] = Mods, Acc) -> + lists:reverse(acc({Mods, ?DEFAULT_TCFG, ?DEFAULT_TTMO}, Acc)); + +pair([], [], Acc) -> + lists:reverse(def(Acc)). + +%% acc/2 + +acc(T, Acc) -> + [T | def(Acc)]. + +%% def/1 +%% +%% Default module of previous pair if none were specified. + +def([{[], Cfg, Tmo} | Acc]) -> + [{[?DEFAULT_TMOD], Cfg, Tmo} | Acc]; +def(Acc) -> + Acc. + +%% start/4 + +start(T, Svc, [{Ms, Cfg, Tmo} | Rest], Errs) -> + start(T, Ms, Cfg, Svc, Tmo, Rest, Errs); + +%% One transport: return the bare error for backwards compatibility. +start(_, _, [], [Err]) -> + {Err}; + +%% Or not: return list of errors. +start(_, _, [], Errs) -> + {{error, Errs}}. + +%% start/7 + +start(T, [], _, Svc, _, Rest, Errs) -> + start(T, Svc, Rest, Errs); + +start(T, [M|Ms], Cfg, Svc, Tmo, Rest, Errs) -> + case start(M, [T, Svc, Cfg]) of + {ok, TPid} -> + {TPid, [], Tmo, {{T, M, Cfg}, Ms, Rest, Errs}}; + {ok, TPid, [_|_] = Addrs} -> + {TPid, Addrs, Tmo, {{T, M, Cfg}, Ms, Rest, Errs}}; + E -> + start(T, Ms, Cfg, Svc, Tmo, Rest, [E|Errs]) + end. + +%% start/2 + +start(Mod, Args) -> + apply(Mod, start, Args). %%% --------------------------------------------------------------------------- %%% # up/[12] @@ -204,21 +292,6 @@ bang(undefined = No, _) -> bang(Pid, T) -> Pid ! T. -%% split_transport/1 -%% -%% Split options into transport module, transport config and -%% remaining options. - -split_transport(Opts) -> - {[M,C], _} = proplists:split(Opts, [transport_module, - transport_config]), - {value(M, diameter_tcp), value(C, [])}. - -value([{_,V}], _) -> - V; -value([], V) -> - V. - %% call/1 call(Request) -> diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl index 99644814d2..d5e2d690cf 100644 --- a/lib/diameter/src/base/diameter_peer_fsm.erl +++ b/lib/diameter/src/base/diameter_peer_fsm.erl @@ -54,6 +54,12 @@ -define(NO_INBAND_SECURITY, 0). -define(TLS, 1). +%% Keys in process dictionary. +-define(CB_KEY, cb). %% capabilities callback +-define(DWA_KEY, dwa). %% outgoing DWA +-define(Q_KEY, q). %% transport start queue +-define(START_KEY, start). %% start of connected transport + %% A 2xxx series Result-Code. Not necessarily 2001. -define(IS_SUCCESS(N), 2 == (N) div 1000). @@ -143,18 +149,17 @@ init(T) -> proc_lib:init_ack({ok, self()}), gen_server:enter_loop(?MODULE, [], i(T)). -i({WPid, T, Opts, #diameter_service{capabilities = Caps} = Svc0}) -> - putr(dwa, dwa(Caps)), +i({WPid, T, Opts, #diameter_service{capabilities = Caps} = Svc}) -> + putr(?DWA_KEY, dwa(Caps)), {M, Ref} = T, {[Ts], Rest} = proplists:split(Opts, [capabilities_cb]), - putr(capabilities_cb, {Ref, [F || {_,F} <- Ts]}), - {ok, TPid, Svc} = start_transport(T, Rest, Svc0), - erlang:monitor(process, TPid), + putr(?CB_KEY, {Ref, [F || {_,F} <- Ts]}), erlang:monitor(process, WPid), + {TPid, Addrs} = start_transport(T, Rest, Svc), #state{parent = WPid, transport = TPid, mode = M, - service = Svc}. + service = svc(Svc, Addrs)}. %% The transport returns its local ip addresses so that different %% transports on the same service can use different local addresses. %% The local addresses are put into Host-IP-Address avps here when @@ -164,18 +169,53 @@ i({WPid, T, Opts, #diameter_service{capabilities = Caps} = Svc0}) -> %% watchdog start (start/2) succeeds regardless so as not to crash the %% service. -start_transport(T, Opts, Svc) -> - case diameter_peer:start(T, Opts, Svc) of - {ok, TPid} -> - {ok, TPid, Svc}; - {ok, TPid, [_|_] = Addrs} -> - #diameter_service{capabilities = Caps0} = Svc, - Caps = Caps0#diameter_caps{host_ip_address = Addrs}, - {ok, TPid, Svc#diameter_service{capabilities = Caps}}; - No -> +start_transport(T, Opts, #diameter_service{capabilities = Caps} = Svc) -> + Addrs0 = Caps#diameter_caps.host_ip_address, + start_transport(Addrs0, {T, Opts, Svc}). + +start_transport(Addrs0, T) -> + case diameter_peer:start(T) of + {TPid, Addrs, Tmo, Data} -> + erlang:monitor(process, TPid), + q_next(TPid, Addrs0, Tmo, Data), + {TPid, addrs(Addrs, Addrs0)}; + {No} -> exit({shutdown, No}) end. +addrs([], Addrs0) -> + Addrs0; +addrs(Addrs, _) -> + Addrs. + +svc(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}. + +%% The 4-tuple Data returned from diameter_peer:start/1 identifies the +%% transport module/config use to start the transport process in +%% question as well as any alternates to try if a connection isn't +%% established within Tmo. +q_next(TPid, Addrs0, Tmo, {_,_,_,_} = Data) -> + send_after(Tmo, {connection_timeout, TPid}), + putr(?Q_KEY, {Addrs0, Tmo, Data}). + +%% Connection has been established: retain the started module/config +%% in the process dictionary. +keep_transport() -> + {_, _, {{_,_,_} = T, _, _, _}} = eraser(?Q_KEY), + putr(?START_KEY, T). + +send_after(infinity, _) -> + ok; +send_after(Tmo, T) -> + erlang:send_after(Tmo, self(), T). + %% handle_call/3 handle_call(_, _, State) -> @@ -240,25 +280,48 @@ eraser(Key) -> %% Connection to peer. transition({diameter, {TPid, connected, Remote}}, - #state{state = PS, + #state{transport = TPid, + state = PS, mode = M} = S) -> 'Wait-Conn-Ack' = PS, %% assert connect = M, %% - send_CER(S#state{mode = {M, Remote}, - transport = TPid}); + keep_transport(), + send_CER(S#state{mode = {M, Remote}}); %% Connection from peer. transition({diameter, {TPid, connected}}, - #state{state = PS, + #state{transport = TPid, + state = PS, mode = M, parent = Pid} = S) -> 'Wait-Conn-Ack' = PS, %% assert accept = M, %% + keep_transport(), Pid ! {accepted, self()}, - start_timer(S#state{state = recv_CER, - transport = TPid}); + start_timer(S#state{state = recv_CER}); + +%% Connection established after receiving a connection_timeout +%% message. This may be followed by an incoming message which arrived +%% before the transport was killed and this can't be distinguished +%% from one from the transport that's been started to replace it. +transition({diameter, {_, connected}}, _) -> + {stop, connection_timeout}; +transition({diameter, {_, connected, _}}, _) -> + {stop, connection_timeout}; + +%% Connection has timed out: start an alternate. +transition({connection_timeout = T, TPid}, + #state{transport = TPid, + state = 'Wait-Conn-Ack'} + = S) -> + exit(TPid, {shutdown, T}), + start_next(S); + +%% Connect timeout after connection or alternate start: ignore. +transition({connection_timeout, _}, _) -> + ok; %% Incoming message from the transport. transition({diameter, {recv, Pkt}}, S) -> @@ -305,14 +368,21 @@ transition({resolve_port, _Pid} = T, #state{transport = TPid}) -> TPid ! T, ok; -%% Parent or transport has died. -transition({'DOWN', _, process, P, _}, - #state{parent = Pid, - transport = TPid}) - when P == Pid; - P == TPid -> +%% Parent has died. +transition({'DOWN', _, process, WPid, _}, + #state{parent = WPid}) -> stop; +%% Transport has died before connection timeout. +transition({'DOWN', _, process, TPid, _}, + #state{transport = TPid} + = S) -> + start_next(S); + +%% Transport has died after connection timeout. +transition({'DOWN', _, process, _, _}, _) -> + ok; + %% State query. transition({state, Pid}, #state{state = S, transport = TPid}) -> Pid ! {self(), [S, TPid]}, @@ -320,6 +390,19 @@ transition({state, Pid}, #state{state = S, transport = TPid}) -> %% Crash on anything unexpected. +%% start_next/1 + +start_next(#state{service = Svc0} = S) -> + case getr(?Q_KEY) of + {Addrs0, Tmo, Data} -> + Svc = readdr(Svc0, Addrs0), + {TPid, Addrs} = start_transport(Addrs0, {Svc, Tmo, Data}), + S#state{transport = TPid, + service = svc(Svc, Addrs)}; + undefined -> + stop + end. + %% send_CER/1 send_CER(#state{mode = {connect, Remote}, @@ -649,7 +732,7 @@ rc([RC|_]) -> %% answer/2 answer('DWR', _) -> - getr(dwa); + getr(?DWA_KEY); answer(Name, #state{service = #diameter_service{capabilities = Caps}}) -> a(Name, Caps). @@ -749,15 +832,15 @@ caps(#state{service = Svc}) -> %% caps_cb/1 caps_cb(Caps) -> - {Ref, Ts} = eraser(capabilities_cb), - ccb(Ts, [Ref, Caps]). + {Ref, Ts} = eraser(?CB_KEY), + caps_cb(Ts, [Ref, Caps]). -ccb([], _) -> +caps_cb([], _) -> ok; -ccb([F | Rest], T) -> +caps_cb([F | Rest], T) -> case diameter_lib:eval([F|T]) of ok -> - ccb(Rest, T); + caps_cb(Rest, T); N when ?IS_SUCCESS(N) -> %% 2xxx result code: accept immediately N; Res -> |