Merge pull request #203 from basho/mv-expiry

integrate open source expiry features into eleveldb option processing

c_src/eleveldb.cc

@@ -43,6 +43,7 @@
 #include "leveldb/perf_count.h"
 #define LEVELDB_PLATFORM_POSIX
 #include "util/hot_threads.h"
+#include "leveldb_os/expiry_os.h"
 
 #ifndef INCL_WORKITEMS_H
     #include "workitems.h"
@@ -133,6 +134,9 @@ 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;
+ERL_NIF_TERM ATOM_EXPIRY_ENABLED;
+ERL_NIF_TERM ATOM_EXPIRY_MINUTES;
+ERL_NIF_TERM ATOM_WHOLE_FILE_EXPIRY;
 }   // namespace eleveldb
 
 
@@ -217,6 +221,8 @@ class eleveldb_priv_data
 public:
     EleveldbOptions m_Opts;
     leveldb::HotThreadPool thread_pool;
+    leveldb::ExpiryPtr_t m_ExpiryModule;    // only populated if expiry options seen
+                                            //  (self deleting pointer)
 
     explicit eleveldb_priv_data(EleveldbOptions & Options)
     : m_Opts(Options),
@@ -459,6 +465,45 @@ ERL_NIF_TERM parse_open_option(ErlNifEnv* env, ERL_NIF_TERM item, leveldb::Optio
                 opts.cache_object_warming = false;
         }
 
+        else if (option[0] == eleveldb::ATOM_EXPIRY_ENABLED)
+        {
+            if (option[1] == eleveldb::ATOM_TRUE)
+            {
+                if (NULL==opts.expiry_module.get())
+                    opts.expiry_module.assign(new leveldb::ExpiryModuleOS);
+                ((leveldb::ExpiryModuleOS *)opts.expiry_module.get())->expiry_enabled = true;
+            }   // if
+            else
+            {
+                if (NULL!=opts.expiry_module.get())
+                    ((leveldb::ExpiryModuleOS *)opts.expiry_module.get())->expiry_enabled = false;
+            }   // else
+        }   // else if
+        else if (option[0] == eleveldb::ATOM_EXPIRY_MINUTES)
+        {
+            unsigned long minutes(0);
+            if (enif_get_ulong(env, option[1], &minutes))
+            {
+                if (NULL==opts.expiry_module.get())
+                    opts.expiry_module.assign(new leveldb::ExpiryModuleOS);
+                ((leveldb::ExpiryModuleOS *)opts.expiry_module.get())->expiry_minutes = minutes;
+            }   // if
+        }   // else if
+        else if (option[0] == eleveldb::ATOM_WHOLE_FILE_EXPIRY)
+        {
+            if (option[1] == eleveldb::ATOM_TRUE)
+            {
+                if (NULL==opts.expiry_module.get())
+                    opts.expiry_module.assign(new leveldb::ExpiryModuleOS);
+                ((leveldb::ExpiryModuleOS *)opts.expiry_module.get())->whole_file_expiry = true;
+            }   // if
+            else
+            {
+                if (NULL!=opts.expiry_module.get())
+                    ((leveldb::ExpiryModuleOS *)opts.expiry_module.get())->whole_file_expiry = false;
+            }   // else
+        }   // else if
+
     }
 
     return eleveldb::ATOM_OK;
