Merge branch 'slf/basho_bench_driver'

priv/basho_bench.append-example.config

@@ -0,0 +1,41 @@
+%% Mandatory: adjust this code path to top of your compiled Machi source distro
+{code_paths, ["/Users/fritchie/b/src/machi"]}.
+{driver, machi_basho_bench_driver}.
+
+%% Chose your maximum rate (per worker proc, see 'concurrent' below)
+%{mode, {rate,10}}.
+%{mode, {rate,20}}.
+{mode, max}.
+
+%% Runtime & reporting interval (seconds)
+{duration, 10}.
+{report_interval, 1}.
+
+%% Choose your number of worker procs
+%{concurrent, 1}.
+{concurrent, 5}.
+%{concurrent, 10}.
+
+%% Here's a chain of (up to) length 3, all on localhost
+{machi_server_info,
+ [
+  {p_srvr,a,machi_flu1_client,"localhost",4444,[]},
+  {p_srvr,b,machi_flu1_client,"localhost",4445,[]},
+  {p_srvr,c,machi_flu1_client,"localhost",4446,[]}
+ ]}.
+{machi_ets_key_tab_type, set}.   % set | ordered_set
+
+%% Workload-specific definitions follow....
+
+%% 10 parts 'append' operation + 0 parts anything else = 100% 'append' ops
+{operations, [{append, 10}]}.
+
+%% For append, key = Machi file prefix name
+{key_generator, {concat_binary, <<"prefix">>,
+                                {to_binstr, "~w", {uniform_int, 30}}}}.
+
+%% Increase size of value_generator_source_size if value_generator is big!!
+{value_generator_source_size, 2111000}.
+{value_generator, {fixed_bin, 32768}}.   %  32 KB
+%{value_generator, {fixed_bin, 1048576}}. % 1024 KB
+

priv/basho_bench.read-example.config

@@ -0,0 +1,41 @@
+%% Mandatory: adjust this code path to top of your compiled Machi source distro
+{code_paths, ["/Users/fritchie/b/src/machi"]}.
+{driver, machi_basho_bench_driver}.
+
+%% Chose your maximum rate (per worker proc, see 'concurrent' below)
+%{mode, {rate,10}}.
+%{mode, {rate,20}}.
+{mode, max}.
+
+%% Runtime & reporting interval (seconds)
+{duration, 10}.
+{report_interval, 1}.
+
+%% Choose your number of worker procs
+%{concurrent, 1}.
+{concurrent, 5}.
+%{concurrent, 10}.
+
+%% Here's a chain of (up to) length 3, all on localhost
+{machi_server_info,
+ [
+  {p_srvr,a,machi_flu1_client,"localhost",4444,[]},
+  {p_srvr,b,machi_flu1_client,"localhost",4445,[]},
+  {p_srvr,c,machi_flu1_client,"localhost",4446,[]}
+ ]}.
+{machi_ets_key_tab_type, set}.   % set | ordered_set
+
+%% Workload-specific definitions follow....
+
+%% 10 parts 'read' operation + 0 parts anything else = 100% 'read' ops
+{operations, [{read, 10}]}.
+
+%% For read, key = integer index into Machi's chunk ETS table, modulo the
+%%                ETS table size, so a huge number here is OK.
+{key_generator, {uniform_int, 999999999999}}.
+
+%% For read, value_generator_* isn't used, so leave these defaults as-is.
+{value_generator_source_size, 2111000}.
+{value_generator, {fixed_bin, 32768}}.   %  32 KB
+
+

src/machi_basho_bench_driver.erl

