hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zhang...@apache.org
Subject [33/50] [abbrv] hadoop git commit: HADOOP-12475. Replace guava Cache with ConcurrentHashMap for caching Connection in ipc Client (Walter Su via sjlee)
Date Sat, 17 Oct 2015 03:43:34 GMT
HADOOP-12475. Replace guava Cache with ConcurrentHashMap for caching Connection in ipc Client
(Walter Su via sjlee)


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8d2d3eb7
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8d2d3eb7
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8d2d3eb7

Branch: refs/heads/HDFS-7966
Commit: 8d2d3eb7bb938cc06ea3cc74040cfe0be13a2ba8
Parents: 7a98d94
Author: Sangjin Lee <sjlee@apache.org>
Authored: Thu Oct 15 11:43:07 2015 -0700
Committer: Sangjin Lee <sjlee@apache.org>
Committed: Thu Oct 15 11:43:07 2015 -0700

----------------------------------------------------------------------
 hadoop-common-project/hadoop-common/CHANGES.txt |  3 +
 .../main/java/org/apache/hadoop/ipc/Client.java | 64 +++++++++-----------
 2 files changed, 33 insertions(+), 34 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/8d2d3eb7/hadoop-common-project/hadoop-common/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/CHANGES.txt b/hadoop-common-project/hadoop-common/CHANGES.txt
index d6ddfd9..7d4cfd3 100644
--- a/hadoop-common-project/hadoop-common/CHANGES.txt
+++ b/hadoop-common-project/hadoop-common/CHANGES.txt
@@ -510,6 +510,9 @@ Trunk (Unreleased)
     HADOOP-12364. Deleting pid file after stop is causing the daemons to
     keep restarting (Siqi Li via aw)
 
+    HADOOP-12475. Replace guava Cache with ConcurrentHashMap for caching
+    Connection in ipc Client (Walter Su via sjlee)
+
   OPTIMIZATIONS
 
     HADOOP-7761. Improve the performance of raw comparisons. (todd)

http://git-wip-us.apache.org/repos/asf/hadoop/blob/8d2d3eb7/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
index bfa85aa..7b80740 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
@@ -43,7 +43,8 @@ import java.util.Iterator;
 import java.util.Map.Entry;
 import java.util.Random;
 import java.util.Set;
-import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -57,8 +58,6 @@ import java.util.concurrent.atomic.AtomicLong;
 import javax.net.SocketFactory;
 import javax.security.sasl.Sasl;
 
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -128,8 +127,8 @@ public class Client {
     retryCount.set(rc);
   }
 
-  private final Cache<ConnectionId, Connection> connections =
-      CacheBuilder.newBuilder().build();
+  private ConcurrentMap<ConnectionId, Connection> connections =
+      new ConcurrentHashMap<>();
 
   private Class<? extends Writable> valueClass;   // class of call values
   private AtomicBoolean running = new AtomicBoolean(true); // if client runs
@@ -1178,7 +1177,10 @@ public class Client {
         return;
       }
 
-      connections.invalidate(remoteId);
+      // We have marked this connection as closed. Other thread could have
+      // already known it and replace this closedConnection with a new one.
+      // We should only remove this closedConnection.
+      connections.remove(remoteId, this);
 
       // close the streams and therefore the socket
       IOUtils.closeStream(out);
@@ -1265,12 +1267,12 @@ public class Client {
     }
     
     // wake up all connections
-    for (Connection conn : connections.asMap().values()) {
+    for (Connection conn : connections.values()) {
       conn.interrupt();
     }
     
     // wait until all connections are closed
-    while (connections.size() > 0) {
+    while (!connections.isEmpty()) {
       try {
         Thread.sleep(100);
       } catch (InterruptedException e) {
@@ -1289,7 +1291,6 @@ public class Client {
     ConnectionId remoteId = ConnectionId.getConnectionId(address, null, null, 0,
         conf);
     return call(RpcKind.RPC_BUILTIN, param, remoteId);
-
   }
 
   /**
@@ -1465,14 +1466,13 @@ public class Client {
   @InterfaceAudience.Private
   @InterfaceStability.Unstable
   Set<ConnectionId> getConnectionIds() {
-    return connections.asMap().keySet();
+    return connections.keySet();
   }
   
   /** Get a connection from the pool, or create a new one and add it to the
    * pool.  Connections to a given ConnectionId are reused. */
-  private Connection getConnection(
-      final ConnectionId remoteId,
-      Call call, final int serviceClass, AtomicBoolean fallbackToSimpleAuth)
+  private Connection getConnection(ConnectionId remoteId,
+      Call call, int serviceClass, AtomicBoolean fallbackToSimpleAuth)
       throws IOException {
     if (!running.get()) {
       // the client is stopped
@@ -1483,34 +1483,30 @@ public class Client {
      * connectionsId object and with set() method. We need to manage the
      * refs for keys in HashMap properly. For now its ok.
      */
-    while(true) {
-      try {
-        connection = connections.get(remoteId, new Callable<Connection>() {
-          @Override
-          public Connection call() throws Exception {
-            return new Connection(remoteId, serviceClass);
-          }
-        });
-      } catch (ExecutionException e) {
-        Throwable cause = e.getCause();
-        // the underlying exception should normally be IOException
-        if (cause instanceof IOException) {
-          throw (IOException) cause;
-        } else {
-          throw new IOException(cause);
+    while (true) {
+      // These lines below can be shorten with computeIfAbsent in Java8
+      connection = connections.get(remoteId);
+      if (connection == null) {
+        connection = new Connection(remoteId, serviceClass);
+        Connection existing = connections.putIfAbsent(remoteId, connection);
+        if (existing != null) {
+          connection = existing;
         }
       }
+
       if (connection.addCall(call)) {
         break;
       } else {
-        connections.invalidate(remoteId);
+        // This connection is closed, should be removed. But other thread could
+        // have already known this closedConnection, and replace it with a new
+        // connection. So we should call conditional remove to make sure we only
+        // remove this closedConnection.
+        connections.remove(remoteId, connection);
       }
     }
-    
-    //we don't invoke the method below inside "synchronized (connections)"
-    //block above. The reason for that is if the server happens to be slow,
-    //it will take longer to establish a connection and that will slow the
-    //entire system down.
+
+    // If the server happens to be slow, the method below will take longer to
+    // establish a connection.
     connection.setupIOstreams(fallbackToSimpleAuth);
     return connection;
   }


Mime
View raw message