You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sj...@apache.org on 2015/10/15 20:43:46 UTC
hadoop git commit: HADOOP-12475. Replace guava Cache with
ConcurrentHashMap for caching Connection in ipc Client (Walter Su via sjlee)
Repository: hadoop
Updated Branches:
refs/heads/trunk 7a98d94b7 -> 8d2d3eb7b
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/trunk
Commit: 8d2d3eb7bb938cc06ea3cc74040cfe0be13a2ba8
Parents: 7a98d94
Author: Sangjin Lee <sj...@apache.org>
Authored: Thu Oct 15 11:43:07 2015 -0700
Committer: Sangjin Lee <sj...@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;
}