@@ -0,0 +1,150 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2007-2015 Basho Technologies, Inc.  All Rights Reserved.
+%%
+%% This file is provided to you under the Apache License,
+%% Version 2.0 (the "License"); you may not use this file
+%% except in compliance with the License.  You may obtain
+%% a copy of the License at
+%%
+%%   http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing,
+%% software distributed under the License is distributed on an
+%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+%% KIND, either express or implied.  See the License for the
+%% specific language governing permissions and limitations
+%% under the License.
+%%
+%% -------------------------------------------------------------------
+
+%% @doc A simple basho_bench driver for Machi
+%%
+%% Basho_bench was originally developed to stress test key-value
+%% stores (as was YCSB and several other mechmarking tools).  A person
+%% can consider the UNIX file system to be a key-value store and thus
+%% use basho_bench to measure its performance under a certain
+%% workload.  Machi is a bit different than most KV stores in that the
+%% client has no direct control over the keys -- Machi servers always
+%% assign the keys.  The schemes typically used by basho_bench & YCSB
+%% to use/mimic key naming conventions used internally ... are
+%% difficult to adapt to Machi.
+%%
+%% So, we'll try to manage key reading by using a common ETS table
+%% that is populated with:
+%%
+%% 1. Key: `non_neg_integer()`
+%% 2. Value: The `{File,Offset,Size}` for a chunk previously written.
+%%
+%% At startup time, basho_bench can use the `list_files' and
+%% `checksum_list' API operations to fetch all of the
+%% `{File,Offset,Size}` tuples that currently reside in the cluster.
+%% Also, optionally (?), each new `append' operation by the b_b driver
+%% could add new entries to this ETS table.
+%%
+%% Now we can use various integer-centric key generators that are
+%% already bundled with basho_bench.
+%%
+%% TODO: Add CRC checking, when feasible and when supported on the
+%% server side.
+%%
+%% TODO: As an alternate idea, if we know that the chunks written are
+%% always the same size, and if we don't care about CRC checking, then
+%% all we need to know are the file names & file sizes on the server:
+%% we can then pick any valid offset within that file.  That would
+%% certainly be more scalable than the zillion-row-ETS-table, which is
+%% definitely RAM-hungry.
+
+-module(machi_basho_bench_driver).
+
+-export([new/1, run/4]).
+
+-record(m, {
+          conn,
+          max_key
+         }).
+
+-define(ETS_TAB, machi_keys).
+
+-define(INFO(Str, Args), lager:info(Str, Args)).
+-define(WARN(Str, Args), lager:warning(Str, Args)).
+-define(ERROR(Str, Args), lager:error(Str, Args)).
+
+new(Id) ->
+    Ps = find_server_info(Id),
+    {ok, Conn} = machi_cr_client:start_link(Ps),
+    if Id == 1 ->
+            ?INFO("Key preload: starting", []),
+            TabType = basho_bench_config:get(machi_ets_key_tab_type, set),
+            ETS = ets:new(?ETS_TAB, [public, named_table, TabType,
+                                     {read_concurrency, true}]),
+            ets:insert(ETS, {max_key, 0}),
+            ets:insert(ETS, {total_bytes, 0}),
+            MaxKeys = load_ets_table(Conn, ETS),
+            ?INFO("Key preload: finished, ~w keys loaded", [MaxKeys]),
+            Bytes = ets:lookup_element(ETS, total_bytes, 2),
+            ?INFO("Key preload: finished, chunk list specifies ~s MBytes of chunks",
+                  [machi_util:mbytes(Bytes)]),
+            ok;
+       true ->
+            ok
+    end,
+    {ok, #m{conn=Conn}}.
+
+run(append, KeyGen, ValueGen, #m{conn=Conn}=S) ->
+    Prefix = KeyGen(),
+    Value = ValueGen(),
+    case machi_cr_client:append_chunk(Conn, Prefix, Value) of
+        {ok, Pos} ->
+            EtsKey = ets:update_counter(?ETS_TAB, max_key, 1),
+            true = ets:insert(?ETS_TAB, {EtsKey, Pos}),
+            {ok, S};
+        {error, _}=Err ->
+            ?ERROR("append ~w bytes to prefix ~w: ~p\n",
+                   [iolist_size(ValueGen), Prefix, Err]),
+            {error, Err, S}
+    end;
+run(read, KeyGen, ValueGen, #m{max_key=undefined}=S) ->
+    MaxKey = ets:update_counter(?ETS_TAB, max_key, 0),
+    run(read, KeyGen, ValueGen, S#m{max_key=MaxKey});
+run(read, KeyGen, _ValueGen, #m{conn=Conn, max_key=MaxKey}=S) ->
+    Idx = KeyGen() rem MaxKey,
+    %% {File, Offset, Size, _CSum} = ets:lookup_element(?ETS_TAB, Idx, 2),
+    {File, Offset, Size} = ets:lookup_element(?ETS_TAB, Idx, 2),
+    case machi_cr_client:read_chunk(Conn, File, Offset, Size) of
+        {ok, _Chunk} ->
+            {ok, S};
+        {error, _}=Err ->
+            ?ERROR("read file ~p offset ~w size ~w: ~w\n",
+                   [File, Offset, Size]),
+            {error, Err, S}
+    end.
+
+find_server_info(_Id) ->
+    Key = machi_server_info,
+    case basho_bench_config:get(Key, undefined) of
+        undefined ->
+            ?ERROR("Please define '~w' in your basho_bench config.\n", [Key]),
+            timer:sleep(500),
+            exit(bad_config);
+        Ps ->
+            Ps
+    end.
+
+load_ets_table(Conn, ETS) ->
+    {ok, Fs} = machi_cr_client:list_files(Conn),
+    [begin
+         {ok, PosList} = machi_cr_client:checksum_list(Conn, File),
+         StartKey = ets:update_counter(ETS, max_key, 0),
+         %% _EndKey = lists:foldl(fun({Off,Sz,CSum}, K) ->
+         %%                               V = {File, Off, Sz, CSum},
+         {_, Bytes} = lists:foldl(fun({Off,Sz,_CSum}, {K, Bs}) ->
+                                          V = {File, Off, Sz},
+                                          ets:insert(ETS, {K, V}),
+                                          {K + 1, Bs + Sz}
+                                  end, {StartKey, 0}, PosList),
+         ets:update_counter(ETS, max_key, length(PosList)),
+         ets:update_counter(ETS, total_bytes, Bytes)
+     end || {_Size, File} <- Fs],
+    ets:update_counter(?ETS_TAB, max_key, 0).
+

src/machi_chain_repair.erl

@@ -361,10 +361,8 @@ execute_repair_directive({File, Cmds}, {ProxiesDict, EpochID, Verb, ETS}=Acc) ->
         {L_K, T_K} <- EtsKeys],
     Acc.
 
-mbytes(0) ->
+mbytes(N) ->
-    "0.0";
+    machi_util:mbytes(N).
-mbytes(Size) ->
-    lists:flatten(io_lib:format("~.1.0f", [max(0.1, Size / (1024*1024))])).
 
 -ifdef(TEST).
 

src/machi_cr_client.erl

@@ -135,6 +135,8 @@
 -define(TIMEOUT, 2*1000).
 -define(DEFAULT_TIMEOUT, 10*1000).
 -define(MAX_RUNTIME, 8*1000).
+-define(WORST_PROJ, #projection_v1{epoch_number=-1,epoch_csum= <<>>,
+                                   members_dict=[]}).
 
 -record(state, {
           members_dict    :: p_srvr_dict(),
@@ -636,6 +638,10 @@ update_proj2(Count, #state{bad_proj=BadProj, proxies_dict=ProxiesDict}=S) ->
     %% then it is possible that choose_best_projs() can incorrectly choose
     %% b's projection.
     case choose_best_proj(Rs) of
+        P when P == ?WORST_PROJ ->
+            io:format(user, "TODO: Using ?WORST_PROJ, chain is not available\n", []),
+            sleep_a_while(Count),
+            update_proj2(Count + 1, S);
         P when P >= BadProj ->
             #projection_v1{epoch_number=Epoch, epoch_csum=CSum,
                            members_dict=NewMembersDict} = P,
@@ -680,13 +686,12 @@ gather_worker_statuses([{Pid,Ref}|Rest], Timeout) ->
     end.
 
 choose_best_proj(Rs) ->
-    WorstEpoch = #projection_v1{epoch_number=-1,epoch_csum= <<>>},
+    lists:foldl(fun({ok, NewProj}, BestProj)
-    lists:foldl(fun({ok, NewEpoch}, BestEpoch)
+                      when NewProj > BestProj ->
-                      when NewEpoch > BestEpoch ->
+                        NewProj;
-                        NewEpoch;
+                   (_, BestProj) ->
-                   (_, BestEpoch) ->
+                        BestProj
-                        BestEpoch
+                end, ?WORST_PROJ, Rs).
-                end, WorstEpoch, Rs).
 
 try_to_find_chunk(Eligible, File, Offset, Size,
                   #state{epoch_id=EpochID, proxies_dict=PD}) ->

src/machi_util.erl

@@ -34,6 +34,7 @@
          make_projection_filename/2,
          read_max_filenum/2, increment_max_filenum/2,
          info_msg/2, verb/1, verb/2,
+         mbytes/1,
          %% TCP protocol helpers
          connect/2, connect/3
         ]).
@@ -231,6 +232,11 @@ verb(Fmt, Args) ->
         _          -> ok
     end.
 
+mbytes(0) ->
+    "0.0";
+mbytes(Size) ->
+    lists:flatten(io_lib:format("~.1.0f", [max(0.1, Size / (1024*1024))])).
+
 %% @doc Log an 'info' level message.
 
 -spec info_msg(string(), list()) -> term().