[Java] Generalize dis-own native handle and refine dispose framework.

Summary:
1. Move disOwnNativeHandle() function from RocksDB to RocksObject
to allow other RocksObject to use disOwnNativeHandle() when its
ownership of native handle has been transferred.

2. RocksObject now has an abstract implementation of dispose(),
which does the following two things.  First, it checks whether
both isOwningNativeHandle() and isInitialized() return true.
If so, it will call the protected abstract function dispose0(),
which all the subclasses of RocksObject should implement.  Second,
it sets nativeHandle_ = 0.  This redesign ensure all subclasses
of RocksObject have the same dispose behavior.

3. All subclasses of RocksObject now should implement dispose0()
instead of dispose(), and dispose0() will be called only when
isInitialized() returns true.

Test Plan:
make rocksdbjava
make jtest

Reviewers: dhruba, sdong, ankgup87, rsumbaly, swapnilghike, zzbennett, haobo

Reviewed By: haobo

Subscribers: leveldb

Differential Revision: https://reviews.facebook.net/D18801

java/org/rocksdb/BackupableDB.java

@@ -32,7 +32,7 @@ public class BackupableDB extends RocksDB {
 
     // Prevent the RocksDB object from attempting to delete
     // the underly C++ DB object.
-    db.disOwnNativeObject();
+    db.disOwnNativeHandle();
 
     return bdb;
   }

java/org/rocksdb/BackupableDBOptions.java

