You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@couchdb.apache.org by davisp <gi...@git.apache.org> on 2016/04/21 22:44:54 UTC
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
GitHub user davisp opened a pull request:
https://github.com/apache/couchdb-fabric/pull/47
Fix fabric_doc_open_revs
When a user specified multiple revisions on a single branch to
fabric_doc_open_revs it would throw a function clause exception in
lists:zipwith/3. This was due to a bad assumption that there would only
ever be exactly one revision for every input revision.
Due to the possibility of having zero or more revisions for a given
revision when using latest=true this code had to be changed fairly
significantly.
COUCHDB-2863
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/cloudant/couchdb-fabric 2863-fix-fabric-doc-open-revs
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/couchdb-fabric/pull/47.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #47
----
commit a1b252e6c40e746ae52d72b371c51f53228c7c52
Author: Paul J. Davis <pa...@gmail.com>
Date: 2016-04-21T20:42:27Z
Fix fabric_doc_open_revs
When a user specified multiple revisions on a single branch to
fabric_doc_open_revs it would throw a function clause exception in
lists:zipwith/3. This was due to a bad assumption that there would only
ever be exactly one revision for every input revision.
Due to the possibility of having zero or more revisions for a given
revision when using latest=true this code had to be changed fairly
significantly.
COUCHDB-2863
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by davisp <gi...@git.apache.org>.
Github user davisp commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r63547840
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
+ SameNumRevs = length(RawReplies) == NumLeafs,
+ QMet = AllInternal andalso SameNumRevs andalso ReplyCount + 1 >= R,
+ {NewReplies0, QMet, Repair0};
+ false ->
+ {NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
+ {NewReplies0, MinCount >= R, false}
end,
- case maybe_reply(DbName, Reduced, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers = lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
- end;
-handle_message({ok, RawReplies0}, Worker, State) ->
- % we've got an explicit revision list, but if latest=true the workers may
- % return a descendant of the requested revision. Take advantage of the
- % fact that revisions are returned in order to keep track.
- RawReplies = strip_not_found_missing(RawReplies0),
- #state{
- dbname = DbName,
- reply_count = ReplyCount,
- worker_count = WorkerCount,
- workers = Workers,
- replies = All0,
- r = R
- } = State,
- All = lists:zipwith(fun({Rev, D}, Reply) ->
- if Reply =:= error -> {Rev, D}; true ->
- {Rev, fabric_util:update_counter(Reply, 1, D)}
- end
- end, All0, RawReplies),
- Reduced = [fabric_util:remove_ancestors(X, []) || {_, X} <- All],
- FinalReplies = [choose_winner(X, R) || X <- Reduced, X =/= []],
+
Complete = (ReplyCount =:= (WorkerCount - 1)),
- case is_repair_needed(All, FinalReplies) of
- true ->
- Repair = [D || {_,{{ok,D}, _}} <- lists:flatten(Reduced)];
- false ->
- Repair = false
- end,
- case maybe_reply(DbName, FinalReplies, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers=lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
+
+ case QuorumMet orelse Complete of
+ true ->
+ fabric_util:cleanup(lists:delete(Worker, Workers)),
+ maybe_read_repair(
+ DbName,
+ IsTree,
+ NewReplies,
+ ReplyCount + 1,
+ InRepair orelse Repair
+ ),
+ {stop, format_reply(IsTree, NewReplies)};
+ false ->
+ {ok, State#state{
+ replies = NewReplies,
+ reply_count = ReplyCount + 1,
+ workers = lists:delete(Worker, Workers),
+ repair = InRepair orelse Repair
+ }}
+ end.
+
+
+tree_replies(RevTree, []) ->
+ {RevTree, true, false};
+
+tree_replies(RevTree0, [{ok, Doc} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Path = couch_doc:to_path(Doc),
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, Repair};
+ {RevTree2, new_leaf} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, true}
+ end;
+
+tree_replies(RevTree0, [{{not_found, missing}, {Pos, Rev}} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Node = {Rev, ?REV_MISSING, []},
+ Path = {Pos, Node},
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, Repair}
end.
-skip(#state{revs=all} = State) ->
- handle_message({ok, []}, nil, State);
-skip(#state{revs=Revs} = State) ->
- handle_message({ok, [error || _Rev <- Revs]}, nil, State).
-
-maybe_reply(_, [], false, _, _) ->
- noreply;
-maybe_reply(_, [], true, _, _) ->
- {reply, {ok, []}};
-maybe_reply(DbName, ReplyDict, Complete, RepairDocs, R) ->
- case Complete orelse lists:all(fun({_,{_, C}}) -> C >= R end, ReplyDict) of
- true ->
- maybe_execute_read_repair(DbName, RepairDocs),
- {reply, unstrip_not_found_missing(extract_replies(ReplyDict))};
- false ->
- noreply
+
+tree_sort(Replies) ->
+ SortFun = fun(A, B) -> sort_key(A) =< sort_key(B) end,
+ lists:sort(SortFun, Replies).
+
+
+sort_key({ok, #doc{revs = {Pos, [Rev | _]}}}) ->
+ {Pos, Rev};
+sort_key({{not_found, _}, {Pos, Rev}}) ->
+ {Pos, Rev}.
+
+
+dict_replies(Dict, []) ->
+ Counts = [Count || {_Key, {_Reply, Count}} <- Dict],
+ {Dict, lists:min(Counts)};
+
+dict_replies(Dict, [Reply | Rest]) ->
+ NewDict = fabric_util:update_counter(Reply, 1, Dict),
+ dict_replies(NewDict, Rest).
+
+
+maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+ Docs = case IsTree of
+ true -> tree_repair_docs(Replies, DoRepair);
+ false -> dict_repair_docs(Replies, ReplyCount)
+ end,
+ case length(Docs) > 0 of
+ true ->
+ erlang:spawn(fun() -> read_repair(Db, Docs) end);
+ false ->
+ ok
end.
-extract_replies(Replies) ->
- lists:map(fun({_,{Reply,_}}) -> Reply end, Replies).
-
-choose_winner(Options, R) ->
- case lists:dropwhile(fun({_,{_Reply, C}}) -> C < R end, Options) of
- [] ->
- case [Elem || {_,{{ok, #doc{}}, _}} = Elem <- Options] of
- [] ->
- hd(Options);
- Docs ->
- lists:last(lists:sort(Docs))
- end;
- [QuorumMet | _] ->
- QuorumMet
+
+tree_repair_docs(_Replies, false) ->
+ [];
+
+tree_repair_docs(Replies, true) ->
+ Leafs = couch_key_tree:get_all_leafs(Replies),
+ [Doc || {Doc, {_Pos, _}} <- Leafs, is_record(Doc, doc)].
+
+
+dict_repair_docs(Replies, ReplyCount) ->
+ NeedsRepair = lists:any(fun({_, {_, C}}) -> C < ReplyCount end, Replies),
+ if not NeedsRepair -> []; true ->
+ [Doc || {_, {{ok, Doc}, _}} <- Replies]
end.
-% repair needed if any reply other than the winner has been received for a rev
-is_repair_needed([], []) ->
- false;
-is_repair_needed([{_Rev, [Reply]} | Tail1], [Reply | Tail2]) ->
- is_repair_needed(Tail1, Tail2);
-is_repair_needed(_, _) ->
- true.
-
-maybe_execute_read_repair(_Db, false) ->
- ok;
-maybe_execute_read_repair(Db, Docs) ->
- [#doc{id=Id} | _] = Docs,
+
+read_repair(Db, Docs) ->
Res = fabric:update_docs(Db, Docs, [replicated_changes, ?ADMIN_CTX]),
case Res of
{ok, []} ->
couch_stats:increment_counter([fabric, read_repairs, success]);
_ ->
couch_stats:increment_counter([fabric, read_repairs, failure]),
+ [#doc{id=Id} | _] = Docs,
couch_log:notice("read_repair ~s ~s ~p", [Db, Id, Res])
end.
-% hackery required so that not_found sorts first
-strip_not_found_missing([]) ->
- [];
-strip_not_found_missing([{{not_found, missing}, Rev} | Rest]) ->
- [{not_found, Rev} | strip_not_found_missing(Rest)];
-strip_not_found_missing([Else | Rest]) ->
- [Else | strip_not_found_missing(Rest)].
-unstrip_not_found_missing([]) ->
- [];
-unstrip_not_found_missing([{not_found, Rev} | Rest]) ->
- [{{not_found, missing}, Rev} | unstrip_not_found_missing(Rest)];
-unstrip_not_found_missing([Else | Rest]) ->
- [Else | unstrip_not_found_missing(Rest)].
+format_reply(true, Replies) ->
+ tree_format_replies(Replies);
+
+format_reply(false, Replies) ->
+ dict_format_replies(Replies).
+
+
+tree_format_replies(RevTree) ->
+ Leafs = couch_key_tree:get_all_leafs(RevTree),
+ lists:sort(lists:map(fun(Reply) ->
+ case Reply of
+ {?REV_MISSING, {Pos, [Rev]}} ->
+ {{not_found, missing}, {Pos, Rev}};
+ {Doc, _} when is_record(Doc, doc) ->
+ {ok, Doc}
+ end
+ end, Leafs)).
+
+
+dict_format_replies(Dict) ->
+ lists:sort([Reply || {_, {Reply, _}} <- Dict]).
+
+
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
-all_revs_test() ->
+setup() ->
config:start_link([]),
- meck:new([fabric, couch_stats]),
+ meck:new([fabric, couch_stats, couch_log]),
meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:new(couch_log),
- meck:expect(couch_log, notice, fun(_,_) -> ok end),
-
- State0 = #state{worker_count = 3, workers=[nil,nil,nil], r = 2, revs = all},
- Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
- Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
- Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
-
- % an empty worker response does not count as meeting quorum
- ?assertMatch(
- {ok, #state{workers=[nil,nil]}},
- handle_message({ok, []}, nil, State0)
- ),
-
- ?assertMatch(
- {ok, #state{workers=[nil, nil]}},
- handle_message({ok, [Foo1, Bar1]}, nil, State0)
- ),
- {ok, State1} = handle_message({ok, [Foo1, Bar1]}, nil, State0),
-
- % the normal case - workers agree
- ?assertEqual(
- {stop, [Bar1, Foo1]},
- handle_message({ok, [Foo1, Bar1]}, nil, State1)
- ),
-
- % a case where the 2nd worker has a newer Foo - currently we're considering
- % Foo to have reached quorum and execute_read_repair()
- ?assertEqual(
- {stop, [Bar1, Foo2]},
- handle_message({ok, [Foo2, Bar1]}, nil, State1)
- ),
-
- % a case where quorum has not yet been reached for Foo
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Bar1]}, nil, State1)
- ),
- {ok, State2} = handle_message({ok, [Bar1]}, nil, State1),
-
- % still no quorum, but all workers have responded. We include Foo1 in the
- % response and execute_read_repair()
- ?assertEqual(
- {stop, [Bar1, Foo1]},
- handle_message({ok, [Bar1]}, nil, State2)
- ),
- meck:unload([fabric, couch_log, couch_stats]),
+ meck:expect(couch_log, notice, fun(_, _) -> ok end).
+
+
+teardown(_) ->
+ (catch meck:unload([fabric, couch_stats, couch_log])),
config:stop().
-specific_revs_test() ->
- config:start_link([]),
- meck:new([fabric, couch_stats]),
- meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
- meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:new(couch_log),
- meck:expect(couch_log, notice, fun(_,_) -> ok end),
- Revs = [{1,<<"foo">>}, {1,<<"bar">>}, {1,<<"baz">>}],
- State0 = #state{
+state0(Revs, Latest) ->
+ #state{
worker_count = 3,
- workers = [nil, nil, nil],
+ workers = [w1, w2, w3],
r = 2,
revs = Revs,
- latest = false,
- replies = [{Rev,[]} || Rev <- Revs]
- },
- Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
- Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
- Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
- Baz1 = {{not_found, missing}, {1,<<"baz">>}},
- Baz2 = {ok, #doc{revs = {1, [<<"baz">>]}}},
-
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0)
- ),
- {ok, State1} = handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0),
-
- % the normal case - workers agree
- ?assertEqual(
- {stop, [Foo1, Bar1, Baz1]},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1)
- ),
-
- % latest=true, worker responds with Foo2 and we return it
- State0L = State0#state{latest = true},
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L)
- ),
- {ok, State1L} = handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L),
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz1]},
- handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State1L)
- ),
-
- % Foo1 is included in the read quorum for Foo2
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz1]},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1L)
- ),
-
- % {not_found, missing} is included in the quorum for any found revision
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz2]},
- handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State1L)
- ),
-
- % a worker failure is skipped
- ?assertMatch(
- {ok, #state{}},
- handle_message({rexi_EXIT, foo}, nil, State1L)
- ),
- {ok, State2L} = handle_message({rexi_EXIT, foo}, nil, State1L),
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz2]},
- handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State2L)
- ),
- meck:unload([fabric, couch_log, couch_stats]),
- config:stop().
+ latest = Latest
+ }.
+
+
+revs() -> [{1,<<"foo">>}, {1,<<"bar">>}, {1,<<"baz">>}].
+
+
+foo1() -> {ok, #doc{revs = {1, [<<"foo">>]}}}.
+foo2() -> {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}}.
+bar1() -> {ok, #doc{revs = {1, [<<"bar">>]}}}.
+bazNF() -> {{not_found, missing}, {1,<<"baz">>}}.
+baz1() -> {ok, #doc{revs = {1, [<<"baz">>]}}}.
+
+
+
+open_doc_revs_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
--- End diff --
I don't think that's something we should guarantee. It may be a result of the current implementation but may end up changing if we ever change it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60657132
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,391 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ case IsTree of
--- End diff --
This would work and it is known pattern in erlang community. But it binds values inside the case forcing the reader to read the case statement in all it's entirety. In order to figure out the source. Wouldn't it be better to write it as:
```
{NewReplies, QuorumMet, Repair} =
case IsTree of
true ->
...
{NewReplies, QuorumMet, Repair};
false ->
...
{NewReplies, MinCount >= R, false}
end
```
However this a matter of a personal opinion so feel free to ignore.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60653902
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,391 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ case IsTree of
+ true ->
+ {NewReplies, AllInternal, Repair} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
--- End diff --
indentation
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60668402
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
--- End diff --
Would `couch_key_tree:count_leafs/1` work here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by rnewson <gi...@git.apache.org>.
Github user rnewson commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r63328874
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
--- End diff --
what's the verdict then, @davisp?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by kxepal <gi...@git.apache.org>.
Github user kxepal commented on the pull request:
https://github.com/apache/couchdb-fabric/pull/47#issuecomment-213891700
Awesome! Thank you, @davisp!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by davisp <gi...@git.apache.org>.
Github user davisp commented on the pull request:
https://github.com/apache/couchdb-fabric/pull/47#issuecomment-213107256
This requires these two PRs to be merged:
https://github.com/apache/couchdb-couch/pull/166
https://github.com/apache/couchdb-couch/pull/167
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by rnewson <gi...@git.apache.org>.
Github user rnewson commented on the pull request:
https://github.com/apache/couchdb-fabric/pull/47#issuecomment-219493529
the dependent PR's are merged.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60671795
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
+ SameNumRevs = length(RawReplies) == NumLeafs,
+ QMet = AllInternal andalso SameNumRevs andalso ReplyCount + 1 >= R,
+ {NewReplies0, QMet, Repair0};
+ false ->
+ {NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
+ {NewReplies0, MinCount >= R, false}
end,
- case maybe_reply(DbName, Reduced, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers = lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
- end;
-handle_message({ok, RawReplies0}, Worker, State) ->
- % we've got an explicit revision list, but if latest=true the workers may
- % return a descendant of the requested revision. Take advantage of the
- % fact that revisions are returned in order to keep track.
- RawReplies = strip_not_found_missing(RawReplies0),
- #state{
- dbname = DbName,
- reply_count = ReplyCount,
- worker_count = WorkerCount,
- workers = Workers,
- replies = All0,
- r = R
- } = State,
- All = lists:zipwith(fun({Rev, D}, Reply) ->
- if Reply =:= error -> {Rev, D}; true ->
- {Rev, fabric_util:update_counter(Reply, 1, D)}
- end
- end, All0, RawReplies),
- Reduced = [fabric_util:remove_ancestors(X, []) || {_, X} <- All],
- FinalReplies = [choose_winner(X, R) || X <- Reduced, X =/= []],
+
Complete = (ReplyCount =:= (WorkerCount - 1)),
- case is_repair_needed(All, FinalReplies) of
- true ->
- Repair = [D || {_,{{ok,D}, _}} <- lists:flatten(Reduced)];
- false ->
- Repair = false
- end,
- case maybe_reply(DbName, FinalReplies, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers=lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
+
+ case QuorumMet orelse Complete of
+ true ->
+ fabric_util:cleanup(lists:delete(Worker, Workers)),
+ maybe_read_repair(
+ DbName,
+ IsTree,
+ NewReplies,
+ ReplyCount + 1,
+ InRepair orelse Repair
+ ),
+ {stop, format_reply(IsTree, NewReplies)};
+ false ->
+ {ok, State#state{
+ replies = NewReplies,
+ reply_count = ReplyCount + 1,
+ workers = lists:delete(Worker, Workers),
+ repair = InRepair orelse Repair
+ }}
+ end.
+
+
+tree_replies(RevTree, []) ->
+ {RevTree, true, false};
+
+tree_replies(RevTree0, [{ok, Doc} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Path = couch_doc:to_path(Doc),
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, Repair};
+ {RevTree2, new_leaf} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, true}
+ end;
+
+tree_replies(RevTree0, [{{not_found, missing}, {Pos, Rev}} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Node = {Rev, ?REV_MISSING, []},
+ Path = {Pos, Node},
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, Repair}
end.
-skip(#state{revs=all} = State) ->
- handle_message({ok, []}, nil, State);
-skip(#state{revs=Revs} = State) ->
- handle_message({ok, [error || _Rev <- Revs]}, nil, State).
-
-maybe_reply(_, [], false, _, _) ->
- noreply;
-maybe_reply(_, [], true, _, _) ->
- {reply, {ok, []}};
-maybe_reply(DbName, ReplyDict, Complete, RepairDocs, R) ->
- case Complete orelse lists:all(fun({_,{_, C}}) -> C >= R end, ReplyDict) of
- true ->
- maybe_execute_read_repair(DbName, RepairDocs),
- {reply, unstrip_not_found_missing(extract_replies(ReplyDict))};
- false ->
- noreply
+
+tree_sort(Replies) ->
+ SortFun = fun(A, B) -> sort_key(A) =< sort_key(B) end,
+ lists:sort(SortFun, Replies).
+
+
+sort_key({ok, #doc{revs = {Pos, [Rev | _]}}}) ->
+ {Pos, Rev};
+sort_key({{not_found, _}, {Pos, Rev}}) ->
+ {Pos, Rev}.
+
+
+dict_replies(Dict, []) ->
+ Counts = [Count || {_Key, {_Reply, Count}} <- Dict],
+ {Dict, lists:min(Counts)};
+
+dict_replies(Dict, [Reply | Rest]) ->
+ NewDict = fabric_util:update_counter(Reply, 1, Dict),
+ dict_replies(NewDict, Rest).
+
+
+maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+ Docs = case IsTree of
+ true -> tree_repair_docs(Replies, DoRepair);
+ false -> dict_repair_docs(Replies, ReplyCount)
+ end,
+ case length(Docs) > 0 of
+ true ->
+ erlang:spawn(fun() -> read_repair(Db, Docs) end);
+ false ->
+ ok
end.
-extract_replies(Replies) ->
- lists:map(fun({_,{Reply,_}}) -> Reply end, Replies).
-
-choose_winner(Options, R) ->
- case lists:dropwhile(fun({_,{_Reply, C}}) -> C < R end, Options) of
- [] ->
- case [Elem || {_,{{ok, #doc{}}, _}} = Elem <- Options] of
- [] ->
- hd(Options);
- Docs ->
- lists:last(lists:sort(Docs))
- end;
- [QuorumMet | _] ->
- QuorumMet
+
+tree_repair_docs(_Replies, false) ->
+ [];
+
+tree_repair_docs(Replies, true) ->
+ Leafs = couch_key_tree:get_all_leafs(Replies),
+ [Doc || {Doc, {_Pos, _}} <- Leafs, is_record(Doc, doc)].
+
+
+dict_repair_docs(Replies, ReplyCount) ->
+ NeedsRepair = lists:any(fun({_, {_, C}}) -> C < ReplyCount end, Replies),
+ if not NeedsRepair -> []; true ->
+ [Doc || {_, {{ok, Doc}, _}} <- Replies]
end.
-% repair needed if any reply other than the winner has been received for a rev
-is_repair_needed([], []) ->
- false;
-is_repair_needed([{_Rev, [Reply]} | Tail1], [Reply | Tail2]) ->
- is_repair_needed(Tail1, Tail2);
-is_repair_needed(_, _) ->
- true.
-
-maybe_execute_read_repair(_Db, false) ->
- ok;
-maybe_execute_read_repair(Db, Docs) ->
- [#doc{id=Id} | _] = Docs,
+
+read_repair(Db, Docs) ->
Res = fabric:update_docs(Db, Docs, [replicated_changes, ?ADMIN_CTX]),
case Res of
{ok, []} ->
couch_stats:increment_counter([fabric, read_repairs, success]);
_ ->
couch_stats:increment_counter([fabric, read_repairs, failure]),
+ [#doc{id=Id} | _] = Docs,
--- End diff --
nitpick: Spaces `#doc{id = Id}`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60672502
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
+ SameNumRevs = length(RawReplies) == NumLeafs,
+ QMet = AllInternal andalso SameNumRevs andalso ReplyCount + 1 >= R,
+ {NewReplies0, QMet, Repair0};
+ false ->
+ {NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
+ {NewReplies0, MinCount >= R, false}
end,
- case maybe_reply(DbName, Reduced, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers = lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
- end;
-handle_message({ok, RawReplies0}, Worker, State) ->
- % we've got an explicit revision list, but if latest=true the workers may
- % return a descendant of the requested revision. Take advantage of the
- % fact that revisions are returned in order to keep track.
- RawReplies = strip_not_found_missing(RawReplies0),
- #state{
- dbname = DbName,
- reply_count = ReplyCount,
- worker_count = WorkerCount,
- workers = Workers,
- replies = All0,
- r = R
- } = State,
- All = lists:zipwith(fun({Rev, D}, Reply) ->
- if Reply =:= error -> {Rev, D}; true ->
- {Rev, fabric_util:update_counter(Reply, 1, D)}
- end
- end, All0, RawReplies),
- Reduced = [fabric_util:remove_ancestors(X, []) || {_, X} <- All],
- FinalReplies = [choose_winner(X, R) || X <- Reduced, X =/= []],
+
Complete = (ReplyCount =:= (WorkerCount - 1)),
- case is_repair_needed(All, FinalReplies) of
- true ->
- Repair = [D || {_,{{ok,D}, _}} <- lists:flatten(Reduced)];
- false ->
- Repair = false
- end,
- case maybe_reply(DbName, FinalReplies, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers=lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
+
+ case QuorumMet orelse Complete of
+ true ->
+ fabric_util:cleanup(lists:delete(Worker, Workers)),
+ maybe_read_repair(
+ DbName,
+ IsTree,
+ NewReplies,
+ ReplyCount + 1,
+ InRepair orelse Repair
+ ),
+ {stop, format_reply(IsTree, NewReplies)};
+ false ->
+ {ok, State#state{
+ replies = NewReplies,
+ reply_count = ReplyCount + 1,
+ workers = lists:delete(Worker, Workers),
+ repair = InRepair orelse Repair
+ }}
+ end.
+
+
+tree_replies(RevTree, []) ->
+ {RevTree, true, false};
+
+tree_replies(RevTree0, [{ok, Doc} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Path = couch_doc:to_path(Doc),
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, Repair};
+ {RevTree2, new_leaf} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, true}
+ end;
+
+tree_replies(RevTree0, [{{not_found, missing}, {Pos, Rev}} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Node = {Rev, ?REV_MISSING, []},
+ Path = {Pos, Node},
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, Repair}
end.
-skip(#state{revs=all} = State) ->
- handle_message({ok, []}, nil, State);
-skip(#state{revs=Revs} = State) ->
- handle_message({ok, [error || _Rev <- Revs]}, nil, State).
-
-maybe_reply(_, [], false, _, _) ->
- noreply;
-maybe_reply(_, [], true, _, _) ->
- {reply, {ok, []}};
-maybe_reply(DbName, ReplyDict, Complete, RepairDocs, R) ->
- case Complete orelse lists:all(fun({_,{_, C}}) -> C >= R end, ReplyDict) of
- true ->
- maybe_execute_read_repair(DbName, RepairDocs),
- {reply, unstrip_not_found_missing(extract_replies(ReplyDict))};
- false ->
- noreply
+
+tree_sort(Replies) ->
+ SortFun = fun(A, B) -> sort_key(A) =< sort_key(B) end,
+ lists:sort(SortFun, Replies).
+
+
+sort_key({ok, #doc{revs = {Pos, [Rev | _]}}}) ->
+ {Pos, Rev};
+sort_key({{not_found, _}, {Pos, Rev}}) ->
+ {Pos, Rev}.
+
+
+dict_replies(Dict, []) ->
+ Counts = [Count || {_Key, {_Reply, Count}} <- Dict],
+ {Dict, lists:min(Counts)};
+
+dict_replies(Dict, [Reply | Rest]) ->
+ NewDict = fabric_util:update_counter(Reply, 1, Dict),
+ dict_replies(NewDict, Rest).
+
+
+maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+ Docs = case IsTree of
+ true -> tree_repair_docs(Replies, DoRepair);
+ false -> dict_repair_docs(Replies, ReplyCount)
+ end,
+ case length(Docs) > 0 of
+ true ->
+ erlang:spawn(fun() -> read_repair(Db, Docs) end);
+ false ->
+ ok
end.
-extract_replies(Replies) ->
- lists:map(fun({_,{Reply,_}}) -> Reply end, Replies).
-
-choose_winner(Options, R) ->
- case lists:dropwhile(fun({_,{_Reply, C}}) -> C < R end, Options) of
- [] ->
- case [Elem || {_,{{ok, #doc{}}, _}} = Elem <- Options] of
- [] ->
- hd(Options);
- Docs ->
- lists:last(lists:sort(Docs))
- end;
- [QuorumMet | _] ->
- QuorumMet
+
+tree_repair_docs(_Replies, false) ->
+ [];
+
+tree_repair_docs(Replies, true) ->
+ Leafs = couch_key_tree:get_all_leafs(Replies),
+ [Doc || {Doc, {_Pos, _}} <- Leafs, is_record(Doc, doc)].
+
+
+dict_repair_docs(Replies, ReplyCount) ->
+ NeedsRepair = lists:any(fun({_, {_, C}}) -> C < ReplyCount end, Replies),
+ if not NeedsRepair -> []; true ->
+ [Doc || {_, {{ok, Doc}, _}} <- Replies]
end.
-% repair needed if any reply other than the winner has been received for a rev
-is_repair_needed([], []) ->
- false;
-is_repair_needed([{_Rev, [Reply]} | Tail1], [Reply | Tail2]) ->
- is_repair_needed(Tail1, Tail2);
-is_repair_needed(_, _) ->
- true.
-
-maybe_execute_read_repair(_Db, false) ->
- ok;
-maybe_execute_read_repair(Db, Docs) ->
- [#doc{id=Id} | _] = Docs,
+
+read_repair(Db, Docs) ->
Res = fabric:update_docs(Db, Docs, [replicated_changes, ?ADMIN_CTX]),
case Res of
{ok, []} ->
couch_stats:increment_counter([fabric, read_repairs, success]);
_ ->
couch_stats:increment_counter([fabric, read_repairs, failure]),
+ [#doc{id=Id} | _] = Docs,
couch_log:notice("read_repair ~s ~s ~p", [Db, Id, Res])
end.
-% hackery required so that not_found sorts first
-strip_not_found_missing([]) ->
- [];
-strip_not_found_missing([{{not_found, missing}, Rev} | Rest]) ->
- [{not_found, Rev} | strip_not_found_missing(Rest)];
-strip_not_found_missing([Else | Rest]) ->
- [Else | strip_not_found_missing(Rest)].
-unstrip_not_found_missing([]) ->
- [];
-unstrip_not_found_missing([{not_found, Rev} | Rest]) ->
- [{{not_found, missing}, Rev} | unstrip_not_found_missing(Rest)];
-unstrip_not_found_missing([Else | Rest]) ->
- [Else | unstrip_not_found_missing(Rest)].
+format_reply(true, Replies) ->
+ tree_format_replies(Replies);
+
+format_reply(false, Replies) ->
+ dict_format_replies(Replies).
+
+
+tree_format_replies(RevTree) ->
+ Leafs = couch_key_tree:get_all_leafs(RevTree),
+ lists:sort(lists:map(fun(Reply) ->
+ case Reply of
+ {?REV_MISSING, {Pos, [Rev]}} ->
+ {{not_found, missing}, {Pos, Rev}};
+ {Doc, _} when is_record(Doc, doc) ->
+ {ok, Doc}
+ end
+ end, Leafs)).
+
+
+dict_format_replies(Dict) ->
+ lists:sort([Reply || {_, {Reply, _}} <- Dict]).
+
+
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
-all_revs_test() ->
+setup() ->
config:start_link([]),
- meck:new([fabric, couch_stats]),
+ meck:new([fabric, couch_stats, couch_log]),
meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:new(couch_log),
- meck:expect(couch_log, notice, fun(_,_) -> ok end),
-
- State0 = #state{worker_count = 3, workers=[nil,nil,nil], r = 2, revs = all},
- Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
- Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
- Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
-
- % an empty worker response does not count as meeting quorum
- ?assertMatch(
- {ok, #state{workers=[nil,nil]}},
- handle_message({ok, []}, nil, State0)
- ),
-
- ?assertMatch(
- {ok, #state{workers=[nil, nil]}},
- handle_message({ok, [Foo1, Bar1]}, nil, State0)
- ),
- {ok, State1} = handle_message({ok, [Foo1, Bar1]}, nil, State0),
-
- % the normal case - workers agree
- ?assertEqual(
- {stop, [Bar1, Foo1]},
- handle_message({ok, [Foo1, Bar1]}, nil, State1)
- ),
-
- % a case where the 2nd worker has a newer Foo - currently we're considering
- % Foo to have reached quorum and execute_read_repair()
- ?assertEqual(
- {stop, [Bar1, Foo2]},
- handle_message({ok, [Foo2, Bar1]}, nil, State1)
- ),
-
- % a case where quorum has not yet been reached for Foo
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Bar1]}, nil, State1)
- ),
- {ok, State2} = handle_message({ok, [Bar1]}, nil, State1),
-
- % still no quorum, but all workers have responded. We include Foo1 in the
- % response and execute_read_repair()
- ?assertEqual(
- {stop, [Bar1, Foo1]},
- handle_message({ok, [Bar1]}, nil, State2)
- ),
- meck:unload([fabric, couch_log, couch_stats]),
+ meck:expect(couch_log, notice, fun(_, _) -> ok end).
+
+
+teardown(_) ->
+ (catch meck:unload([fabric, couch_stats, couch_log])),
config:stop().
-specific_revs_test() ->
- config:start_link([]),
- meck:new([fabric, couch_stats]),
- meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
- meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:new(couch_log),
- meck:expect(couch_log, notice, fun(_,_) -> ok end),
- Revs = [{1,<<"foo">>}, {1,<<"bar">>}, {1,<<"baz">>}],
- State0 = #state{
+state0(Revs, Latest) ->
+ #state{
worker_count = 3,
- workers = [nil, nil, nil],
+ workers = [w1, w2, w3],
r = 2,
revs = Revs,
- latest = false,
- replies = [{Rev,[]} || Rev <- Revs]
- },
- Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
- Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
- Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
- Baz1 = {{not_found, missing}, {1,<<"baz">>}},
- Baz2 = {ok, #doc{revs = {1, [<<"baz">>]}}},
-
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0)
- ),
- {ok, State1} = handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0),
-
- % the normal case - workers agree
- ?assertEqual(
- {stop, [Foo1, Bar1, Baz1]},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1)
- ),
-
- % latest=true, worker responds with Foo2 and we return it
- State0L = State0#state{latest = true},
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L)
- ),
- {ok, State1L} = handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L),
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz1]},
- handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State1L)
- ),
-
- % Foo1 is included in the read quorum for Foo2
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz1]},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1L)
- ),
-
- % {not_found, missing} is included in the quorum for any found revision
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz2]},
- handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State1L)
- ),
-
- % a worker failure is skipped
- ?assertMatch(
- {ok, #state{}},
- handle_message({rexi_EXIT, foo}, nil, State1L)
- ),
- {ok, State2L} = handle_message({rexi_EXIT, foo}, nil, State1L),
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz2]},
- handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State2L)
- ),
- meck:unload([fabric, couch_log, couch_stats]),
- config:stop().
+ latest = Latest
+ }.
+
+
+revs() -> [{1,<<"foo">>}, {1,<<"bar">>}, {1,<<"baz">>}].
+
+
+foo1() -> {ok, #doc{revs = {1, [<<"foo">>]}}}.
+foo2() -> {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}}.
+bar1() -> {ok, #doc{revs = {1, [<<"bar">>]}}}.
+bazNF() -> {{not_found, missing}, {1,<<"baz">>}}.
+baz1() -> {ok, #doc{revs = {1, [<<"baz">>]}}}.
+
+
+
+open_doc_revs_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
--- End diff --
It does look like we return ordered results. Should we add test to check that results are sorted?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by nickva <gi...@git.apache.org>.
Github user nickva commented on the pull request:
https://github.com/apache/couchdb-fabric/pull/47#issuecomment-213864498
+1
Verified this along with https://github.com/apache/couchdb-couch/pull/166 and https://github.com/apache/couchdb-couch/pull/167
Don't fully understand this code (if that is a hard requirement for +1-ing then disregard this). However checked EUnit tests pass, saw new ones in fabric and couch, those pas, and most importantly verified function_clause error from https://issues.apache.org/jira/browse/COUCHDB-2863 doesn't happen anymore (they still happen on master as Jira issue describes).
```
$ http 'http://adm:pass@localhost:15984/db/doc?open_revs=["1-967a00dff5e02add41819138abb3284d", "2-7051cbe5c8faecd085a3fa619e6e6337"]&latest=true'
HTTP/1.1 200 OK
Content-Type: multipart/mixed; boundary="958cf62657e616ab7d2e799d0d268bf6"
Date: Sun, 24 Apr 2016 01:04:15 GMT
Server: CouchDB/b49d069 (Erlang OTP/17)
Transfer-Encoding: chunked
--958cf62657e616ab7d2e799d0d268bf6
Content-Type: application/json
{"_id":"doc","_rev":"2-7051cbe5c8faecd085a3fa619e6e6337"}
--958cf62657e616ab7d2e799d0d268bf6--
```
Deleting the doc and getting latest works as expected as well:
```
$ http delete http://adm:pass@localhost:15984/db/doc?rev=2-7051cbe5c8faecd085a3fa619e6e6337
{
"id": "doc",
"ok": true,
"rev": "3-7379b9e515b161226c6559d90c4dc49f"
}
$ http 'http://localhost:15984/db/doc?open_revs=["1-967a00dff5e02add41819138abb3284d", "2-7051cbe5c8faecd085a3fa619e6e6337"]&latest=true'
HTTP/1.1 200 OK
Content-Type: multipart/mixed; boundary="85b637594584dc5bb88940a5513ebc74"
Date: Sun, 24 Apr 2016 01:32:20 GMT
Server: CouchDB/b49d069 (Erlang OTP/17)
Transfer-Encoding: chunked
--85b637594584dc5bb88940a5513ebc74
Content-Type: application/json
{"_id":"doc","_rev":"3-7379b9e515b161226c6559d90c4dc49f","_deleted":true}
--85b637594584dc5bb88940a5513ebc74--
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by rnewson <gi...@git.apache.org>.
Github user rnewson commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r63328918
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
+ SameNumRevs = length(RawReplies) == NumLeafs,
+ QMet = AllInternal andalso SameNumRevs andalso ReplyCount + 1 >= R,
+ {NewReplies0, QMet, Repair0};
+ false ->
+ {NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
+ {NewReplies0, MinCount >= R, false}
end,
- case maybe_reply(DbName, Reduced, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers = lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
- end;
-handle_message({ok, RawReplies0}, Worker, State) ->
- % we've got an explicit revision list, but if latest=true the workers may
- % return a descendant of the requested revision. Take advantage of the
- % fact that revisions are returned in order to keep track.
- RawReplies = strip_not_found_missing(RawReplies0),
- #state{
- dbname = DbName,
- reply_count = ReplyCount,
- worker_count = WorkerCount,
- workers = Workers,
- replies = All0,
- r = R
- } = State,
- All = lists:zipwith(fun({Rev, D}, Reply) ->
- if Reply =:= error -> {Rev, D}; true ->
- {Rev, fabric_util:update_counter(Reply, 1, D)}
- end
- end, All0, RawReplies),
- Reduced = [fabric_util:remove_ancestors(X, []) || {_, X} <- All],
- FinalReplies = [choose_winner(X, R) || X <- Reduced, X =/= []],
+
Complete = (ReplyCount =:= (WorkerCount - 1)),
- case is_repair_needed(All, FinalReplies) of
- true ->
- Repair = [D || {_,{{ok,D}, _}} <- lists:flatten(Reduced)];
- false ->
- Repair = false
- end,
- case maybe_reply(DbName, FinalReplies, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers=lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
+
+ case QuorumMet orelse Complete of
+ true ->
+ fabric_util:cleanup(lists:delete(Worker, Workers)),
+ maybe_read_repair(
+ DbName,
+ IsTree,
+ NewReplies,
+ ReplyCount + 1,
+ InRepair orelse Repair
+ ),
+ {stop, format_reply(IsTree, NewReplies)};
+ false ->
+ {ok, State#state{
+ replies = NewReplies,
+ reply_count = ReplyCount + 1,
+ workers = lists:delete(Worker, Workers),
+ repair = InRepair orelse Repair
+ }}
+ end.
+
+
+tree_replies(RevTree, []) ->
+ {RevTree, true, false};
+
+tree_replies(RevTree0, [{ok, Doc} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Path = couch_doc:to_path(Doc),
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, Repair};
+ {RevTree2, new_leaf} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, true}
+ end;
+
+tree_replies(RevTree0, [{{not_found, missing}, {Pos, Rev}} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Node = {Rev, ?REV_MISSING, []},
+ Path = {Pos, Node},
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, Repair}
end.
-skip(#state{revs=all} = State) ->
- handle_message({ok, []}, nil, State);
-skip(#state{revs=Revs} = State) ->
- handle_message({ok, [error || _Rev <- Revs]}, nil, State).
-
-maybe_reply(_, [], false, _, _) ->
- noreply;
-maybe_reply(_, [], true, _, _) ->
- {reply, {ok, []}};
-maybe_reply(DbName, ReplyDict, Complete, RepairDocs, R) ->
- case Complete orelse lists:all(fun({_,{_, C}}) -> C >= R end, ReplyDict) of
- true ->
- maybe_execute_read_repair(DbName, RepairDocs),
- {reply, unstrip_not_found_missing(extract_replies(ReplyDict))};
- false ->
- noreply
+
+tree_sort(Replies) ->
+ SortFun = fun(A, B) -> sort_key(A) =< sort_key(B) end,
+ lists:sort(SortFun, Replies).
+
+
+sort_key({ok, #doc{revs = {Pos, [Rev | _]}}}) ->
+ {Pos, Rev};
+sort_key({{not_found, _}, {Pos, Rev}}) ->
+ {Pos, Rev}.
+
+
+dict_replies(Dict, []) ->
+ Counts = [Count || {_Key, {_Reply, Count}} <- Dict],
+ {Dict, lists:min(Counts)};
+
+dict_replies(Dict, [Reply | Rest]) ->
+ NewDict = fabric_util:update_counter(Reply, 1, Dict),
+ dict_replies(NewDict, Rest).
+
+
+maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+ Docs = case IsTree of
+ true -> tree_repair_docs(Replies, DoRepair);
+ false -> dict_repair_docs(Replies, ReplyCount)
+ end,
+ case length(Docs) > 0 of
--- End diff --
agreed. @davisp?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by davisp <gi...@git.apache.org>.
Github user davisp commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60661239
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,391 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ case IsTree of
+ true ->
+ {NewReplies, AllInternal, Repair} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
--- End diff --
Its a continuation which I've always doubled indented. What would you prefer?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by davisp <gi...@git.apache.org>.
Github user davisp commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60661526
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,391 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ case IsTree of
--- End diff --
Yarp. I originally had it that way but thought it was worse than the current way. I think I'll change it back cause I tend to prefer the explicit route as well.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on the pull request:
https://github.com/apache/couchdb-fabric/pull/47#issuecomment-220324348
The tests pass locally. +1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60673770
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
+ SameNumRevs = length(RawReplies) == NumLeafs,
+ QMet = AllInternal andalso SameNumRevs andalso ReplyCount + 1 >= R,
+ {NewReplies0, QMet, Repair0};
+ false ->
+ {NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
+ {NewReplies0, MinCount >= R, false}
end,
- case maybe_reply(DbName, Reduced, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers = lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
- end;
-handle_message({ok, RawReplies0}, Worker, State) ->
- % we've got an explicit revision list, but if latest=true the workers may
- % return a descendant of the requested revision. Take advantage of the
- % fact that revisions are returned in order to keep track.
- RawReplies = strip_not_found_missing(RawReplies0),
- #state{
- dbname = DbName,
- reply_count = ReplyCount,
- worker_count = WorkerCount,
- workers = Workers,
- replies = All0,
- r = R
- } = State,
- All = lists:zipwith(fun({Rev, D}, Reply) ->
- if Reply =:= error -> {Rev, D}; true ->
- {Rev, fabric_util:update_counter(Reply, 1, D)}
- end
- end, All0, RawReplies),
- Reduced = [fabric_util:remove_ancestors(X, []) || {_, X} <- All],
- FinalReplies = [choose_winner(X, R) || X <- Reduced, X =/= []],
+
Complete = (ReplyCount =:= (WorkerCount - 1)),
- case is_repair_needed(All, FinalReplies) of
- true ->
- Repair = [D || {_,{{ok,D}, _}} <- lists:flatten(Reduced)];
- false ->
- Repair = false
- end,
- case maybe_reply(DbName, FinalReplies, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers=lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
+
+ case QuorumMet orelse Complete of
+ true ->
+ fabric_util:cleanup(lists:delete(Worker, Workers)),
+ maybe_read_repair(
+ DbName,
+ IsTree,
+ NewReplies,
+ ReplyCount + 1,
+ InRepair orelse Repair
+ ),
+ {stop, format_reply(IsTree, NewReplies)};
+ false ->
+ {ok, State#state{
+ replies = NewReplies,
+ reply_count = ReplyCount + 1,
+ workers = lists:delete(Worker, Workers),
+ repair = InRepair orelse Repair
+ }}
+ end.
+
+
+tree_replies(RevTree, []) ->
+ {RevTree, true, false};
+
+tree_replies(RevTree0, [{ok, Doc} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Path = couch_doc:to_path(Doc),
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, Repair};
+ {RevTree2, new_leaf} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, true}
+ end;
+
+tree_replies(RevTree0, [{{not_found, missing}, {Pos, Rev}} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Node = {Rev, ?REV_MISSING, []},
+ Path = {Pos, Node},
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, Repair}
end.
-skip(#state{revs=all} = State) ->
- handle_message({ok, []}, nil, State);
-skip(#state{revs=Revs} = State) ->
- handle_message({ok, [error || _Rev <- Revs]}, nil, State).
-
-maybe_reply(_, [], false, _, _) ->
- noreply;
-maybe_reply(_, [], true, _, _) ->
- {reply, {ok, []}};
-maybe_reply(DbName, ReplyDict, Complete, RepairDocs, R) ->
- case Complete orelse lists:all(fun({_,{_, C}}) -> C >= R end, ReplyDict) of
- true ->
- maybe_execute_read_repair(DbName, RepairDocs),
- {reply, unstrip_not_found_missing(extract_replies(ReplyDict))};
- false ->
- noreply
+
+tree_sort(Replies) ->
+ SortFun = fun(A, B) -> sort_key(A) =< sort_key(B) end,
+ lists:sort(SortFun, Replies).
+
+
+sort_key({ok, #doc{revs = {Pos, [Rev | _]}}}) ->
+ {Pos, Rev};
+sort_key({{not_found, _}, {Pos, Rev}}) ->
+ {Pos, Rev}.
+
+
+dict_replies(Dict, []) ->
+ Counts = [Count || {_Key, {_Reply, Count}} <- Dict],
+ {Dict, lists:min(Counts)};
+
+dict_replies(Dict, [Reply | Rest]) ->
+ NewDict = fabric_util:update_counter(Reply, 1, Dict),
+ dict_replies(NewDict, Rest).
+
+
+maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+ Docs = case IsTree of
+ true -> tree_repair_docs(Replies, DoRepair);
+ false -> dict_repair_docs(Replies, ReplyCount)
+ end,
+ case length(Docs) > 0 of
+ true ->
+ erlang:spawn(fun() -> read_repair(Db, Docs) end);
+ false ->
+ ok
end.
-extract_replies(Replies) ->
- lists:map(fun({_,{Reply,_}}) -> Reply end, Replies).
-
-choose_winner(Options, R) ->
- case lists:dropwhile(fun({_,{_Reply, C}}) -> C < R end, Options) of
- [] ->
- case [Elem || {_,{{ok, #doc{}}, _}} = Elem <- Options] of
- [] ->
- hd(Options);
- Docs ->
- lists:last(lists:sort(Docs))
- end;
- [QuorumMet | _] ->
- QuorumMet
+
+tree_repair_docs(_Replies, false) ->
+ [];
+
+tree_repair_docs(Replies, true) ->
+ Leafs = couch_key_tree:get_all_leafs(Replies),
+ [Doc || {Doc, {_Pos, _}} <- Leafs, is_record(Doc, doc)].
+
+
+dict_repair_docs(Replies, ReplyCount) ->
+ NeedsRepair = lists:any(fun({_, {_, C}}) -> C < ReplyCount end, Replies),
+ if not NeedsRepair -> []; true ->
+ [Doc || {_, {{ok, Doc}, _}} <- Replies]
end.
-% repair needed if any reply other than the winner has been received for a rev
-is_repair_needed([], []) ->
- false;
-is_repair_needed([{_Rev, [Reply]} | Tail1], [Reply | Tail2]) ->
- is_repair_needed(Tail1, Tail2);
-is_repair_needed(_, _) ->
- true.
-
-maybe_execute_read_repair(_Db, false) ->
- ok;
-maybe_execute_read_repair(Db, Docs) ->
- [#doc{id=Id} | _] = Docs,
+
+read_repair(Db, Docs) ->
Res = fabric:update_docs(Db, Docs, [replicated_changes, ?ADMIN_CTX]),
case Res of
{ok, []} ->
couch_stats:increment_counter([fabric, read_repairs, success]);
_ ->
couch_stats:increment_counter([fabric, read_repairs, failure]),
+ [#doc{id=Id} | _] = Docs,
couch_log:notice("read_repair ~s ~s ~p", [Db, Id, Res])
end.
-% hackery required so that not_found sorts first
-strip_not_found_missing([]) ->
- [];
-strip_not_found_missing([{{not_found, missing}, Rev} | Rest]) ->
- [{not_found, Rev} | strip_not_found_missing(Rest)];
-strip_not_found_missing([Else | Rest]) ->
- [Else | strip_not_found_missing(Rest)].
-unstrip_not_found_missing([]) ->
- [];
-unstrip_not_found_missing([{not_found, Rev} | Rest]) ->
- [{{not_found, missing}, Rev} | unstrip_not_found_missing(Rest)];
-unstrip_not_found_missing([Else | Rest]) ->
- [Else | unstrip_not_found_missing(Rest)].
+format_reply(true, Replies) ->
+ tree_format_replies(Replies);
+
+format_reply(false, Replies) ->
+ dict_format_replies(Replies).
+
+
+tree_format_replies(RevTree) ->
+ Leafs = couch_key_tree:get_all_leafs(RevTree),
+ lists:sort(lists:map(fun(Reply) ->
+ case Reply of
+ {?REV_MISSING, {Pos, [Rev]}} ->
+ {{not_found, missing}, {Pos, Rev}};
+ {Doc, _} when is_record(Doc, doc) ->
+ {ok, Doc}
+ end
+ end, Leafs)).
+
+
+dict_format_replies(Dict) ->
+ lists:sort([Reply || {_, {Reply, _}} <- Dict]).
+
+
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
-all_revs_test() ->
+setup() ->
config:start_link([]),
- meck:new([fabric, couch_stats]),
+ meck:new([fabric, couch_stats, couch_log]),
meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:new(couch_log),
- meck:expect(couch_log, notice, fun(_,_) -> ok end),
-
- State0 = #state{worker_count = 3, workers=[nil,nil,nil], r = 2, revs = all},
- Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
- Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
- Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
-
- % an empty worker response does not count as meeting quorum
- ?assertMatch(
- {ok, #state{workers=[nil,nil]}},
- handle_message({ok, []}, nil, State0)
- ),
-
- ?assertMatch(
- {ok, #state{workers=[nil, nil]}},
- handle_message({ok, [Foo1, Bar1]}, nil, State0)
- ),
- {ok, State1} = handle_message({ok, [Foo1, Bar1]}, nil, State0),
-
- % the normal case - workers agree
- ?assertEqual(
- {stop, [Bar1, Foo1]},
- handle_message({ok, [Foo1, Bar1]}, nil, State1)
- ),
-
- % a case where the 2nd worker has a newer Foo - currently we're considering
- % Foo to have reached quorum and execute_read_repair()
- ?assertEqual(
- {stop, [Bar1, Foo2]},
- handle_message({ok, [Foo2, Bar1]}, nil, State1)
- ),
-
- % a case where quorum has not yet been reached for Foo
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Bar1]}, nil, State1)
- ),
- {ok, State2} = handle_message({ok, [Bar1]}, nil, State1),
-
- % still no quorum, but all workers have responded. We include Foo1 in the
- % response and execute_read_repair()
- ?assertEqual(
- {stop, [Bar1, Foo1]},
- handle_message({ok, [Bar1]}, nil, State2)
- ),
- meck:unload([fabric, couch_log, couch_stats]),
+ meck:expect(couch_log, notice, fun(_, _) -> ok end).
+
+
+teardown(_) ->
+ (catch meck:unload([fabric, couch_stats, couch_log])),
config:stop().
-specific_revs_test() ->
- config:start_link([]),
- meck:new([fabric, couch_stats]),
- meck:expect(fabric, update_docs, fun(_, _, _) -> {ok, nil} end),
- meck:expect(couch_stats, increment_counter, fun(_) -> ok end),
- meck:new(couch_log),
- meck:expect(couch_log, notice, fun(_,_) -> ok end),
- Revs = [{1,<<"foo">>}, {1,<<"bar">>}, {1,<<"baz">>}],
- State0 = #state{
+state0(Revs, Latest) ->
+ #state{
worker_count = 3,
- workers = [nil, nil, nil],
+ workers = [w1, w2, w3],
r = 2,
revs = Revs,
- latest = false,
- replies = [{Rev,[]} || Rev <- Revs]
- },
- Foo1 = {ok, #doc{revs = {1, [<<"foo">>]}}},
- Foo2 = {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}},
- Bar1 = {ok, #doc{revs = {1, [<<"bar">>]}}},
- Baz1 = {{not_found, missing}, {1,<<"baz">>}},
- Baz2 = {ok, #doc{revs = {1, [<<"baz">>]}}},
-
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0)
- ),
- {ok, State1} = handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State0),
-
- % the normal case - workers agree
- ?assertEqual(
- {stop, [Foo1, Bar1, Baz1]},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1)
- ),
-
- % latest=true, worker responds with Foo2 and we return it
- State0L = State0#state{latest = true},
- ?assertMatch(
- {ok, #state{}},
- handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L)
- ),
- {ok, State1L} = handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State0L),
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz1]},
- handle_message({ok, [Foo2, Bar1, Baz1]}, nil, State1L)
- ),
-
- % Foo1 is included in the read quorum for Foo2
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz1]},
- handle_message({ok, [Foo1, Bar1, Baz1]}, nil, State1L)
- ),
-
- % {not_found, missing} is included in the quorum for any found revision
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz2]},
- handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State1L)
- ),
-
- % a worker failure is skipped
- ?assertMatch(
- {ok, #state{}},
- handle_message({rexi_EXIT, foo}, nil, State1L)
- ),
- {ok, State2L} = handle_message({rexi_EXIT, foo}, nil, State1L),
- ?assertEqual(
- {stop, [Foo2, Bar1, Baz2]},
- handle_message({ok, [Foo2, Bar1, Baz2]}, nil, State2L)
- ),
- meck:unload([fabric, couch_log, couch_stats]),
- config:stop().
+ latest = Latest
+ }.
+
+
+revs() -> [{1,<<"foo">>}, {1,<<"bar">>}, {1,<<"baz">>}].
+
+
+foo1() -> {ok, #doc{revs = {1, [<<"foo">>]}}}.
+foo2() -> {ok, #doc{revs = {2, [<<"foo2">>, <<"foo">>]}}}.
+bar1() -> {ok, #doc{revs = {1, [<<"bar">>]}}}.
+bazNF() -> {{not_found, missing}, {1,<<"baz">>}}.
+baz1() -> {ok, #doc{revs = {1, [<<"baz">>]}}}.
+
+
+
+open_doc_revs_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ check_empty_response_not_quorum(),
+ check_basic_response(),
+ check_finish_quorum(),
+ check_finish_quorum_newer(),
+ check_no_quorum_on_second(),
+ check_done_on_third(),
+ check_specific_revs_first_msg(),
+ check_revs_done_on_agreement(),
+ check_latest_true(),
+ check_ancestor_counted_in_quorum(),
+ check_not_found_counts_for_descendant(),
+ check_worker_error_skipped()
+ ]
+ }.
+
+
+% Tests for revs=all
+
+
+check_empty_response_not_quorum() ->
+ % Simple smoke test that we don't think we're
+ % done with a first empty response
+ ?_assertMatch(
+ {ok, #state{workers = [w2, w3]}},
+ handle_message({ok, []}, w1, state0(all, false))
+ ).
+
+
+check_basic_response() ->
+ % Check that we've handle a response
+ ?_assertMatch(
+ {ok, #state{reply_count = 1, workers = [w2, w3]}},
+ handle_message({ok, [foo1(), bar1()]}, w1, state0(all, false))
+ ).
+
+
+check_finish_quorum() ->
+ % Two messages with the same revisions means we're done
+ ?_test(begin
+ S0 = state0(all, false),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ Expect = {stop, [bar1(), foo1()]},
+ ?assertEqual(Expect, handle_message({ok, [foo1(), bar1()]}, w2, S1))
+ end).
+
+
+check_finish_quorum_newer() ->
+ % We count a descendant of a revision for quorum so
+ % foo1 should count for foo2 which means we're finished.
+ % We also validate that read_repair was triggered.
+ ?_test(begin
+ S0 = state0(all, false),
+ {ok, S1} = handle_message({ok, [foo1(), bar1()]}, w1, S0),
+ Expect = {stop, [bar1(), foo2()]},
+ ?assertEqual(Expect, handle_message({ok, [foo2(), bar1()]}, w2, S1)),
+ ?assertEqual(true, meck:validate(fabric))
--- End diff --
`meck:validate -> true` doesn't mean the mocked function was used. You could use meck:history instead
```
3> meck:new([fabric]).
ok
4> meck:expect(fabric, all_dbs, fun() -> [] end).
ok
5> meck:validate(fabric).
true # <------------------ It is true even though it never being called
6> meck:history(fabric).
[]
7> fabric:all_dbs().
[]
8> meck:history(fabric).
[{<0.35.0>,{fabric,all_dbs,[]},[]}]
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by davisp <gi...@git.apache.org>.
Github user davisp commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60668584
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
--- End diff --
Oooh! Probably so. I'll look and update tomorrow. Good call.
> On Apr 21, 2016, at 5:57 PM, iilyak <no...@github.com> wrote:
>
> In src/fabric_doc_open_revs.erl:
>
> > - All0, RawReplies),
> > - Reduced = fabric_util:remove_ancestors(All, []),
> > - Complete = (ReplyCount =:= (WorkerCount - 1)),
> > - QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
> > - case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
> > - Repair = false;
> > - _ ->
> > - Repair = [D || {_,{{ok,D}, _}} <- Reduced]
> > +
> > + IsTree = Revs == all orelse Latest,
> > +
> > + {NewReplies, QuorumMet, Repair} = case IsTree of
> > + true ->
> > + {NewReplies0, AllInternal, Repair0} =
> > + tree_replies(PrevReplies, tree_sort(RawReplies)),
> > + NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
> Would couch_key_tree:count_leafs/1 work here?
>
> —
> You are receiving this because you authored the thread.
> Reply to this email directly or view it on GitHub
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by iilyak <gi...@git.apache.org>.
Github user iilyak commented on a diff in the pull request:
https://github.com/apache/couchdb-fabric/pull/47#discussion_r60669159
--- Diff: src/fabric_doc_open_revs.erl ---
@@ -56,266 +57,390 @@ go(DbName, Id, Revs, Options) ->
rexi_monitor:stop(RexiMon)
end.
+
handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, #state{workers=Workers}=State) ->
- NewWorkers = lists:keydelete(NodeRef, #shard.node, Workers),
- skip(State#state{workers=NewWorkers});
+ NewState = State#state{
+ workers = lists:keydelete(NodeRef, #shard.node, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
handle_message({rexi_EXIT, _}, Worker, #state{workers=Workers}=State) ->
- skip(State#state{workers=lists:delete(Worker,Workers)});
-handle_message({ok, RawReplies}, Worker, #state{revs = all} = State) ->
+ NewState = State#state{
+ workers = lists:delete(Worker, Workers)
+ },
+ handle_message({ok, []}, nil, NewState);
+
+handle_message({ok, RawReplies}, Worker, State) ->
#state{
dbname = DbName,
reply_count = ReplyCount,
worker_count = WorkerCount,
workers = Workers,
- replies = All0,
- r = R
+ replies = PrevReplies,
+ r = R,
+ revs = Revs,
+ latest = Latest,
+ repair = InRepair
} = State,
- All = lists:foldl(fun(Reply,D) -> fabric_util:update_counter(Reply,1,D) end,
- All0, RawReplies),
- Reduced = fabric_util:remove_ancestors(All, []),
- Complete = (ReplyCount =:= (WorkerCount - 1)),
- QuorumMet = lists:all(fun({_,{_, C}}) -> C >= R end, Reduced),
- case Reduced of All when QuorumMet andalso ReplyCount =:= (R-1) ->
- Repair = false;
- _ ->
- Repair = [D || {_,{{ok,D}, _}} <- Reduced]
+
+ IsTree = Revs == all orelse Latest,
+
+ {NewReplies, QuorumMet, Repair} = case IsTree of
+ true ->
+ {NewReplies0, AllInternal, Repair0} =
+ tree_replies(PrevReplies, tree_sort(RawReplies)),
+ NumLeafs = length(couch_key_tree:get_all_leafs(PrevReplies)),
+ SameNumRevs = length(RawReplies) == NumLeafs,
+ QMet = AllInternal andalso SameNumRevs andalso ReplyCount + 1 >= R,
+ {NewReplies0, QMet, Repair0};
+ false ->
+ {NewReplies0, MinCount} = dict_replies(PrevReplies, RawReplies),
+ {NewReplies0, MinCount >= R, false}
end,
- case maybe_reply(DbName, Reduced, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers = lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
- end;
-handle_message({ok, RawReplies0}, Worker, State) ->
- % we've got an explicit revision list, but if latest=true the workers may
- % return a descendant of the requested revision. Take advantage of the
- % fact that revisions are returned in order to keep track.
- RawReplies = strip_not_found_missing(RawReplies0),
- #state{
- dbname = DbName,
- reply_count = ReplyCount,
- worker_count = WorkerCount,
- workers = Workers,
- replies = All0,
- r = R
- } = State,
- All = lists:zipwith(fun({Rev, D}, Reply) ->
- if Reply =:= error -> {Rev, D}; true ->
- {Rev, fabric_util:update_counter(Reply, 1, D)}
- end
- end, All0, RawReplies),
- Reduced = [fabric_util:remove_ancestors(X, []) || {_, X} <- All],
- FinalReplies = [choose_winner(X, R) || X <- Reduced, X =/= []],
+
Complete = (ReplyCount =:= (WorkerCount - 1)),
- case is_repair_needed(All, FinalReplies) of
- true ->
- Repair = [D || {_,{{ok,D}, _}} <- lists:flatten(Reduced)];
- false ->
- Repair = false
- end,
- case maybe_reply(DbName, FinalReplies, Complete, Repair, R) of
- noreply ->
- {ok, State#state{replies = All, reply_count = ReplyCount+1,
- workers=lists:delete(Worker,Workers)}};
- {reply, FinalReply} ->
- fabric_util:cleanup(lists:delete(Worker,Workers)),
- {stop, FinalReply}
+
+ case QuorumMet orelse Complete of
+ true ->
+ fabric_util:cleanup(lists:delete(Worker, Workers)),
+ maybe_read_repair(
+ DbName,
+ IsTree,
+ NewReplies,
+ ReplyCount + 1,
+ InRepair orelse Repair
+ ),
+ {stop, format_reply(IsTree, NewReplies)};
+ false ->
+ {ok, State#state{
+ replies = NewReplies,
+ reply_count = ReplyCount + 1,
+ workers = lists:delete(Worker, Workers),
+ repair = InRepair orelse Repair
+ }}
+ end.
+
+
+tree_replies(RevTree, []) ->
+ {RevTree, true, false};
+
+tree_replies(RevTree0, [{ok, Doc} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Path = couch_doc:to_path(Doc),
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, Repair};
+ {RevTree2, new_leaf} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, true}
+ end;
+
+tree_replies(RevTree0, [{{not_found, missing}, {Pos, Rev}} | Rest]) ->
+ {RevTree1, Done, Repair} = tree_replies(RevTree0, Rest),
+ Node = {Rev, ?REV_MISSING, []},
+ Path = {Pos, Node},
+ case couch_key_tree:merge(RevTree1, Path) of
+ {RevTree2, internal_node} ->
+ {RevTree2, Done, true};
+ {RevTree2, _} ->
+ {RevTree2, false, Repair}
end.
-skip(#state{revs=all} = State) ->
- handle_message({ok, []}, nil, State);
-skip(#state{revs=Revs} = State) ->
- handle_message({ok, [error || _Rev <- Revs]}, nil, State).
-
-maybe_reply(_, [], false, _, _) ->
- noreply;
-maybe_reply(_, [], true, _, _) ->
- {reply, {ok, []}};
-maybe_reply(DbName, ReplyDict, Complete, RepairDocs, R) ->
- case Complete orelse lists:all(fun({_,{_, C}}) -> C >= R end, ReplyDict) of
- true ->
- maybe_execute_read_repair(DbName, RepairDocs),
- {reply, unstrip_not_found_missing(extract_replies(ReplyDict))};
- false ->
- noreply
+
+tree_sort(Replies) ->
+ SortFun = fun(A, B) -> sort_key(A) =< sort_key(B) end,
+ lists:sort(SortFun, Replies).
+
+
+sort_key({ok, #doc{revs = {Pos, [Rev | _]}}}) ->
+ {Pos, Rev};
+sort_key({{not_found, _}, {Pos, Rev}}) ->
+ {Pos, Rev}.
+
+
+dict_replies(Dict, []) ->
+ Counts = [Count || {_Key, {_Reply, Count}} <- Dict],
+ {Dict, lists:min(Counts)};
+
+dict_replies(Dict, [Reply | Rest]) ->
+ NewDict = fabric_util:update_counter(Reply, 1, Dict),
+ dict_replies(NewDict, Rest).
+
+
+maybe_read_repair(Db, IsTree, Replies, ReplyCount, DoRepair) ->
+ Docs = case IsTree of
+ true -> tree_repair_docs(Replies, DoRepair);
+ false -> dict_repair_docs(Replies, ReplyCount)
+ end,
+ case length(Docs) > 0 of
--- End diff --
`Docs =/= []` would perform better in this case. Or you can use match if you like.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by davisp <gi...@git.apache.org>.
Github user davisp commented on the pull request:
https://github.com/apache/couchdb-fabric/pull/47#issuecomment-219763847
Updated as per @iilyak's feedback.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] couchdb-fabric pull request: Fix fabric_doc_open_revs
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/couchdb-fabric/pull/47
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---