port of mv-cache-object-warming from develop branch

c_src/build_deps.sh

@@ -8,7 +8,7 @@ if [ `uname -s` = 'SunOS' -a "${POSIX_SHELL}" != "true" ]; then
 fi
 unset POSIX_SHELL # clear it so if we invoke other scripts, they run as ksh as well
 
-LEVELDB_VSN="2.0.7"
+LEVELDB_VSN="2.0.8"
 
 SNAPPY_VSN="1.0.4"
 

c_src/eleveldb.cc

@@ -134,6 +134,7 @@ ERL_NIF_TERM ATOM_DELETE_THRESHOLD;
 ERL_NIF_TERM ATOM_TIERED_SLOW_LEVEL;
 ERL_NIF_TERM ATOM_TIERED_FAST_PREFIX;
 ERL_NIF_TERM ATOM_TIERED_SLOW_PREFIX;
+ERL_NIF_TERM ATOM_CACHE_OBJECT_WARMING;
 }   // namespace eleveldb
 
 
@@ -449,6 +450,13 @@ ERL_NIF_TERM parse_open_option(ErlNifEnv* env, ERL_NIF_TERM item, leveldb::Optio
             if (0<ret_val && ret_val<256)
                 opts.tiered_slow_prefix = buffer;
         }
+        else if (option[0] == eleveldb::ATOM_CACHE_OBJECT_WARMING)
+        {
+            if (option[1] == eleveldb::ATOM_TRUE)
+                opts.cache_object_warming = true;
+            else
+                opts.cache_object_warming = false;
+        }
 
     }
 
@@ -1259,6 +1267,7 @@ try
     ATOM(eleveldb::ATOM_TIERED_SLOW_LEVEL, "tiered_slow_level");
     ATOM(eleveldb::ATOM_TIERED_FAST_PREFIX, "tiered_fast_prefix");
     ATOM(eleveldb::ATOM_TIERED_SLOW_PREFIX, "tiered_slow_prefix");
+    ATOM(eleveldb::ATOM_CACHE_OBJECT_WARMING, "cache_object_warming");
 #undef ATOM
 
 

priv/eleveldb.schema

@@ -228,3 +228,15 @@
   {datatype, directory},
   hidden
 ]}.
+
+%% @doc This parameter enables/disables logic that saves a list
+%% of cached objects (currently only open .sst file names) at
+%% database close.  The list is used on the next open to pre-populate
+%% the cache.  This typically eliminates heavy latencies associated
+%% with .sst files opening for early user requests.
+{mapping, "leveldb.cache_object_warming", "eleveldb.cache_object_warming", [
+  {default, on},
+  {datatype, flag},
+  hidden
+]}.
+

priv/eleveldb_multi.schema

@@ -155,3 +155,11 @@
   {datatype, directory},
   hidden
 ]}.
+
+%% @see leveldb.cache_object_warming
+{mapping, "multi_backend.$name.leveldb.cache_object_warming", "riak_kv.multi_backend", [
+  {default, on},
+  {datatype, flag},
+  hidden
+]}.
+

rebar.config

@@ -9,7 +9,7 @@
 {erl_opts, [warnings_as_errors, debug_info]}.
 
 {deps, [
-        {cuttlefish, ".*", {git, "git://github.com/basho/cuttlefish.git", {branch, "2.0"}}}
+        {cuttlefish, ".*", {git, "git://github.com/basho/cuttlefish.git", {tag, "2.0.1"}}}
        ]}.
 
 {port_env, [

src/eleveldb.erl

@@ -25,6 +25,7 @@
          close/1,
          get/3,
          put/4,
+         async_put/5,
          delete/3,
          write/3,
          fold/4,
@@ -101,7 +102,8 @@ init() ->
                          {delete_threshold, pos_integer()} |
                          {tiered_slow_level, pos_integer()} |
                          {tiered_fast_prefix, string()} |
-                         {tiered_slow_prefix, string()}].
+                         {tiered_slow_prefix, string()} |
+                         {cache_object_warming, boolean()}].
 
 -type read_options() :: [{verify_checksums, boolean()} |
                          {fill_cache, boolean()} |
@@ -161,6 +163,12 @@ write(Ref, Updates, Opts) ->
     async_write(CallerRef, Ref, Updates, Opts),
     ?WAIT_FOR_REPLY(CallerRef).
 
+-spec async_put(db_ref(), reference(), binary(), binary(), write_options()) -> ok.
+async_put(Ref, Context, Key, Value, Opts) ->
+    Updates = [{put, Key, Value}],
+    async_write(Context, Ref, Updates, Opts),
+    ok.
+
 -spec async_write(reference(), db_ref(), write_actions(), write_options()) -> ok.
 async_write(_CallerRef, _Ref, _Updates, _Opts) ->
     erlang:nif_error({error, not_loaded}).
@@ -293,7 +301,8 @@ option_types(open) ->
      {delete_threshold, integer},
      {tiered_slow_level, integer},
      {tiered_fast_prefix, any},
-     {tiered_slow_prefix, any}];
+     {tiered_slow_prefix, any},
+     {cache_object_warming, bool}];
 
 option_types(read) ->
     [{verify_checksums, bool},
@@ -480,13 +489,23 @@ values() ->
     eqc_gen:non_empty(list(binary())).
 
 ops(Keys, Values) ->
-    {oneof([put, delete]), oneof(Keys), oneof(Values)}.
+    {oneof([put, async_put, delete]), oneof(Keys), oneof(Values)}.
 
 apply_kv_ops([], _Ref, Acc0) ->
     Acc0;
 apply_kv_ops([{put, K, V} | Rest], Ref, Acc0) ->
     ok = eleveldb:put(Ref, K, V, []),
     apply_kv_ops(Rest, Ref, orddict:store(K, V, Acc0));
+apply_kv_ops([{async_put, K, V} | Rest], Ref, Acc0) ->
+    MyRef = make_ref(),
+    Context = {my_context, MyRef},
+    ok = eleveldb:async_put(Ref, Context, K, V, []),
+    receive
+        {Context, ok} ->
+            apply_kv_ops(Rest, Ref, orddict:store(K, V, Acc0));
+        Msg ->
+            error({unexpected_msg, Msg})
+    end;
 apply_kv_ops([{delete, K, _} | Rest], Ref, Acc0) ->
     ok = eleveldb:delete(Ref, K, []),
     apply_kv_ops(Rest, Ref, orddict:store(K, deleted, Acc0)).