264 lines
11 KiB
Erlang
264 lines
11 KiB
Erlang
%% -------------------------------------------------------------------
|
|
%%
|
|
%% Machi: a small village of replicated files
|
|
%%
|
|
%% Copyright (c) 2014 Basho Technologies, Inc. All Rights Reserved.
|
|
%%
|
|
%% This file is provided to you under the Apache License,
|
|
%% Version 2.0 (the "License"); you may not use this file
|
|
%% except in compliance with the License. You may obtain
|
|
%% a copy of the License at
|
|
%%
|
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
%%
|
|
%% Unless required by applicable law or agreed to in writing,
|
|
%% software distributed under the License is distributed on an
|
|
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
%% KIND, either express or implied. See the License for the
|
|
%% specific language governing permissions and limitations
|
|
%% under the License.
|
|
%%
|
|
%% -------------------------------------------------------------------
|
|
-module(machi_chain_manager1_test).
|
|
|
|
-include("machi.hrl").
|
|
-include("machi_projection.hrl").
|
|
|
|
-define(MGR, machi_chain_manager1).
|
|
|
|
-define(D(X), io:format(user, "~s ~p\n", [??X, X])).
|
|
-define(Dw(X), io:format(user, "~s ~w\n", [??X, X])).
|
|
-define(FLU_C, machi_flu1_client).
|
|
-define(FLU_PC, machi_proxy_flu1_client).
|
|
|
|
-export([]).
|
|
|
|
-ifdef(TEST).
|
|
|
|
-ifdef(EQC).
|
|
-include_lib("eqc/include/eqc.hrl").
|
|
%% -include_lib("eqc/include/eqc_statem.hrl").
|
|
-define(QC_OUT(P),
|
|
eqc:on_output(fun(Str, Args) -> io:format(user, Str, Args) end, P)).
|
|
-endif.
|
|
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
-compile(export_all).
|
|
|
|
unanimous_report(Namez) ->
|
|
UniquePrivateEs =
|
|
lists:usort(lists:flatten(
|
|
[element(2, ?FLU_PC:list_all_projections(FLU, private)) ||
|
|
{_FLUName, FLU} <- Namez])),
|
|
[unanimous_report(Epoch, Namez) || Epoch <- UniquePrivateEs].
|
|
|
|
unanimous_report(Epoch, Namez) ->
|
|
Projs = [{FLUName, case ?FLU_PC:read_projection(FLU, private, Epoch) of
|
|
{ok, T} -> T;
|
|
_Else -> not_in_this_epoch
|
|
end} || {FLUName, FLU} <- Namez],
|
|
UPI_R_Sums = [{Proj#projection_v1.upi, Proj#projection_v1.repairing,
|
|
Proj#projection_v1.epoch_csum} ||
|
|
{_FLUname, Proj} <- Projs,
|
|
is_record(Proj, projection_v1)],
|
|
UniqueUPIs = lists:usort([UPI || {UPI, _Repairing, _CSum} <- UPI_R_Sums]),
|
|
Res =
|
|
[begin
|
|
case lists:usort([CSum || {U, _Repairing, CSum} <- UPI_R_Sums,
|
|
U == UPI]) of
|
|
[_1CSum] ->
|
|
%% Yay, there's only 1 checksum. Let's check
|
|
%% that all FLUs are in agreement.
|
|
{UPI, Repairing, _CSum} =
|
|
lists:keyfind(UPI, 1, UPI_R_Sums),
|
|
%% TODO: make certain that this subtlety doesn't get
|
|
%% last in later implementations.
|
|
|
|
%% So, this is a bit of a tricky thing. If we're at
|
|
%% upi=[c] and repairing=[a,b], then the transition
|
|
%% (eventually!) to upi=[c,a] does not currently depend
|
|
%% on b being an active participant in the repair.
|
|
%%
|
|
%% Yes, b's state is very important for making certain
|
|
%% that all repair operations succeed both to a & b.
|
|
%% However, in this simulation, we only consider that
|
|
%% the head(Repairing) is sane. Therefore, we use only
|
|
%% the "HeadOfRepairing" in our considerations here.
|
|
HeadOfRepairing = case Repairing of
|
|
[H_Rep|_] ->
|
|
[H_Rep];
|
|
_ ->
|
|
[]
|
|
end,
|
|
Tmp = [{FLU, case proplists:get_value(FLU, Projs) of
|
|
P when is_record(P, projection_v1) ->
|
|
P#projection_v1.epoch_csum;
|
|
Else ->
|
|
Else
|
|
end} || FLU <- UPI ++ HeadOfRepairing],
|
|
case lists:usort([CSum || {_FLU, CSum} <- Tmp]) of
|
|
[_] ->
|
|
{agreed_membership, {UPI, Repairing}};
|
|
Else2 ->
|
|
{not_agreed, {UPI, Repairing}, Else2}
|
|
end;
|
|
_Else ->
|
|
{UPI, not_unique, Epoch, _Else}
|
|
end
|
|
end || UPI <- UniqueUPIs],
|
|
AgreedResUPI_Rs = [UPI++Repairing ||
|
|
{agreed_membership, {UPI, Repairing}} <- Res],
|
|
Tag = case lists:usort(lists:flatten(AgreedResUPI_Rs)) ==
|
|
lists:sort(lists:flatten(AgreedResUPI_Rs)) of
|
|
true ->
|
|
ok_disjoint;
|
|
false ->
|
|
bummer_NOT_DISJOINT
|
|
end,
|
|
{Epoch, {Tag, Res}}.
|
|
|
|
all_reports_are_disjoint(Report) ->
|
|
[] == [X || {_Epoch, Tuple}=X <- Report,
|
|
element(1, Tuple) /= ok_disjoint].
|
|
|
|
extract_chains_relative_to_flu(FLU, Report) ->
|
|
{FLU, [{Epoch, UPI, Repairing} ||
|
|
{Epoch, {ok_disjoint, Es}} <- Report,
|
|
{agreed_membership, {UPI, Repairing}} <- Es,
|
|
lists:member(FLU, UPI) orelse lists:member(FLU, Repairing)]}.
|
|
|
|
chain_to_projection(MyName, Epoch, UPI_list, Repairing_list, All_list) ->
|
|
exit({todo_broken_fixme,?MODULE,?LINE}),
|
|
machi_projection:new(Epoch, MyName, All_list,
|
|
All_list -- (UPI_list ++ Repairing_list),
|
|
UPI_list, Repairing_list, []).
|
|
|
|
-ifndef(PULSE).
|
|
|
|
smoke0_test() ->
|
|
{ok, _} = machi_partition_simulator:start_link({1,2,3}, 50, 50),
|
|
Host = "localhost",
|
|
TcpPort = 6623,
|
|
{ok, FLUa} = machi_flu1:start_link([{a,TcpPort,"./data.a"}]),
|
|
Pa = #p_srvr{name=a, address=Host, port=TcpPort},
|
|
Members_Dict = machi_projection:make_members_dict([Pa]),
|
|
%% Egadz, more racing on startup, yay. TODO fix.
|
|
timer:sleep(1),
|
|
{ok, FLUaP} = ?FLU_PC:start_link(Pa),
|
|
{ok, M0} = ?MGR:start_link(a, Members_Dict, [{active_mode, false}]),
|
|
_SockA = machi_util:connect(Host, TcpPort),
|
|
try
|
|
pong = ?MGR:ping(M0)
|
|
after
|
|
ok = ?MGR:stop(M0),
|
|
ok = machi_flu1:stop(FLUa),
|
|
ok = ?FLU_PC:quit(FLUaP),
|
|
ok = machi_partition_simulator:stop()
|
|
end.
|
|
|
|
smoke1_test() ->
|
|
machi_partition_simulator:start_link({1,2,3}, 100, 0),
|
|
TcpPort = 62777,
|
|
FluInfo = [{a,TcpPort+0,"./data.a"}, {b,TcpPort+1,"./data.b"}, {c,TcpPort+2,"./data.c"}],
|
|
P_s = [#p_srvr{name=Name, address="localhost", port=Port} ||
|
|
{Name,Port,_Dir} <- FluInfo],
|
|
|
|
[machi_flu1_test:clean_up_data_dir(Dir) || {_,_,Dir} <- FluInfo],
|
|
FLUs = [element(2, machi_flu1:start_link([{Name,Port,Dir}])) ||
|
|
{Name,Port,Dir} <- FluInfo],
|
|
MembersDict = machi_projection:make_members_dict(P_s),
|
|
{ok, M0} = ?MGR:start_link(a, MembersDict, [{active_mode,false}]),
|
|
try
|
|
{ok, P1} = ?MGR:test_calc_projection(M0, false),
|
|
% DERP! Check for race with manager's proxy vs. proj listener
|
|
case ?MGR:test_read_latest_public_projection(M0, false) of
|
|
{error, partition} -> timer:sleep(500);
|
|
_ -> ok
|
|
end,
|
|
{local_write_result, ok,
|
|
{remote_write_results, [{b,ok},{c,ok}]}} =
|
|
?MGR:test_write_public_projection(M0, P1),
|
|
{unanimous, P1, Extra1} = ?MGR:test_read_latest_public_projection(M0, false),
|
|
|
|
ok
|
|
after
|
|
ok = ?MGR:stop(M0),
|
|
[ok = machi_flu1:stop(X) || X <- FLUs],
|
|
ok = machi_partition_simulator:stop()
|
|
end.
|
|
|
|
nonunanimous_setup_and_fix_test() ->
|
|
%% TODO attack list:
|
|
%% __ Add start option to chain manager to be "passive" only, i.e.,
|
|
%% not immediately go to work on
|
|
%% 1. Start FLUs with full complement of FLU+proj+chmgr.
|
|
%% 2. Put each of them under a supervisor?
|
|
%% - Sup proc could be a created-specifically-for-test thing, perhaps?
|
|
%% Rather than relying on a supervisor with reg name + OTP app started
|
|
%% plus plus more more yaddayadda?
|
|
%% 3. Add projection catalog/orddict of #p_srvr records??
|
|
%% 4. Fix this test, etc etc.
|
|
machi_partition_simulator:start_link({1,2,3}, 100, 0),
|
|
TcpPort = 62877,
|
|
FluInfo = [{a,TcpPort+0,"./data.a"}, {b,TcpPort+1,"./data.b"}],
|
|
P_s = [#p_srvr{name=Name, address="localhost", port=Port} ||
|
|
{Name,Port,_Dir} <- FluInfo],
|
|
|
|
[machi_flu1_test:clean_up_data_dir(Dir) || {_,_,Dir} <- FluInfo],
|
|
FLUs = [element(2, machi_flu1:start_link([{Name,Port,Dir}])) ||
|
|
{Name,Port,Dir} <- FluInfo],
|
|
[Proxy_a, Proxy_b] = Proxies =
|
|
[element(2,?FLU_PC:start_link(P)) || P <- P_s],
|
|
MembersDict = machi_projection:make_members_dict(P_s),
|
|
XX = [],
|
|
%% XX = [{private_write_verbose,true}],
|
|
{ok, Ma} = ?MGR:start_link(a, MembersDict, [{active_mode, false}]++XX),
|
|
{ok, Mb} = ?MGR:start_link(b, MembersDict, [{active_mode, false}]++XX),
|
|
try
|
|
{ok, P1} = ?MGR:test_calc_projection(Ma, false),
|
|
|
|
P1a = machi_projection:update_checksum(
|
|
P1#projection_v1{down=[b], upi=[a], dbg=[{hackhack, ?LINE}]}),
|
|
P1b = machi_projection:update_checksum(
|
|
P1#projection_v1{author_server=b, creation_time=now(),
|
|
down=[a], upi=[b], dbg=[{hackhack, ?LINE}]}),
|
|
%% Scribble different projections
|
|
ok = ?FLU_PC:write_projection(Proxy_a, public, P1a),
|
|
ok = ?FLU_PC:write_projection(Proxy_b, public, P1b),
|
|
|
|
%% ?D(x),
|
|
{not_unanimous,_,_}=_XX = ?MGR:test_read_latest_public_projection(Ma, false),
|
|
%% ?Dw(_XX),
|
|
{not_unanimous,_,_}=_YY = ?MGR:test_read_latest_public_projection(Ma, true),
|
|
%% The read repair here doesn't automatically trigger the creation of
|
|
%% a new projection (to try to create a unanimous projection). So
|
|
%% we expect nothing to change when called again.
|
|
{not_unanimous,_,_}=_YY = ?MGR:test_read_latest_public_projection(Ma, true),
|
|
|
|
{now_using, _, EpochNum_a} = ?MGR:test_react_to_env(Ma),
|
|
{no_change, _, EpochNum_a} = ?MGR:test_react_to_env(Ma),
|
|
{unanimous,P2,_E2} = ?MGR:test_read_latest_public_projection(Ma, false),
|
|
{ok, P2pa} = ?FLU_PC:read_latest_projection(Proxy_a, private),
|
|
P2 = P2pa#projection_v1{dbg2=[]},
|
|
|
|
%% FLUb should have nothing written to private because it hasn't
|
|
%% reacted yet.
|
|
{error, not_written} = ?FLU_PC:read_latest_projection(Proxy_b, private),
|
|
|
|
%% Poke FLUb to react ... should be using the same private proj
|
|
%% as FLUa.
|
|
{now_using, _, EpochNum_a} = ?MGR:test_react_to_env(Mb),
|
|
{ok, P2pb} = ?FLU_PC:read_latest_projection(Proxy_b, private),
|
|
P2 = P2pb#projection_v1{dbg2=[]},
|
|
|
|
ok
|
|
after
|
|
ok = ?MGR:stop(Ma),
|
|
ok = ?MGR:stop(Mb),
|
|
[ok = ?FLU_PC:quit(X) || X <- Proxies],
|
|
[ok = machi_flu1:stop(X) || X <- FLUs],
|
|
ok = machi_partition_simulator:stop()
|
|
end.
|
|
|
|
-endif. % not PULSE
|
|
-endif. % TEST
|