From fc12cb83f25476e38f7ded6458036895705c680c Mon Sep 17 00:00:00 2001
From: Adam Retter <adam.retter@googlemail.com>
Date: Fri, 15 Aug 2014 13:34:10 +0100
Subject: [PATCH] Add locking to comparator jni callback methods which must be
 thread-safe

---
 java/Makefile                           |  2 +-
 java/org/rocksdb/Comparator.java        |  6 +--
 java/org/rocksdb/ComparatorOptions.java | 49 +++++++++++++++++++++++++
 java/org/rocksdb/DirectComparator.java  |  6 +--
 java/rocksjni/comparator.cc             | 32 ++++++++++++++--
 java/rocksjni/comparatorjnicallback.cc  | 27 ++++++++------
 java/rocksjni/comparatorjnicallback.h   | 35 ++++++++++++++++--
 java/rocksjni/options.cc                | 48 ++++++++++++++++++++++++
 java/rocksjni/portal.h                  | 27 ++++++++++++++
 9 files changed, 206 insertions(+), 26 deletions(-)
 create mode 100644 java/org/rocksdb/ComparatorOptions.java

diff --git a/java/Makefile b/java/Makefile
index 5c20032d2..19df0fa69 100644
--- a/java/Makefile
+++ b/java/Makefile
@@ -1,4 +1,4 @@
-NATIVE_JAVA_CLASSES = org.rocksdb.RocksDB org.rocksdb.Options org.rocksdb.WriteBatch org.rocksdb.WriteBatchInternal org.rocksdb.WriteBatchTest org.rocksdb.WriteOptions org.rocksdb.BackupableDB org.rocksdb.BackupableDBOptions org.rocksdb.Statistics org.rocksdb.RocksIterator org.rocksdb.VectorMemTableConfig org.rocksdb.SkipListMemTableConfig org.rocksdb.HashLinkedListMemTableConfig org.rocksdb.HashSkipListMemTableConfig org.rocksdb.PlainTableConfig org.rocksdb.BlockBasedTableConfig org.rocksdb.ReadOptions org.rocksdb.Filter org.rocksdb.BloomFilter org.rocksdb.AbstractComparator org.rocksdb.Comparator org.rocksdb.DirectComparator org.rocksdb.AbstractSlice org.rocksdb.Slice org.rocksdb.DirectSlice org.rocksdb.RestoreOptions org.rocksdb.RestoreBackupableDB org.rocksdb.RocksEnv org.rocksdb.GenericRateLimiterConfig org.rocksdb.ColumnFamilyHandle
+NATIVE_JAVA_CLASSES = org.rocksdb.RocksDB org.rocksdb.Options org.rocksdb.WriteBatch org.rocksdb.WriteBatchInternal org.rocksdb.WriteBatchTest org.rocksdb.WriteOptions org.rocksdb.BackupableDB org.rocksdb.BackupableDBOptions org.rocksdb.Statistics org.rocksdb.RocksIterator org.rocksdb.VectorMemTableConfig org.rocksdb.SkipListMemTableConfig org.rocksdb.HashLinkedListMemTableConfig org.rocksdb.HashSkipListMemTableConfig org.rocksdb.PlainTableConfig org.rocksdb.BlockBasedTableConfig org.rocksdb.ReadOptions org.rocksdb.Filter org.rocksdb.BloomFilter org.rocksdb.ComparatorOptions org.rocksdb.AbstractComparator org.rocksdb.Comparator org.rocksdb.DirectComparator org.rocksdb.AbstractSlice org.rocksdb.Slice org.rocksdb.DirectSlice org.rocksdb.RestoreOptions org.rocksdb.RestoreBackupableDB org.rocksdb.RocksEnv org.rocksdb.GenericRateLimiterConfig org.rocksdb.ColumnFamilyHandle
 
 ROCKSDB_MAJOR = $(shell egrep "ROCKSDB_MAJOR.[0-9]" ../include/rocksdb/version.h | cut -d ' ' -f 3)
 ROCKSDB_MINOR = $(shell egrep "ROCKSDB_MINOR.[0-9]" ../include/rocksdb/version.h | cut -d ' ' -f 3)
