From c14ef2dbd4e0e259fc401acce2ee43dc9fd4f4f9 Mon Sep 17 00:00:00 2001 From: Anders Svensson Date: Tue, 9 Oct 2012 11:03:25 +0200 Subject: Add reopen message from watchdog This makes capabilities available to service_info as soon as capabilities exchange has been completed. In particular, before state OKAY is reached. --- lib/diameter/src/base/diameter_service.erl | 103 ++++++++++++++++++---------- lib/diameter/src/base/diameter_watchdog.erl | 5 +- 2 files changed, 70 insertions(+), 38 deletions(-) (limited to 'lib/diameter/src/base') diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index ef660a10b9..72b886bcdd 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -119,15 +119,15 @@ {id = now(), service_name, %% as passed to start_service/2, key in ?STATE_TABLE service :: #diameter_service{}, - peerT = ets_new(peers) :: ets:tid(), %% #peer{} at start_fsm - connT = ets_new(conns) :: ets:tid(), %% #conn{} at connection_up + peerT = ets_new(peers) :: ets:tid(),%% #peer{} at start_fsm + connT = ets_new(conns) :: ets:tid(),%% #conn{} at connection_up/reopen shared_peers = ?Dict:new(), %% Alias -> [{TPid, Caps}, ...] local_peers = ?Dict:new(), %% Alias -> [{TPid, Caps}, ...] monitor = false :: false | pid(), %% process to die with - options :: [{atom(), term()}]}). -% :: [{sequence, diameter:sequence()} %% sequence mask -% | {share_peers, boolean()} %% broadcast peers to remote nodes? -% | {use_shared_peers, boolean()}]}).%% use broadcasted peers? + options + :: [{sequence, diameter:sequence()} %% sequence mask + | {share_peers, boolean()} %% broadcast peers to remote nodes? + | {use_shared_peers, boolean()}]}).%% use broadcasted peers? %% shared_peers reflects the peers broadcast from remote nodes. Note %% that the state term itself doesn't change, which is relevant for %% the stateless application callbacks since the state is retrieved @@ -145,7 +145,7 @@ :: match(op_state() | {op_state(), wd_state()}), started = now(), %% at process start conn = false :: match(boolean() | pid())}). - %% true at accept, pid() at connection_up (connT key) + %% true at accepted, pid() at connection_up or reopen %% Record representing a peer_fsm process. -record(conn, @@ -524,8 +524,6 @@ handle_info(T, #state{} = S) -> case transition(T,S) of ok -> {noreply, S}; - #state{} = NS -> - {noreply, NS}; {stop, Reason} -> {stop, {shutdown, Reason}, S} end; @@ -542,15 +540,26 @@ transition({accepted, Pid, TPid}, S) -> %% Peer process has a new open connection. transition({connection_up, Pid, T}, S) -> - connection_up(Pid, T, S); + connection_up(Pid, T, S), + ok; + +%% Peer process has a new connection that will be opened after +%% watchdog exchange. This message was added long after connection_up, +%% to communicate the information as soon as it's available. Leave +%% connection_up as is it for now, duplicated information and all. +transition({reopen, Pid, T}, S) -> + reopen(Pid, T, S), + ok; %% Peer process has left state open. transition({connection_down, Pid}, S) -> - connection_down(Pid, S); + connection_down(Pid, S), + ok; %% Peer process has returned to state open. transition({connection_up, Pid}, S) -> - connection_up(Pid, S); + connection_up(Pid, S), + ok; %% Accepting transport has lost connectivity. transition({close, Pid}, S) -> @@ -599,7 +608,8 @@ transition({'DOWN', MRef, process, _, Reason}, #state{monitor = MRef}) -> %% Local peer process has died. transition({'DOWN', _, process, Pid, Reason}, S) when node(Pid) == node() -> - peer_down(Pid, Reason, S); + peer_down(Pid, Reason, S), + ok; %% Remote service wants to know about shared transports. transition({service, Pid}, S) -> @@ -775,16 +785,20 @@ shutdown(#state{peerT = PeerT}) -> wait(Fun, T) -> diameter_lib:wait(ets:foldl(Fun, [], T)). -st(#peer{conn = B}, Acc) - when is_boolean(B) -> - Acc; -st(#peer{conn = Pid}, Acc) -> +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]. + [Pid | Acc]; +st(#peer{}, Acc) -> + Acc. -sw(#peer{pid = Pid}, Acc) -> +sw(#peer{pid = Pid}, Acc) + when is_pid(Pid) -> exit(Pid, shutdown), - [Pid | Acc]. + [Pid | Acc]; +sw(#peer{}, Acc) -> + Acc. %%% --------------------------------------------------------------------------- %%% # call_service/2 @@ -999,8 +1013,6 @@ fetch(Tid, Key) -> %%% --------------------------------------------------------------------------- %%% # connection_up/3 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- %% Peer process has reached the open state. @@ -1017,10 +1029,30 @@ connection_up(Pid, {TPid, {Caps, SApps, Pkt}}, #state{peerT = PeerT, insert(ConnT, C), connection_up([Pkt], P#peer{conn = TPid}, C, S). +%%% --------------------------------------------------------------------------- +%%% # reopen/3 +%%% --------------------------------------------------------------------------- + +%% Note that this connection_up/3 rewrites the same #conn{} now +%% written here. Both do so in case reopen has not happened in old +%% code. + +reopen(Pid, {TPid, {Caps, SApps, _Pkt}}, #state{peerT = PeerT, + connT = ConnT}) -> + P = fetch(PeerT, Pid), + C = #conn{pid = TPid, + apps = SApps, + caps = Caps, + peer = Pid}, + + insert(ConnT, C), + #peer{op_state = {?STATE_DOWN, _}} + = P, + insert(PeerT, P#peer{op_state = {?STATE_DOWN, ?WD_REOPEN}, + conn = TPid}). + %%% --------------------------------------------------------------------------- %%% # connection_up/2 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- %% Peer process has transitioned back into the open state. Note that there @@ -1050,8 +1082,7 @@ connection_up(T, P, C, #state{peerT = PeerT, request_peer_up(TPid), insert_local_peer(SApps, {{TPid, Caps}, {SvcName, Apps}}, LDict), - report_status(up, P, C, S, T), - S. + report_status(up, P, C, S, T). insert_local_peer(SApps, T, LDict) -> lists:foldl(fun(A,D) -> ilp(A, T, D) end, LDict, SApps). @@ -1093,8 +1124,6 @@ peer_cb(MFA, Alias) -> %%% --------------------------------------------------------------------------- %%% # connection_down/2 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- %% Peer process has transitioned out of the open state. @@ -1113,8 +1142,8 @@ connection_down(Pid, #state{peerT = PeerT, %% connection_down/3 -connection_down(#peer{op_state = {?STATE_DOWN, _}}, _, S) -> - S; +connection_down(#peer{op_state = {?STATE_DOWN, _}}, _, _) -> + ok; connection_down(#peer{conn = TPid, op_state = {?STATE_UP, _}} @@ -1128,8 +1157,7 @@ connection_down(#peer{conn = TPid, = S) -> report_status(down, P, C, S, []), remove_local_peer(SApps, {{TPid, Caps}, {SvcName, Apps}}, LDict), - request_peer_down(TPid, S), - S. + request_peer_down(TPid, S). remove_local_peer(SApps, T, LDict) -> lists:foldl(fun(A,D) -> rlp(A, T, D) end, LDict, SApps). @@ -1148,8 +1176,6 @@ down_conn(Id, Alias, TC, {SvcName, Apps}) -> %%% --------------------------------------------------------------------------- %%% # peer_down/3 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- %% Peer process has died. @@ -1173,11 +1199,11 @@ closed(_, _, _) -> ok. %% The peer has never come up ... -peer_down(#peer{conn = B}, S) +peer_down(#peer{conn = B}, _) when is_boolean(B) -> - S; + ok; -%% ... or it has. +%% ... or maybe it has. peer_down(#peer{conn = TPid} = P, #state{connT = ConnT} = S) -> #conn{} = C = fetch(ConnT, TPid), ets:delete_object(ConnT, C), @@ -3167,6 +3193,9 @@ peer_acc(ConnT, Acc, #peer{pid = Pid, | info_conn(ConnT, TPid, WS /= ?WD_DOWN)], Acc). +info_conn(ConnT, [TPid], B) -> + info_conn(ConnT, TPid, B); + info_conn(ConnT, TPid, true) when is_pid(TPid) -> try ets:lookup(ConnT, TPid) of diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl index 1045d9ad9f..b37a1a10e9 100644 --- a/lib/diameter/src/base/diameter_watchdog.erl +++ b/lib/diameter/src/base/diameter_watchdog.erl @@ -272,12 +272,15 @@ transition({open, TPid, Hosts, T} = Open, transition({open = P, TPid, _Hosts, T}, #watchdog{transport = TPid, + parent = Pid, status = down} = S) -> %% Store the info we need to notify the parent to reopen the %% connection after the requisite DWA's are received, at which - %% time we eraser(open). + %% time we eraser(open). The reopen message is a later addition, + %% to communicate the new capabilities as soon as they're known. putr(P, {TPid, T}), + Pid ! {reopen, self(), {TPid, T}}, set_watchdog(send_watchdog(S#watchdog{status = reopen, num_dwa = 0})); -- cgit v1.2.3