From 2bf7ad4e4fc40d024c29ccb6d110c1f5ddeab0ea Mon Sep 17 00:00:00 2001
From: binlijin <binlijin@apache.org>
Date: Wed, 22 May 2019 16:19:08 +0800
Subject: [PATCH] HBASE-22447 Check refCount before free block in BucketCache

---
 .../hbase/io/hfile/bucket/BucketCache.java    | 44 +++++--------------
 1 file changed, 10 insertions(+), 34 deletions(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
index 1c418f5a3b0..009b29446e3 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
@@ -557,37 +557,6 @@ public class BucketCache implements BlockCache, HeapSize {
     return evictBlock(cacheKey, true);
   }
 
-  // does not check for the ref count. Just tries to evict it if found in the
-  // bucket map
-  private boolean forceEvict(BlockCacheKey cacheKey) {
-    if (!cacheEnabled) {
-      return false;
-    }
-    RAMQueueEntry removedBlock = checkRamCache(cacheKey);
-    BucketEntry bucketEntry = backingMap.get(cacheKey);
-    if (bucketEntry == null) {
-      if (removedBlock != null) {
-        cacheStats.evicted(0, cacheKey.isPrimary());
-        return true;
-      } else {
-        return false;
-      }
-    }
-    ReentrantReadWriteLock lock = offsetLock.getLock(bucketEntry.offset());
-    try {
-      lock.writeLock().lock();
-      if (backingMap.remove(cacheKey, bucketEntry)) {
-        blockEvicted(cacheKey, bucketEntry, removedBlock == null);
-      } else {
-        return false;
-      }
-    } finally {
-      lock.writeLock().unlock();
-    }
-    cacheStats.evicted(bucketEntry.getCachedTime(), cacheKey.isPrimary());
-    return true;
-  }
-
   private RAMQueueEntry checkRamCache(BlockCacheKey cacheKey) {
     RAMQueueEntry removedBlock = ramCache.remove(cacheKey);
     if (removedBlock != null) {
@@ -1049,8 +1018,15 @@ public class BucketCache implements BlockCache, HeapSize {
           ReentrantReadWriteLock lock = offsetLock.getLock(bucketEntries[i].offset());
           try {
             lock.writeLock().lock();
-            if (backingMap.remove(key, bucketEntries[i])) {
-              blockEvicted(key, bucketEntries[i], false);
+            int refCount = bucketEntries[i].getRefCount();
+            if (refCount == 0) {
+              if (backingMap.remove(key, bucketEntries[i])) {
+                blockEvicted(key, bucketEntries[i], false);
+              } else {
+                bucketEntries[i].markForEvict();
+              }
+            } else {
+              bucketEntries[i].markForEvict();
             }
           } finally {
             lock.writeLock().unlock();
@@ -1693,7 +1669,7 @@ public class BucketCache implements BlockCache, HeapSize {
       if (bucketEntry != null) {
         int refCount = bucketEntry.decrementRefCountAndGet();
         if (refCount == 0 && bucketEntry.isMarkedForEvict()) {
-          forceEvict(cacheKey);
+          evictBlock(cacheKey);
         }
       }
     }