Merge pull request #519 from basho/1.3_to_master

1.3 to master

src/hashtree.erl

@@ -506,7 +506,27 @@ share_segment_store(State, #state{ref=Ref, path=Path}) ->
 -spec hash(term()) -> binary().
 hash(X) ->
     %% erlang:phash2(X).
-    crypto:sha(term_to_binary(X)).
+    sha(term_to_binary(X)).
+
+sha(Bin) ->
+    Chunk = app_helper:get_env(riak_kv, anti_entropy_sha_chunk, 4096),
+    sha(Chunk, Bin).
+
+sha(Chunk, Bin) ->
+    Ctx1 = crypto:sha_init(),
+    Ctx2 = sha(Chunk, Bin, Ctx1),
+    SHA = crypto:sha_final(Ctx2),
+    SHA.
+
+sha(Chunk, Bin, Ctx) ->
+    case Bin of
+        <<Data:Chunk/binary, Rest/binary>> ->
+            Ctx2 = crypto:sha_update(Ctx, Data),
+            sha(Chunk, Rest, Ctx2);
+        Data ->
+            Ctx2 = crypto:sha_update(Ctx, Data),
+            Ctx2
+    end.
 
 -spec update_levels(integer(),
                     [{integer(), [{integer(), binary()}]}],
@@ -1026,6 +1046,19 @@ delta_test() ->
 %%%===================================================================
 
 -ifdef(EQC).
+sha_test_() ->
+    {timeout, 60,
+     fun() ->
+             ?assert(eqc:quickcheck(eqc:testing_time(4, prop_sha())))
+     end
+    }.
+
+prop_sha() ->
+    ?FORALL(Size, choose(256, 1024*1024),
+            ?FORALL(Chunk, choose(1, Size),
+                    ?FORALL(Bin, binary(Size),
+                            sha(Chunk, Bin) =:= crypto:sha(Bin)))).
+
 eqc_test_() ->
     {timeout, 5,
      fun() ->

src/riak_kv.app.src

@@ -3,7 +3,7 @@
 {application, riak_kv,
  [
   {description, "Riak Key/Value Store"},
-  {vsn, "1.3.0"},
+  {vsn, "1.3.1"},
   {applications, [
                   kernel,
                   stdlib,

src/riak_kv_console.erl

@@ -36,6 +36,7 @@
          cluster_info/1,
          down/1,
          aae_status/1,
+         reformat_indexes/1,
          reload_code/1]).
 
 %% Arrow is 24 chars wide
@@ -421,6 +422,73 @@ format_timestamp(_Now, undefined) ->
 format_timestamp(Now, TS) ->
     riak_core_format:human_time_fmt("~.1f", timer:now_diff(Now, TS)).
 
+parse_int(IntStr) ->
+    try
+        list_to_integer(IntStr)
+    catch
+        error:badarg ->
+            undefined
+    end.
+
+index_reformat_options([], Opts) ->
+    Defaults = [{concurrency, 2}, {batch_size, 100}],
+    AddIfAbsent =
+        fun({Name,Val}, Acc) ->
+            case lists:keymember(Name, 1, Acc) of
+                true ->
+                    Acc;
+                false ->
+                    [{Name, Val} | Acc]
+            end
+        end,
+    lists:foldl(AddIfAbsent, Opts, Defaults);
+index_reformat_options(["--downgrade"], Opts) ->
+    [{downgrade, true} | Opts];
+index_reformat_options(["--downgrade" | More], _Opts) ->
+    io:format("Invalid arguments after downgrade switch : ~p~n", [More]),
+    undefined;
+index_reformat_options([IntStr | Rest], Opts) ->
+    HasConcurrency = lists:keymember(concurrency, 1, Opts),
+    HasBatchSize = lists:keymember(batch_size, 1, Opts),
+    case {parse_int(IntStr), HasConcurrency, HasBatchSize} of
+        {_, true, true} ->
+            io:format("Expected --downgrade instead of ~p~n", [IntStr]),
+            undefined;
+        {undefined, _, _ } ->
+            io:format("Expected integer parameter instead of ~p~n", [IntStr]),
+            undefined;
+        {IntVal, false, false} ->
+            index_reformat_options(Rest, [{concurrency, IntVal} | Opts]);
+        {IntVal, true, false} ->
+            index_reformat_options(Rest, [{batch_size, IntVal} | Opts])
+    end;
+index_reformat_options(_, _) ->
+    undefined.
+
+reformat_indexes(Args) ->
+    Opts = index_reformat_options(Args, []),
+    case Opts of
+        undefined ->
+            io:format("Expected options: <concurrency> <batch size> [--downgrade]~n"),
+            ok;
+        _ ->
+            start_index_reformat(Opts),
+            io:format("index reformat started with options ~p ~n", [Opts]),
+            io:format("check console.log for status information~n"),
+            ok
+    end.
+
+start_index_reformat(Opts) ->
+    spawn(fun() -> run_index_reformat(Opts) end).
+
+run_index_reformat(Opts) ->
+    try riak_kv_util:fix_incorrect_index_entries(Opts)
+    catch
+        Err:Reason ->
+            lager:error("index reformat crashed with error type ~p and reason: ~p",
+                        [Err, Reason])
+    end.
+
 %%%===================================================================
 %%% Private
 %%%===================================================================