hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hair...@apache.org
Subject svn commit: r1038918 - in /hadoop/common/trunk: ./ src/java/org/apache/hadoop/ipc/ src/test/core/org/apache/hadoop/ipc/
Date Thu, 25 Nov 2010 04:36:17 GMT
Author: hairong
Date: Thu Nov 25 04:36:17 2010
New Revision: 1038918

URL: http://svn.apache.org/viewvc?rev=1038918&view=rev
Log:
HADOOP-6764. Add number of reader threads and queue length as configuration parameters in
RPC.getServer. Contributed by Dmytro Molkov.

Modified:
    hadoop/common/trunk/CHANGES.txt
    hadoop/common/trunk/src/java/org/apache/hadoop/ipc/AvroRpcEngine.java
    hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RPC.java
    hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RpcEngine.java
    hadoop/common/trunk/src/java/org/apache/hadoop/ipc/Server.java
    hadoop/common/trunk/src/java/org/apache/hadoop/ipc/WritableRpcEngine.java
    hadoop/common/trunk/src/test/core/org/apache/hadoop/ipc/TestRPC.java

Modified: hadoop/common/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/CHANGES.txt?rev=1038918&r1=1038917&r2=1038918&view=diff
==============================================================================
--- hadoop/common/trunk/CHANGES.txt (original)
+++ hadoop/common/trunk/CHANGES.txt Thu Nov 25 04:36:17 2010
@@ -15,7 +15,10 @@ Trunk (unreleased changes)
     improve other messaging. (nigel)
 
     HADOOP-7001.  Configuration changes can occur via the Reconfigurable
-    interface. (Patrick Kline via dhruba)
+    interface. (Patrick Kling via dhruba)
+
+    HADOOP-6764. Add number of reader threads and queue length as
+    configuration parameters in RPC.getServer. (Dmytro Molkov via hairong)
 
   OPTIMIZATIONS
 

Modified: hadoop/common/trunk/src/java/org/apache/hadoop/ipc/AvroRpcEngine.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/ipc/AvroRpcEngine.java?rev=1038918&r1=1038917&r2=1038918&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/ipc/AvroRpcEngine.java (original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/ipc/AvroRpcEngine.java Thu Nov 25 04:36:17
2010
@@ -211,14 +211,15 @@ public class AvroRpcEngine implements Rp
   /** Construct a server for a protocol implementation instance listening on a
    * port and address. */
   public RPC.Server getServer(Class<?> iface, Object impl, String bindAddress,
-                              int port, int numHandlers, boolean verbose,
+                              int port, int numHandlers, int numReaders,
+                              int queueSizePerHandler, boolean verbose,
                               Configuration conf, 
                        SecretManager<? extends TokenIdentifier> secretManager
                               ) throws IOException {
     return ENGINE.getServer(TunnelProtocol.class,
                             new TunnelResponder(iface, impl),
-                            bindAddress, port, numHandlers, verbose, conf, 
-                            secretManager);
+                            bindAddress, port, numHandlers, numReaders,
+                            queueSizePerHandler, verbose, conf, secretManager);
   }
 
 }

