HADOOP 2909. Improve IPC idle connection management. Contributed by Hairong Kuang.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@662913 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -65,6 +65,12 @@ Trunk (unreleased changes)
     recovery. The block gets a new  generation stamp.
     (Tsz Wo (Nicholas), SZE via dhruba)
 
+    HADOOP-2909. Improve IPC idle connection management. Property
+    ipc.client.maxidletime is removed from the default configuration,
+    instead it is defined as twice of the ipc.client.connection.maxidletime.
+    A connection with outstanding requests won't be treated as idle.
+    (hairong)
+
   NEW FEATURES
 
     HADOOP-3074. Provides a UrlStreamHandler for DFS and other FS,

conf/hadoop-default.xml

@@ -1080,14 +1080,6 @@ creations/deletions), or "all".</description>
   </description>
 </property>
 
-<property>
-  <name>ipc.client.maxidletime</name>
-  <value>120000</value>
-  <description>Defines the maximum idle time in msec for a connected client after 
-               which it may be disconnected.
-  </description>
-</property>
-
 <property>
   <name>ipc.client.kill.max</name>
   <value>10</value>

src/java/org/apache/hadoop/ipc/Server.java

@@ -589,6 +589,7 @@ public abstract class Server {
             return true;
           }
           if (!call.response.hasRemaining()) {
+            call.connection.decRpcCount();
             if (numElements == 1) {    // last call fully processes.
               done = true;             // no more data for this channel.
             } else {
@@ -678,6 +679,7 @@ public abstract class Server {
     private ByteBuffer data;
     private ByteBuffer dataLengthBuffer;
     private LinkedList<Call> responseQueue;
+    private volatile int rpcCount = 0; // number of outstanding rpcs
     private long lastContact;
     private int dataLength;
     private Socket socket;
@@ -729,8 +731,23 @@ public abstract class Server {
       return lastContact;
     }
 
+    /* Return true if the connection has no outstanding rpc */
+    private boolean isIdle() {
+      return rpcCount == 0;
+    }
+    
+    /* Decrement the outstanding RPC count */
+    private void decRpcCount() {
+      rpcCount--;
+    }
+    
+    /* Increment the outstanding RPC count */
+    private void incRpcCount() {
+      rpcCount++;
+    }
+    
     private boolean timedOut(long currentTime) {
-      if (currentTime -  lastContact > maxIdleTime)
+      if (isIdle() && currentTime -  lastContact > maxIdleTime)
         return true;
       return false;
     }
@@ -779,6 +796,7 @@ public abstract class Server {
             return 0;  //ping message
           }
           data = ByteBuffer.allocate(dataLength);
+          incRpcCount();  // Increment the rpc count
         }
         
         count = channel.read(data);
@@ -925,7 +943,7 @@ public abstract class Server {
     this.socketSendBufferSize = 0;
     this.maxQueueSize = handlerCount * MAX_QUEUE_SIZE_PER_HANDLER;
     this.callQueue  = new LinkedBlockingQueue<Call>(maxQueueSize); 
-    this.maxIdleTime = conf.getInt("ipc.client.maxidletime", 120000);
+    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);