add rocksdb options manual_wal_flush support (#123)

librocksdb_sys/crocksdb/c.cc

@@ -1109,6 +1109,13 @@ void crocksdb_flush_cf(
   SaveError(errptr, db->rep->Flush(options->rep, column_family->rep));
 }
 
+void crocksdb_flush_wal(
+    crocksdb_t* db,
+    unsigned char sync,
+    char** errptr) {
+  SaveError(errptr, db->rep->FlushWAL(sync));
+}
+
 void crocksdb_sync_wal(
     crocksdb_t* db,
     char** errptr) {
@@ -2133,6 +2140,10 @@ void crocksdb_options_set_allow_concurrent_memtable_write(crocksdb_options_t* op
   opt->rep.allow_concurrent_memtable_write = v;
 }
 
+void crocksdb_options_set_manual_wal_flush(crocksdb_options_t* opt, unsigned char v) {
+  opt->rep.manual_wal_flush = v;
+}
+
 void crocksdb_options_set_enable_write_thread_adaptive_yield(
     crocksdb_options_t* opt, unsigned char v) {
   opt->rep.enable_write_thread_adaptive_yield = v;

librocksdb_sys/crocksdb/crocksdb/c.h

@@ -415,6 +415,9 @@ extern C_ROCKSDB_LIBRARY_API void crocksdb_flush_cf(
     crocksdb_t* db, crocksdb_column_family_handle_t* column_family,
     const crocksdb_flushoptions_t* options, char** errptr);
 
+extern C_ROCKSDB_LIBRARY_API void crocksdb_flush_wal(
+    crocksdb_t* db, unsigned char sync, char** errptr);
+
 extern C_ROCKSDB_LIBRARY_API void crocksdb_sync_wal(
     crocksdb_t* db, char** errptr);
 
@@ -880,6 +883,8 @@ crocksdb_options_set_enable_pipelined_write(crocksdb_options_t *,
 extern C_ROCKSDB_LIBRARY_API void
 crocksdb_options_set_allow_concurrent_memtable_write(crocksdb_options_t *,
                                                      unsigned char);
+extern C_ROCKSDB_LIBRARY_API void crocksdb_options_set_manual_wal_flush(
+    crocksdb_options_t *, unsigned char);
 extern C_ROCKSDB_LIBRARY_API void
 crocksdb_options_set_enable_write_thread_adaptive_yield(crocksdb_options_t*,
                                                        unsigned char);

librocksdb_sys/src/lib.rs

@@ -357,6 +357,7 @@ extern "C" {
     pub fn crocksdb_options_set_bytes_per_sync(options: *mut Options, bytes: u64);
     pub fn crocksdb_options_set_enable_pipelined_write(options: *mut Options, v: bool);
     pub fn crocksdb_options_set_allow_concurrent_memtable_write(options: *mut Options, v: bool);
+    pub fn crocksdb_options_set_manual_wal_flush(options: *mut Options, v: bool);
     pub fn crocksdb_options_optimize_for_point_lookup(
         options: *mut Options,
         block_cache_size_mb: u64,
@@ -848,6 +849,7 @@ extern "C" {
         options: *const DBFlushOptions,
         err: *mut *mut c_char,
     );
+    pub fn crocksdb_flush_wal(db: *mut DBInstance, sync: bool, err: *mut *mut c_char);
     pub fn crocksdb_sync_wal(db: *mut DBInstance, err: *mut *mut c_char);
 
     pub fn crocksdb_approximate_sizes(

src/rocksdb.rs

@@ -810,6 +810,15 @@ impl DB {
         }
     }
 
+    /// Flush the WAL memory buffer to the file. If sync is true, it calls SyncWAL
+    /// afterwards.
+    pub fn flush_wal(&self, sync: bool) -> Result<(), String> {
+        unsafe {
+            ffi_try!(crocksdb_flush_wal(self.inner, sync));
+            Ok(())
+        }
+    }
+
     /// Sync the wal. Note that Write() followed by SyncWAL() is not exactly the
     /// same as Write() with sync=true: in the latter case the changes won't be
     /// visible until the sync is done.

src/rocksdb_options.rs

@@ -724,6 +724,12 @@ impl DBOptions {
         }
     }
 
+    pub fn manual_wal_flush(&self, v: bool) {
+        unsafe {
+            crocksdb_ffi::crocksdb_options_set_manual_wal_flush(self.inner, v);
+        }
+    }
+
     /// the second parameter is a slice which contains tuples (path, target_size).
     pub fn set_db_paths<T: AsRef<Path>>(&self, val: &[(T, u64)]) {
         let num_paths = val.len();

tests/test_rocksdb_options.rs

@@ -169,6 +169,19 @@ fn test_set_wal_opt() {
     drop(db);
 }
 
+#[cfg(not(windows))]
+#[test]
+fn test_flush_wal() {
+    let path = TempDir::new("_rust_rocksdb_test_flush_wal").expect("");
+    let mut opts = DBOptions::new();
+    opts.create_if_missing(true);
+    opts.manual_wal_flush(true);
+    let db = DB::open(opts, path.path().to_str().unwrap()).unwrap();
+    db.put(b"key", b"value").unwrap();
+    db.flush_wal(false).unwrap();
+    drop(db);
+}
+
 #[cfg(not(windows))]
 #[test]
 fn test_sync_wal() {
@@ -411,6 +424,18 @@ fn test_allow_concurrent_memtable_write() {
     }
 }
 
+#[test]
+fn test_manual_wal_flush() {
+    let path = TempDir::new("_rust_rocksdb_manual_wal_flush").expect("");
+    let mut opts = DBOptions::new();
+    opts.create_if_missing(true);
+    opts.manual_wal_flush(true);
+    let db = DB::open(opts, path.path().to_str().unwrap()).unwrap();
+    for i in 0..200 {
+        db.put(format!("k_{}", i).as_bytes(), b"v").unwrap();
+    }
+}
+
 #[test]
 fn test_enable_pipelined_write() {
     let path = TempDir::new("_rust_rocksdb_enable_pipelined_write").expect("");