HBASE-13980 Distinguish blockedFlushCount vs unblockedFlushCount when tuning heap memory (Abhilash)

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DefaultHeapMemoryTuner.java

@@ -60,7 +60,7 @@ import org.apache.hadoop.hbase.util.RollingStatCalculator;
  * <i>hbase.regionserver.heapmemory.autotuner.step.max</i>. If we are reverting the previous step
  * then we decrease step size to half. This decrease is similar to binary search where we try to
  * reach the most desired value. The minimum step size can be specified  in config by
- * <i>hbase.regionserver.heapmemory.autotuner.step.max</i>. In other cases we leave step size
+ * <i>hbase.regionserver.heapmemory.autotuner.step.min</i>. In other cases we leave step size
  * unchanged.
  */
 @InterfaceAudience.Private
@@ -217,6 +217,11 @@ class DefaultHeapMemoryTuner implements HeapMemoryTuner {
           // more flushes , increasing memstore size
           newTuneDirection = StepDirection.INCREASE_MEMSTORE_SIZE;
           tunerLog += "Increasing memstore size as observed increase in number of flushes.";
+        } else if (blockedFlushCount > 0 && prevTuneDirection == StepDirection.NEUTRAL) {
+          // we do not want blocked flushes
+          newTuneDirection = StepDirection.INCREASE_MEMSTORE_SIZE;
+          tunerLog += "Increasing memstore size as observed "
+                      + blockedFlushCount + " blocked flushes.";
         } else {
           // Default. Not enough facts to do tuning.
           newTuneDirection = StepDirection.NEUTRAL;

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHeapMemoryManager.java

@@ -187,12 +187,11 @@ public class TestHeapMemoryManager {
     long oldBlockCacheSize = blockCache.maxSize;
     final ChoreService choreService = new ChoreService("TEST_SERVER_NAME");
     heapMemoryManager.start(choreService);
-    memStoreFlusher.flushType = FlushType.ABOVE_HIGHER_MARK;
-    memStoreFlusher.requestFlush(null, false);
-    memStoreFlusher.requestFlush(null, false);
-    memStoreFlusher.requestFlush(null, false);
     memStoreFlusher.flushType = FlushType.ABOVE_LOWER_MARK;
     memStoreFlusher.requestFlush(null, false);
+    memStoreFlusher.requestFlush(null, false);
+    memStoreFlusher.requestFlush(null, false);
+    memStoreFlusher.requestFlush(null, false);
     Thread.sleep(1500); // Allow the tuner to run once and do necessary memory up
     assertHeapSpaceDelta(DefaultHeapMemoryTuner.DEFAULT_MAX_STEP_VALUE, oldMemstoreHeapSize,
         memStoreFlusher.memstoreSize);
@@ -201,7 +200,7 @@ public class TestHeapMemoryManager {
     oldMemstoreHeapSize = memStoreFlusher.memstoreSize;
     oldBlockCacheSize = blockCache.maxSize;
     // Do some more flushes before the next run of HeapMemoryTuner
-    memStoreFlusher.flushType = FlushType.ABOVE_HIGHER_MARK;
+    memStoreFlusher.flushType = FlushType.ABOVE_LOWER_MARK;
     memStoreFlusher.requestFlush(null, false);
     memStoreFlusher.requestFlush(null, false);
     Thread.sleep(1500);
@@ -275,21 +274,67 @@ public class TestHeapMemoryManager {
     long oldBlockCacheSize = blockCache.maxSize;
     final ChoreService choreService = new ChoreService("TEST_SERVER_NAME");
     heapMemoryManager.start(choreService);
-    memStoreFlusher.flushType = FlushType.ABOVE_HIGHER_MARK;
+    memStoreFlusher.flushType = FlushType.ABOVE_LOWER_MARK;
     memStoreFlusher.requestFlush(null, false);
     memStoreFlusher.requestFlush(null, false);
     memStoreFlusher.requestFlush(null, false);
     blockCache.evictBlock(null);
-    memStoreFlusher.flushType = FlushType.ABOVE_LOWER_MARK;
-    memStoreFlusher.requestFlush(null, false);
     Thread.sleep(1500); // Allow the tuner to run once and do necessary memory up
     // No changes should happen as there is undefined increase in flushes and evictions
     assertEquals(oldMemstoreHeapSize, memStoreFlusher.memstoreSize);
     assertEquals(oldBlockCacheSize, blockCache.maxSize);
     // Do some more flushes before the next run of HeapMemoryTuner
+    memStoreFlusher.flushType = FlushType.ABOVE_LOWER_MARK;
+    memStoreFlusher.requestFlush(null, false);
+    memStoreFlusher.requestFlush(null, false);
+    memStoreFlusher.requestFlush(null, false);
+    Thread.sleep(1500);
+    assertHeapSpaceDelta(DefaultHeapMemoryTuner.DEFAULT_MAX_STEP_VALUE, oldMemstoreHeapSize,
+        memStoreFlusher.memstoreSize);
+    assertHeapSpaceDelta(-(DefaultHeapMemoryTuner.DEFAULT_MAX_STEP_VALUE), oldBlockCacheSize,
+        blockCache.maxSize);
+  }
+
+  @Test
+  public void testBlockedFlushesIncreaseMemstoreInSteadyState() throws Exception {
+    BlockCacheStub blockCache = new BlockCacheStub((long) (maxHeapSize * 0.4));
+    MemstoreFlusherStub memStoreFlusher = new MemstoreFlusherStub((long) (maxHeapSize * 0.4));
+    RegionServerAccountingStub regionServerAccounting = new RegionServerAccountingStub();
+    // Both memstore and block cache are nearly filled
+    blockCache.setTestBlockSize(0);
+    regionServerAccounting.setTestMemstoreSize((long) (maxHeapSize * 0.4 * 0.8));
+    blockCache.setTestBlockSize((long) (maxHeapSize * 0.4 * 0.8));
+    Configuration conf = HBaseConfiguration.create();
+    conf.setFloat(HeapMemoryManager.MEMSTORE_SIZE_MAX_RANGE_KEY, 0.75f);
+    conf.setFloat(HeapMemoryManager.MEMSTORE_SIZE_MIN_RANGE_KEY, 0.10f);
+    conf.setFloat(HeapMemoryManager.BLOCK_CACHE_SIZE_MAX_RANGE_KEY, 0.7f);
+    conf.setFloat(HeapMemoryManager.BLOCK_CACHE_SIZE_MIN_RANGE_KEY, 0.05f);
+    conf.setLong(HeapMemoryManager.HBASE_RS_HEAP_MEMORY_TUNER_PERIOD, 1000);
+    conf.setInt(DefaultHeapMemoryTuner.NUM_PERIODS_TO_IGNORE, 0);
+    // Let the system start with default values for memstore heap and block cache size.
+    HeapMemoryManager heapMemoryManager = new HeapMemoryManager(blockCache, memStoreFlusher,
+        new RegionServerStub(conf), regionServerAccounting);
+    long oldMemstoreHeapSize = memStoreFlusher.memstoreSize;
+    long oldBlockCacheSize = blockCache.maxSize;
+    final ChoreService choreService = new ChoreService("TEST_SERVER_NAME");
+    heapMemoryManager.start(choreService);
+    memStoreFlusher.flushType = FlushType.ABOVE_LOWER_MARK;
+    memStoreFlusher.requestFlush(null, false);
+    memStoreFlusher.requestFlush(null, false);
+    memStoreFlusher.requestFlush(null, false);
+    blockCache.evictBlock(null);
+    blockCache.evictBlock(null);
+    Thread.sleep(1500); // Allow the tuner to run once and do necessary memory up
+    // No changes should happen as there is undefined increase in flushes and evictions
+    assertEquals(oldMemstoreHeapSize, memStoreFlusher.memstoreSize);
+    assertEquals(oldBlockCacheSize, blockCache.maxSize);
+    // Flushes that block updates
     memStoreFlusher.flushType = FlushType.ABOVE_HIGHER_MARK;
     memStoreFlusher.requestFlush(null, false);
-    memStoreFlusher.requestFlush(null, false);
+    blockCache.evictBlock(null);
+    blockCache.evictBlock(null);
+    blockCache.evictBlock(null);
+    blockCache.evictBlock(null);
     Thread.sleep(1500);
     assertHeapSpaceDelta(DefaultHeapMemoryTuner.DEFAULT_MAX_STEP_VALUE, oldMemstoreHeapSize,
         memStoreFlusher.memstoreSize);