This is an automated email from the ASF dual-hosted git repository. nickva pushed a commit to branch optimize-doc-updates in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit ce4bb2e521512881b035f8dbe67cb56dc2be95ca Author: Nick Vatamaniuc <[email protected]> AuthorDate: Mon Jun 15 23:57:25 2026 -0400 Speed up bulk doc updates Previously in `fabric_doc_update` we used a dict keyed by the whole `#doc{}` record. Updating a large batch meant spening a decent amount of CPU time repeatedly hashing the document bodies. To avoid that we use small and lightweight integers as key and while at it, also switch to use maps instead of dicts. A few more details on the two optimizations 1) Use a lightweight integer tag to match and index documents instead of using their whole `#doc{}` records Using `#doc{}` as a key means applying the `phash/2` function to it, which, in turn, traverses the whole term to produce the hash slot number for the dict implementation https://github.com/erlang/otp/blob/OTP-29.0.2/lib/stdlib/src/dict.erl#L676. A seemingly obvious choice to use might be a the DocId, however in a batch we can have duplicate DocIds so that won't work. So we borrow an already used pattern in couch_db local doc updater which tags docs in a batch with the `tag` field in the `meta` proplist. This coordinator tag lives only in the coordinator we strip away before sending the data to workers or returning it. 2) Use a map instead of a dict. This one is a no-brainer. A map scales better and uses an internal C based structures. Dict is as just phashes of tuples essentially I benchmarked lazily with the built-in fabric bench and at last on my laptop the difference is noticeable for the default 1000 doc batches. I ran the benchmarks 3 times alternating between main and PR code. Environment ---- ``` fabric_bench:go(). *** Parameters * batch_size : 1000 * doc_size : medium * docs : 100000 * individual_docs : 1000 * n : default * q : default *** Environment * Nodes : 1 * Bench ver. : 1 * N : 1 * Q : 2 * OS : unix/darwin * Couch ver. : 3.5.2-25b1b9a * Couch git sha: 25b1b9a * VM details : Erlang/OTP 27 [erts-15.2.7.9] [source] [64-bit] [smp:12:12] * [ds:12:12:16] [async-threads:1] ``` Main ---- ``` *** Inserting 100000 docs * Add 100000 docs, ok:100/accepted:0 (Hz): 440 * Get random doc 100000X (Hz): 1600 * All docs (Hz): 49000 * All docs w/ include_docs (Hz): 3900 * Changes (Hz): 24000 * Single doc updates 1000X (Hz): 280 * Time to run all benchmarks (sec): 300 *** Inserting 100000 docs * Add 100000 docs, ok:100/accepted:0 (Hz): 440 * Get random doc 100000X (Hz): 1600 * All docs (Hz): 46000 * All docs w/ include_docs (Hz): 3900 * Changes (Hz): 16000 * Single doc updates 1000X (Hz): 290 * Time to run all benchmarks (sec): 301 *** Inserting 100000 docs * Add 100000 docs, ok:100/accepted:0 (Hz): 460 * Get random doc 100000X (Hz): 1700 * All docs (Hz): 48000 * All docs w/ include_docs (Hz): 3900 * Changes (Hz): 21000 * Single doc updates 1000X (Hz): 300 * Time to run all benchmarks (sec): 288 ``` PR --- ``` *** Inserting 100000 docs * Add 100000 docs, ok:100/accepted:0 (Hz): 540 * Get random doc 100000X (Hz): 1600 * All docs (Hz): 49000 * All docs w/ include_docs (Hz): 3900 * Changes (Hz): 25000 * Single doc updates 1000X (Hz): 290 * Time to run all benchmarks (sec): 257 *** Inserting 100000 docs * Add 100000 docs, ok:100/accepted:0 (Hz): 550 * Get random doc 100000X (Hz): 1700 * All docs (Hz): 45000 * All docs w/ include_docs (Hz): 3900 * Changes (Hz): 22000 * Single doc updates 1000X (Hz): 300 * Time to run all benchmarks (sec): 253 *** Inserting 100000 docs * Add 100000 docs, ok:100/accepted:0 (Hz): 560 * Get random doc 100000X (Hz): 1600 * All docs (Hz): 46000 * All docs w/ include_docs (Hz): 4000 * Changes (Hz): 29000 * Single doc updates 1000X (Hz): 300 * Time to run all benchmarks (sec): 249 ``` If we approximate it's about 20% or so. --- src/fabric/src/fabric.erl | 4 +- src/fabric/src/fabric_doc_update.erl | 265 +++++++++++++++++++++-------------- 2 files changed, 165 insertions(+), 104 deletions(-) diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl index 965f59c33..47895374f 100644 --- a/src/fabric/src/fabric.erl +++ b/src/fabric/src/fabric.erl @@ -838,7 +838,9 @@ setup() -> meck:expect(mem3, shards, fun(_, _, _) -> [] end), meck:expect(mem3, quorum, fun(_) -> 1 end), meck:expect(rexi, cast, fun(_, _) -> ok end), - meck:expect(rexi_utils, recv, fun(_, _, _, _, _, _) -> {ok, {error, [{Doc, conflict}]}} end), + meck:expect(rexi_utils, recv, fun(_, _, _, _, _, _) -> + {ok, {error, [{Doc#doc{meta = [{ref, 1}]}, conflict}]}} + end), meck:expect(couch_util, reorder_results, fun(_, [{_, Res}], _) -> [Res] end), meck:expect(fabric_util, create_monitors, fun(_) -> ok end), meck:expect(rexi_monitor, stop, fun(_) -> ok end), diff --git a/src/fabric/src/fabric_doc_update.erl b/src/fabric/src/fabric_doc_update.erl index 203f9c89d..70081af73 100644 --- a/src/fabric/src/fabric_doc_update.erl +++ b/src/fabric/src/fabric_doc_update.erl @@ -50,7 +50,7 @@ go(DbName, AllDocs0, Opts) -> doc_count = length(AllDocs), w = fabric_util:w_from_opts(DbName, Options), grouped_docs = GroupedDocs, - reply = dict:new(), + reply = #{}, dbname = DbName, update_options = Options, serialize_worker_startup = serialize_worker_startup(Options) @@ -71,7 +71,7 @@ go(DbName, AllDocs0, Opts) -> } = Acc, {DefunctWorkers, _} = lists:unzip(GroupedDocs1), fabric_util:log_timeout(DefunctWorkers, "update_docs"), - {Health, _, _, Resp} = dict:fold( + {Health, _, _, Resp} = fold_replies( fun force_reply/3, {ok, W1, SWS, []}, DocReplDict @@ -119,10 +119,10 @@ handle_message({ok, Replies}, Worker, #acc{} = Acc0) -> Conflicts = collect_conflicts(Docs, Replies, Conflicts0, SWS), DocReplyDict = append_update_replies(Docs, Replies, DocReplyDict0), Acc1 = Acc0#acc{conflicts = Conflicts}, - case {WaitingCount, dict:size(DocReplyDict)} of + case {WaitingCount, map_size(DocReplyDict)} of {1, _} -> % last message has arrived, we need to conclude things - {Health, W, _, Reply} = dict:fold( + {Health, W, _, Reply} = fold_replies( fun force_reply/3, {ok, W, SWS, []}, DocReplyDict @@ -131,7 +131,7 @@ handle_message({ok, Replies}, Worker, #acc{} = Acc0) -> {stop, {Health, Reply}}; {_, DocCount} -> % we've got at least one reply for each document, let's take a look - case dict:fold(fun maybe_reply/3, {stop, W, SWS, []}, DocReplyDict) of + case fold_replies(fun maybe_reply/3, {stop, W, SWS, []}, DocReplyDict) of continue -> {ok, start_workers(Acc1#acc{ @@ -196,10 +196,20 @@ before_doc_update(DbName, Docs, Opts) -> Docs end. -tag_docs([]) -> +% We can't rely on doc ids for uniqueness as we can have duplicates in a batch. +% Instead we tag each document with a unique lightweight tag so we can match +% and reorder replies later. +tag_docs(Docs) -> + tag_docs(Docs, 1). + +tag_docs([], _Tag) -> []; -tag_docs([#doc{meta = Meta} = Doc | Rest]) -> - [Doc#doc{meta = [{ref, make_ref()} | Meta]} | tag_docs(Rest)]. +tag_docs([#doc{meta = Meta} = Doc | Rest], Tag) -> + [Doc#doc{meta = [{ref, Tag} | Meta]} | tag_docs(Rest, Tag + 1)]. + +doc_tag(#doc{meta = Meta}) -> + {ref, Tag} = lists:keyfind(ref, 1, Meta), + Tag. untag_docs([]) -> []; @@ -265,11 +275,14 @@ maybe_reply(Doc, Replies, {stop, W, SWS, Acc}) -> % this ensures that we got some response for all documents being updated ensure_all_responses(Health, AllDocs, Resp) -> + % Matching docs by their lightweight tag not by #doc{} boides + RespByTag = [{doc_tag(Doc), R} || {Doc, R} <- Resp], + AllTags = [doc_tag(Doc) || Doc <- AllDocs], Results = [ R || R <- couch_util:reorder_results( - AllDocs, - Resp, + AllTags, + RespByTag, {error, internal_server_error} ), R =/= noreply @@ -344,21 +357,19 @@ group_docs_by_shard(DbName, Docs) -> rotate_ranges(DbName, Grouped). group_docs(DocsShards) -> - dict:to_list( - lists:foldl( - fun({Doc, Shards}, D0) -> - lists:foldl( - fun(Shard, D1) -> - dict:append(Shard, Doc, D1) - end, - D0, - Shards - ) - end, - dict:new(), - DocsShards - ) - ). + Groups = lists:foldl( + fun({Doc, Shards}, Acc0) -> + AppendF = fun(DocsAcc) -> [Doc | DocsAcc] end, + FoldF = fun(Shard, Acc) -> maps:update_with(Shard, AppendF, [Doc], Acc) end, + lists:foldl(FoldF, Acc0, Shards) + end, + #{}, + DocsShards + ), + % Docs are prepended above, so reverse to restore the batch order. Worker + % replies are matched positionally against these per-shard doc lists so + % we're ensuring they are in the same order. + [{Shard, lists:reverse(DocsAcc)} || {Shard, DocsAcc} <- maps:to_list(Groups)]. % Deterministically order each range's workers by ownership. When serialized, % workers start with the first entry of each range. @@ -373,17 +384,33 @@ owner_order(DbName, Range, Entries) -> Owners = mem3:owners(DbName, Range, [N || {#shard{node = N}, _} <- Entries]), [E || N <- Owners, {#shard{node = N1}, _} = E <- Entries, N1 =:= N]. +% Reply accumulator is a map #{doc_tag(Doc) => {Doc, Replies}}. The tags +% are small integers, so they hash much faster as keys than whole doc bodies. append_update_replies([], [], DocReplyDict) -> DocReplyDict; append_update_replies([Doc | Rest], [], Dict0) -> % icky, if replicated_changes only errors show up in result - append_update_replies(Rest, [], dict:append(Doc, noreply, Dict0)); + append_update_replies(Rest, [], append_update_reply(Doc, noreply, Dict0)); append_update_replies([Doc | Rest1], [Reply | Rest2], Dict0) -> - append_update_replies(Rest1, Rest2, dict:append(Doc, Reply, Dict0)). + append_update_replies(Rest1, Rest2, append_update_reply(Doc, Reply, Dict0)). + +append_update_reply(#doc{} = Doc, Reply, DocReplyDict) -> + maps:update_with( + doc_tag(Doc), + fun({D, Replies}) -> {D, Replies ++ [Reply]} end, + {Doc, [Reply]}, + DocReplyDict + ). + +% Fold the reply map in reverse tag order so that, with the fold callbacks +% prepending to their result list, replies come out in the original batch order. +fold_replies(Fun, Acc0, DocReplyDict) -> + Sorted = lists:reverse(lists:sort(maps:to_list(DocReplyDict))), + lists:foldl(fun({_Tag, {Doc, Replies}}, Acc) -> Fun(Doc, Replies, Acc) end, Acc0, Sorted). skip_message(#acc{waiting_count = 0} = Acc) -> #acc{w = W, reply = DocReplyDict, serialize_worker_startup = SWS} = Acc, - {Health, W, _, Reply} = dict:fold(fun force_reply/3, {ok, W, SWS, []}, DocReplyDict), + {Health, W, _, Reply} = fold_replies(fun force_reply/3, {ok, W, SWS, []}, DocReplyDict), {stop, {Health, Reply}}; skip_message(#acc{} = Acc0) -> {ok, Acc0}. @@ -530,12 +557,14 @@ doc_update_test_() -> % eunits doc_update1() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, + Docs2 = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Docs = [Doc1], - Docs2 = [Doc1, Doc2], - Dict = dict:from_list([{Doc, []} || Doc <- Docs]), - Dict2 = dict:from_list([{Doc, []} || Doc <- Docs2]), + Dict = reply_map(Docs), + Dict2 = reply_map(Docs2), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), @@ -623,9 +652,11 @@ doc_update1() -> ). doc_update2() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -634,7 +665,7 @@ doc_update2() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -654,9 +685,11 @@ doc_update2() -> ). doc_update3() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -665,7 +698,7 @@ doc_update3() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -682,9 +715,11 @@ doc_update3() -> ?assertEqual({ok, [{Doc1, {ok, Doc1}}, {Doc2, {ok, Doc2}}]}, Reply). early_termination_on_conflict_1() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), [S1 | _] = Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), @@ -694,7 +729,7 @@ early_termination_on_conflict_1() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, ?assertEqual( {stop, {ok, [{Doc1, conflict}, {Doc2, conflict}]}}, @@ -702,9 +737,11 @@ early_termination_on_conflict_1() -> ). early_termination_on_conflict_2() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), [S1, S2 | _] = Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), @@ -714,7 +751,7 @@ early_termination_on_conflict_2() -> doc_count = length(Docs), w = 3, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{} = Acc1} = handle_message({ok, [internal_server_error, internal_server_error]}, S1, Acc0), @@ -724,9 +761,11 @@ early_termination_on_conflict_2() -> ). handle_all_dbs_active() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -735,7 +774,7 @@ handle_all_dbs_active() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -752,9 +791,11 @@ handle_all_dbs_active() -> ?assertEqual({ok, [{Doc1, {ok, Doc1}}, {Doc2, {ok, Doc2}}]}, Reply). handle_two_all_dbs_actives() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -763,7 +804,7 @@ handle_two_all_dbs_actives() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -783,9 +824,11 @@ handle_two_all_dbs_actives() -> ). one_forbid() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -795,7 +838,7 @@ one_forbid() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -820,9 +863,11 @@ one_forbid() -> ). two_forbid() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -832,7 +877,7 @@ two_forbid() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -862,9 +907,11 @@ two_forbid() -> % tree is extended and so the VDU should forbid the document. % Leaving this test here to make sure quorum rules still apply. extend_tree_forbid() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -874,7 +921,7 @@ extend_tree_forbid() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -893,9 +940,11 @@ extend_tree_forbid() -> ?assertEqual({ok, [{Doc1, {ok, Doc1}}, {Doc2, {ok, Doc2}}]}, Reply). other_errors_one_forbid() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -905,7 +954,7 @@ other_errors_one_forbid() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -923,9 +972,11 @@ other_errors_one_forbid() -> ?assertEqual({error, [{Doc1, {ok, Doc1}}, {Doc2, {Doc2, {error, <<"foo">>}}}]}, Reply). one_error_two_forbid() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -935,7 +986,7 @@ one_error_two_forbid() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -957,9 +1008,11 @@ one_error_two_forbid() -> ). one_success_two_forbid() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -969,7 +1022,7 @@ one_success_two_forbid() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, {ok, #acc{waiting_count = WaitingCount1} = Acc1} = @@ -991,8 +1044,7 @@ one_success_two_forbid() -> ). worker_before_doc_update_forbidden() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Docs = [Doc1], + Docs = [Doc1] = tag_docs([#doc{revs = {1, [<<"foo">>]}}]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -1001,13 +1053,12 @@ worker_before_doc_update_forbidden() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, ?assertThrow({forbidden, <<"msg">>}, handle_message({forbidden, <<"msg">>}, hd(Shards), Acc)). handle_bad_request() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Docs = [Doc1], + Docs = [Doc1] = tag_docs([#doc{revs = {1, [<<"foo">>]}}]), Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), GroupedDocs = group_docs_by_shard_hack(<<"foo">>, Shards, Docs), @@ -1016,7 +1067,7 @@ handle_bad_request() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{Doc, []} || Doc <- Docs]) + reply = reply_map(Docs) }, ?assertThrow( {bad_request, msg}, @@ -1043,7 +1094,7 @@ filter_conflicts_drops_seen_docs() -> parallel_in_flight_after_conflict() -> Doc1 = #doc{revs = {1, [<<"foo">>]}}, Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = [Tagged1, Tagged2] = tag_docs([Doc1, Doc2]), [S1, S2 | _] = Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), @@ -1053,21 +1104,21 @@ parallel_in_flight_after_conflict() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{D, []} || D <- Docs]) + reply = reply_map(Docs) }, - % S1 returns Doc1=conflict / Doc2=ok. Doc1 should be added to conflicts. S2 - % entry should be left alone + % S1 returns Doc1=conflict / Doc2=ok. Doc1 (untagged) should be added to + % conflicts. S2 entry should be left alone {ok, #acc{conflicts = [Doc1]} = Acc1} = - handle_message({ok, [conflict, {ok, Doc2}]}, S1, Acc0), + handle_message({ok, [conflict, {ok, Tagged2}]}, S1, Acc0), {S2, S2Docs} = lists:keyfind(S2, 1, Acc1#acc.grouped_docs), - ?assertEqual([Doc1, Doc2], S2Docs), + ?assertEqual([Tagged1, Tagged2], S2Docs), %% S2 reply comes back, and we want lengths to match exactly {stop, {ok, Reply}} = - handle_message({ok, [conflict, {ok, Doc2}]}, S2, Acc1), + handle_message({ok, [conflict, {ok, Tagged2}]}, S2, Acc1), ?assertEqual( - lists:sort([{Doc1, conflict}, {Doc2, {ok, Doc2}}]), + lists:sort([{Tagged1, conflict}, {Tagged2, {ok, Tagged2}}]), lists:sort(Reply) ). @@ -1107,9 +1158,11 @@ serial_filters_conflicts_at_cast() -> ?assertEqual([Tagged2], Stored). sws_false_mode_conflict_not_final() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), [S1, S2 | _] = Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), @@ -1119,7 +1172,7 @@ sws_false_mode_conflict_not_final() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{D, []} || D <- Docs]), + reply = reply_map(Docs), serialize_worker_startup = false }, @@ -1134,9 +1187,11 @@ sws_false_mode_conflict_not_final() -> {ok, _Acc2} = handle_message({ok, [conflict, {ok, Doc2}]}, S2, Acc1). sws_false_mode_ok_can_outvote_conflict() -> - Doc1 = #doc{revs = {1, [<<"foo">>]}}, - Doc2 = #doc{revs = {1, [<<"bar">>]}}, - Docs = [Doc1, Doc2], + Docs = + [Doc1, Doc2] = tag_docs([ + #doc{revs = {1, [<<"foo">>]}}, + #doc{revs = {1, [<<"bar">>]}} + ]), [S1, S2, S3] = Shards = mem3_util:create_partition_map("foo", 3, 1, ["node1", "node2", "node3"]), @@ -1146,7 +1201,7 @@ sws_false_mode_ok_can_outvote_conflict() -> doc_count = length(Docs), w = 2, grouped_docs = GroupedDocs, - reply = dict:from_list([{D, []} || D <- Docs]), + reply = reply_map(Docs), serialize_worker_startup = false }, % With enough peers returning ok for both docs, resolve doc1 as ok @@ -1249,4 +1304,8 @@ mk_shard(Name, Node, Range) -> group_docs_by_shard_hack(DbName, Shards, Docs) -> rotate_ranges(DbName, group_docs([{Doc, Shards} || Doc <- Docs])). +% seed a reply map the way arriving worker replies would build it +reply_map(Docs) -> + maps:from_list([{doc_tag(Doc), {Doc, []}} || Doc <- Docs]). + -endif.
