You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2019/02/20 23:44:33 UTC
[couchdb] branch shard-split updated: Fix fabric_view_changes
find_replacements
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch shard-split
in repository https://gitbox.apache.org/repos/asf/couchdb.git
The following commit(s) were added to refs/heads/shard-split by this push:
new c85a711 Fix fabric_view_changes find_replacements
c85a711 is described below
commit c85a71125aa05bde56a493c6e5521e9f89e2fc32
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Wed Feb 20 18:43:16 2019 -0500
Fix fabric_view_changes find_replacements
Previously it was not properly prioritizing workers over available shards.
Also add unit tests to show that it works.
---
src/fabric/src/fabric_view.erl | 94 ++++++---------------
src/fabric/src/fabric_view_changes.erl | 150 ++++++++++++++++++++++++++++++++-
2 files changed, 172 insertions(+), 72 deletions(-)
diff --git a/src/fabric/src/fabric_view.erl b/src/fabric/src/fabric_view.erl
index e913207..4f6e34c 100644
--- a/src/fabric/src/fabric_view.erl
+++ b/src/fabric/src/fabric_view.erl
@@ -16,8 +16,7 @@
remove_overlapping_shards/3, maybe_send_row/1,
transform_row/1, keydict/1, extract_view/4, get_shards/2,
check_down_shards/2, handle_worker_exit/3,
- get_shard_replacements/2, maybe_update_others/5,
- find_replacements/2
+ get_shard_replacements/2, maybe_update_others/5
]).
-export([fix_skip_and_limit/1]).
@@ -388,8 +387,9 @@ get_shard_replacements(DbName, UsedShards0) ->
% that aren't already used.
AllLiveShards = mem3:live_shards(DbName, [node() | nodes()]),
UsedShards = [S#shard{ref=undefined} || S <- UsedShards0],
- UnusedShards = AllLiveShards -- UsedShards,
+ get_shard_replacements_int(AllLiveShards -- UsedShards, UsedShards).
+get_shard_replacements_int(UnusedShards, UsedShards) ->
% If we have more than one copy of a range then we don't
% want to try and add a replacement to any copy.
RangeCounts = lists:foldl(fun(#shard{range=R}, Acc) ->
@@ -429,71 +429,6 @@ remove_finalizer(Args) ->
couch_mrview_util:set_extra(Args, finalizer, null).
-find_replacements(Workers, AllShards) ->
- % Build map [B, E] => [Worker1, Worker2, ...] for all workers
- WrkMap = lists:foldl(fun({#shard{range = [B, E]}, _} = W, Acc) ->
- maps:update_with({B, E}, fun(Ws) -> [W | Ws] end, [W], Acc)
- end, #{}, fabric_dict:to_list(Workers)),
-
- % Build map [B, E] => [Shard1, Shard2, ...] for all shards
- AllMap = lists:foldl(fun(#shard{range = [B, E]} = S, Acc) ->
- maps:update_with({B, E}, fun(Ss) -> [S | Ss] end, [S], Acc)
- end, #{}, AllShards),
- % The sort function will prioritize workers over other shards.
- % The idea is to not unnecessarily kill workers if we don't have to
- SortFun = fun
- (R1 = {B, E1}, R2 = {B, E2}) ->
- case {maps:is_key(R1, WrkMap), maps:is_key(R2, AllMap)} of
- {true, true} ->
- % Both are workers, larger interval wins
- E1 >= E2;
- {true, false} ->
- % First element is a worker range, it wins
- true;
- {false, true} ->
- % Second element is a worker range, it wins
- false;
- {false, false} ->
- % Neither one is a worker interval, pick larger one
- E1 >= E2
- end;
- ({B1, _}, {B2, _}) ->
- B1 =< B2
- end,
-
- % Call get_ring/2 with all the ranges combined and the custom sort
- % function. It returns a sequence of ranges that complete the ring. As many
- % as possible will be workers but some workers might have been replaced.
- Ring = mem3_util:get_ring(maps:keys(WrkMap) ++ maps:keys(AllMap), SortFun),
-
- % Keep only workers that are members of the ring.
- Workers1 = fabric_dict:filter(fun(#shard{range = [B, E]}, _) ->
- lists:member({B, E}, Ring)
- end, Workers),
-
- % Also return workers that had to be removed
- Removed = fabric_dict:filter(fun(#shard{range = [B, E]}, _) ->
- not lists:member({B, E}, Ring)
- end, Workers),
-
- {Rep, _} = lists:foldl(fun(R, {RepAcc, AllMapAcc}) ->
- case {maps:is_key(R, WrkMap), maps:is_key(R, AllMapAcc)} of
- {true, _} ->
- % There is a worker in that range. Don't change anything.
- {RepAcc, AllMapAcc};
- {false, true} ->
- % No worker for this range. Replace from shards
- [Shard | Rest] = maps:get(R, AllMapAcc),
- {[Shard | RepAcc], AllMapAcc#{R := Rest}}
- end
- end, {[], AllMap}, Ring),
-
- % Return the list of workers that are part of ring, list of removed workers
- % and a list of replacement shards that could be used to make sure the ring
- % completes.
- {Workers1, Removed, Rep}.
-
-
% Unit tests
is_progress_possible_test() ->
@@ -524,7 +459,7 @@ remove_overlapping_shards_test() ->
Shards = mk_cnts([[0, 10], [11, 20], [21, ?RING_END]], 3),
% Simple (exact) overlap
- Shard1 = mk_shard("node-3", [11,20]),
+ Shard1 = mk_shard("node-3", [11, 20]),
Shards1 = fabric_dict:store(Shard1, nil, Shards),
R1 = remove_overlapping_shards(Shard1, Shards1, Cb),
?assertEqual([{0, 10}, {11, 20}, {21, ?RING_END}], get_worker_ranges(R1)),
@@ -538,6 +473,27 @@ remove_overlapping_shards_test() ->
?assert(fabric_dict:is_key(Shard2, R2)).
+get_shard_replacements_test() ->
+ Unused = [mk_shard(N, [B, E]) || {N, B, E} <- [
+ {"n1", 11, 20}, {"n1", 21, ?RING_END},
+ {"n2", 0, 4}, {"n2", 5, 10}, {"n2", 11, 20},
+ {"n3", 0, 21, ?RING_END}
+ ]],
+ Used = [mk_shard(N, [B, E]) || {N, B, E} <- [
+ {"n2", 21, ?RING_END},
+ {"n3", 0, 10}, {"n3", 11, 20}
+ ]],
+ Res = lists:sort(get_shard_replacements_int(Unused, Used)),
+ % Notice that [0, 10] range can be replaces by spawning the
+ % [0, 4] and [5, 10] workers on n1
+ Expect = [
+ {[0, 10], [mk_shard("n2", [0, 4]), mk_shard("n2", [5, 10])]},
+ {[11, 20], [mk_shard("n1", [11, 20]), mk_shard("n2", [11, 20])]},
+ {[21, ?RING_END], [mk_shard("n1", [21, ?RING_END])]}
+ ],
+ ?assertEqual(Expect, Res).
+
+
mk_cnts(Ranges) ->
Shards = lists:map(fun mk_shard/1, Ranges),
orddict:from_list([{Shard,nil} || Shard <- Shards]).
diff --git a/src/fabric/src/fabric_view_changes.erl b/src/fabric/src/fabric_view_changes.erl
index d2dd634..bdabff2 100644
--- a/src/fabric/src/fabric_view_changes.erl
+++ b/src/fabric/src/fabric_view_changes.erl
@@ -128,7 +128,7 @@ send_changes(DbName, ChangesArgs, Callback, PackedSeqs, AccIn, Timeout) ->
LiveNodes = [node() | nodes()],
AllLiveShards = mem3:live_shards(DbName, LiveNodes),
Seqs0 = unpack_seqs(PackedSeqs, DbName),
- {WSeqs0, Dead, Reps} = fabric_view:find_replacements(Seqs0, AllLiveShards),
+ {WSeqs0, Dead, Reps} = find_replacements(Seqs0, AllLiveShards),
% Start workers which didn't need replacements
WSeqs = lists:map(fun({#shard{name = Name, node = N} = S, Seq}) ->
Ref = rexi:cast(N, {fabric_rpc, changes, [Name, ChangesArgs, Seq]}),
@@ -454,8 +454,7 @@ do_unpack_seqs(Opaque, DbName) ->
Unpacked;
false ->
Shards = mem3:shards(DbName),
- {Unpacked1, Dead, Reps} =
- fabric_view:find_replacements(Unpacked, Shards),
+ {Unpacked1, Dead, Reps} = find_replacements(Unpacked, Shards),
{Splits0, Reps1} = find_split_shard_replacements(Dead, Reps),
Splits1 = lists:map(fun({#shard{} = S, Seq}) ->
{S, Seq}
@@ -503,6 +502,75 @@ changes_row(Props0) ->
{change, {Props2}}.
+find_replacements(Workers, AllShards) ->
+ % Build map [B, E] => [Worker1, Worker2, ...] for all workers
+ WrkMap = lists:foldl(fun({#shard{range = [B, E]}, _} = W, Acc) ->
+ maps:update_with({B, E}, fun(Ws) -> [W | Ws] end, [W], Acc)
+ end, #{}, fabric_dict:to_list(Workers)),
+
+ % Build map [B, E] => [Shard1, Shard2, ...] for all shards
+ AllMap = lists:foldl(fun(#shard{range = [B, E]} = S, Acc) ->
+ maps:update_with({B, E}, fun(Ss) -> [S | Ss] end, [S], Acc)
+ end, #{}, AllShards),
+
+ % Custom sort function will prioritize workers over other shards.
+ % The idea is to not unnecessarily kill workers if we don't have to
+ SortFun = fun
+ (R1 = {B, E1}, R2 = {B, E2}) ->
+ case {maps:is_key(R1, WrkMap), maps:is_key(R2, WrkMap)} of
+ {true, true} ->
+ % Both are workers, larger interval wins
+ E1 >= E2;
+ {true, false} ->
+ % First element is a worker range, it wins
+ true;
+ {false, true} ->
+ % Second element is a worker range, it wins
+ false;
+ {false, false} ->
+ % Neither one is a worker interval, pick larger one
+ E1 >= E2
+ end;
+ ({B1, _}, {B2, _}) ->
+ B1 =< B2
+ end,
+ Ring = mem3_util:get_ring(maps:keys(AllMap), SortFun),
+
+ % Keep only workers in the ring and from one of the available nodes
+ Keep = fun(#shard{range = [B, E], node = N}) ->
+ lists:member({B, E}, Ring) andalso lists:keyfind(N, #shard.node,
+ maps:get({B, E}, AllMap)) =/= false
+ end,
+ Workers1 = fabric_dict:filter(fun(S, _) -> Keep(S) end, Workers),
+ Removed = fabric_dict:filter(fun(S, _) -> not Keep(S) end, Workers),
+
+ {Rep, _} = lists:foldl(fun(R, {RepAcc, AllMapAcc}) ->
+ case maps:is_key(R, WrkMap)of
+ true ->
+ % It's a worker and in the map of available shards. Make sure to
+ % keep it only if there is a range available on that node only
+ % (reuse Keep/1 predicate from above)
+ WorkersInRange = maps:get(R, WrkMap),
+ case lists:any(fun({S, _}) -> Keep(S) end, WorkersInRange) of
+ true ->
+ {RepAcc, AllMapAcc};
+ false ->
+ [Shard | Rest] = maps:get(R, AllMapAcc),
+ {[Shard | RepAcc], AllMapAcc#{R := Rest}}
+ end;
+ false ->
+ % No worker for this range. Replace from available shards
+ [Shard | Rest] = maps:get(R, AllMapAcc),
+ {[Shard | RepAcc], AllMapAcc#{R := Rest}}
+ end
+ end, {[], AllMap}, Ring),
+
+ % Return the list of workers that are part of ring, list of removed workers
+ % and a list of replacement shards that could be used to make sure the ring
+ % completes.
+ {Workers1, Removed, Rep}.
+
+
% From the list of dead workers determine if any are a result of a split shard. In that
% case perhaps there is a way to not rewind the changes feed back to 0
find_split_shard_replacements(DeadWorkers, Shards) ->
@@ -600,3 +668,79 @@ unpack_seqs_test() ->
assert_shards(Packed) ->
?assertMatch([{#shard{},_}|_], unpack_seqs(Packed, <<"foo">>)).
+
+
+find_replacements_test() ->
+ % None of the workers are in the live list of shard but there is a
+ % replacement on n3 for the full range. It should get picked instead of
+ % the two smaller one on n2.
+ Workers1 = mk_workers([{"n1", 0, 10}, {"n2", 11, ?RING_END}]),
+ AllShards1 = [
+ mk_shard("n1", 11, ?RING_END),
+ mk_shard("n2", 0, 4),
+ mk_shard("n2", 5, 10),
+ mk_shard("n3", 0, ?RING_END)
+ ],
+ {WorkersRes1, Dead1, Reps1} = find_replacements(Workers1, AllShards1),
+ ?assertEqual([], WorkersRes1),
+ ?assertEqual(Workers1, Dead1),
+ ?assertEqual([mk_shard("n3", 0, ?RING_END)], Reps1),
+
+ % None of the workers are in the live list of shards and there is a
+ % split replacement from n2 (range [0, 10] replaced with [0, 4], [5, 10])
+ Workers2 = mk_workers([{"n1", 0, 10}, {"n2", 11, ?RING_END}]),
+ AllShards2 = [
+ mk_shard("n1", 11, ?RING_END),
+ mk_shard("n2", 0, 4),
+ mk_shard("n2", 5, 10)
+ ],
+ {WorkersRes2, Dead2, Reps2} = find_replacements(Workers2, AllShards2),
+ ?assertEqual([], WorkersRes2),
+ ?assertEqual(Workers2, Dead2),
+ ?assertEqual([
+ mk_shard("n1", 11, ?RING_END),
+ mk_shard("n2", 0, 4),
+ mk_shard("n2", 5, 10)
+ ], lists:sort(Reps2)),
+
+ % One worker is available and one needs to be replaced. Replacement will be
+ % from two split shards
+ Workers3 = mk_workers([{"n1", 0, 10}, {"n2", 11, ?RING_END}]),
+ AllShards3 = [
+ mk_shard("n1", 11, ?RING_END),
+ mk_shard("n2", 0, 4),
+ mk_shard("n2", 5, 10),
+ mk_shard("n2", 11, ?RING_END)
+ ],
+ {WorkersRes3, Dead3, Reps3} = find_replacements(Workers3, AllShards3),
+ ?assertEqual(mk_workers([{"n2", 11, ?RING_END}]), WorkersRes3),
+ ?assertEqual(mk_workers([{"n1", 0, 10}]), Dead3),
+ ?assertEqual([
+ mk_shard("n2", 0, 4),
+ mk_shard("n2", 5, 10)
+ ], lists:sort(Reps3)),
+
+ % All workers are available. Make sure they are not killed even if there is
+ % a longer (single) shard to replace them.
+ Workers4 = mk_workers([{"n1", 0, 10}, {"n1", 11, ?RING_END}]),
+ AllShards4 = [
+ mk_shard("n1", 0, 10),
+ mk_shard("n1", 11, ?RING_END),
+ mk_shard("n2", 0, 4),
+ mk_shard("n2", 5, 10),
+ mk_shard("n3", 0, ?RING_END)
+ ],
+ {WorkersRes4, Dead4, Reps4} = find_replacements(Workers4, AllShards4),
+ ?assertEqual(Workers4, WorkersRes4),
+ ?assertEqual([], Dead4),
+ ?assertEqual([], Reps4).
+
+
+mk_workers(NodesRanges) ->
+ orddict:from_list([{mk_shard(N, B, E), nil} || {N, B, E} <- NodesRanges]).
+
+
+mk_shard(Name, B, E) ->
+ Node = list_to_atom(Name),
+ BName = list_to_binary(Name),
+ #shard{name=BName, node=Node, range=[B, E]}.