@@ -568,8 +613,13 @@ async_open(
     eleveldb_priv_data& priv = *static_cast<eleveldb_priv_data *>(enif_priv_data(env));
 
     leveldb::Options *opts = new leveldb::Options;
+    opts->expiry_module.assign(priv.m_ExpiryModule.get());
+
     fold(env, argv[2], parse_open_option, *opts);
+
     opts->fadvise_willneed = priv.m_Opts.m_FadviseWillNeed;
+    if (NULL==priv.m_ExpiryModule.get() && NULL!=opts->expiry_module.get())
+        priv.m_ExpiryModule.assign(opts->expiry_module.get());
 
     // convert total_leveldb_mem to byte count if it arrived as percent
     //  This happens now because there is no guarantee as to when the total_memory
@@ -1282,6 +1332,9 @@ try
     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");
+    ATOM(eleveldb::ATOM_EXPIRY_ENABLED, "expiry_enabled");
+    ATOM(eleveldb::ATOM_EXPIRY_MINUTES, "expiry_minutes");
+    ATOM(eleveldb::ATOM_WHOLE_FILE_EXPIRY, "whole_file_expiry");
 #undef ATOM
 
 

c_src/refobjects.cc

@@ -415,12 +415,13 @@ LevelIteratorWrapper::LogIterator()
     struct tm created;
 
     localtime_r(&m_IteratorCreated, &created);
+#if 0 // available in different branch
     leveldb::Log(m_DbPtr->m_Db->GetLogger(),
                  "Iterator created %d/%d/%d %d:%d:%d, move operations %zd (%p)",
                  created.tm_mon, created.tm_mday, created.tm_year-100,
                  created.tm_hour, created.tm_min, created.tm_sec,
                  m_MoveCount, m_Iterator);
-
+#endif
 }   // LevelIteratorWrapper::LogIterator()
 
 

priv/eleveldb.schema

@@ -239,3 +239,31 @@
   {datatype, {enum, [true, false]}},
   hidden
 ]}.
+
+%% @doc Enable global expiry.  All leveldb databases / vnodes
+%% will use same expiry_minutes and whole_file_expiry settings.
+%% Same settings apply to all leveldb instances in multi_backend.
+{mapping, "leveldb.expiry_enabled", "eleveldb.expiry_enabled", [
+  {default, off},
+  {datatype, flag},
+  hidden
+]}.
+
+%% @doc Minutes until object expires.  Gives number of minutes
+%% a stored key/value will stay within the database before automatic
+%% deletion.  Zero disables expiry by age (minutes) feature
+{mapping, "leveldb.expiry_minutes", "eleveldb.expiry_minutes", [
+  {default, 0},
+  {datatype, integer},
+  hidden
+]}.
+
+%% @doc Expire entire .sst table file.  Authorizes leveldb to
+%% eliminate entire files that contain expired data (delete files
+%% instead of removing expired key/values during compaction).
+{mapping, "leveldb.whole_file_expiry", "eleveldb.whole_file_expiry", [
+  {default, off},
+  {datatype, flag},
+  hidden
+]}.
+

priv/eleveldb_multi.schema

@@ -165,3 +165,24 @@
   hidden
 ]}.
 
+%% @see leveldb.expiry_enabled
+{mapping, "multi_backend.$name.leveldbexpiry_enabled", "riak_kv.multi_backend", [
+  {default, off},
+  {datatype, flag},
+  hidden
+]}.
+
+%% @see leveldb.expiry_minutes
+{mapping, "multi_backend.$name.leveldbexpiry_minutes", "riak_kv.multi_backend", [
+  {default, 0},
+  {datatype, integer},
+  hidden
+]}.
+
+%% @see leveldb.whole_file_expiry
+{mapping, "multi_backend.$name.leveldbwhole_file_expiry", "riak_kv.multi_backend", [
+  {default, off},
+  {datatype, flag},
+  hidden
+]}.
+

src/eleveldb.erl

@@ -103,7 +103,11 @@ init() ->
                          {tiered_slow_level, pos_integer()} |
                          {tiered_fast_prefix, string()} |
                          {tiered_slow_prefix, string()} |
-                         {cache_object_warming, boolean()}].
+                         {cache_object_warming, boolean()} |
+                         {expiry_enabled, boolean()} |
+                         {expiry_minutes, pos_integer()} |
+                         {whole_file_expiry, boolean()}
+                        ].
 
 -type read_option() :: {verify_checksums, boolean()} |
                        {fill_cache, boolean()} |
@@ -307,7 +311,10 @@ option_types(open) ->
      {tiered_slow_level, integer},
      {tiered_fast_prefix, any},
      {tiered_slow_prefix, any},
-     {cache_object_warming, bool}];
+     {cache_object_warming, bool},
+     {expiry_enabled, bool},
+     {expiry_minutes, integer},
+     {whole_file_expiry, bool}];
 
 option_types(read) ->
     [{verify_checksums, bool},