umgeher/leveled
0
Fork 0
Code Issues Pull requests Releases Activity Actions

Move snapshot inside of the fold function

Browse source 
riak_kv_sweeper gets the async fold function, then determines if the
function can be called.  If the system is busy the fold may be queued,
and may never be acted upon.

This may cause issues with snapshot timeouts etc.
This commit is contained in:
martinsumner 2017-04-17 15:03:03 +01:00
parent 60b0b88226
commit 50d95ef6aa
1 changed files with 25 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

40
src/leveled_bookie.erl
View file

@ -849,27 +849,37 @@ foldobjects_byindex(State, Tag, Bucket,
foldobjects(State, Tag, StartKey, EndKey, FoldObjectsFun, DeferredFetch) ->
{ok, LedgerSnapshot, JournalSnapshot} = snapshot_store(State, store),
{FoldFun, InitAcc} = case is_tuple(FoldObjectsFun) of
true ->
FoldObjectsFun;
false ->
{FoldObjectsFun, []}
end,
Folder = fun() ->
AccFun = accumulate_objects(FoldFun,
JournalSnapshot,
Tag,
DeferredFetch),
Acc = leveled_penciller:pcl_fetchkeys(LedgerSnapshot,
StartKey,
EndKey,
AccFun,
InitAcc),
ok = leveled_penciller:pcl_close(LedgerSnapshot),
ok = leveled_inker:ink_close(JournalSnapshot),
Acc
end,
% For fold_objects the snapshot has been moved inside of the Folder
% function.
%
% fold_objects and fold_heads are called by the riak_kv_sweeper in Riak,
% and the sweeper prompts the fold before checking to see if the fold is
% ready to be run. This may lead to the fold being called on an old
% snapshot.
Folder =
fun() ->
{ok,
LedgerSnapshot,
JournalSnapshot} = snapshot_store(State, store),
AccFun = accumulate_objects(FoldFun,
JournalSnapshot,
Tag,
DeferredFetch),
Acc = leveled_penciller:pcl_fetchkeys(LedgerSnapshot,
StartKey,
EndKey,
AccFun,
InitAcc),
ok = leveled_penciller:pcl_close(LedgerSnapshot),
ok = leveled_inker:ink_close(JournalSnapshot),
Acc
end,
{async, Folder}.