From 5dba2c71f7f4b48ec1d8e6b54e2efe59e77f4b77 Mon Sep 17 00:00:00 2001
From: anoopsjohn <anoopsamjohn@gmail.com>
Date: Tue, 13 Oct 2015 07:59:11 +0530
Subject: [PATCH] HBASE-13819 Make RPC layer CellBlock buffer a
 DirectByteBuffer.

---
 .../apache/hadoop/hbase/io/BoundedByteBufferPool.java    | 9 +++++++--
 .../hadoop/hbase/io/TestBoundedByteBufferPool.java       | 4 ++--
 .../main/java/org/apache/hadoop/hbase/ipc/RpcServer.java | 4 +++-
 3 files changed, 12 insertions(+), 5 deletions(-)

diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/BoundedByteBufferPool.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/BoundedByteBufferPool.java
index c685a925001..a8ff598ed5f 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/io/BoundedByteBufferPool.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/io/BoundedByteBufferPool.java
@@ -67,16 +67,20 @@ public class BoundedByteBufferPool {
 
   private ReentrantLock lock =  new ReentrantLock();
 
+  private boolean createDirectByteBuffer;
+
   /**
    * @param maxByteBufferSizeToCache
    * @param initialByteBufferSize
    * @param maxToCache
+   * @param createDirectByteBuffer whether the buffers created by this pool to be off heap
    */
   public BoundedByteBufferPool(final int maxByteBufferSizeToCache, final int initialByteBufferSize,
-      final int maxToCache) {
+      final int maxToCache, final boolean createDirectByteBuffer) {
     this.maxByteBufferSizeToCache = maxByteBufferSizeToCache;
     this.runningAverage = initialByteBufferSize;
     this.buffers = new BoundedArrayQueue<ByteBuffer>(maxToCache);
+    this.createDirectByteBuffer = createDirectByteBuffer;
   }
 
   public ByteBuffer getBuffer() {
@@ -94,7 +98,8 @@ public class BoundedByteBufferPool {
       // Clear sets limit == capacity. Postion == 0.
       bb.clear();
     } else {
-      bb = ByteBuffer.allocate(this.runningAverage);
+      bb = this.createDirectByteBuffer ? ByteBuffer.allocateDirect(this.runningAverage)
+          : ByteBuffer.allocate(this.runningAverage);
       this.allocations.incrementAndGet();
     }
     if (LOG.isTraceEnabled()) {
diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestBoundedByteBufferPool.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestBoundedByteBufferPool.java
index 5074b4cdf67..987a2ff43c1 100644
--- a/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestBoundedByteBufferPool.java
+++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/io/TestBoundedByteBufferPool.java
@@ -37,8 +37,8 @@ public class TestBoundedByteBufferPool {
 
   @Before
   public void before() {
-    this.reservoir =
-      new BoundedByteBufferPool(maxByteBufferSizeToCache, initialByteBufferSize, maxToCache);
+    this.reservoir = new BoundedByteBufferPool(maxByteBufferSizeToCache, initialByteBufferSize,
+        maxToCache, false);
   }
 
   @After
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java
index ac428dad5c6..2b678171b18 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java
@@ -1967,7 +1967,9 @@ public class RpcServer implements RpcServerInterface {
       // Make the max twice the number of handlers to be safe.
       conf.getInt("hbase.ipc.server.reservoir.initial.max",
         conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
-          HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT) * 2));
+          HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT) * 2),
+      // By default make direct byte buffers from the buffer pool.
+      conf.getBoolean("hbase.ipc.server.reservoir.direct.buffer", true));
     this.server = server;
     this.services = services;
     this.bindAddress = bindAddress;