svn commit: r662913 - in /hadoop/core/trunk: CHANGES.txt conf/hadoop-default.xml src/java/org/apache/hadoop/ipc/Server.java

[hairong]

Author: hairong
Date: Tue Jun  3 16:05:50 2008
New Revision: 662913

URL: http://svn.apache.org/viewvc?rev=662913&view=rev
Log:
HADOOP 2909. Improve IPC idle connection management. Contributed by Hairong 
Kuang.

Modified:
    hadoop/core/trunk/CHANGES.txt
    hadoop/core/trunk/conf/hadoop-default.xml
    hadoop/core/trunk/src/java/org/apache/hadoop/ipc/Server.java

Modified: hadoop/core/trunk/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hadoop/core/trunk/CHANGES.txt?rev=662913&r1=662912&r2=662913&view=diff
==============================================================================
--- hadoop/core/trunk/CHANGES.txt (original)
+++ hadoop/core/trunk/CHANGES.txt Tue Jun  3 16:05:50 2008
@@ -65,6 +65,12 @@
     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,

Modified: hadoop/core/trunk/conf/hadoop-default.xml
URL: 
http://svn.apache.org/viewvc/hadoop/core/trunk/conf/hadoop-default.xml?rev=662913&r1=662912&r2=662913&view=diff
==============================================================================
--- hadoop/core/trunk/conf/hadoop-default.xml (original)
+++ hadoop/core/trunk/conf/hadoop-default.xml Tue Jun  3 16:05:50 2008
@@ -1081,14 +1081,6 @@
 </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>
   <description>Defines the maximum number of clients to disconnect in one go.

Modified: hadoop/core/trunk/src/java/org/apache/hadoop/ipc/Server.java
URL: 
http://svn.apache.org/viewvc/hadoop/core/trunk/src/java/org/apache/hadoop/ipc/Server.java?rev=662913&r1=662912&r2=662913&view=diff
==============================================================================
--- hadoop/core/trunk/src/java/org/apache/hadoop/ipc/Server.java (original)
+++ hadoop/core/trunk/src/java/org/apache/hadoop/ipc/Server.java Tue Jun  3 
16:05:50 2008
@@ -589,6 +589,7 @@
             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 @@
     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 @@
       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 @@
             return 0;  //ping message
           }
           data = ByteBuffer.allocate(dataLength);
+          incRpcCount();  // Increment the rpc count
         }
         
         count = channel.read(data);
@@ -925,7 +943,7 @@
     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);