add the possibility to do live backups

See
http://rocksdb.org/blog/2609/use-checkpoints-for-efficient-snapshots/

for more infos

Author: benoitc

c_src/erocksdb.cc

@@ -74,7 +74,7 @@ static ErlNifFunc nif_funcs[] =
 
     {"async_snapshot", 2, erocksdb::async_snapshot},
     {"async_release_snapshot", 2, erocksdb::async_release_snapshot},
-
+    {"async_checkpoint", 3, erocksdb::async_checkpoint},
 
     {"async_iterator", 3, erocksdb::async_iterator},
     {"async_iterator", 4, erocksdb::async_iterator},
@@ -1112,11 +1112,8 @@ async_iterator(
 
     // Parse out the read options
     rocksdb::ReadOptions *opts = new rocksdb::ReadOptions;
-
-    if(NULL!=snapshot)
-        opts->snapshot = snapshot;
-
     fold(env, options_ref, parse_read_option, *opts);
+    opts->snapshot = snapshot;
 
     erocksdb::WorkTask *work_item = new erocksdb::IterTask(env, caller_ref,
                                                            db_ptr.get(), keys_only, opts,
@@ -1289,6 +1286,52 @@ async_iterator_move(
 }   // async_iter_move
 
 
+ERL_NIF_TERM
+async_checkpoint(
+    ErlNifEnv* env,
+    int argc,
+    const ERL_NIF_TERM argv[])
+{
+    const ERL_NIF_TERM& caller_ref = argv[0];
+    const ERL_NIF_TERM& handle_ref = argv[1];
+
+    char path[4096];
+
+
+    ReferencePtr<DbObject> db_ptr;
+
+    db_ptr.assign(DbObject::RetrieveDbObject(env, handle_ref));
+
+    if(NULL==db_ptr.get())
+    {
+        return enif_make_badarg(env);
+    }
+
+    // is this even possible?
+    if(NULL == db_ptr->m_Db)
+        return send_reply(env, caller_ref, error_einval(env));
+
+    if(!enif_get_string(env, argv[2], path, sizeof(path), ERL_NIF_LATIN1))
+    {
+        return enif_make_badarg(env);
+    }
+
+    erocksdb_priv_data& priv = *static_cast<erocksdb_priv_data *>(enif_priv_data(env));
+
+    erocksdb::WorkTask *work_item = new erocksdb::CheckpointTask(env, caller_ref, db_ptr.get(), path);
+
+    if(false == priv.thread_pool.submit(work_item))
+    {
+        delete work_item;
+        return send_reply(env, caller_ref,
+                          enif_make_tuple2(env, erocksdb::ATOM_ERROR, caller_ref));
+    }
+
+    return erocksdb::ATOM_OK;
+
+}   // async_checkpoint
+
+
 } // namespace erocksdb
 
 

c_src/erocksdb.h

@@ -39,6 +39,7 @@ ERL_NIF_TERM erocksdb_status(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]
 ERL_NIF_TERM erocksdb_destroy(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
 ERL_NIF_TERM erocksdb_repair(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
 ERL_NIF_TERM erocksdb_is_empty(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
+ERL_NIF_TERM erocksdb_checkpoint(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
 }
 
 namespace erocksdb {
@@ -48,6 +49,7 @@ ERL_NIF_TERM async_snapshot(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
 ERL_NIF_TERM async_release_snapshot(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
 ERL_NIF_TERM async_write(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
 ERL_NIF_TERM async_get(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
+ERL_NIF_TERM async_checkpoint(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
 
 ERL_NIF_TERM async_iterator(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);
 ERL_NIF_TERM async_iterator_move(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]);

c_src/workitems.h

@@ -27,6 +27,7 @@
 
 #include "rocksdb/db.h"
 #include "rocksdb/write_batch.h"
+#include "rocksdb/utilities/checkpoint.h"
 
 
 #ifndef INCL_MUTEX_H
@@ -197,6 +198,49 @@ class ReleaseSnapshotTask : public WorkTask
 };  // class GetSnapshotTask
 
 
+/**
+ * Background object for async snapshot creation
+ */
+
+class CheckpointTask : public WorkTask
+{
+protected:
+    std::string path;
+
+public:
+    CheckpointTask(ErlNifEnv *_caller_env,
+                    ERL_NIF_TERM _caller_ref,
+                    DbObject *_db_handle,
+                    const std::string& path_)
+                : WorkTask(_caller_env, _caller_ref, _db_handle), 
+                path(path_)
+    {
+
+    };
+
+    virtual ~CheckpointTask() {};
+
+    virtual work_result operator()()
+    {
+        rocksdb::Checkpoint* checkpoint;
+        rocksdb::Status status;
+
+        status = rocksdb::Checkpoint::Create(m_DbPtr->m_Db, &checkpoint);
+
+        if (status.ok()) {
+            status = checkpoint->CreateCheckpoint(path);
+            if (status.ok())
+            {
+                return work_result(ATOM_OK);
+            }
+        }
+        checkpoint = NULL;
+
+        return work_result(local_env(), ATOM_ERROR, status);
+    }   // operator()
+
+};  // class CheckpointTask
+
 /**
  * Background object for async write
  */

src/erocksdb.erl

@@ -31,6 +31,7 @@
 -export([iterator/2, iterator/3, iterator_with_cf/3, iterator_move/2, iterator_close/1]).
 -export([fold/4, fold/5, fold_keys/4, fold_keys/5]).
 -export([destroy/2, repair/2, is_empty/1]).
+-export([checkpoint/2]).
 -export([count/1, count/2, status/1, status/2, status/3]).
 
 -export_type([db_handle/0,
@@ -473,6 +474,16 @@ destroy(_Name, _DBOpts) ->
 repair(_Name, _DBOpts) ->
     erlang:nif_error({error, not_loaded}).
 
+
+async_checkpoint(_Callerfef, _DbHandle, Path) ->
+    erlang:nif_error({error, not_loaded}).
+
+checkpoint(DbHandle, Path) ->
+    CallerRef = make_ref(),
+    async_checkpoint(CallerRef, DbHandle, Path),
+    ?WAIT_FOR_REPLY(CallerRef).
+
+
 %% @doc
 %% Return the approximate number of keys in the default column family.
 %% Implemented by calling GetIntProperty with "rocksdb.estimate-num-keys"