@@ -32,13 +32,12 @@ public class BackupableDBOptions extends RocksObject {
    * Release the memory allocated for the current instance
    * in the c++ side.
    */
-  @Override public synchronized void dispose() {
+  @Override protected void disposeInternal() {
-    if (isInitialized()) {
+    assert(isInitialized());
-      dispose(nativeHandle_);
+    disposeInternal(nativeHandle_);
-    }
   }
 
   private native void newBackupableDBOptions(String path);
   private native String backupDir(long handle);
-  private native void dispose(long handle);
+  private native void disposeInternal(long handle);
 }

java/org/rocksdb/Filter.java

@@ -22,11 +22,10 @@ public abstract class Filter extends RocksObject {
    * RocksDB instances referencing the filter are closed.
    * Otherwise an undefined behavior will occur.
    */
-  @Override public synchronized void dispose() {
+  @Override protected void disposeInternal() {
-    if (isInitialized()) {
+    assert(isInitialized());
-      dispose0(nativeHandle_);
+    disposeInternal(nativeHandle_);
-    }
   }
 
-  private native void dispose0(long handle);
+  private native void disposeInternal(long handle);
 }

java/org/rocksdb/Options.java

@@ -2311,10 +2311,9 @@ public class Options extends RocksObject {
    * Release the memory allocated for the current instance
    * in the c++ side.
    */
-  @Override public synchronized void dispose() {
+  @Override protected void disposeInternal() {
-    if (isInitialized()) {
+    assert(isInitialized());
-      dispose0();
+    disposeInternal(nativeHandle_);
-    }
   }
 
   static final int DEFAULT_PLAIN_TABLE_BLOOM_BITS_PER_KEY = 10;
@@ -2322,7 +2321,7 @@ public class Options extends RocksObject {
   static final int DEFAULT_PLAIN_TABLE_INDEX_SPARSENESS = 16;
 
   private native void newOptions();
-  private native void dispose0();
+  private native void disposeInternal(long handle);
   private native void setCreateIfMissing(long handle, boolean flag);
   private native boolean createIfMissing(long handle);
   private native void setWriteBufferSize(long handle, long writeBufferSize);

java/org/rocksdb/ReadOptions.java

@@ -18,19 +18,6 @@ public class ReadOptions extends RocksObject {
   }
   private native void newReadOptions();
 
-  /**
-   * Release the memory allocated for the current instance
-   * in the c++ side.
-   *
-   * Calling other methods after dispose() leads to undefined behavior.
-   */
-  @Override public synchronized void dispose() {
-    if (isInitialized()) {
-      dispose(nativeHandle_);
-    }
-  }
-  private native void dispose(long handle);
-
   /**
    * If true, all data read from underlying storage will be
    * verified against corresponding checksums.
@@ -127,4 +114,12 @@ public class ReadOptions extends RocksObject {
   }
   private native void setTailing(
       long handle, boolean tailing);
+
+
+  @Override protected void disposeInternal() {
+    assert(isInitialized());
+    disposeInternal(nativeHandle_);
+  }
+  private native void disposeInternal(long handle);
+
 }

java/org/rocksdb/RocksDB.java

@@ -114,11 +114,9 @@ public class RocksDB extends RocksObject {
     return db;
   }
 
-  @Override public synchronized void dispose() {
+  @Override protected void disposeInternal() {
-    if (isInitialized()) {
+    assert(isInitialized());
-      dispose(nativeHandle_);
+    disposeInternal(nativeHandle_);
-      nativeHandle_ = 0;
-    }
   }
 
   /**
@@ -315,10 +313,6 @@ public class RocksDB extends RocksObject {
     return new RocksIterator(iterator0(nativeHandle_));
   }
 
-  @Override protected void finalize() {
-    close();
-  }
-
   /**
    * Private constructor.
    */
@@ -337,18 +331,6 @@ public class RocksDB extends RocksObject {
     opt.filter_ = null;
   }
 
-  /**
-   * Revoke ownership of the native object.
-   *
-   * This will prevent the object from attempting to delete the underlying
-   * native object in its finalizer. This must be used when another object
-   * (e.g. BackupableDB) takes over ownership of the native object or both
-   * will attempt to delete the underlying object when garbage collected.
-   */
-  protected void disOwnNativeObject() {
-    nativeHandle_ = 0;
-  }
-
   // native methods
   protected native void open(
       long optionsHandle, long cacheSize, String path) throws RocksDBException;
@@ -382,7 +364,7 @@ public class RocksDB extends RocksObject {
       long handle, long writeOptHandle,
       byte[] key, int keyLen) throws RocksDBException;
   protected native long iterator0(long optHandle);
-  protected native void dispose(long handle);
+  private native void disposeInternal(long handle);
 
   protected Filter filter_;
 }

java/org/rocksdb/RocksIterator.java

@@ -118,15 +118,13 @@ public class RocksIterator extends RocksObject {
   /**
    * Deletes underlying C++ iterator pointer.
    */
-  @Override public synchronized void dispose() {
+  @Override protected void disposeInternal() {
-    if(isInitialized()) {
+    assert(isInitialized());
-      dispose(nativeHandle_);
+    disposeInternal(nativeHandle_);
-      nativeHandle_ = 0;
-    }
   }
 
   private native boolean isValid0(long handle);
-  private native void dispose(long handle);
+  private native void disposeInternal(long handle);
   private native void seekToFirst0(long handle);
   private native void seekToLast0(long handle);
   private native void next0(long handle);

java/org/rocksdb/RocksObject.java

@@ -16,12 +16,48 @@ package org.rocksdb;
 public abstract class RocksObject {
   protected RocksObject() {
     nativeHandle_ = 0;
+    owningHandle_ = true;
   }
 
   /**
    * Release the c++ object pointed by the native handle.
+   *
+   * Note that once an instance of RocksObject has been disposed,
+   * calling its function will lead undefined behavior.
    */
-  public abstract void dispose();
+  public final synchronized void dispose() {
+    if (isOwningNativeHandle() && isInitialized()) {
+      disposeInternal();
+    }
+    nativeHandle_ = 0;
+    disOwnNativeHandle();
+  }
+
+  /**
+   * The helper function of dispose() which all subclasses of RocksObject
+   * must implement to release their associated C++ resource.
+   */
+  protected abstract void disposeInternal();
+
+  /**
+   * Revoke ownership of the native object.
+   *
+   * This will prevent the object from attempting to delete the underlying
+   * native object in its finalizer. This must be used when another object
+   * takes over ownership of the native object or both will attempt to delete
+   * the underlying object when garbage collected.
+   *
+   * When disOwnNativeHandle is called, dispose() will simply set nativeHandle_
+   * to 0 without releasing its associated C++ resource.  As a result,
+   * incorrectly use this function may cause memory leak.
+   */
+  protected void disOwnNativeHandle() {
+    owningHandle_ = false;
+  }
+
+  protected boolean isOwningNativeHandle() {
+    return owningHandle_;
+  }
 
   protected boolean isInitialized() {
     return (nativeHandle_ != 0);
@@ -32,4 +68,5 @@ public abstract class RocksObject {
   }
 
   protected long nativeHandle_;
+  private boolean owningHandle_;
 }

java/org/rocksdb/WriteBatch.java

@@ -86,10 +86,9 @@ public class WriteBatch extends RocksObject {
   /**
    * Delete the c++ side pointer.
    */
-  @Override public synchronized void dispose() {
+  @Override protected void disposeInternal() {
-    if (isInitialized()) {
+    assert(isInitialized());
-      dispose0();
+    disposeInternal(nativeHandle_);
-    }
   }
 
   private native void newWriteBatch(int reserved_bytes);
@@ -99,7 +98,7 @@ public class WriteBatch extends RocksObject {
                             byte[] value, int valueLen);
   private native void remove(byte[] key, int keyLen);
   private native void putLogData(byte[] blob, int blobLen);
-  private native void dispose0();
+  private native void disposeInternal(long handle);
 }
 
 /**

java/org/rocksdb/WriteOptions.java

@@ -17,10 +17,9 @@ public class WriteOptions extends RocksObject {
     newWriteOptions();
   }
 
-  @Override public synchronized void dispose() {
+  @Override protected void disposeInternal() {
-    if (isInitialized()) {
+    assert(isInitialized());
-      dispose0(nativeHandle_);
+    disposeInternal(nativeHandle_);
-    }
   }
 
   /**
@@ -96,5 +95,5 @@ public class WriteOptions extends RocksObject {
   private native boolean sync(long handle);
   private native void setDisableWAL(long handle, boolean flag);
   private native boolean disableWAL(long handle);
-  private native void dispose0(long handle);
+  private native void disposeInternal(long handle);
 }

java/rocksjni/backupablejni.cc

@@ -72,10 +72,10 @@ jstring Java_org_rocksdb_BackupableDBOptions_backupDir(
 
 /*
  * Class:     org_rocksdb_BackupableDBOptions
- * Method:    dispose
+ * Method:    disposeInternal
  * Signature: (J)V
  */
-void Java_org_rocksdb_BackupableDBOptions_dispose(
+void Java_org_rocksdb_BackupableDBOptions_disposeInternal(
     JNIEnv* env, jobject jopt, jlong jhandle) {
   auto bopt = reinterpret_cast<rocksdb::BackupableDBOptions*>(jhandle);
   assert(bopt);

java/rocksjni/filter.cc

@@ -29,13 +29,10 @@ void Java_org_rocksdb_BloomFilter_createNewFilter0(
 
 /*
  * Class:     org_rocksdb_Filter
- * Method:    dispose0
+ * Method:    disposeInternal
  * Signature: (J)V
  */
-void Java_org_rocksdb_Filter_dispose0(
+void Java_org_rocksdb_Filter_disposeInternal(
     JNIEnv* env, jobject jobj, jlong handle) {
-  auto fp = reinterpret_cast<rocksdb::FilterPolicy*>(handle);
+  delete reinterpret_cast<rocksdb::FilterPolicy*>(handle);
-  delete fp;
-
-  rocksdb::FilterJni::setHandle(env, jobj, nullptr);
 }

java/rocksjni/iterator.cc

@@ -135,10 +135,10 @@ void Java_org_rocksdb_RocksIterator_status0(
 
 /*
  * Class:     org_rocksdb_RocksIterator
- * Method:    dispose
+ * Method:    disposeInternal
  * Signature: (J)V
  */
-void Java_org_rocksdb_RocksIterator_dispose(
+void Java_org_rocksdb_RocksIterator_disposeInternal(
     JNIEnv* env, jobject jobj, jlong handle) {
   auto it = reinterpret_cast<rocksdb::Iterator*>(handle);
   delete it;

java/rocksjni/options.cc

@@ -35,14 +35,12 @@ void Java_org_rocksdb_Options_newOptions(JNIEnv* env, jobject jobj) {
 
 /*
  * Class:     org_rocksdb_Options
- * Method:    dispose0
+ * Method:    disposeInternal
- * Signature: ()V
+ * Signature: (J)V
  */
-void Java_org_rocksdb_Options_dispose0(JNIEnv* env, jobject jobj) {
+void Java_org_rocksdb_Options_disposeInternal(
-  rocksdb::Options* op = rocksdb::OptionsJni::getHandle(env, jobj);
+    JNIEnv* env, jobject jobj, jlong handle) {
-  delete op;
+  delete reinterpret_cast<rocksdb::Options*>(handle);
-
-  rocksdb::OptionsJni::setHandle(env, jobj, nullptr);
 }
 
 /*
@@ -1665,10 +1663,10 @@ void Java_org_rocksdb_WriteOptions_newWriteOptions(
 
 /*
  * Class:     org_rocksdb_WriteOptions
- * Method:    dispose0
+ * Method:    disposeInternal
  * Signature: ()V
  */
-void Java_org_rocksdb_WriteOptions_dispose0(
+void Java_org_rocksdb_WriteOptions_disposeInternal(
     JNIEnv* env, jobject jwrite_options, jlong jhandle) {
   auto write_options = reinterpret_cast<rocksdb::WriteOptions*>(jhandle);
   delete write_options;
@@ -1732,10 +1730,10 @@ void Java_org_rocksdb_ReadOptions_newReadOptions(
 
 /*
  * Class:     org_rocksdb_ReadOptions
- * Method:    dispose
+ * Method:    disposeInternal
  * Signature: (J)V
  */
-void Java_org_rocksdb_ReadOptions_dispose(
+void Java_org_rocksdb_ReadOptions_disposeInternal(
     JNIEnv* env, jobject jobj, jlong jhandle) {
   delete reinterpret_cast<rocksdb::ReadOptions*>(jhandle);
   rocksdb::ReadOptionsJni::setHandle(env, jobj, nullptr);

java/rocksjni/rocksjni.cc

@@ -419,14 +419,12 @@ void Java_org_rocksdb_RocksDB_remove__JJ_3BI(
 
 /*
  * Class:     org_rocksdb_RocksDB
- * Method:    dispose
+ * Method:    disposeInternal
  * Signature: (J)V
  */
-void Java_org_rocksdb_RocksDB_dispose(
+void Java_org_rocksdb_RocksDB_disposeInternal(
     JNIEnv* env, jobject java_db, jlong jhandle) {
-  auto db = reinterpret_cast<rocksdb::DB*>(jhandle);
+  delete reinterpret_cast<rocksdb::DB*>(jhandle);
-  assert(db != nullptr);
-  delete db;
 }
 
 /*

java/rocksjni/write_batch.cc

@@ -134,15 +134,12 @@ void Java_org_rocksdb_WriteBatch_putLogData(
 
 /*
  * Class:     org_rocksdb_WriteBatch
- * Method:    dispose0
+ * Method:    disposeInternal
- * Signature: ()V
+ * Signature: (J)V
  */
-void Java_org_rocksdb_WriteBatch_dispose0(JNIEnv* env, jobject jobj) {
+void Java_org_rocksdb_WriteBatch_disposeInternal(
-  rocksdb::WriteBatch* wb = rocksdb::WriteBatchJni::getHandle(env, jobj);
+    JNIEnv* env, jobject jobj, jlong handle) {
-  assert(wb != nullptr);
+  delete reinterpret_cast<rocksdb::WriteBatch*>(handle);
-  delete wb;
-
-  rocksdb::WriteBatchJni::setHandle(env, jobj, nullptr);
 }
 
 /*