Use iterator_refresh backend option for some folds

src/riak_kv_eleveldb_backend.erl

@@ -59,7 +59,8 @@
 -endif.
 
 -define(API_VERSION, 1).
--define(CAPABILITIES, [async_fold, indexes, index_reformat, size]).
+-define(CAPABILITIES, [async_fold, indexes, index_reformat, size,
+        iterator_refresh]).
 -define(FIXED_INDEXES_KEY, fixed_indexes).
 
 -record(state, {ref :: reference(),
@@ -466,9 +467,15 @@ fold_objects(FoldObjectsFun, Acc, Opts, #state{fold_opts=FoldOpts,
            true            -> undefined
         end,
 
+    IteratorRefresh =
+        case lists:keyfind(iterator_refresh, 1, Opts) of
+            false -> [];
+            Tuple -> [Tuple]
+        end,
+
     %% Set up the fold...
     FirstKey = to_first_key(Limiter),
-    FoldOpts1 = [{first_key, FirstKey} | FoldOpts],
+    FoldOpts1 = IteratorRefresh ++ [{first_key, FirstKey} | FoldOpts],
     FoldFun = fold_objects_fun(FoldObjectsFun, Limiter),
 
     ObjectFolder =

src/riak_kv_index_hashtree.erl

@@ -444,8 +444,11 @@ hash_index_data(IndexData) when is_list(IndexData) ->
 -spec fold_keys(index(), pid(), boolean()) -> ok.
 fold_keys(Partition, Tree, HasIndexTree) ->
     FoldFun = fold_fun(Tree, HasIndexTree),
+    Refresh = app_helper:get_env(riak_kv, iterator_refresh, true),
     Req = riak_core_util:make_fold_req(FoldFun,
-                                       0, false, [aae_reconstruction]),
+                                       0, false, 
+                                       [aae_reconstruction,
+                                        {iterator_refresh, Refresh}]),
     riak_core_vnode_master:sync_command({Partition, node()},
                                         Req,
                                         riak_kv_vnode_master, infinity),

src/riak_kv_multi_backend.erl

@@ -52,7 +52,7 @@
 
 -define(API_VERSION, 1).
 -define(CAPABILITIES, [async_fold, index_reformat]).
--define(ANY_CAPABILITIES, [indexes]).
+-define(ANY_CAPABILITIES, [indexes, iterator_refresh]).
 
 -record (state, {backends :: [{atom(), atom(), term()}],
                  default_backend :: atom()}).

src/riak_kv_vnode.erl

@@ -37,6 +37,7 @@
          readrepair/6,
          list_keys/4,
          fold/3,
+         fold/4,
          get_vclocks/2,
          vnode_status/1,
          ack_keys/1,
@@ -305,6 +306,12 @@ fold(Preflist, Fun, Acc0) ->
                                               Req,
                                               riak_kv_vnode_master).
 
+fold(Preflist, Fun, Acc0, Options) ->
+    Req = riak_core_util:make_fold_req(Fun, Acc0, false, Options),
+    riak_core_vnode_master:sync_spawn_command(Preflist,
+                                              Req,
+                                              riak_kv_vnode_master).
+
 get_vclocks(Preflist, BKeyList) ->
     riak_core_vnode_master:sync_spawn_command(Preflist,
                                               ?KV_VCLOCK_REQ{bkeys=BKeyList},
@@ -926,10 +933,13 @@ handoff_starting({_HOType, TargetNode}=_X, State) ->
     end.
 
 %% @doc Optional callback that is exported, but not part of the behaviour.
-handoff_started({SrcPartition, _SrcNode}, WorkerPid) ->
+handoff_started(SrcPartition, WorkerPid) ->
     case maybe_get_vnode_lock(SrcPartition, WorkerPid) of
-        ok -> true;
-        Error -> Error
+        ok ->
+            Refresh = app_helper:get_env(riak_kv, iterator_refresh, true),
+            FoldOpts = [{iterator_refresh, Refresh}],
+            {ok, FoldOpts};
+        max_concurrency -> {error, max_concurrency}
     end.
 
 handoff_cancelled(State) ->
@@ -1714,13 +1724,8 @@ 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|ReqOpts];
-        false ->
-            Opts = ReqOpts
-    end,
+    Opts0 = maybe_enable_async_fold(AsyncFolding, Capabilities, ReqOpts),
+    Opts = maybe_enable_iterator_refresh(Capabilities, Opts0),
     case Mod:fold_objects(Fun, Acc0, Opts, ModState) of
         {ok, Acc} ->
             {reply, Acc, State};
@@ -1734,6 +1739,25 @@ do_fold(Fun, Acc0, Sender, ReqOpts, State=#state{async_folding=AsyncFolding,
             {reply, ER, State}
     end.
 
+%% @private
+maybe_enable_async_fold(AsyncFolding, Capabilities, Opts) ->
+    AsyncBackend = lists:member(async_fold, Capabilities),
+    case AsyncFolding andalso AsyncBackend of
+        true ->
+            [async_fold|Opts];
+        false ->
+            Opts
+    end.
+
+%% @private
+maybe_enable_iterator_refresh(Capabilities, Opts) ->
+    case lists:member(iterator_refresh, Capabilities) of
+        true ->
+            Opts;
+        false ->
+            lists:keydelete(iterator_refresh, 1, Opts)
+    end.
+
 %% @private
 do_get_vclocks(KeyList,_State=#state{mod=Mod,modstate=ModState}) ->
     [{BKey, do_get_vclock(BKey,Mod,ModState)} || BKey <- KeyList].