svn commit: r1303512 - /hbase/trunk/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java

[st...@apache.org]

Author: stack
Date: Wed Mar 21 19:17:59 2012
New Revision: 1303512

URL: http://svn.apache.org/viewvc?rev=1303512&view=rev
Log:
HBASE-5560 Avoid RegionServer GC caused by timed-out calls

Modified:
    hbase/trunk/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java

Modified: hbase/trunk/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java
URL: http://svn.apache.org/viewvc/hbase/trunk/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java?rev=1303512&r1=1303511&r2=1303512&view=diff
==============================================================================
--- hbase/trunk/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java (original)
+++ hbase/trunk/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java Wed Mar 21 19:17:59 2012
@@ -59,6 +59,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.io.HbaseObjectWritable;
 import org.apache.hadoop.hbase.io.WritableWithSize;
 import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler;
@@ -196,6 +197,7 @@ public abstract class HBaseServer implem
   protected int socketSendBufferSize;
   protected final boolean tcpNoDelay;   // if T then disable Nagle's Algorithm
   protected final boolean tcpKeepAlive; // if T then use keepalives
+  protected final long purgeTimeout;    // in milliseconds
 
   volatile protected boolean running = true;         // true while server runs
   protected BlockingQueue<Call> callQueue; // queued calls
@@ -753,8 +755,6 @@ public abstract class HBaseServer implem
     private final Selector writeSelector;
     private int pending;         // connections waiting to register
 
-    final static int PURGE_INTERVAL = 900000; // 15mins
-
     Responder() throws IOException {
       this.setName("IPC Server Responder");
       this.setDaemon(true);
@@ -784,7 +784,7 @@ public abstract class HBaseServer implem
       while (running) {
         try {
           waitPending();     // If a channel is being registered, wait.
-          writeSelector.select(PURGE_INTERVAL);
+          writeSelector.select(purgeTimeout);
           Iterator<SelectionKey> iter = writeSelector.selectedKeys().iterator();
           while (iter.hasNext()) {
             SelectionKey key = iter.next();
@@ -798,7 +798,7 @@ public abstract class HBaseServer implem
             }
           }
           long now = System.currentTimeMillis();
-          if (now < lastPurgeTime + PURGE_INTERVAL) {
+          if (now < lastPurgeTime + purgeTimeout) {
             continue;
           }
           lastPurgeTime = now;
@@ -886,7 +886,7 @@ public abstract class HBaseServer implem
         Iterator<Call> iter = call.connection.responseQueue.listIterator(0);
         while (iter.hasNext()) {
           Call nextCall = iter.next();
-          if (now > nextCall.timestamp + PURGE_INTERVAL) {
+          if (now > nextCall.timestamp + purgeTimeout) {
             closeConnection(nextCall.connection);
             break;
           }
@@ -1459,6 +1459,8 @@ public abstract class HBaseServer implem
     this.maxIdleTime = 2*conf.getInt("ipc.client.connection.maxidletime", 1000);
     this.maxConnectionsToNuke = conf.getInt("ipc.client.kill.max", 10);
     this.thresholdIdleConnections = conf.getInt("ipc.client.idlethreshold", 4000);
+    this.purgeTimeout = conf.getLong("ipc.client.call.purge.timeout",
+                                     2 * HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
 
     // Start the listener here and let it bind to the port
     listener = new Listener();