Merge pull request #430 from adamretter/increase-parallelism

Added setIncreaseParallelism() to Java API Options

java/org/rocksdb/DBOptions.java

@@ -72,6 +72,13 @@ public class DBOptions extends RocksObject implements DBOptionsInterface {
     return dbOptions;
   }
 
+  @Override
+  public DBOptions setIncreaseParallelism(int totalThreads) {
+    assert (isInitialized());
+    setIncreaseParallelism(nativeHandle_, totalThreads);
+    return this;
+  }
+
   @Override
   public DBOptions setCreateIfMissing(boolean flag) {
     assert(isInitialized());
@@ -547,6 +554,7 @@ public class DBOptions extends RocksObject implements DBOptionsInterface {
   private native void newDBOptions();
   private native void disposeInternal(long handle);
 
+  private native void setIncreaseParallelism(long handle, int totalThreads);
   private native void setCreateIfMissing(long handle, boolean flag);
   private native boolean createIfMissing(long handle);
   private native void setCreateMissingColumnFamilies(

java/org/rocksdb/DBOptionsInterface.java

@@ -7,6 +7,21 @@ package org.rocksdb;
 
 public interface DBOptionsInterface {
 
+  /**
+   * <p>By default, RocksDB uses only one background thread for flush and
+   * compaction. Calling this function will set it up such that total of
+   * `total_threads` is used.</p>
+   *
+   * <p>You almost definitely want to call this function if your system is
+   * bottlenecked by RocksDB.</p>
+   *
+   * @param The total number of threads to be used by RocksDB. A good value
+   *            is the number of cores.
+   *
+   * @return the instance of the current Options
+   */
+  Object setIncreaseParallelism(int totalThreads);
+
   /**
    * If this value is set to true, then the database will be created
    * if it is missing during {@code RocksDB.open()}.

java/org/rocksdb/Options.java

@@ -43,6 +43,13 @@ public class Options extends RocksObject
     env_ = RocksEnv.getDefault();
   }
 
+  @Override
+  public Options setIncreaseParallelism(int totalThreads) {
+    assert(isInitialized());
+    setIncreaseParallelism(nativeHandle_, totalThreads);
+    return this;
+  }
+
   @Override
   public Options setCreateIfMissing(boolean flag) {
     assert(isInitialized());
@@ -1032,6 +1039,7 @@ public class Options extends RocksObject
   private native void prepareForBulkLoad(long handle);
 
   // DB native handles
+  private native void setIncreaseParallelism(long handle, int totalThreads);
   private native void setCreateIfMissing(long handle, boolean flag);
   private native boolean createIfMissing(long handle);
   private native void setCreateMissingColumnFamilies(

java/org/rocksdb/test/DBOptionsTest.java

@@ -73,6 +73,20 @@ public class DBOptionsTest {
         new Properties());
   }
 
+  @Test
+  public void setIncreaseParallelism() {
+    DBOptions opt = null;
+    try {
+      opt = new DBOptions();
+      final int threads = Runtime.getRuntime().availableProcessors() * 2;
+      opt.setIncreaseParallelism(threads);
+    } finally {
+      if (opt != null) {
+        opt.dispose();
+      }
+    }
+  }
+
   @Test
   public void createIfMissing() {
     DBOptions opt = null;

java/org/rocksdb/test/OptionsTest.java

@@ -22,6 +22,20 @@ public class OptionsTest {
   public static final Random rand = PlatformRandomHelper.
       getPlatformSpecificRandomFactory();
 
+  @Test
+  public void setIncreaseParallelism() {
+    Options opt = null;
+    try {
+      opt = new Options();
+      final int threads = Runtime.getRuntime().availableProcessors() * 2;
+      opt.setIncreaseParallelism(threads);
+    } finally {
+      if (opt != null) {
+        opt.dispose();
+      }
+    }
+  }
+
   @Test
   public void writeBufferSize() throws RocksDBException {
     Options opt = null;

java/rocksjni/options.cc

@@ -68,6 +68,17 @@ void Java_org_rocksdb_Options_disposeInternal(
   delete reinterpret_cast<rocksdb::Options*>(handle);
 }
 
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    setIncreaseParallelism
+ * Signature: (JI)V
+ */
+void Java_org_rocksdb_Options_setIncreaseParallelism(
+    JNIEnv * evnv, jobject jobj, jlong jhandle, jint totalThreads) {
+  reinterpret_cast<rocksdb::Options*>
+      (jhandle)->IncreaseParallelism(static_cast<int>(totalThreads));
+}
+
 /*
  * Class:     org_rocksdb_Options
  * Method:    setCreateIfMissing
@@ -2816,6 +2827,18 @@ void Java_org_rocksdb_DBOptions_disposeInternal(
   delete reinterpret_cast<rocksdb::DBOptions*>(handle);
 }
 
+/*
+ * Class:     org_rocksdb_DBOptions
+ * Method:    setIncreaseParallelism
+ * Signature: (JI)V
+ */
+void Java_org_rocksdb_DBOptions_setIncreaseParallelism(
+    JNIEnv * env, jobject jobj, jlong jhandle, jint totalThreads) {
+  reinterpret_cast<rocksdb::DBOptions*>
+      (jhandle)->IncreaseParallelism(static_cast<int>(totalThreads));
+}
+
+
 /*
  * Class:     org_rocksdb_DBOptions
  * Method:    setCreateIfMissing