added sync mechanism

src/datastore/berkeleydb_datastore.cc

@@ -180,7 +180,11 @@ bool BerkeleyDBDataStore::erase(const ds_bulk_t &key) {
     Dbt db_key((void*)key.data(), key.size());
     int status = _dbm->del(NULL, &db_key, 0);
     return status == 0;
-} 
+}
+
+void BerkeleyDBDataStore::sync() {
+    _dbm->sync(0);
+}
 
 // In the case where Duplicates::ALLOW, this will return the first
 // value found using key.

src/datastore/berkeleydb_datastore.h

@@ -39,6 +39,7 @@ class BerkeleyDBDataStore : public AbstractDataStore {
         virtual void set_no_overwrite() {
             _no_overwrite = true;
         }
+        virtual void sync();
         remi_fileset_t create_and_populate_fileset() const;
     protected:
         virtual std::vector<ds_bulk_t> vlist_keys(

src/datastore/datastore.h

@@ -28,6 +28,7 @@ class AbstractDataStore {
         virtual void set_in_memory(bool enable)=0; // enable/disable in-memory mode (where supported)
         virtual void set_comparison_function(const std::string& name, comparator_fn less)=0;
         virtual void set_no_overwrite()=0;
+        virtual void sync() = 0;
         virtual remi_fileset_t create_and_populate_fileset() const = 0;
 
         const std::string& get_path() const {

src/datastore/leveldb_datastore.cc

@@ -35,6 +35,10 @@ LevelDBDataStore::~LevelDBDataStore() {
   //leveldb::Env::Shutdown(); // Riak version only
 };
 
+void LevelDBDataStore::sync() {
+
+}
+
 bool LevelDBDataStore::openDatabase(const std::string& db_name, const std::string& db_path) {
     _name = db_name;
     _path = db_path;

src/datastore/leveldb_datastore.h

@@ -52,6 +52,7 @@ class LevelDBDataStore : public AbstractDataStore {
         virtual void set_no_overwrite() {
             _no_overwrite = true;
         }
+        virtual void sync();
         remi_fileset_t create_and_populate_fileset() const;
     protected:
         virtual std::vector<ds_bulk_t> vlist_keys(

src/datastore/map_datastore.h

@@ -44,6 +44,8 @@ class MapDataStore : public AbstractDataStore {
             return true;
         }
 
+        virtual void sync() {}
+
         virtual bool put(const ds_bulk_t &key, const ds_bulk_t &data) {
             auto x = _map.count(key);
             if(_no_overwrite && (x != 0)) {

src/sdskv-server.cc

@@ -2016,7 +2016,7 @@ static void sdskv_migrate_database_ult(hg_handle_t handle)
             break;
         }
         auto database = it->second;
-
+        database->sync();
         /* lookup the address of the destination REMI provider */
         hret = margo_addr_lookup(mid, in.dest_remi_addr, &dest_addr);
         if(hret != HG_SUCCESS) {