greg/machi
1
0
Fork 0
Code Issues 11 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Avoid using host/port combo for machi_fitness (ab)use of machi_projection

Browse source
This commit is contained in:
Scott Lystig Fritchie 2015-09-17 16:43:08 +09:00
parent 09ae2db0ba
commit d695f30e4f
2 changed files with 28 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
src/machi_fitness.erl
View file

@ -51,6 +51,7 @@
local_down=[] :: list(), local_down=[] :: list(),
admin_down=[] :: list({term(),term()}), admin_down=[] :: list({term(),term()}),
members_dict=orddict:new() :: orddict:orddict(), members_dict=orddict:new() :: orddict:orddict(),
proxies_dict=orddict:new() :: orddict:orddict(),
active_unfit=[] :: list(), active_unfit=[] :: list(),
pending_map=?MAP:new() :: ?MAP:riak_dt_map(), pending_map=?MAP:new() :: ?MAP:riak_dt_map(),
partition_simulator_p :: boolean() partition_simulator_p :: boolean()
@ -250,14 +251,16 @@ send_spam(NewMap, DontSendList, MembersDict, #state{my_flu_name=MyFluName}=S) ->
_Sent = orddict:fold(F, [], MembersDict), _Sent = orddict:fold(F, [], MembersDict),
ok. ok.
send_projection(FLU, Host, TcpPort, SpamProj, send_projection(FLU, _Host, _TcpPort, SpamProj,
#state{my_flu_name=MyFluName, members_dict=MembersDict, #state{my_flu_name=MyFluName, members_dict=MembersDict,
partition_simulator_p=SimulatorP}) -> partition_simulator_p=SimulatorP}=S) ->
%% At the moment, we're using utterly-temporary-hack method of tunneling %% At the moment, we're using utterly-temporary-hack method of tunneling
%% our messages through the write_projection API. Eventually the PB %% our messages through the write_projection API. Eventually the PB
%% API should be expanded to accomodate this new fitness service. %% API should be expanded to accomodate this new fitness service.
DoIt = fun(_ProxyPid) -> %% This is "best effort" only, use catch to ignore failures.
machi_flu1_client:write_projection(Host, TcpPort, ProxyPid = (catch proxy_pid(FLU, S)),
DoIt = fun(_ArgIgnored) ->
machi_proxy_flu1_client:write_projection(ProxyPid,
public, SpamProj) public, SpamProj)
end, end,
ProxyPidPlaceholder = proxy_pid_unused, ProxyPidPlaceholder = proxy_pid_unused,
@ -265,9 +268,9 @@ send_projection(FLU, Host, TcpPort, SpamProj,
AllMembers = [K || {K,_V} <- orddict:to_list(MembersDict)], AllMembers = [K || {K,_V} <- orddict:to_list(MembersDict)],
{Partitions, _Islands} = machi_partition_simulator:get(AllMembers), {Partitions, _Islands} = machi_partition_simulator:get(AllMembers),
machi_chain_manager1:init_remember_down_list(), machi_chain_manager1:init_remember_down_list(),
Res = machi_chain_manager1:perhaps_call(ProxyPidPlaceholder, Res = (catch machi_chain_manager1:perhaps_call(ProxyPidPlaceholder,
MyFluName, MyFluName,
Partitions, FLU, DoIt), Partitions, FLU, DoIt)),
%% case machi_chain_manager1:get_remember_down_list() of %% case machi_chain_manager1:get_remember_down_list() of
%% [] -> %% [] ->
%% ok; %% ok;
@ -277,9 +280,12 @@ send_projection(FLU, Host, TcpPort, SpamProj,
%% end, %% end,
Res; Res;
true -> true ->
DoIt(ProxyPidPlaceholder) (catch DoIt(ProxyPidPlaceholder))
end. end.
proxy_pid(Name, #state{proxies_dict=ProxiesDict}) ->
orddict:fetch(Name, ProxiesDict).
calc_unfit(All_list, HosedAnnotations) -> calc_unfit(All_list, HosedAnnotations) ->
G = digraph:new(), G = digraph:new(),
[digraph:add_vertex(G, V) || V <- All_list], [digraph:add_vertex(G, V) || V <- All_list],
@ -342,7 +348,17 @@ do_map_change(NewMap, DontSendList, MembersDict,
%% _MapV = map_value(NewMap), %% _MapV = map_value(NewMap),
%% io:format(user, "TODO: ~w async tick trigger/scheduling... ~w for:\n" %% io:format(user, "TODO: ~w async tick trigger/scheduling... ~w for:\n"
%% " ~p\n ~p\n",[_MyFluName,ChangedServers,_OldMapV,_MapV]), %% " ~p\n ~p\n",[_MyFluName,ChangedServers,_OldMapV,_MapV]),
S#state{pending_map=NewMap, members_dict=MembersDict}. S2 = perhaps_adjust_members_proxies_dicts(MembersDict, S),
S2#state{pending_map=NewMap}.
perhaps_adjust_members_proxies_dicts(SameMembersDict,
#state{members_dict=SameMembersDict}=S) ->
S;
perhaps_adjust_members_proxies_dicts(MembersDict,
#state{proxies_dict=OldProxiesDict}=S) ->
_ = machi_proxy_flu1_client:stop_proxies(OldProxiesDict),
ProxiesDict = machi_proxy_flu1_client:start_proxies(MembersDict),
S#state{members_dict=MembersDict, proxies_dict=ProxiesDict}.
find_changed_servers(OldMap, NewMap, _MyFluName) -> find_changed_servers(OldMap, NewMap, _MyFluName) ->
AddBad = fun({_Who, {_Time, BadList, AdminDown, _Props}}, Acc) -> AddBad = fun({_Who, {_Time, BadList, AdminDown, _Props}}, Acc) ->

4
src/machi_flu1_client.erl
View file

@ -687,7 +687,7 @@ filter_sock_error_result(Error) ->
%%%%%%%%%%%%%%%%%%%%%%%%%%% %%%%%%%%%%%%%%%%%%%%%%%%%%%
w_connect(#p_srvr{proto_mod=?MODULE, address=Host, port=Port, props=Props})-> w_connect(#p_srvr{proto_mod=?MODULE, address=Host, port=Port, props=Props}=_P)->
try try
case proplists:get_value(session_proto, Props, tcp) of case proplists:get_value(session_proto, Props, tcp) of
tcp -> tcp ->
@ -705,7 +705,7 @@ w_connect(#p_srvr{proto_mod=?MODULE, address=Host, port=Port, props=Props})->
{w,ssl,SslSock} {w,ssl,SslSock}
end end
catch catch
_:_ -> _X:_Y ->
undefined undefined
end. end.