HDFS-13274. RBF: Extend RouterRpcClient to use multiple sockets

Author: zengqiang.xu

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionContext.java

@@ -53,9 +53,12 @@ public class ConnectionContext {
   private long lastActiveTs = 0;
   /** The connection's active status would expire after this window. */
   private final static long ACTIVE_WINDOW_TIME = TimeUnit.SECONDS.toMillis(30);
+  /** The maximum number of requests that this connection can handle concurrently. **/
+  private final int maxConcurrencyPerConn;
 
-  public ConnectionContext(ProxyAndInfo<?> connection) {
+  public ConnectionContext(ProxyAndInfo<?> connection, int maxConcurrencyPerConn) {
     this.client = connection;
+    this.maxConcurrencyPerConn = maxConcurrencyPerConn;
   }
 
   /**
@@ -93,6 +96,23 @@ public synchronized boolean isClosed() {
    * @return True if the connection can be used.
    */
   public synchronized boolean isUsable() {
+    return hasAvailableConcurrency() && !isClosed();
+  }
+
+  /**
+   * Return true if this connection context still has available concurrency,
+   * else return false.
+   */
+  private synchronized boolean hasAvailableConcurrency() {
+    return this.numThreads < maxConcurrencyPerConn;
+  }
+
+  /**
+   *  Check if the connection is idle. It checks if the connection is not used
+   *  by another thread.
+   * @return True if the connection is not used by another thread.
+   */
+  public synchronized boolean isIdle() {
     return !isActive() && !isClosed();
   }
 

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/ConnectionPool.java

@@ -77,7 +77,6 @@ public class ConnectionPool {
   private static final Logger LOG =
       LoggerFactory.getLogger(ConnectionPool.class);
 
-
   /** Configuration settings for the connection pool. */
   private final Configuration conf;
 
@@ -94,6 +93,8 @@ public class ConnectionPool {
   private volatile List<ConnectionContext> connections = new ArrayList<>();
   /** Connection index for round-robin. */
   private final AtomicInteger clientIndex = new AtomicInteger(0);
+  /** Underlying socket index. **/
+  private final AtomicInteger socketIndex = new AtomicInteger(0);
 
   /** Min number of connections per user. */
   private final int minSize;
@@ -105,6 +106,12 @@ public class ConnectionPool {
   /** The last time a connection was active. */
   private volatile long lastActiveTime = 0;
 
+  /** Enable using multiple physical socket or not. **/
+  private final boolean enableMultiSocket;
+
+  /** Max Concurrency of each connection. */
+  private final int maxConcurrencyPerConn;
+
   /** Map for the protocols and their protobuf implementations. */
   private final static Map<Class<?>, ProtoImpl> PROTO_MAP = new HashMap<>();
   static {
@@ -149,9 +156,15 @@ protected ConnectionPool(Configuration config, String address,
     this.minSize = minPoolSize;
     this.maxSize = maxPoolSize;
     this.minActiveRatio = minActiveRatio;
+    this.enableMultiSocket = conf.getBoolean(
+        RBFConfigKeys.DFS_ROUTER_NAMENODE_ENABLE_MULTIPLE_SOCKET_KEY,
+        RBFConfigKeys.DFS_ROUTER_NAMENODE_ENABLE_MULTIPLE_SOCKET_DEFAULT);
+    this.maxConcurrencyPerConn = conf.getInt(
+        RBFConfigKeys.DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_KEY,
+        RBFConfigKeys.DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_DEFAULT);
 
     // Add minimum connections to the pool
-    for (int i=0; i<this.minSize; i++) {
+    for (int i = 0; i < this.minSize; i++) {
       ConnectionContext newConnection = newConnection();
       this.connections.add(newConnection);
     }
@@ -210,24 +223,22 @@ public AtomicInteger getClientIndex() {
    * @return Connection context.
    */
   protected ConnectionContext getConnection() {
-
     this.lastActiveTime = Time.now();
-
-    // Get a connection from the pool following round-robin
-    ConnectionContext conn = null;
     List<ConnectionContext> tmpConnections = this.connections;
-    int size = tmpConnections.size();
-    // Inc and mask off sign bit, lookup index should be non-negative int
-    int threadIndex = this.clientIndex.getAndIncrement() & 0x7FFFFFFF;
-    for (int i=0; i<size; i++) {
-      int index = (threadIndex + i) % size;
-      conn = tmpConnections.get(index);
-      if (conn != null && conn.isUsable()) {
-        return conn;
+    for (ConnectionContext tmpConnection : tmpConnections) {
+      if (tmpConnection != null && tmpConnection.isUsable()) {
+        return tmpConnection;
       }
     }
 
-    // We return a connection even if it's active
+    ConnectionContext conn = null;
+    // We return a connection even if it's busy
+    int size = tmpConnections.size();
+    if (size > 0) {
+      // Get a connection from the pool following round-robin
+      int threadIndex = this.clientIndex.getAndIncrement() & 0x7FFFFFFF;
+      conn = tmpConnections.get(threadIndex % size);
+    }
     return conn;
   }
 
@@ -256,19 +267,18 @@ public synchronized List<ConnectionContext> removeConnections(int num) {
       int targetCount = Math.min(num, this.connections.size() - this.minSize);
       // Remove and close targetCount of connections
       List<ConnectionContext> tmpConnections = new ArrayList<>();
-      for (int i = 0; i < this.connections.size(); i++) {
-        ConnectionContext conn = this.connections.get(i);
+      for (ConnectionContext conn : this.connections) {
         // Only pick idle connections to close
-        if (removed.size() < targetCount && conn.isUsable()) {
+        if (removed.size() < targetCount && conn.isIdle()) {
           removed.add(conn);
         } else {
           tmpConnections.add(conn);
         }
       }
       this.connections = tmpConnections;
     }
-    LOG.debug("Expected to remove {} connection " +
-        "and actually removed {} connections", num, removed.size());
+    LOG.debug("Expected to remove {} connection and actually removed {} connections",
+        num, removed.size());
     return removed;
   }
 
@@ -303,7 +313,6 @@ protected int getNumConnections() {
    */
   protected int getNumActiveConnections() {
     int ret = 0;
-
     List<ConnectionContext> tmpConnections = this.connections;
     for (ConnectionContext conn : tmpConnections) {
       if (conn.isActive()) {
@@ -320,10 +329,9 @@ protected int getNumActiveConnections() {
    */
   protected int getNumIdleConnections() {
     int ret = 0;
-
     List<ConnectionContext> tmpConnections = this.connections;
     for (ConnectionContext conn : tmpConnections) {
-      if (conn.isUsable()) {
+      if (conn.isIdle()) {
         ret++;
       }
     }
@@ -393,28 +401,34 @@ public String getJSON() {
    * @throws IOException If it cannot get a new connection.
    */
   public ConnectionContext newConnection() throws IOException {
-    return newConnection(
-        this.conf, this.namenodeAddress, this.ugi, this.protocol);
+    return newConnection(this.conf, this.namenodeAddress,
+        this.ugi, this.protocol, this.enableMultiSocket,
+        this.socketIndex.incrementAndGet(),
+        this.maxConcurrencyPerConn);
   }
 
   /**
    * Creates a proxy wrapper for a client NN connection. Each proxy contains
    * context for a single user/security context. To maximize throughput it is
    * recommended to use multiple connection per user+server, allowing multiple
    * writes and reads to be dispatched in parallel.
-   * @param <T>
+   * @param <T> Input type T.
    *
    * @param conf Configuration for the connection.
    * @param nnAddress Address of server supporting the ClientProtocol.
    * @param ugi User context.
    * @param proto Interface of the protocol.
+   * @param enableMultiSocket Enable multiple socket or not.
+   * @param maxConcurrencyPerConn The maximum number of requests that
+   *                              this connection can handle concurrently.
    * @return proto for the target ClientProtocol that contains the user's
    *         security context.
    * @throws IOException If it cannot be created.
    */
   protected static <T> ConnectionContext newConnection(Configuration conf,
-      String nnAddress, UserGroupInformation ugi, Class<T> proto)
-      throws IOException {
+      String nnAddress, UserGroupInformation ugi, Class<T> proto,
+      boolean enableMultiSocket, int socketIndex,
+      int maxConcurrencyPerConn) throws IOException {
     if (!PROTO_MAP.containsKey(proto)) {
       String msg = "Unsupported protocol for connection to NameNode: "
           + ((proto != null) ? proto.getName() : "null");
@@ -437,23 +451,31 @@ protected static <T> ConnectionContext newConnection(Configuration conf,
     }
     InetSocketAddress socket = NetUtils.createSocketAddr(nnAddress);
     final long version = RPC.getProtocolVersion(classes.protoPb);
-    Object proxy = RPC.getProtocolProxy(classes.protoPb, version, socket, ugi,
-        conf, factory, RPC.getRpcTimeout(conf), defaultPolicy, null).getProxy();
+    Object proxy;
+    if (enableMultiSocket) {
+      FederationConnectionId connectionId = new FederationConnectionId(
+          socket, classes.protoPb, ugi, RPC.getRpcTimeout(conf),
+          defaultPolicy, conf, socketIndex);
+      proxy = RPC.getProtocolProxy(classes.protoPb, version, connectionId,
+          conf, factory).getProxy();
+    } else {
+      proxy = RPC.getProtocolProxy(classes.protoPb, version, socket, ugi,
+          conf, factory, RPC.getRpcTimeout(conf), defaultPolicy, null).getProxy();
+    }
+
     T client = newProtoClient(proto, classes, proxy);
     Text dtService = SecurityUtil.buildTokenService(socket);
 
-    ProxyAndInfo<T> clientProxy =
-        new ProxyAndInfo<T>(client, dtService, socket);
-    ConnectionContext connection = new ConnectionContext(clientProxy);
-    return connection;
+    ProxyAndInfo<T> clientProxy = new ProxyAndInfo<T>(client, dtService, socket);
+    return new ConnectionContext(clientProxy, maxConcurrencyPerConn);
   }
 
   private static <T> T newProtoClient(Class<T> proto, ProtoImpl classes,
       Object proxy) {
     try {
       Constructor<?> constructor =
           classes.protoClientPb.getConstructor(classes.protoPb);
-      Object o = constructor.newInstance(new Object[] {proxy});
+      Object o = constructor.newInstance(proxy);
       if (proto.isAssignableFrom(o.getClass())) {
         @SuppressWarnings("unchecked")
         T client = (T) o;

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/FederationConnectionId.java

@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.ipc.Client;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import java.net.InetSocketAddress;
+
+public class FederationConnectionId extends Client.ConnectionId {
+  private static final int PRIME = 16777619;
+  private final int index;
+
+  public FederationConnectionId(InetSocketAddress address, Class<?> protocol,
+      UserGroupInformation ticket, int rpcTimeout,
+      RetryPolicy connectionRetryPolicy, Configuration conf, int index) {
+    super(address, protocol, ticket, rpcTimeout, connectionRetryPolicy, conf);
+    this.index = index;
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder()
+        .append(PRIME * super.hashCode())
+        .append(this.index)
+        .toHashCode();
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    if (!super.equals(obj)) {
+      return false;
+    }
+    if (obj instanceof FederationConnectionId) {
+      FederationConnectionId other = (FederationConnectionId)obj;
+      return new EqualsBuilder()
+          .append(this.index, other.index)
+          .isEquals();
+    }
+    return false;
+  }
+}

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RBFConfigKeys.java

@@ -135,6 +135,13 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
       FEDERATION_ROUTER_PREFIX + "connection.clean.ms";
   public static final long DFS_ROUTER_NAMENODE_CONNECTION_CLEAN_MS_DEFAULT =
       TimeUnit.SECONDS.toMillis(10);
+  public static final String DFS_ROUTER_NAMENODE_ENABLE_MULTIPLE_SOCKET_KEY =
+      FEDERATION_ROUTER_PREFIX + "enable.multiple.socket";
+  public static final boolean DFS_ROUTER_NAMENODE_ENABLE_MULTIPLE_SOCKET_DEFAULT
+      = false;
+  public static final String DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_KEY =
+      FEDERATION_ROUTER_PREFIX + "max.concurrency.per.connection";
+  public static final int DFS_ROUTER_MAX_CONCURRENCY_PER_CONNECTION_DEFAULT = 1;
 
   // HDFS Router RPC client
   public static final String DFS_ROUTER_CLIENT_THREADS_SIZE =

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/resources/hdfs-rbf-default.xml

@@ -134,6 +134,22 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.federation.router.enable.multiple.socket</name>
+    <value>false</value>
+    <description>
+      If enable multiple downstream socket or not.
+    </description>
+  </property>
+
+  <property>
+    <name>dfs.federation.router.max.concurrency.per.connection</name>
+    <value>1</value>
+    <description>
+      The maximum number of requests that a connection can handle concurrently.
+    </description>
+  </property>
+
   <property>
     <name>dfs.federation.router.connection.pool.clean.ms</name>
     <value>60000</value>