HBASE-17782 Extend IdReadWriteLock to support using both weak and soft reference

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java

@@ -72,6 +72,7 @@ import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.HasThread;
 import org.apache.hadoop.hbase.util.IdReadWriteLock;
+import org.apache.hadoop.hbase.util.IdReadWriteLock.ReferenceType;
 import org.apache.hadoop.util.StringUtils;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -185,9 +186,11 @@ public class BucketCache implements BlockCache, HeapSize {
   /**
    * A ReentrantReadWriteLock to lock on a particular block identified by offset.
    * The purpose of this is to avoid freeing the block which is being read.
+   * <p>
+   * Key set of offsets in BucketCache is limited so soft reference is the best choice here.
    */
   @VisibleForTesting
-  final IdReadWriteLock offsetLock = new IdReadWriteLock();
+  final IdReadWriteLock offsetLock = new IdReadWriteLock(ReferenceType.SOFT);
 
   private final NavigableSet<BlockCacheKey> blocksByHFile =
       new ConcurrentSkipListSet<>(new Comparator<BlockCacheKey>() {

hbase-server/src/main/java/org/apache/hadoop/hbase/util/IdReadWriteLock.java

@@ -18,6 +18,7 @@
  */
 package org.apache.hadoop.hbase.util;
 
+import java.lang.ref.Reference;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
@@ -44,16 +45,48 @@ import com.google.common.annotations.VisibleForTesting;
 public class IdReadWriteLock {
   // The number of lock we want to easily support. It's not a maximum.
   private static final int NB_CONCURRENT_LOCKS = 1000;
-  // The pool to get entry from, entries are mapped by soft reference and will be
+  /**
-  // automatically garbage-collected when JVM memory pressure is high
+   * The pool to get entry from, entries are mapped by {@link Reference} and will be automatically
-  private final ObjectPool<Long, ReentrantReadWriteLock> lockPool =
+   * garbage-collected by JVM
-      new SoftObjectPool<>(
+   */
-          new ObjectPool.ObjectFactory<Long, ReentrantReadWriteLock>() {
+  private final ObjectPool<Long, ReentrantReadWriteLock> lockPool;
-            @Override
+  private final ReferenceType refType;
-            public ReentrantReadWriteLock createObject(Long id) {
+
-              return new ReentrantReadWriteLock();
+  public IdReadWriteLock() {
-            }
+    this(ReferenceType.WEAK);
-          }, NB_CONCURRENT_LOCKS);
+  }
+
+  /**
+   * Constructor of IdReadWriteLock
+   * @param referenceType type of the reference used in lock pool, {@link ReferenceType#WEAK} by
+   *          default. Use {@link ReferenceType#SOFT} if the key set is limited and the locks will
+   *          be reused with a high frequency
+   */
+  public IdReadWriteLock(ReferenceType referenceType) {
+    this.refType = referenceType;
+    switch (referenceType) {
+    case SOFT:
+      lockPool = new SoftObjectPool<>(new ObjectPool.ObjectFactory<Long, ReentrantReadWriteLock>() {
+        @Override
+        public ReentrantReadWriteLock createObject(Long id) {
+          return new ReentrantReadWriteLock();
+        }
+      }, NB_CONCURRENT_LOCKS);
+      break;
+    case WEAK:
+    default:
+      lockPool = new WeakObjectPool<>(new ObjectPool.ObjectFactory<Long, ReentrantReadWriteLock>() {
+        @Override
+        public ReentrantReadWriteLock createObject(Long id) {
+          return new ReentrantReadWriteLock();
+        }
+      }, NB_CONCURRENT_LOCKS);
+    }
+  }
+
+  public static enum ReferenceType {
+    WEAK, SOFT
+  }
 
   /**
    * Get the ReentrantReadWriteLock corresponding to the given id
@@ -93,4 +126,9 @@ public class IdReadWriteLock {
       Thread.sleep(50);
     }
   }
+
+  @VisibleForTesting
+  public ReferenceType getReferenceType() {
+    return this.refType;
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java

@@ -27,7 +27,6 @@ import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.locks.Lock;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -36,7 +35,7 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience;
 // imports for classes still in regionserver.wal
 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
 import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.IdReadWriteLock;
+import org.apache.hadoop.hbase.util.IdLock;
 
 /**
  * A WAL Provider that returns a WAL per group of regions.
@@ -132,7 +131,7 @@ public class RegionGroupingProvider implements WALProvider {
   /** A group-provider mapping, make sure one-one rather than many-one mapping */
   private final ConcurrentMap<String, WALProvider> cached = new ConcurrentHashMap<>();
 
-  private final IdReadWriteLock createLock = new IdReadWriteLock();
+  private final IdLock createLock = new IdLock();
 
   private RegionGroupingStrategy strategy = null;
   private WALFactory factory = null;
@@ -181,16 +180,18 @@ public class RegionGroupingProvider implements WALProvider {
   private WAL getWAL(final String group) throws IOException {
     WALProvider provider = cached.get(group);
     if (provider == null) {
-      Lock lock = createLock.getLock(group.hashCode()).writeLock();
+      IdLock.Entry lockEntry = null;
-      lock.lock();
       try {
+        lockEntry = createLock.getLockEntry(group.hashCode());
         provider = cached.get(group);
         if (provider == null) {
           provider = createProvider(group);
           cached.put(group, provider);
         }
       } finally {
-        lock.unlock();
+        if (lockEntry != null) {
+          createLock.releaseLockEntry(lockEntry);
+        }
       }
     }
     return provider.getWAL(null, null);

hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestIdReadWriteLock.java

@@ -22,6 +22,7 @@ package org.apache.hadoop.hbase.util;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertEquals;
 
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.Callable;
@@ -38,9 +39,13 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.testclassification.MiscTests;
+import org.apache.hadoop.hbase.util.IdReadWriteLock.ReferenceType;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
+@RunWith(Parameterized.class)
 @Category({MiscTests.class, MediumTests.class})
 // Medium as it creates 100 threads; seems better to run it isolated
 public class TestIdReadWriteLock {
@@ -51,7 +56,14 @@ public class TestIdReadWriteLock {
   private static final int NUM_THREADS = 128;
   private static final int NUM_SECONDS = 15;
 
-  private IdReadWriteLock idLock = new IdReadWriteLock();
+  @Parameterized.Parameter
+  public IdReadWriteLock idLock;
+
+  @Parameterized.Parameters
+  public static Iterable<Object[]> data() {
+    return Arrays.asList(new Object[][] { { new IdReadWriteLock(ReferenceType.WEAK) },
+        { new IdReadWriteLock(ReferenceType.SOFT) } });
+  }
 
   private Map<Long, String> idOwner = new ConcurrentHashMap<>();
 
@@ -111,11 +123,22 @@ public class TestIdReadWriteLock {
         Future<Boolean> result = ecs.take();
         assertTrue(result.get());
       }
-      // make sure the entry pool won't be cleared when JVM memory is enough
-      // even after GC and purge call
       int entryPoolSize = idLock.purgeAndGetEntryPoolSize();
       LOG.debug("Size of entry pool after gc and purge: " + entryPoolSize);
-      assertEquals(NUM_IDS, entryPoolSize);
+      ReferenceType refType = idLock.getReferenceType();
+      switch (refType) {
+      case WEAK:
+        // make sure the entry pool will be cleared after GC and purge call
+        assertEquals(0, entryPoolSize);
+        break;
+      case SOFT:
+        // make sure the entry pool won't be cleared when JVM memory is enough
+        // even after GC and purge call
+        assertEquals(NUM_IDS, entryPoolSize);
+        break;
+      default:
+        break;
+      }
     } finally {
       exec.shutdown();
       exec.awaitTermination(5000, TimeUnit.MILLISECONDS);