Skip to content

Commit

Permalink
Add #riak_core_fold_req_v2 record, use new options for AAE tree building
Browse files Browse the repository at this point in the history
Add `aae_reconstruction` annotation to the fold request sent by
riak_kv_index_hashtree:fold_keys().

Also add not-forwardable annotation to the AAE hash tree fold,
to address GitHub issue #621.

If riak_kv_vnode:handle_command() gets a #riak_core_fold_req_v1 command,
that command is converted to a v2 command before continuing.  The
?FOLD_REQ.opts proplist is appended to the end of the `Opts` proplist
that is passed to the KV storage manager's Mod:fold_objects/4 function.
  • Loading branch information
slfritchie authored and buddhisthead committed Sep 12, 2013
1 parent c005138 commit 0309526
Show file tree
Hide file tree
Showing 2 changed files with 25 additions and 17 deletions.
15 changes: 8 additions & 7 deletions src/riak_kv_index_hashtree.erl
Original file line number Diff line number Diff line change
Expand Up @@ -401,13 +401,14 @@ hash_object({Bucket, Key}, RObjBin) ->
%% key/hash pair will be ignored.
-spec fold_keys(index(), pid()) -> ok.
fold_keys(Partition, Tree) ->
Req = ?FOLD_REQ{foldfun=fun(BKey={Bucket,Key}, RObj, _) ->
IndexN = riak_kv_util:get_index_n({Bucket, Key}),
insert(IndexN, term_to_binary(BKey), hash_object(BKey, RObj),
Tree, [if_missing]),
ok
end,
acc0=ok},
Req = riak_core_util:make_fold_req(
fun(BKey={Bucket,Key}, RObj, _) ->
IndexN = riak_kv_util:get_index_n({Bucket, Key}),
insert(IndexN, term_to_binary(BKey), hash_object(BKey, RObj),
Tree, [if_missing]),
ok
end,
ok, false, [aae_reconstruction]),
riak_core_vnode_master:sync_command({Partition, node()},
Req,
riak_kv_vnode_master, infinity),
Expand Down
27 changes: 17 additions & 10 deletions src/riak_kv_vnode.erl
Original file line number Diff line number Diff line change
Expand Up @@ -258,10 +258,9 @@ list_keys(Preflist, ReqId, Caller, Bucket) ->
riak_kv_vnode_master).

fold(Preflist, Fun, Acc0) ->
Req = riak_core_util:make_fold_req(Fun, Acc0),
riak_core_vnode_master:sync_spawn_command(Preflist,
?FOLD_REQ{
foldfun=Fun,
acc0=Acc0},
Req,
riak_kv_vnode_master).

get_vclocks(Preflist, BKeyList) ->
Expand Down Expand Up @@ -475,15 +474,23 @@ handle_command(?KV_DELETE_REQ{bkey=BKey, req_id=ReqId}, _Sender, State) ->
do_delete(BKey, ReqId, State);
handle_command(?KV_VCLOCK_REQ{bkeys=BKeys}, _Sender, State) ->
{reply, do_get_vclocks(BKeys, State), State};
handle_command(?FOLD_REQ{foldfun=FoldFun, acc0=Acc0}, Sender, State) ->
handle_command(#riak_core_fold_req_v1{} = ReqV1,
Sender, State) ->
%% Use make_fold_req() to upgrade to the most recent ?FOLD_REQ
handle_command(riak_core_util:make_fold_req(ReqV1), Sender, State);
handle_command(?FOLD_REQ{foldfun=FoldFun, acc0=Acc0,
forwardable=_Forwardable, opts=Opts}, Sender, State) ->
%% The riak_core layer takes care of forwarding/not forwarding, so
%% we ignore forwardable here.
%%
%% The function in riak_core used for object folding expects the
%% bucket and key pair to be passed as the first parameter, but in
%% riak_kv the bucket and key have been separated. This function
%% wrapper is to address this mismatch.
FoldWrapper = fun(Bucket, Key, Value, Acc) ->
FoldFun({Bucket, Key}, Value, Acc)
end,
do_fold(FoldWrapper, Acc0, Sender, State);
do_fold(FoldWrapper, Acc0, Sender, Opts, State);

%% entropy exchange commands
handle_command({hashtree_pid, Node}, _, State=#state{hashtrees=HT}) ->
Expand Down Expand Up @@ -1365,16 +1372,16 @@ do_delete(BKey, ReqId, State) ->
end.

%% @private
do_fold(Fun, Acc0, Sender, State=#state{async_folding=AsyncFolding,
mod=Mod,
modstate=ModState}) ->
do_fold(Fun, Acc0, Sender, ReqOpts, State=#state{async_folding=AsyncFolding,
mod=Mod,
modstate=ModState}) ->
{ok, Capabilities} = Mod:capabilities(ModState),
AsyncBackend = lists:member(async_fold, Capabilities),
case AsyncFolding andalso AsyncBackend of
true ->
Opts = [async_fold];
Opts = [async_fold|ReqOpts];
false ->
Opts = []
Opts = ReqOpts
end,
case Mod:fold_objects(Fun, Acc0, Opts, ModState) of
{ok, Acc} ->
Expand Down

0 comments on commit 0309526

Please sign in to comment.