diff --git a/java/org/rocksdb/Comparator.java b/java/org/rocksdb/Comparator.java
index 8466cfd8e..7272a555a 100644
--- a/java/org/rocksdb/Comparator.java
+++ b/java/org/rocksdb/Comparator.java
@@ -16,10 +16,10 @@ package org.rocksdb;
  */
 public abstract class Comparator extends AbstractComparator<Slice> {
 
-  public Comparator() {
+  public Comparator(final ComparatorOptions copt) {
     super();
-    createNewComparator0();
+    createNewComparator0(copt.nativeHandle_);
   }
 
-  private native void createNewComparator0();
+  private native void createNewComparator0(final long comparatorOptionsHandle);
 }
diff --git a/java/org/rocksdb/ComparatorOptions.java b/java/org/rocksdb/ComparatorOptions.java
new file mode 100644
index 000000000..a55091dfa
--- /dev/null
+++ b/java/org/rocksdb/ComparatorOptions.java
@@ -0,0 +1,49 @@
+package org.rocksdb;
+
+public class ComparatorOptions extends RocksObject {
+
+  public ComparatorOptions() {
+    super();
+    newComparatorOptions();
+  }
+
+  /**
+   * Use adaptive mutex, which spins in the user space before resorting
+   * to kernel. This could reduce context switch when the mutex is not
+   * heavily contended. However, if the mutex is hot, we could end up
+   * wasting spin time.
+   * Default: false
+   *
+   * @return true if adaptive mutex is used.
+   */
+  public boolean useAdaptiveMutex() {
+    assert(isInitialized());
+    return useAdaptiveMutex(nativeHandle_);
+  }
+
+  /**
+   * Use adaptive mutex, which spins in the user space before resorting
+   * to kernel. This could reduce context switch when the mutex is not
+   * heavily contended. However, if the mutex is hot, we could end up
+   * wasting spin time.
+   * Default: false
+   *
+   * @param useAdaptiveMutex true if adaptive mutex is used.
+   * @return the reference to the current comparator options.
+   */
+  public ComparatorOptions setUseAdaptiveMutex(final boolean useAdaptiveMutex) {
+    assert (isInitialized());
+    setUseAdaptiveMutex(nativeHandle_, useAdaptiveMutex);
+    return this;
+  }
+
+  @Override protected void disposeInternal() {
+    assert(isInitialized());
+    disposeInternal(nativeHandle_);
+  }
+
+  private native void newComparatorOptions();
+  private native boolean useAdaptiveMutex(final long handle);
+  private native void setUseAdaptiveMutex(final long handle, final boolean useAdaptiveMutex);
+  private native void disposeInternal(long handle);
+}
diff --git a/java/org/rocksdb/DirectComparator.java b/java/org/rocksdb/DirectComparator.java
index 25b4058ae..86476c40e 100644
--- a/java/org/rocksdb/DirectComparator.java
+++ b/java/org/rocksdb/DirectComparator.java
@@ -16,10 +16,10 @@ package org.rocksdb;
  */
 public abstract class DirectComparator extends AbstractComparator<DirectSlice> {
 
-  public DirectComparator() {
+  public DirectComparator(final ComparatorOptions copt) {
     super();
-    createNewDirectComparator0();
+    createNewDirectComparator0(copt.nativeHandle_);
   }
 
-  private native void createNewDirectComparator0();
+  private native void createNewDirectComparator0(final long comparatorOptionsHandle);
 }
diff --git a/java/rocksjni/comparator.cc b/java/rocksjni/comparator.cc
index 54d6137cd..8dcda6aa6 100644
--- a/java/rocksjni/comparator.cc
+++ b/java/rocksjni/comparator.cc
@@ -18,6 +18,28 @@
 #include "rocksjni/comparatorjnicallback.h"
 #include "rocksjni/portal.h"
 
+//<editor-fold desc="org.rocksdb.ComparatorOptions">
+
+void Java_org_rocksdb_ComparatorOptions_newComparatorOptions(
+    JNIEnv* env, jobject jobj, jstring jpath, jboolean jshare_table_files,
+    jboolean jsync, jboolean jdestroy_old_data, jboolean jbackup_log_files,
+    jlong jbackup_rate_limit, jlong jrestore_rate_limit) {
+  jbackup_rate_limit = (jbackup_rate_limit <= 0) ? 0 : jbackup_rate_limit;
+  jrestore_rate_limit = (jrestore_rate_limit <= 0) ? 0 : jrestore_rate_limit;
+
+  const char* cpath = env->GetStringUTFChars(jpath, 0);
+
+  auto bopt = new rocksdb::BackupableDBOptions(cpath, nullptr,
+      jshare_table_files, nullptr, jsync, jdestroy_old_data, jbackup_log_files,
+      jbackup_rate_limit, jrestore_rate_limit);
+
+  env->ReleaseStringUTFChars(jpath, cpath);
+
+  rocksdb::BackupableDBOptionsJni::setHandle(env, jobj, bopt);
+}
+
+//</editor-fold>
+
 //<editor-fold desc="org.rocksdb.AbstractComparator>
 
 /*
@@ -40,8 +62,9 @@ void Java_org_rocksdb_AbstractComparator_disposeInternal(
  * Signature: ()V
  */
 void Java_org_rocksdb_Comparator_createNewComparator0(
-    JNIEnv* env, jobject jobj) {
-  const rocksdb::ComparatorJniCallback* c = new rocksdb::ComparatorJniCallback(env, jobj);
+    JNIEnv* env, jobject jobj, jlong copt_handle) {
+  const rocksdb::ComparatorJniCallbackOptions* copt = reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(copt_handle);
+  const rocksdb::ComparatorJniCallback* c = new rocksdb::ComparatorJniCallback(env, jobj, copt);
   rocksdb::AbstractComparatorJni::setHandle(env, jobj, c);
 }
 
@@ -55,8 +78,9 @@ void Java_org_rocksdb_Comparator_createNewComparator0(
  * Signature: ()V
  */
 void Java_org_rocksdb_DirectComparator_createNewDirectComparator0(
-    JNIEnv* env, jobject jobj) {
-  const rocksdb::DirectComparatorJniCallback* c = new rocksdb::DirectComparatorJniCallback(env, jobj);
+    JNIEnv* env, jobject jobj, jlong copt_handle) {
+  const rocksdb::ComparatorJniCallbackOptions* copt = reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(copt_handle);
+  const rocksdb::DirectComparatorJniCallback* c = new rocksdb::DirectComparatorJniCallback(env, jobj, copt);
   rocksdb::AbstractComparatorJni::setHandle(env, jobj, c);
 }
 
diff --git a/java/rocksjni/comparatorjnicallback.cc b/java/rocksjni/comparatorjnicallback.cc
index 6d80547d5..22bba334c 100644
--- a/java/rocksjni/comparatorjnicallback.cc
+++ b/java/rocksjni/comparatorjnicallback.cc
@@ -11,7 +11,11 @@
 
 namespace rocksdb {
 BaseComparatorJniCallback::BaseComparatorJniCallback(
-    JNIEnv* env, jobject jComparator) {
+    JNIEnv* env, jobject jComparator, const ComparatorJniCallbackOptions* copt) {
+
+  //mutex is used for synchronisation when we are re-using
+  //the global java slice objects
+  mutex_ = new port::Mutex(copt->use_adaptive_mutex);
 
   // Note: Comparator methods may be accessed by multiple threads,
   // so we ref the jvm not the env
@@ -51,11 +55,14 @@ int BaseComparatorJniCallback::Compare(const Slice& a, const Slice& b) const {
 
   JNIEnv* m_env = getJniEnv();
 
+  mutex_->Lock();
+
   AbstractSliceJni::setHandle(m_env, m_jSliceA, &a);
   AbstractSliceJni::setHandle(m_env, m_jSliceB, &b);
-
   jint result = m_env->CallIntMethod(m_jComparator, m_jCompareMethodId, m_jSliceA, m_jSliceB);
 
+  mutex_->Unlock();
+
   m_jvm->DetachCurrentThread();
 
   return result;
@@ -72,10 +79,13 @@ void BaseComparatorJniCallback::FindShortestSeparator(std::string* start, const
   const char* startUtf = start->c_str();
   jstring jsStart = m_env->NewStringUTF(startUtf);
 
-  AbstractSliceJni::setHandle(m_env, m_jSliceLimit, &limit);
+  mutex_->Lock();
 
+  AbstractSliceJni::setHandle(m_env, m_jSliceLimit, &limit);
   jstring jsResultStart = (jstring)m_env->CallObjectMethod(m_jComparator, m_jFindShortestSeparatorMethodId, jsStart, m_jSliceLimit);
 
+  mutex_->Unlock();
+
   m_env->DeleteLocalRef(jsStart);
 
   if(jsResultStart != nullptr) {
@@ -110,13 +120,6 @@ void BaseComparatorJniCallback::FindShortSuccessor(std::string* key) const {
 }
 
 BaseComparatorJniCallback::~BaseComparatorJniCallback() {
-
-  // NOTE: we do not need to delete m_name here,
-  // I am not yet sure why, but doing so causes the error:
-  //   java(13051,0x109f54000) malloc: *** error for object 0x109f52fa9: pointer being freed was not allocated
-  //   *** set a breakpoint in malloc_error_break to debug
-  //delete[] m_name;
-
   JNIEnv* m_env = getJniEnv();
   
   m_env->DeleteGlobalRef(m_jComparator);
@@ -131,7 +134,7 @@ BaseComparatorJniCallback::~BaseComparatorJniCallback() {
 }
 
 ComparatorJniCallback::ComparatorJniCallback(
-    JNIEnv* env, jobject jComparator) : BaseComparatorJniCallback(env, jComparator) {
+    JNIEnv* env, jobject jComparator, const ComparatorJniCallbackOptions* copt) : BaseComparatorJniCallback(env, jComparator, copt) {
 
   m_jSliceA = env->NewGlobalRef(SliceJni::construct0(env));
   m_jSliceB = env->NewGlobalRef(SliceJni::construct0(env));
@@ -139,7 +142,7 @@ ComparatorJniCallback::ComparatorJniCallback(
 }
 
 DirectComparatorJniCallback::DirectComparatorJniCallback(
-    JNIEnv* env, jobject jComparator) : BaseComparatorJniCallback(env, jComparator) {
+    JNIEnv* env, jobject jComparator, const ComparatorJniCallbackOptions* copt) : BaseComparatorJniCallback(env, jComparator, copt) {
 
   m_jSliceA = env->NewGlobalRef(DirectSliceJni::construct0(env));
   m_jSliceB = env->NewGlobalRef(DirectSliceJni::construct0(env));
diff --git a/java/rocksjni/comparatorjnicallback.h b/java/rocksjni/comparatorjnicallback.h
index f188dce86..8ca0ac64f 100644
--- a/java/rocksjni/comparatorjnicallback.h
+++ b/java/rocksjni/comparatorjnicallback.h
@@ -12,11 +12,39 @@
 #include <jni.h>
 #include "rocksdb/comparator.h"
 #include "rocksdb/slice.h"
+#include "port/port.h"
 
 namespace rocksdb {
+
+struct ComparatorJniCallbackOptions {
+  // Use adaptive mutex, which spins in the user space before resorting
+  // to kernel. This could reduce context switch when the mutex is not
+  // heavily contended. However, if the mutex is hot, we could end up
+  // wasting spin time.
+  // Default: false
+  bool use_adaptive_mutex;
+
+  ComparatorJniCallbackOptions() : use_adaptive_mutex(false) {
+  }
+};
+
+/**
+ * This class acts as a bridge between C++
+ * and Java. The methods in this class will be
+ * called back from the RocksDB storage engine (C++)
+ * we then callback to the appropriate Java method
+ * this enables Comparators to be implemented in Java.
+ *
+ * The design of this Comparator caches the Java Slice
+ * objects that are used in the compare and findShortestSeparator
+ * method callbacks. Instead of creating new objects for each callback
+ * of those functions, by reuse via setHandle we are a lot
+ * faster; Unfortunately this means that we have to
+ * introduce locking in regions of those methods via mutex_.
+ */
 class BaseComparatorJniCallback : public Comparator {
   public:
-    BaseComparatorJniCallback(JNIEnv* env, jobject jComparator);
+    BaseComparatorJniCallback(JNIEnv* env, jobject jComparator, const ComparatorJniCallbackOptions* copt);
     virtual ~BaseComparatorJniCallback();
     virtual const char* Name() const;
     virtual int Compare(const Slice& a, const Slice& b) const;
@@ -24,6 +52,7 @@ class BaseComparatorJniCallback : public Comparator {
     virtual void FindShortSuccessor(std::string* key) const;
 
   private:
+    port::Mutex* mutex_;
     JavaVM* m_jvm;
     jobject m_jComparator;
     std::string m_name;
@@ -40,12 +69,12 @@ class BaseComparatorJniCallback : public Comparator {
 
 class ComparatorJniCallback : public BaseComparatorJniCallback {
     public:
-      ComparatorJniCallback(JNIEnv* env, jobject jComparator);
+      ComparatorJniCallback(JNIEnv* env, jobject jComparator, const ComparatorJniCallbackOptions* copt);
 };
 
 class DirectComparatorJniCallback : public BaseComparatorJniCallback {
     public:
-      DirectComparatorJniCallback(JNIEnv* env, jobject jComparator);
+      DirectComparatorJniCallback(JNIEnv* env, jobject jComparator, const ComparatorJniCallbackOptions* copt);
 };
 }  // namespace rocksdb
 
diff --git a/java/rocksjni/options.cc b/java/rocksjni/options.cc
index 8e94f965b..4367fc708 100644
--- a/java/rocksjni/options.cc
+++ b/java/rocksjni/options.cc
@@ -14,6 +14,8 @@
 #include "include/org_rocksdb_Options.h"
 #include "include/org_rocksdb_WriteOptions.h"
 #include "include/org_rocksdb_ReadOptions.h"
+#include "include/org_rocksdb_ComparatorOptions.h"
+
 #include "rocksjni/portal.h"
 #include "rocksdb/db.h"
 #include "rocksdb/options.h"
@@ -23,6 +25,7 @@
 #include "rocksdb/slice_transform.h"
 #include "rocksdb/rate_limiter.h"
 #include "rocksdb/comparator.h"
+#include "comparatorjnicallback.h"
 
 /*
  * Class:     org_rocksdb_Options
@@ -1770,3 +1773,48 @@ void Java_org_rocksdb_ReadOptions_setTailing(
   reinterpret_cast<rocksdb::ReadOptions*>(jhandle)->tailing =
       static_cast<bool>(jtailing);
 }
+
+/////////////////////////////////////////////////////////////////////
+// rocksdb::ComparatorOptions
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    newComparatorOptions
+ * Signature: ()V
+ */
+void Java_org_rocksdb_ComparatorOptions_newComparatorOptions(
+    JNIEnv* env, jobject jobj) {
+  auto comparator_opt = new rocksdb::ComparatorJniCallbackOptions();
+  rocksdb::ComparatorOptionsJni::setHandle(env, jobj, comparator_opt);
+}
+
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    useAdaptiveMutex
+ * Signature: (J)Z
+ */
+jboolean Java_org_rocksdb_ComparatorOptions_useAdaptiveMutex(
+    JNIEnv * env, jobject jobj, jlong jhandle) {
+  return reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(jhandle)->use_adaptive_mutex;
+}
+
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    setUseAdaptiveMutex
+ * Signature: (JZ)V
+ */
+void Java_org_rocksdb_ComparatorOptions_setUseAdaptiveMutex(
+    JNIEnv * env, jobject jobj, jlong jhandle, jboolean juse_adaptive_mutex) {
+  reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(jhandle)->use_adaptive_mutex =
+      static_cast<bool>(juse_adaptive_mutex);
+}
+
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_ComparatorOptions_disposeInternal(
+    JNIEnv * env, jobject jobj, jlong jhandle) {
+  delete reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(jhandle);
+  rocksdb::ComparatorOptionsJni::setHandle(env, jobj, nullptr);
+}
diff --git a/java/rocksjni/portal.h b/java/rocksjni/portal.h
index 68403ebff..bd40d4290 100644
--- a/java/rocksjni/portal.h
+++ b/java/rocksjni/portal.h
@@ -363,6 +363,33 @@ class ColumnFamilyHandleJni {
   }
 };
 
+class ComparatorOptionsJni {
+  public:
+    // Get the java class id of org.rocksdb.ComparatorOptions.
+    static jclass getJClass(JNIEnv* env) {
+      jclass jclazz = env->FindClass("org/rocksdb/ComparatorOptions");
+      assert(jclazz != nullptr);
+      return jclazz;
+    }
+
+    // Get the field id of the member variable of org.rocksdb.ComparatorOptions
+    // that stores the pointer to rocksdb::ComparatorJniCallbackOptions.
+    static jfieldID getHandleFieldID(JNIEnv* env) {
+      static jfieldID fid = env->GetFieldID(
+          getJClass(env), "nativeHandle_", "J");
+      assert(fid != nullptr);
+      return fid;
+    }
+
+    // Pass the ComparatorJniCallbackOptions pointer to the java side.
+    static void setHandle(
+        JNIEnv* env, jobject jobj, const rocksdb::ComparatorJniCallbackOptions* op) {
+      env->SetLongField(
+          jobj, getHandleFieldID(env),
+          reinterpret_cast<jlong>(op));
+    }
+};
+
 class AbstractComparatorJni {
  public:
   // Get the java class id of org.rocksdb.Comparator.