summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorEric Avdey <eiri@eiri.ca>2016-11-24 14:00:17 -0400
committerEric Avdey <eiri@eiri.ca>2016-11-24 14:00:17 -0400
commitc3c5429180de14a2b139f7741c934143ef73988c (patch)
tree567dec1b0f2135b1b8ae662f9aba3a343496bce7
parent252467cb4a27637090b5f9006483f5b7ab551699 (diff)
parentc4da61c8eb98cedd3cf7a28c293cb1f6d3ec8571 (diff)
downloadcouchdb-c3c5429180de14a2b139f7741c934143ef73988c.tar.gz
Merge remote branch 'cloudant:79066-port-chunkified-replicate_batch'
This closes #26 Signed-off-by: Eric Avdey <eiri@eiri.ca>
-rw-r--r--src/mem3_rep.erl32
1 files changed, 30 insertions, 2 deletions
diff --git a/src/mem3_rep.erl b/src/mem3_rep.erl
index 938260da6..ad7ac55f5 100644
--- a/src/mem3_rep.erl
+++ b/src/mem3_rep.erl
@@ -252,8 +252,10 @@ replicate_batch(#acc{target = #shard{node=Node, name=Name}} = Acc) ->
[] ->
ok;
Missing ->
- Docs = open_docs(Acc, Missing),
- ok = save_on_target(Node, Name, Docs)
+ lists:map(fun(Chunk) ->
+ Docs = open_docs(Acc, Chunk),
+ ok = save_on_target(Node, Name, Docs)
+ end, chunk_revs(Missing))
end,
update_locals(Acc),
{ok, Acc#acc{revcount=0, infos=[]}}.
@@ -271,6 +273,32 @@ find_missing_revs(Acc) ->
]).
+chunk_revs(Revs) ->
+ Limit = list_to_integer(config:get("mem3", "rev_chunk_size", "5000")),
+ chunk_revs(Revs, Limit).
+
+chunk_revs(Revs, Limit) ->
+ chunk_revs(Revs, {0, []}, [], Limit).
+
+chunk_revs([], {_Count, Chunk}, Chunks, _Limit) ->
+ [Chunk|Chunks];
+chunk_revs([{Id, R, A}|Revs], {Count, Chunk}, Chunks, Limit) when length(R) =< Limit - Count ->
+ chunk_revs(
+ Revs,
+ {Count + length(R), [{Id, R, A}|Chunk]},
+ Chunks,
+ Limit
+ );
+chunk_revs([{Id, R, A}|Revs], {Count, Chunk}, Chunks, Limit) ->
+ {This, Next} = lists:split(Limit - Count, R),
+ chunk_revs(
+ [{Id, Next, A}|Revs],
+ {0, []},
+ [[{Id, This, A}|Chunk]|Chunks],
+ Limit
+ ).
+
+
open_docs(#acc{source=Source, infos=Infos}, Missing) ->
lists:flatmap(fun({Id, Revs, _}) ->
FDI = lists:keyfind(Id, #full_doc_info.id, Infos),