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 49bd8ba..c685a92 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 @@ -47,7 +47,6 @@ import com.google.common.annotations.VisibleForTesting; */ @InterfaceAudience.Private public class BoundedByteBufferPool { - private boolean debug = false; private static final Log LOG = LogFactory.getLog(BoundedByteBufferPool.class); @VisibleForTesting @@ -67,7 +66,7 @@ public class BoundedByteBufferPool { private AtomicLong allocations = new AtomicLong(0); private ReentrantLock lock = new ReentrantLock(); - + /** * @param maxByteBufferSizeToCache * @param initialByteBufferSize @@ -78,7 +77,6 @@ public class BoundedByteBufferPool { this.maxByteBufferSizeToCache = maxByteBufferSizeToCache; this.runningAverage = initialByteBufferSize; this.buffers = new BoundedArrayQueue(maxToCache); - if (maxToCache == 10 * 3) debug = true; } public ByteBuffer getBuffer() { @@ -97,7 +95,6 @@ public class BoundedByteBufferPool { bb.clear(); } else { bb = ByteBuffer.allocate(this.runningAverage); - if (debug) LOG.info("#CREATE#"+bb.capacity()); this.allocations.incrementAndGet(); } if (LOG.isTraceEnabled()) { @@ -110,10 +107,7 @@ public class BoundedByteBufferPool { public void putBuffer(ByteBuffer bb) { // If buffer is larger than we want to keep around, just let it go. - if (bb.capacity() > this.maxByteBufferSizeToCache) { - if (debug) LOG.info("#DISCARD#"+bb.capacity()); - return; - } + if (bb.capacity() > this.maxByteBufferSizeToCache) return; boolean success = false; int average = 0; lock.lock(); @@ -127,7 +121,6 @@ public class BoundedByteBufferPool { lock.unlock(); } if (!success) { - if (debug) LOG.info("#DISCARD#"+bb.capacity()); LOG.warn("At capacity: " + this.buffers.size()); } else { if (average > this.runningAverage && average < this.maxByteBufferSizeToCache) { 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 f1a4f01..b90c542 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 @@ -23,7 +23,6 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; import java.io.IOException; -import java.lang.ref.SoftReference; import java.net.BindException; import java.net.InetAddress; import java.net.InetSocketAddress; @@ -33,7 +32,6 @@ import java.net.SocketException; import java.net.UnknownHostException; import java.nio.ByteBuffer; import java.nio.channels.CancelledKeyException; -import java.nio.channels.Channels; import java.nio.channels.ClosedChannelException; import java.nio.channels.GatheringByteChannel; import java.nio.channels.ReadableByteChannel; @@ -275,7 +273,7 @@ public class RpcServer implements RpcServerInterface { private final BoundedByteBufferPool reservoir; private final BoundedByteBufferPool reqBufPool; - + /** * Datastructure that holds all necessary to a method invocation and then afterward, carries * the result. @@ -594,14 +592,14 @@ public class RpcServer implements RpcServerInterface { // Create a new server socket and set to non blocking mode acceptChannel = ServerSocketChannel.open(); acceptChannel.configureBlocking(false); - + // Bind the server socket to the binding addrees (can be different from the default interface) bind(acceptChannel.socket(), bindAddress, backlogLength); port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port address = (InetSocketAddress)acceptChannel.socket().getLocalSocketAddress(); // create a selector; selector= Selector.open(); - + readers = new Reader[readThreads]; readPool = Executors.newFixedThreadPool(readThreads, new ThreadFactoryBuilder().setNameFormat( @@ -1581,6 +1579,8 @@ public class RpcServer implements RpcServerInterface { } } + // We have read a length and we have read the preamble. It is either the connection header + // or it is a request. if (data == null) { dataLengthBuffer.flip(); int dataLength = dataLengthBuffer.getInt(); @@ -1594,17 +1594,10 @@ public class RpcServer implements RpcServerInterface { throw new IllegalArgumentException("Unexpected data length " + dataLength + "!! from " + getHostAddress()); } - data = reqBufPool.getBuffer(); - - // We have read a length and we have read the preamble. It is either the connection header - // or it is a request. if (data.capacity() < dataLength) { - LOG.info("#DISCARD#"+data.capacity()); data = ByteBuffer.allocate(dataLength); - LOG.info("#CREATE#"+data.capacity()); } else { - data.clear(); data.limit(dataLength); } // Increment the rpc count. This counter will be decreased when we write