You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2018/05/24 16:56:31 UTC
[couchdb] 01/02: SQUERGE: fabric read-repair
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor-3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit c0b04dfbbd9491e76855c383171333f9c774a138
Author: Paul J. Davis <pa...@gmail.com>
AuthorDate: Wed May 23 12:26:31 2018 -0500
SQUERGE: fabric read-repair
---
src/fabric/src/fabric_doc_open.erl | 16 +--
src/fabric/src/fabric_doc_open_revs.erl | 24 ++---
src/fabric/src/fabric_rpc.erl | 153 +++++++++++++----------------
src/fabric/test/fabric_rpc_purge_tests.erl | 98 +++++++++---------
4 files changed, 131 insertions(+), 160 deletions(-)
diff --git a/src/fabric/src/fabric_doc_open.erl b/src/fabric/src/fabric_doc_open.erl
index 0f8e377..f363dba 100644
--- a/src/fabric/src/fabric_doc_open.erl
+++ b/src/fabric/src/fabric_doc_open.erl
@@ -25,7 +25,7 @@
r,
state,
replies,
- node_id_revs = [],
+ node_revs = [],
q_reply
}).
@@ -84,13 +84,13 @@ handle_message({rexi_EXIT, _Reason}, Worker, Acc) ->
end;
handle_message(Reply, Worker, Acc) ->
NewReplies = fabric_util:update_counter(Reply, 1, Acc#acc.replies),
- NewNodeIdRevs = case Reply of
- {ok, #doc{id = Id, revs = {Pos, [Rev | _]}}} ->
- [{Worker#shard.node, {Id, [{Pos, Rev}]}} | Acc#acc.node_id_revs];
+ NewNodeRevs = case Reply of
+ {ok, #doc{revs = {Pos, [Rev | _]}}} ->
+ [{Worker#shard.node, [{Pos, Rev}]} | Acc#acc.node_revs];
_ ->
- Acc#acc.node_id_revs
+ Acc#acc.node_revs
end,
- NewAcc = Acc#acc{replies = NewReplies, node_id_revs = NewNodeIdRevs},
+ NewAcc = Acc#acc{replies = NewReplies, node_revs = NewNodeRevs},
case is_r_met(Acc#acc.workers, NewReplies, Acc#acc.r) of
{true, QuorumReply} ->
fabric_util:cleanup(lists:delete(Worker, Acc#acc.workers)),
@@ -129,14 +129,14 @@ is_r_met(Workers, Replies, R) ->
no_more_workers
end.
-read_repair(#acc{dbname=DbName, replies=Replies, node_id_revs=NodeIdRevs}) ->
+read_repair(#acc{dbname=DbName, replies=Replies, node_revs=NodeRevs}) ->
Docs = [Doc || {_, {{ok, #doc{}=Doc}, _}} <- Replies],
case Docs of
% omit local docs from read repair
[#doc{id = <<?LOCAL_DOC_PREFIX, _/binary>>} | _] ->
choose_reply(Docs);
[#doc{id=Id} | _] ->
- Opts = [replicated_changes, ?ADMIN_CTX, {read_repair, NodeIdRevs}],
+ Opts = [?ADMIN_CTX, {read_repair, NodeRevs}],
Res = fabric:update_docs(DbName, Docs, Opts),
case Res of
{ok, []} ->
diff --git a/src/fabric/src/fabric_doc_open_revs.erl b/src/fabric/src/fabric_doc_open_revs.erl
index bb93568..0dc2677 100644
--- a/src/fabric/src/fabric_doc_open_revs.erl
+++ b/src/fabric/src/fabric_doc_open_revs.erl
@@ -29,7 +29,7 @@
revs,
latest,
replies = [],
- node_id_revs = [],
+ node_revs = [],
repair = false
}).
@@ -83,7 +83,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
worker_count = WorkerCount,
workers = Workers,
replies = PrevReplies,
- node_id_revs = PrevNodeIdRevs,
+ node_revs = PrevNodeRevs,
r = R,
revs = Revs,
latest = Latest,
@@ -109,14 +109,14 @@ handle_message({ok, RawReplies}, Worker, State) ->
{NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
{NewReplies0, MinCount >= R, false}
end,
- NewNodeIdRevs = if Worker == nil -> PrevNodeIdRevs; true ->
+ NewNodeRevs = if Worker == nil -> PrevNodeRevs; true ->
IdRevs = lists:foldl(fun
- ({ok, #doc{id = Id, revs = {Pos, [Rev | _]}}}, Acc) ->
- [{Id, {Pos, Rev}} | Acc];
+ ({ok, #doc{revs = {Pos, [Rev | _]}}}, Acc) ->
+ [{Pos, Rev} | Acc];
(_, Acc) ->
Acc
end, [], RawReplies),
- [{Worker#shard.node, IdRevs} | PrevNodeIdRevs]
+ [{Worker#shard.node, IdRevs} | PrevNodeRevs]
end,
Complete = (ReplyCount =:= (WorkerCount - 1)),
@@ -128,7 +128,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
DbName,
IsTree,
NewReplies,
- NewNodeIdRevs,
+ NewNodeRevs,
ReplyCount + 1,
InRepair orelse Repair
),
@@ -136,7 +136,7 @@ handle_message({ok, RawReplies}, Worker, State) ->
false ->
{ok, State#state{
replies = NewReplies,
- node_id_revs = NewNodeIdRevs,
+ node_revs = NewNodeRevs,
reply_count = ReplyCount + 1,
workers = lists:delete(Worker, Workers),
repair = InRepair orelse Repair
@@ -193,7 +193,7 @@ dict_replies(Dict, [Reply | Rest]) ->
dict_replies(NewDict, Rest).
-maybe_read_repair(Db, IsTree, Replies, NodeIdRevs, ReplyCount, DoRepair) ->
+maybe_read_repair(Db, IsTree, Replies, NodeRevs, ReplyCount, DoRepair) ->
Docs = case IsTree of
true -> tree_repair_docs(Replies, DoRepair);
false -> dict_repair_docs(Replies, ReplyCount)
@@ -202,7 +202,7 @@ maybe_read_repair(Db, IsTree, Replies, NodeIdRevs, ReplyCount, DoRepair) ->
[] ->
ok;
_ ->
- erlang:spawn(fun() -> read_repair(Db, Docs, NodeIdRevs) end)
+ erlang:spawn(fun() -> read_repair(Db, Docs, NodeRevs) end)
end.
@@ -221,8 +221,8 @@ dict_repair_docs(Replies, ReplyCount) ->
end.
-read_repair(Db, Docs, NodeIdRevs) ->
- Opts = [replicated_changes, ?ADMIN_CTX, {read_repair, NodeIdRevs}],
+read_repair(Db, Docs, NodeRevs) ->
+ Opts = [?ADMIN_CTX, {read_repair, NodeRevs}],
Res = fabric:update_docs(Db, Docs, Opts),
case Res of
{ok, []} ->
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index da9a936..c6f6165 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -240,19 +240,19 @@ get_missing_revs(DbName, IdRevsList, Options) ->
end).
update_docs(DbName, Docs0, Options) ->
- X = case proplists:get_value(replicated_changes, Options) of
- true -> replicated_changes;
- _ -> interactive_edit
- end,
- NodeIdRevs = couch_util:get_value(read_repair, Options),
- Docs1 = case {X, is_list(NodeIdRevs)} of
- {replicated_changes, true} ->
- read_repair_filter(DbName, NodeIdRevs, Docs0, Options);
- _ ->
- Docs0
+ {Docs1, Type} = case couch_util:get_value(read_repair, Options) of
+ NodeRevs when is_list(NodeRevs) ->
+ Filtered = read_repair_filter(DbName, Docs0, NodeRevs, Options),
+ {Filtered, replicated_changes};
+ undefined ->
+ X = case proplists:get_value(replicated_changes, Options) of
+ true -> replicated_changes;
+ _ -> interactive_edit
+ end,
+ {Docs0, X}
end,
Docs2 = make_att_readers(Docs1),
- with_db(DbName, Options, {couch_db, update_docs, [Docs2, Options, X]}).
+ with_db(DbName, Options, {couch_db, update_docs, [Docs2, Options, Type]}).
get_purge_seq(DbName, Options) ->
@@ -315,12 +315,12 @@ with_db(DbName, Options, {M,F,A}) ->
end.
-read_repair_filter(DbName, NodeIdRevs, Docs, Options) ->
+read_repair_filter(DbName, Docs, NodeRevs, Options) ->
set_io_priority(DbName, Options),
case get_or_create_db(DbName, Options) of
{ok, Db} ->
try
- filter_purged_revs(Db, NodeIdRevs, Docs)
+ read_repair_filter(Db, Docs, NodeRevs)
after
couch_db:close(Db)
end;
@@ -342,93 +342,74 @@ read_repair_filter(DbName, NodeIdRevs, Docs, Options) ->
% need to look at the purge infos that we have locally but
% have not yet sent to the remote copy.
%
-% NodeIdRevs are the list of {node(), {docid(), [rev()]}}
-% tuples passed as the read_repair option to update_docs.
-filter_purged_revs(Db, NodeIdRevs, Docs) ->
- Nodes0 = [Node || {Node, _IdRevs} <- NodeIdRevs, Node /= node()],
- Nodes = lists:usort(Nodes0),
+% NodeRevs is a list of the {node(), [rev()]} tuples passed
+% as the read_repair option to update_docs.
+read_repair_filter(Db, Docs, NodeRevs) ->
+ [#doc{id = DocId} | _] = Docs,
+ Nodes = lists:usort([Node || {Node, _} <- NodeRevs, Node /= node()]),
+ NodeSeqs = get_node_seqs(Db, Nodes),
+
+ {ok, DbPSeq} = couch_db:get_purge_seq(Db),
+ Lag = config:get_integer("couchdb", "read_repair_lag", 100),
+
+ % Filter out read-repair updates from any node that is
+ % so out of date that it would force us to scan a large
+ % number of purge infos
+ NodeFiltFun = fun({Node, _Revs}) ->
+ {Node, NodeSeq} = lists:keyfind(Node, 1, NodeSeqs),
+ NodeSeq >= DbPSeq - Lag
+ end,
+ RecentNodeRevs = lists:filter(NodeFiltFun, NodeRevs),
+
+ % For each node we scan the purge infos to filter out any
+ % revisions that have been locally purged since we last
+ % replicated to the remote node's shard copy.
+ AllowableRevs = lists:foldl(fun({Node, Revs}, RevAcc) ->
+ {Node, StartSeq} = lists:keyfind(Node, 1, NodeSeqs),
+ FoldFun = fun({_PSeq, _UUID, PDocId, PRevs}, InnerAcc) ->
+ if PDocId /= DocId -> {ok, InnerAcc}; true ->
+ {ok, InnerAcc -- PRevs}
+ end
+ end,
+ {ok, FiltRevs} = couch_db:fold_purge_infos(Db, StartSeq, FoldFun, Revs),
+ lists:usort(FiltRevs ++ RevAcc)
+ end, [], RecentNodeRevs),
+
+ % Finally, filter the doc updates to only include revisions
+ % that have not been purged locally.
+ DocFiltFun = fun(#doc{revs = {Pos, [Rev | _]}}) ->
+ lists:member({Pos, Rev}, AllowableRevs)
+ end,
+ lists:filter(DocFiltFun, Docs).
+
+get_node_seqs(Db, Nodes) ->
% Gather the list of {Node, PurgeSeq} pairs for all nodes
% that are present in our read repair group
- StartKey = <<?LOCAL_DOC_PREFIX, "/purge-mem3-">>,
- Opts = [{start_key, StartKey}],
FoldFun = fun(#doc{id = Id, body = {Props}}, Acc) ->
case Id of
<<?LOCAL_DOC_PREFIX, "purge-mem3-", _/binary>> ->
- TargetNodeBin = couch_util:get_value(<<"target_node">>, Props),
+ TgtNode = couch_util:get_value(<<"target_node">>, Props),
PurgeSeq = couch_util:get_value(<<"purge_seq">>, Props),
- try
- TargetNode = binary_to_existing_atom(TargetNodeBin, latin1),
- case lists:member(TargetNode, Nodes) of
- true ->
- {ok, [{TargetNode, PurgeSeq} | Acc]};
- false ->
- {ok, Acc}
- end
- catch error:badarg ->
- % A really old doc referring to a node that's
- % no longer in the cluster
- {ok, Acc}
+ case lists:keyfind(TgtNode, 1, Acc) of
+ {_, OldSeq} ->
+ NewSeq = erlang:max(OldSeq, PurgeSeq),
+ NewEntry = {TgtNode, NewSeq},
+ NewAcc = lists:keyreplace(TgtNode, 1, Acc, NewEntry),
+ {ok, NewAcc};
+ false ->
+ {ok, Acc}
end;
_ ->
% We've processed all _local mem3 purge docs
{stop, Acc}
end
end,
- {ok, NodeSeqs} = couch_db:fold_local_docs(Db, FoldFun, [], Opts),
+ InitAcc = [{list_to_binary(atom_to_list(Node)), 0} || Node <- Nodes],
+ Opts = [{start_key, <<?LOCAL_DOC_PREFIX, "/purge-mem3-">>}],
+ {ok, NodeBinSeqs} = couch_db:fold_local_docs(Db, FoldFun, InitAcc, Opts),
+ [{list_to_existing_atom(binary_to_list(N)), S} || {N, S} <- NodeBinSeqs].
- {ok, DbPSeq} = couch_db:get_purge_seq(Db),
- Lag = config:get_integer("couchdb", "read_repair_lag", 100),
-
- CheckSeqFun = fun({Node, IdRevs}, {GoodToGo, MaybeGood}) ->
- NodeSeq = case lists:keyfind(Node, 1, NodeSeqs) of
- {Node, PS} -> PS;
- false -> 0
- end,
- case NodeSeq of
- DbPSeq ->
- {DocId, Revs} = IdRevs,
- NewGTG = [{DocId, Rev} || Rev <- Revs] ++ GoodToGo,
- {NewGTG, MaybeGood};
- _ when NodeSeq >= DbPSeq - Lag ->
- {GoodToGo, [{NodeSeq, IdRevs} | MaybeGood]};
- _ ->
- % The remote node `Node` is so far out of date
- % we'll just ignore its read-repair updates rather
- % than scan an unbounded number of purge infos
- {GoodToGo, MaybeGood}
- end
- end,
- {TotesGood, NeedChecking} = lists:foldl(CheckSeqFun, {[], []}, NodeIdRevs),
-
- % For any node that's not up to date with internal
- % replication we have to check if any of the revisions
- % have been purged before running our updates
- RestGood = if NeedChecking == [] -> []; true ->
- CheckFoldFun = fun({PSeq, _UUID, DocId, Revs}, Acc) ->
- FilterFun = fun({NS, {FiltDocId, FiltRev}}) ->
- % The `NS =< PSeq` portion of this translates to the
- % fact that we haven't yet replicated PSeq to the
- % target node, hence we would need to filter this read
- % repair update or risk undoing a purge operation.
- NS =< PSeq andalso FiltDocId == DocId
- andalso lists:member(FiltRev, Revs)
- end,
- {ok, lists:filter(FilterFun, Acc)}
- end,
- StartSeq = lists:min([S || {S, _} <- NeedChecking]),
- {ok, Result} = couch_db:fold_purge_infos(
- Db, StartSeq, CheckFoldFun, NeedChecking),
- [{DocId, Rev} || {_NSeq, DocId, Rev} <- Result]
- end,
-
- % Finally, only return docs that have a revision that
- % has not been filtered out of the initial set
- AllGood = lists:usort(TotesGood ++ RestGood),
- DocFiltFun = fun(#doc{id = Id, revs = {Pos, [Rev | _]}}) ->
- lists:member({Id, {Pos, Rev}}, AllGood)
- end,
- lists:filter(DocFiltFun, Docs).
get_or_create_db(DbName, Options) ->
diff --git a/src/fabric/test/fabric_rpc_purge_tests.erl b/src/fabric/test/fabric_rpc_purge_tests.erl
index 0a909cd..953913b 100644
--- a/src/fabric/test/fabric_rpc_purge_tests.erl
+++ b/src/fabric/test/fabric_rpc_purge_tests.erl
@@ -21,6 +21,7 @@
% TODO: Add tests:
% - filter some updates
+% - allow for an update that was filtered by a node
% - ignore lagging nodes
main_test_() ->
@@ -44,12 +45,21 @@ main_test_() ->
fun teardown_single_purge/1,
lists:map(fun wrap/1, [
?TDEF(t_filter),
- ?TDEF(t_no_filter_interactive_edit),
- ?TDEF(t_no_filter_unknown_node),
+ ?TDEF(t_filter_unknown_node),
+ ?TDEF(t_no_filter_old_node),
+ ?TDEF(t_no_filter_different_node),
+ ?TDEF(t_no_filter_after_repl)
+ ])
+ },
+ {
+ foreach,
+ fun setup_multi_purge/0,
+ fun teardown_multi_purge/1,
+ lists:map(fun wrap/1, [
+ ?TDEF(t_filter),
+ ?TDEF(t_filter_unknown_node),
?TDEF(t_no_filter_old_node),
?TDEF(t_no_filter_different_node),
- ?TDEF(t_no_filter_different_docid),
- ?TDEF(t_no_filter_different_rev),
?TDEF(t_no_filter_after_repl)
])
}
@@ -80,13 +90,28 @@ setup_single_purge() ->
DocId = <<"0003">>,
{ok, OldDoc} = open_doc(DbName, DocId),
purge_doc(DbName, DocId),
- {DbName, DocId, OldDoc}.
+ {DbName, DocId, OldDoc, 1}.
-teardown_single_purge({DbName, _, _}) ->
+teardown_single_purge({DbName, _, _, _}) ->
teardown_no_purge(DbName).
+setup_multi_purge() ->
+ DbName = setup_no_purge(),
+ DocId = <<"0003">>,
+ {ok, OldDoc} = open_doc(DbName, DocId),
+ lists:foreach(fun(I) ->
+ PDocId = iolist_to_binary(io_lib:format("~4..0b", [I])),
+ purge_doc(DbName, PDocId)
+ end, lists:seq(1, 5)),
+ {DbName, DocId, OldDoc, 3}.
+
+
+teardown_multi_purge(Ctx) ->
+ teardown_single_purge(Ctx).
+
+
t_no_purge_no_filter(DbName) ->
DocId = <<"0003">>,
@@ -100,7 +125,7 @@ t_no_purge_no_filter(DbName) ->
?assert(CurrDoc == NewDoc).
-t_filter({DbName, DocId, OldDoc}) ->
+t_filter({DbName, DocId, OldDoc, _PSeq}) ->
?assertEqual({not_found, missing}, open_doc(DbName, DocId)),
create_purge_checkpoint(DbName, 0),
@@ -109,24 +134,21 @@ t_filter({DbName, DocId, OldDoc}) ->
?assertEqual({not_found, missing}, open_doc(DbName, DocId)).
-t_no_filter_interactive_edit(_) ->
- ok.
-
-
-t_no_filter_unknown_node({DbName, DocId, OldDoc}) ->
+t_filter_unknown_node({DbName, DocId, OldDoc, _PSeq}) ->
+ % Unknown nodes are assumed to start at PurgeSeq = 0
?assertEqual({not_found, missing}, open_doc(DbName, DocId)),
create_purge_checkpoint(DbName, 0),
{Pos, [Rev | _]} = OldDoc#doc.revs,
- RROpt = {read_repair, [{'blargh@127.0.0.1', {DocId, [{Pos, Rev}]}}]},
+ RROpt = {read_repair, [{'blargh@127.0.0.1', [{Pos, Rev}]}]},
rpc_update_doc(DbName, OldDoc, [RROpt]),
- ?assertEqual({ok, OldDoc}, open_doc(DbName, DocId)).
+ ?assertEqual({not_found, missing}, open_doc(DbName, DocId)).
-t_no_filter_old_node({DbName, DocId, OldDoc}) ->
+t_no_filter_old_node({DbName, DocId, OldDoc, PSeq}) ->
?assertEqual({not_found, missing}, open_doc(DbName, DocId)),
- create_purge_checkpoint(DbName, 1),
+ create_purge_checkpoint(DbName, PSeq),
% The random UUID is to generate a badarg exception when
% we try and convert it to an existing atom.
@@ -137,9 +159,9 @@ t_no_filter_old_node({DbName, DocId, OldDoc}) ->
?assertEqual({ok, OldDoc}, open_doc(DbName, DocId)).
-t_no_filter_different_node({DbName, DocId, OldDoc}) ->
+t_no_filter_different_node({DbName, DocId, OldDoc, PSeq}) ->
?assertEqual({not_found, missing}, open_doc(DbName, DocId)),
- create_purge_checkpoint(DbName, 1),
+ create_purge_checkpoint(DbName, PSeq),
% Create a valid purge for a different node
TgtNode = list_to_binary(atom_to_list('notfoo@127.0.0.1')),
@@ -150,37 +172,9 @@ t_no_filter_different_node({DbName, DocId, OldDoc}) ->
?assertEqual({ok, OldDoc}, open_doc(DbName, DocId)).
-% This really shows that I should rewrite how read_repair
-% works as we'll never actually call this with different
-% doc ids or even multiple docids
-t_no_filter_different_docid({DbName, DocId, OldDoc}) ->
- ?assertEqual({not_found, missing}, open_doc(DbName, DocId)),
- create_purge_checkpoint(DbName, 0),
-
- {Pos, [Rev | _]} = OldDoc#doc.revs,
- RROpt = {read_repair, [{tgt_node(), {<<"bang">>, [{Pos, Rev}]}}]},
- rpc_update_doc(DbName, OldDoc, [RROpt]),
-
- ?assertEqual({ok, OldDoc}, open_doc(DbName, DocId)).
-
-
-% Similar to the different docid rev we're not going to be
-% passing a different doc update with a different revision
-% to the read_repair function
-t_no_filter_different_rev({DbName, DocId, OldDoc}) ->
- ?assertEqual({not_found, missing}, open_doc(DbName, DocId)),
- create_purge_checkpoint(DbName, 0),
-
- Rev = crypto:hash(md5, couch_uuids:random()),
- RROpt = {read_repair, [{tgt_node(), {DocId, [{3, Rev}]}}]},
- rpc_update_doc(DbName, OldDoc, [RROpt]),
-
- ?assertEqual({ok, OldDoc}, open_doc(DbName, DocId)).
-
-
-t_no_filter_after_repl({DbName, DocId, OldDoc}) ->
+t_no_filter_after_repl({DbName, DocId, OldDoc, PSeq}) ->
?assertEqual({not_found, missing}, open_doc(DbName, DocId)),
- create_purge_checkpoint(DbName, 1),
+ create_purge_checkpoint(DbName, PSeq),
rpc_update_doc(DbName, OldDoc),
@@ -259,15 +253,11 @@ create_purge_checkpoint(DbName, PurgeSeq, TgtNode) when is_binary(TgtNode) ->
rpc_update_doc(DbName, Doc) ->
{Pos, [Rev | _]} = Doc#doc.revs,
- RROpt = {read_repair, [{tgt_node(), {Doc#doc.id, [{Pos, Rev}]}}]},
+ RROpt = {read_repair, [{tgt_node(), [{Pos, Rev}]}]},
rpc_update_doc(DbName, Doc, [RROpt]).
-rpc_update_doc(DbName, Doc, Opts0) ->
- Opts = case lists:member(interactive_edit, Opts0) of
- true -> Opts0;
- false -> [replicated_changes | Opts0]
- end,
+rpc_update_doc(DbName, Doc, Opts) ->
Ref = erlang:make_ref(),
put(rexi_from, {self(), Ref}),
fabric_rpc:update_docs(DbName, [Doc], Opts),
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.