Modified: hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RPC.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RPC.java?rev=1038918&r1=1038917&r2=1038918&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RPC.java (original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RPC.java Thu Nov 25 04:36:17 2010
@@ -380,18 +380,33 @@ public class RPC {
     throws IOException {
     
     return getProtocolEngine(protocol, conf)
-      .getServer(protocol, instance, bindAddress, port, numHandlers, verbose,
-                 conf, secretManager);
+      .getServer(protocol, instance, bindAddress, port, numHandlers, -1, -1,
+                 verbose, conf, secretManager);
+  }
+
+  /** Construct a server for a protocol implementation instance. */
+  public static Server getServer(Class<?> protocol,
+                                 Object instance, String bindAddress, int port,
+                                 int numHandlers, int numReaders, int queueSizePerHandler,
+                                 boolean verbose, Configuration conf,
+                                 SecretManager<? extends TokenIdentifier> secretManager)

+    throws IOException {
+    
+    return getProtocolEngine(protocol, conf)
+      .getServer(protocol, instance, bindAddress, port, numHandlers,
+                 numReaders, queueSizePerHandler, verbose, conf, secretManager);
   }
 
   /** An RPC Server. */
   public abstract static class Server extends org.apache.hadoop.ipc.Server {
   
     protected Server(String bindAddress, int port, 
-                     Class<? extends Writable> paramClass, int handlerCount, 
+                     Class<? extends Writable> paramClass, int handlerCount,
+                     int numReaders, int queueSizePerHandler,
                      Configuration conf, String serverName, 
                      SecretManager<? extends TokenIdentifier> secretManager) throws
IOException {
-      super(bindAddress, port, paramClass, handlerCount, conf, serverName, secretManager);
+      super(bindAddress, port, paramClass, handlerCount, numReaders, queueSizePerHandler,
+            conf, serverName, secretManager);
     }
   }
 

Modified: hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RpcEngine.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RpcEngine.java?rev=1038918&r1=1038917&r2=1038918&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RpcEngine.java (original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/ipc/RpcEngine.java Thu Nov 25 04:36:17
2010
@@ -50,7 +50,8 @@ public interface RpcEngine {
 
   /** Construct a server for a protocol implementation instance. */
   RPC.Server getServer(Class<?> protocol, Object instance, String bindAddress,
-                       int port, int numHandlers, boolean verbose,
+                       int port, int numHandlers, int numReaders,
+                       int queueSizePerHandler, boolean verbose,
                        Configuration conf, 
                        SecretManager<? extends TokenIdentifier> secretManager
                        ) throws IOException;

Modified: hadoop/common/trunk/src/java/org/apache/hadoop/ipc/Server.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/ipc/Server.java?rev=1038918&r1=1038917&r2=1038918&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/ipc/Server.java (original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/ipc/Server.java Thu Nov 25 04:36:17 2010
@@ -1451,16 +1451,18 @@ public abstract class Server {
                   Configuration conf)
     throws IOException 
   {
-    this(bindAddress, port, paramClass, handlerCount,  conf, Integer.toString(port), null);
+    this(bindAddress, port, paramClass, handlerCount, -1, -1, conf, Integer.toString(port),
null);
   }
+  
   /** Constructs a server listening on the named port and address.  Parameters passed must
    * be of the named class.  The <code>handlerCount</handlerCount> determines
    * the number of handler threads that will be used to process calls.
-   * 
+   * If queueSizePerHandler or numReaders are not -1 they will be used instead of parameters
+   * from configuration. Otherwise the configuration will be picked up.
    */
   @SuppressWarnings("unchecked")
   protected Server(String bindAddress, int port, 
-                  Class<? extends Writable> paramClass, int handlerCount, 
+                  Class<? extends Writable> paramClass, int handlerCount, int numReaders,
int queueSizePerHandler,
                   Configuration conf, String serverName, SecretManager<? extends TokenIdentifier>
secretManager) 
     throws IOException {
     this.bindAddress = bindAddress;
@@ -1469,15 +1471,23 @@ public abstract class Server {
     this.paramClass = paramClass;
     this.handlerCount = handlerCount;
     this.socketSendBufferSize = 0;
-    this.maxQueueSize = handlerCount * conf.getInt(
-        CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_KEY,
-        CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_DEFAULT);
+    if (queueSizePerHandler != -1) {
+      this.maxQueueSize = queueSizePerHandler;
+    } else {
+      this.maxQueueSize = handlerCount * conf.getInt(
+          CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_KEY,
+          CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_DEFAULT);      
+    }
     this.maxRespSize = conf.getInt(
         CommonConfigurationKeys.IPC_SERVER_RPC_MAX_RESPONSE_SIZE_KEY,
         CommonConfigurationKeys.IPC_SERVER_RPC_MAX_RESPONSE_SIZE_DEFAULT);
-    this.readThreads = conf.getInt(
-        CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
-        CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_DEFAULT);
+    if (numReaders != -1) {
+      this.readThreads = numReaders;
+    } else {
+      this.readThreads = conf.getInt(
+          CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
+          CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_DEFAULT);      
+    }
     this.callQueue  = new LinkedBlockingQueue<Call>(maxQueueSize); 
     this.maxIdleTime = 2*conf.getInt("ipc.client.connection.maxidletime", 1000);
     this.maxConnectionsToNuke = conf.getInt("ipc.client.kill.max", 10);
@@ -1691,7 +1701,22 @@ public abstract class Server {
     return callQueue.size();
   }
   
-  
+  /**
+   * The maximum size of the rpc call queue of this server.
+   * @return The maximum size of the rpc call queue.
+   */
+  public int getMaxQueueSize() {
+    return maxQueueSize;
+  }
+
+  /**
+   * The number of reader threads for this server.
+   * @return The number of reader threads.
+   */
+  public int getNumReaders() {
+    return readThreads;
+  }
+
   /**
    * When the read or write buffer size is larger than this limit, i/o will be 
    * done in chunks of this size. Most RPC requests and responses would be

Modified: hadoop/common/trunk/src/java/org/apache/hadoop/ipc/WritableRpcEngine.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/src/java/org/apache/hadoop/ipc/WritableRpcEngine.java?rev=1038918&r1=1038917&r2=1038918&view=diff
==============================================================================
--- hadoop/common/trunk/src/java/org/apache/hadoop/ipc/WritableRpcEngine.java (original)
+++ hadoop/common/trunk/src/java/org/apache/hadoop/ipc/WritableRpcEngine.java Thu Nov 25 04:36:17
2010
@@ -285,11 +285,12 @@ public class WritableRpcEngine implement
    * port and address. */
   public Server getServer(Class<?> protocol,
                           Object instance, String bindAddress, int port,
-                          int numHandlers, boolean verbose, Configuration conf,
+                          int numHandlers, int numReaders, int queueSizePerHandler,
+                          boolean verbose, Configuration conf,
                       SecretManager<? extends TokenIdentifier> secretManager) 
     throws IOException {
     return new Server(instance, conf, bindAddress, port, numHandlers, 
-        verbose, secretManager);
+        numReaders, queueSizePerHandler, verbose, secretManager);
   }
 
   /** An RPC Server. */
@@ -305,7 +306,7 @@ public class WritableRpcEngine implement
      */
     public Server(Object instance, Configuration conf, String bindAddress, int port) 
       throws IOException {
-      this(instance, conf,  bindAddress, port, 1, false, null);
+      this(instance, conf,  bindAddress, port, 1, -1, -1, false, null);
     }
     
     private static String classNameBase(String className) {
@@ -325,10 +326,11 @@ public class WritableRpcEngine implement
      * @param verbose whether each call should be logged
      */
     public Server(Object instance, Configuration conf, String bindAddress,  int port,
-                  int numHandlers, boolean verbose, 
+                  int numHandlers, int numReaders, int queueSizePerHandler, boolean verbose,

                   SecretManager<? extends TokenIdentifier> secretManager) 
         throws IOException {
-      super(bindAddress, port, Invocation.class, numHandlers, conf, 
+      super(bindAddress, port, Invocation.class, numHandlers, numReaders,
+          queueSizePerHandler, conf,
           classNameBase(instance.getClass().getName()), secretManager);
       this.instance = instance;
       this.verbose = verbose;

Modified: hadoop/common/trunk/src/test/core/org/apache/hadoop/ipc/TestRPC.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/src/test/core/org/apache/hadoop/ipc/TestRPC.java?rev=1038918&r1=1038917&r2=1038918&view=diff
==============================================================================
--- hadoop/common/trunk/src/test/core/org/apache/hadoop/ipc/TestRPC.java (original)
+++ hadoop/common/trunk/src/test/core/org/apache/hadoop/ipc/TestRPC.java Thu Nov 25 04:36:17
2010
@@ -190,6 +190,28 @@ public class TestRPC extends TestCase {
       }
     }
   }
+  
+  public void testConfRpc() throws Exception {
+    Server server = RPC.getServer(TestProtocol.class,
+                                  new TestImpl(), ADDRESS, 0, 1, false, conf, null);
+    // Just one handler
+    int confQ = conf.getInt(
+              CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_KEY,
+              CommonConfigurationKeys.IPC_SERVER_HANDLER_QUEUE_SIZE_DEFAULT);
+    assertEquals(confQ, server.getMaxQueueSize());
+
+    int confReaders = conf.getInt(
+              CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
+              CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_DEFAULT);
+    assertEquals(confReaders, server.getNumReaders());
+    server.stop();
+    
+    server = RPC.getServer(TestProtocol.class,
+                                  new TestImpl(), ADDRESS, 0, 1, 3, 200, false, conf, null);
+    assertEquals(3, server.getNumReaders());
+    assertEquals(200, server.getMaxQueueSize());
+    server.stop();    
+  }
 
   public void testSlowRpc() throws Exception {
     System.out.println("Testing Slow RPC");
@@ -234,6 +256,10 @@ public class TestRPC extends TestCase {
       System.out.println("Down slow rpc testing");
     }
   }
+  
+  public void testRPCConf(Configuration conf) throws Exception {
+    
+  }
 
   public void testCalls(Configuration conf) throws Exception {
     Server server = RPC.getServer(TestProtocol.class,



Mime
View raw message