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 ar...@apache.org on 2013/11/15 00:56:59 UTC
svn commit: r1542125 - in
/hadoop/common/branches/HDFS-2832/hadoop-common-project:
hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/client/
hadoop-auth/src/test/java/org/apache/hadoop/security/authentication/client/
hadoop-common/ ha...
Author: arp
Date: Thu Nov 14 23:56:56 2013
New Revision: 1542125
URL: http://svn.apache.org/r1542125
Log:
Merging r1541618 through r1542122 from trunk to branch HDFS-2832
Modified:
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/test/java/org/apache/hadoop/security/authentication/client/AuthenticatorTestCase.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/CHANGES.txt (contents, props changed)
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/pom.xml
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/ (props changed)
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/CodecPool.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/GenericOptionsParser.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ZKUtil.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestGenericOptionsParser.java
hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestZKUtil.java
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/main/java/org/apache/hadoop/security/authentication/client/KerberosAuthenticator.java Thu Nov 14 23:56:56 2013
@@ -185,7 +185,7 @@ public class KerberosAuthenticator imple
conn.setRequestMethod(AUTH_HTTP_METHOD);
conn.connect();
- if (conn.getRequestProperty(AUTHORIZATION) != null && conn.getResponseCode() == HttpURLConnection.HTTP_OK) {
+ if (conn.getResponseCode() == HttpURLConnection.HTTP_OK) {
LOG.debug("JDK performed authentication on our behalf.");
// If the JDK already did the SPNEGO back-and-forth for
// us, just pull out the token.
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/test/java/org/apache/hadoop/security/authentication/client/AuthenticatorTestCase.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/test/java/org/apache/hadoop/security/authentication/client/AuthenticatorTestCase.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/test/java/org/apache/hadoop/security/authentication/client/AuthenticatorTestCase.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-auth/src/test/java/org/apache/hadoop/security/authentication/client/AuthenticatorTestCase.java Thu Nov 14 23:56:56 2013
@@ -136,7 +136,6 @@ public class AuthenticatorTestCase {
TestConnectionConfigurator connConf = new TestConnectionConfigurator();
AuthenticatedURL aUrl = new AuthenticatedURL(authenticator, connConf);
HttpURLConnection conn = aUrl.openConnection(url, token);
- Assert.assertTrue(token.isSet());
Assert.assertTrue(connConf.invoked);
String tokenStr = token.toString();
if (doPost) {
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/CHANGES.txt?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/CHANGES.txt (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/CHANGES.txt Thu Nov 14 23:56:56 2013
@@ -284,7 +284,7 @@ Trunk (Unreleased)
HADOOP-7761. Improve the performance of raw comparisons. (todd)
- HADOOP-8589 ViewFs tests fail when tests and home dirs are nested (sanjay Radia)
+ HADOOP-8589. ViewFs tests fail when tests and home dirs are nested (sanjay Radia)
Release 2.3.0 - UNRELEASED
@@ -375,6 +375,14 @@ Release 2.3.0 - UNRELEASED
HADOOP-10064. Upgrade to maven antrun plugin version 1.7 (Arpit Agarwal via
jeagles)
+ HADOOP-9594. Update apache commons math dependency (Timothy St. Clair via
+ stevel)
+
+ HADOOP-10095. In CodecPool, synchronize pool and codecList separately in
+ order to reduce lock contention. (Nicolas Liochon via szetszwo)
+
+ HADOOP-10067. Missing POM dependency on jsr305 (Robert Rati via stevel)
+
OPTIMIZATIONS
HADOOP-9748. Reduce blocking on UGI.ensureInitialized (daryn)
@@ -426,6 +434,9 @@ Release 2.3.0 - UNRELEASED
HADOOP-10093. hadoop-env.cmd sets HADOOP_CLIENT_OPTS with a max heap size
that is too small. (Shanyu Zhao via cnauroth)
+ HADOOP-10094. NPE in GenericOptionsParser#preProcessForWindows().
+ (Enis Soztutar via cnauroth)
+
Release 2.2.1 - UNRELEASED
INCOMPATIBLE CHANGES
@@ -476,6 +487,8 @@ Release 2.2.1 - UNRELEASED
as [-Dkey, value] which breaks GenericsOptionParser.
(Enis Soztutar via cnauroth)
+ HADOOP-10078. KerberosAuthenticator always does SPNEGO. (rkanter via tucu)
+
Release 2.2.0 - 2013-10-13
INCOMPATIBLE CHANGES
@@ -2298,6 +2311,10 @@ Release 0.23.10 - UNRELEASED
OPTIMIZATIONS
+ HADOOP-9956. RPC listener inefficiently assigns connections to readers (daryn)
+
+ HADOOP-9955. RPC idle connection closing is extremely inefficient (daryn)
+
BUG FIXES
Release 0.23.9 - 2013-07-08
Propchange: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/CHANGES.txt
------------------------------------------------------------------------------
Merged /hadoop/common/branches/branch-2/hadoop-common-project/hadoop-common/CHANGES.txt:r1541861
Merged /hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt:r1541618-1542122
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/pom.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/pom.xml?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/pom.xml (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/pom.xml Thu Nov 14 23:56:56 2013
@@ -55,7 +55,7 @@
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
- <artifactId>commons-math</artifactId>
+ <artifactId>commons-math3</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
@@ -84,6 +84,11 @@
<scope>compile</scope>
</dependency>
<dependency>
+ <groupId>commons-collections</groupId>
+ <artifactId>commons-collections</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
<groupId>javax.servlet</groupId>
<artifactId>servlet-api</artifactId>
<scope>compile</scope>
@@ -213,6 +218,11 @@
<groupId>com.jcraft</groupId>
<artifactId>jsch</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ <scope>compile</scope>
+ </dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
Propchange: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/
------------------------------------------------------------------------------
Merged /hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java:r1541342-1542122
Merged /hadoop/common/branches/branch-2/hadoop-common-project/hadoop-common/src/main/java:r1541667
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java Thu Nov 14 23:56:56 2013
@@ -65,6 +65,13 @@ public class CommonConfigurationKeys ext
/** Default value for IPC_SERVER_RPC_READ_THREADS_KEY */
public static final int IPC_SERVER_RPC_READ_THREADS_DEFAULT = 1;
+ /** Number of pending connections that may be queued per socket reader */
+ public static final String IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_KEY =
+ "ipc.server.read.connection-queue.size";
+ /** Default value for IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE */
+ public static final int IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_DEFAULT =
+ 100;
+
public static final String IPC_MAXIMUM_DATA_LENGTH =
"ipc.maximum.data.length";
@@ -219,4 +226,10 @@ public class CommonConfigurationKeys ext
public static final String IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY = "ipc.client.fallback-to-simple-auth-allowed";
public static final boolean IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_DEFAULT = false;
+ /** How often the server scans for idle connections */
+ public static final String IPC_CLIENT_CONNECTION_IDLESCANINTERVAL_KEY =
+ "ipc.client.connection.idle-scan-interval.ms";
+ /** Default value for IPC_SERVER_CONNECTION_IDLE_SCAN_INTERVAL_KEY */
+ public static final int IPC_CLIENT_CONNECTION_IDLESCANINTERVAL_DEFAULT =
+ 10000;
}
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/CodecPool.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/CodecPool.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/CodecPool.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/CodecPool.java Thu Nov 14 23:56:56 2013
@@ -85,16 +85,15 @@ public class CodecPool {
T codec = null;
// Check if an appropriate codec is available
+ List<T> codecList;
synchronized (pool) {
- if (pool.containsKey(codecClass)) {
- List<T> codecList = pool.get(codecClass);
-
- if (codecList != null) {
- synchronized (codecList) {
- if (!codecList.isEmpty()) {
- codec = codecList.remove(codecList.size()-1);
- }
- }
+ codecList = pool.get(codecClass);
+ }
+
+ if (codecList != null) {
+ synchronized (codecList) {
+ if (!codecList.isEmpty()) {
+ codec = codecList.remove(codecList.size() - 1);
}
}
}
@@ -105,15 +104,17 @@ public class CodecPool {
private static <T> void payback(Map<Class<T>, List<T>> pool, T codec) {
if (codec != null) {
Class<T> codecClass = ReflectionUtils.getClass(codec);
+ List<T> codecList;
synchronized (pool) {
- if (!pool.containsKey(codecClass)) {
- pool.put(codecClass, new ArrayList<T>());
+ codecList = pool.get(codecClass);
+ if (codecList == null) {
+ codecList = new ArrayList<T>();
+ pool.put(codecClass, codecList);
}
+ }
- List<T> codecList = pool.get(codecClass);
- synchronized (codecList) {
- codecList.add(codec);
- }
+ synchronized (codecList) {
+ codecList.add(codec);
}
}
}
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java Thu Nov 14 23:56:56 2013
@@ -114,7 +114,17 @@ public class NativeIO {
public static interface CacheTracker {
public void fadvise(String identifier, long offset, long len, int flags);
}
-
+
+ public static CacheManipulator cacheManipulator = new CacheManipulator();
+
+ @VisibleForTesting
+ public static class CacheManipulator {
+ public void mlock(String identifier, ByteBuffer buffer,
+ long len) throws IOException {
+ POSIX.mlock(buffer, len);
+ }
+ }
+
static {
if (NativeCodeLoader.isNativeCodeLoaded()) {
try {
@@ -249,7 +259,7 @@ public class NativeIO {
*
* @throws NativeIOException
*/
- public static void mlock(ByteBuffer buffer, long len)
+ static void mlock(ByteBuffer buffer, long len)
throws IOException {
assertCodeLoaded();
if (!buffer.isDirect()) {
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java Thu Nov 14 23:56:56 2013
@@ -51,11 +51,13 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Random;
import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
import javax.security.sasl.Sasl;
import javax.security.sasl.SaslException;
@@ -345,17 +347,8 @@ public abstract class Server {
private int port; // port we listen on
private int handlerCount; // number of handler threads
private int readThreads; // number of read threads
+ private int readerPendingConnectionQueue; // number of connections to queue per read thread
private Class<? extends Writable> rpcRequestClass; // class used for deserializing the rpc request
- private int maxIdleTime; // the maximum idle time after
- // which a client may be disconnected
- private int thresholdIdleConnections; // the number of idle connections
- // after which we will start
- // cleaning up idle
- // connections
- int maxConnectionsToNuke; // the max number of
- // connections to nuke
- //during a cleanup
-
protected RpcMetrics rpcMetrics;
protected RpcDetailedMetrics rpcDetailedMetrics;
@@ -373,13 +366,10 @@ public abstract class Server {
volatile private boolean running = true; // true while server runs
private BlockingQueue<Call> callQueue; // queued calls
- private List<Connection> connectionList =
- Collections.synchronizedList(new LinkedList<Connection>());
- //maintain a list
- //of client connections
+ // maintains the set of client connections and handles idle timeouts
+ private ConnectionManager connectionManager;
private Listener listener = null;
private Responder responder = null;
- private int numConnections = 0;
private Handler[] handlers = null;
/**
@@ -449,8 +439,8 @@ public abstract class Server {
}
@VisibleForTesting
- List<Connection> getConnections() {
- return connectionList;
+ Connection[] getConnections() {
+ return connectionManager.toArray();
}
/**
@@ -518,11 +508,6 @@ public abstract class Server {
private Reader[] readers = null;
private int currentReader = 0;
private InetSocketAddress address; //the address we bind at
- private Random rand = new Random();
- private long lastCleanupRunTime = 0; //the last time when a cleanup connec-
- //-tion (for idle connections) ran
- private long cleanupInterval = 10000; //the minimum interval between
- //two cleanup runs
private int backlogLength = conf.getInt(
CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_KEY,
CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_DEFAULT);
@@ -553,12 +538,14 @@ public abstract class Server {
}
private class Reader extends Thread {
- private volatile boolean adding = false;
+ final private BlockingQueue<Connection> pendingConnections;
private final Selector readSelector;
Reader(String name) throws IOException {
super(name);
+ this.pendingConnections =
+ new LinkedBlockingQueue<Connection>(readerPendingConnectionQueue);
this.readSelector = Selector.open();
}
@@ -580,10 +567,14 @@ public abstract class Server {
while (running) {
SelectionKey key = null;
try {
+ // consume as many connections as currently queued to avoid
+ // unbridled acceptance of connections that starves the select
+ int size = pendingConnections.size();
+ for (int i=size; i>0; i--) {
+ Connection conn = pendingConnections.take();
+ conn.channel.register(readSelector, SelectionKey.OP_READ, conn);
+ }
readSelector.select();
- while (adding) {
- this.wait(1000);
- }
Iterator<SelectionKey> iter = readSelector.selectedKeys().iterator();
while (iter.hasNext()) {
@@ -607,26 +598,14 @@ public abstract class Server {
}
/**
- * This gets reader into the state that waits for the new channel
- * to be registered with readSelector. If it was waiting in select()
- * the thread will be woken up, otherwise whenever select() is called
- * it will return even if there is nothing to read and wait
- * in while(adding) for finishAdd call
+ * Updating the readSelector while it's being used is not thread-safe,
+ * so the connection must be queued. The reader will drain the queue
+ * and update its readSelector before performing the next select
*/
- public void startAdd() {
- adding = true;
+ public void addConnection(Connection conn) throws InterruptedException {
+ pendingConnections.put(conn);
readSelector.wakeup();
}
-
- public synchronized SelectionKey registerChannel(SocketChannel channel)
- throws IOException {
- return channel.register(readSelector, SelectionKey.OP_READ);
- }
-
- public synchronized void finishAdd() {
- adding = false;
- this.notify();
- }
void shutdown() {
assert !running;
@@ -638,58 +617,12 @@ public abstract class Server {
}
}
}
- /** cleanup connections from connectionList. Choose a random range
- * to scan and also have a limit on the number of the connections
- * that will be cleanedup per run. The criteria for cleanup is the time
- * for which the connection was idle. If 'force' is true then all
- * connections will be looked at for the cleanup.
- */
- private void cleanupConnections(boolean force) {
- if (force || numConnections > thresholdIdleConnections) {
- long currentTime = Time.now();
- if (!force && (currentTime - lastCleanupRunTime) < cleanupInterval) {
- return;
- }
- int start = 0;
- int end = numConnections - 1;
- if (!force) {
- start = rand.nextInt() % numConnections;
- end = rand.nextInt() % numConnections;
- int temp;
- if (end < start) {
- temp = start;
- start = end;
- end = temp;
- }
- }
- int i = start;
- int numNuked = 0;
- while (i <= end) {
- Connection c;
- synchronized (connectionList) {
- try {
- c = connectionList.get(i);
- } catch (Exception e) {return;}
- }
- if (c.timedOut(currentTime)) {
- if (LOG.isDebugEnabled())
- LOG.debug(getName() + ": disconnecting client " + c.getHostAddress());
- closeConnection(c);
- numNuked++;
- end--;
- c = null;
- if (!force && numNuked == maxConnectionsToNuke) break;
- }
- else i++;
- }
- lastCleanupRunTime = Time.now();
- }
- }
@Override
public void run() {
LOG.info(getName() + ": starting");
SERVER.set(Server.this);
+ connectionManager.startIdleScan();
while (running) {
SelectionKey key = null;
try {
@@ -713,12 +646,11 @@ public abstract class Server {
// some thread(s) a chance to finish
LOG.warn("Out of Memory in server select", e);
closeCurrentConnection(key, e);
- cleanupConnections(true);
+ connectionManager.closeIdle(true);
try { Thread.sleep(60000); } catch (Exception ie) {}
} catch (Exception e) {
closeCurrentConnection(key, e);
}
- cleanupConnections(false);
}
LOG.info("Stopping " + this.getName());
@@ -731,10 +663,9 @@ public abstract class Server {
selector= null;
acceptChannel= null;
- // clean up all connections
- while (!connectionList.isEmpty()) {
- closeConnection(connectionList.remove(0));
- }
+ // close all connections
+ connectionManager.stopIdleScan();
+ connectionManager.closeAll();
}
}
@@ -742,8 +673,6 @@ public abstract class Server {
if (key != null) {
Connection c = (Connection)key.attachment();
if (c != null) {
- if (LOG.isDebugEnabled())
- LOG.debug(getName() + ": disconnecting client " + c.getHostAddress());
closeConnection(c);
c = null;
}
@@ -754,8 +683,7 @@ public abstract class Server {
return (InetSocketAddress)acceptChannel.socket().getLocalSocketAddress();
}
- void doAccept(SelectionKey key) throws IOException, OutOfMemoryError {
- Connection c = null;
+ void doAccept(SelectionKey key) throws InterruptedException, IOException, OutOfMemoryError {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel channel;
while ((channel = server.accept()) != null) {
@@ -765,22 +693,9 @@ public abstract class Server {
channel.socket().setKeepAlive(true);
Reader reader = getReader();
- try {
- reader.startAdd();
- SelectionKey readKey = reader.registerChannel(channel);
- c = new Connection(readKey, channel, Time.now());
- readKey.attach(c);
- synchronized (connectionList) {
- connectionList.add(numConnections, c);
- numConnections++;
- }
- if (LOG.isDebugEnabled())
- LOG.debug("Server connection from " + c.toString() +
- "; # active connections: " + numConnections +
- "; # queued calls: " + callQueue.size());
- } finally {
- reader.finishAdd();
- }
+ Connection c = connectionManager.register(channel);
+ key.attach(c); // so closeCurrentConnection can get the object
+ reader.addConnection(c);
}
}
@@ -808,10 +723,6 @@ public abstract class Server {
count = -1; //so that the (count < 0) block is executed
}
if (count < 0) {
- if (LOG.isDebugEnabled())
- LOG.debug(getName() + ": disconnecting client " +
- c + ". Number of active connections: "+
- numConnections);
closeConnection(c);
c = null;
}
@@ -1190,8 +1101,7 @@ public abstract class Server {
private boolean sentNegotiate = false;
private boolean useWrap = false;
- public Connection(SelectionKey key, SocketChannel channel,
- long lastContact) {
+ public Connection(SocketChannel channel, long lastContact) {
this.channel = channel;
this.lastContact = lastContact;
this.data = null;
@@ -1253,12 +1163,6 @@ public abstract class Server {
rpcCount++;
}
- private boolean timedOut(long currentTime) {
- if (isIdle() && currentTime - lastContact > maxIdleTime)
- return true;
- return false;
- }
-
private UserGroupInformation getAuthorizedUgi(String authorizedId)
throws InvalidToken, AccessControlException {
if (authMethod == AuthMethod.TOKEN) {
@@ -2189,16 +2093,10 @@ public abstract class Server {
CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_KEY,
CommonConfigurationKeys.IPC_SERVER_RPC_READ_THREADS_DEFAULT);
}
+ this.readerPendingConnectionQueue = conf.getInt(
+ CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_KEY,
+ CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_DEFAULT);
this.callQueue = new LinkedBlockingQueue<Call>(maxQueueSize);
- this.maxIdleTime = 2 * conf.getInt(
- CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY,
- CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_DEFAULT);
- this.maxConnectionsToNuke = conf.getInt(
- CommonConfigurationKeysPublic.IPC_CLIENT_KILL_MAX_KEY,
- CommonConfigurationKeysPublic.IPC_CLIENT_KILL_MAX_DEFAULT);
- this.thresholdIdleConnections = conf.getInt(
- CommonConfigurationKeysPublic.IPC_CLIENT_IDLETHRESHOLD_KEY,
- CommonConfigurationKeysPublic.IPC_CLIENT_IDLETHRESHOLD_DEFAULT);
this.secretManager = (SecretManager<TokenIdentifier>) secretManager;
this.authorize =
conf.getBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION,
@@ -2219,6 +2117,7 @@ public abstract class Server {
// Create the responder here
responder = new Responder();
+ connectionManager = new ConnectionManager();
if (secretManager != null) {
SaslRpcServer.init(conf);
@@ -2277,11 +2176,7 @@ public abstract class Server {
}
private void closeConnection(Connection connection) {
- synchronized (connectionList) {
- if (connectionList.remove(connection))
- numConnections--;
- }
- connection.close();
+ connectionManager.close(connection);
}
/**
@@ -2536,7 +2431,7 @@ public abstract class Server {
* @return the number of open rpc connections
*/
public int getNumOpenConnections() {
- return numConnections;
+ return connectionManager.size();
}
/**
@@ -2646,4 +2541,151 @@ public abstract class Server {
int nBytes = initialRemaining - buf.remaining();
return (nBytes > 0) ? nBytes : ret;
}
+
+ private class ConnectionManager {
+ final private AtomicInteger count = new AtomicInteger();
+ final private Set<Connection> connections;
+
+ final private Timer idleScanTimer;
+ final private int idleScanThreshold;
+ final private int idleScanInterval;
+ final private int maxIdleTime;
+ final private int maxIdleToClose;
+
+ ConnectionManager() {
+ this.idleScanTimer = new Timer(
+ "IPC Server idle connection scanner for port " + getPort(), true);
+ this.idleScanThreshold = conf.getInt(
+ CommonConfigurationKeysPublic.IPC_CLIENT_IDLETHRESHOLD_KEY,
+ CommonConfigurationKeysPublic.IPC_CLIENT_IDLETHRESHOLD_DEFAULT);
+ this.idleScanInterval = conf.getInt(
+ CommonConfigurationKeys.IPC_CLIENT_CONNECTION_IDLESCANINTERVAL_KEY,
+ CommonConfigurationKeys.IPC_CLIENT_CONNECTION_IDLESCANINTERVAL_DEFAULT);
+ this.maxIdleTime = 2 * conf.getInt(
+ CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY,
+ CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_DEFAULT);
+ this.maxIdleToClose = conf.getInt(
+ CommonConfigurationKeysPublic.IPC_CLIENT_KILL_MAX_KEY,
+ CommonConfigurationKeysPublic.IPC_CLIENT_KILL_MAX_DEFAULT);
+ // create a set with concurrency -and- a thread-safe iterator, add 2
+ // for listener and idle closer threads
+ this.connections = Collections.newSetFromMap(
+ new ConcurrentHashMap<Connection,Boolean>(
+ maxQueueSize, 0.75f, readThreads+2));
+ }
+
+ private boolean add(Connection connection) {
+ boolean added = connections.add(connection);
+ if (added) {
+ count.getAndIncrement();
+ }
+ return added;
+ }
+
+ private boolean remove(Connection connection) {
+ boolean removed = connections.remove(connection);
+ if (removed) {
+ count.getAndDecrement();
+ }
+ return removed;
+ }
+
+ int size() {
+ return count.get();
+ }
+
+ Connection[] toArray() {
+ return connections.toArray(new Connection[0]);
+ }
+
+ Connection register(SocketChannel channel) {
+ Connection connection = new Connection(channel, Time.now());
+ add(connection);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Server connection from " + connection +
+ "; # active connections: " + size() +
+ "; # queued calls: " + callQueue.size());
+ }
+ return connection;
+ }
+
+ boolean close(Connection connection) {
+ boolean exists = remove(connection);
+ if (exists) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(Thread.currentThread().getName() +
+ ": disconnecting client " + connection +
+ ". Number of active connections: "+ size());
+ }
+ // only close if actually removed to avoid double-closing due
+ // to possible races
+ connection.close();
+ }
+ return exists;
+ }
+
+ // synch'ed to avoid explicit invocation upon OOM from colliding with
+ // timer task firing
+ synchronized void closeIdle(boolean scanAll) {
+ long minLastContact = Time.now() - maxIdleTime;
+ // concurrent iterator might miss new connections added
+ // during the iteration, but that's ok because they won't
+ // be idle yet anyway and will be caught on next scan
+ int closed = 0;
+ for (Connection connection : connections) {
+ // stop if connections dropped below threshold unless scanning all
+ if (!scanAll && size() < idleScanThreshold) {
+ break;
+ }
+ // stop if not scanning all and max connections are closed
+ if (connection.isIdle() &&
+ connection.getLastContact() < minLastContact &&
+ close(connection) &&
+ !scanAll && (++closed == maxIdleToClose)) {
+ break;
+ }
+ }
+ }
+
+ void closeAll() {
+ // use a copy of the connections to be absolutely sure the concurrent
+ // iterator doesn't miss a connection
+ for (Connection connection : toArray()) {
+ close(connection);
+ }
+ }
+
+ void startIdleScan() {
+ scheduleIdleScanTask();
+ }
+
+ void stopIdleScan() {
+ idleScanTimer.cancel();
+ }
+
+ private void scheduleIdleScanTask() {
+ if (!running) {
+ return;
+ }
+ TimerTask idleScanTask = new TimerTask(){
+ @Override
+ public void run() {
+ if (!running) {
+ return;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(Thread.currentThread().getName()+": task running");
+ }
+ try {
+ closeIdle(false);
+ } finally {
+ // explicitly reschedule so next execution occurs relative
+ // to the end of this scan, not the beginning
+ scheduleIdleScanTask();
+ }
+ }
+ };
+ idleScanTimer.schedule(idleScanTask, idleScanInterval);
+ }
+ }
}
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java Thu Nov 14 23:56:56 2013
@@ -38,7 +38,7 @@ import static com.google.common.base.Pre
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.commons.math.util.MathUtils;
+import org.apache.commons.math3.util.ArithmeticUtils;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.metrics2.MetricsInfo;
import org.apache.hadoop.metrics2.MetricsCollector;
@@ -457,7 +457,7 @@ public class MetricsSystemImpl extends M
MetricsConfig conf = entry.getValue();
int sinkPeriod = conf.getInt(PERIOD_KEY, PERIOD_DEFAULT);
confPeriod = confPeriod == 0 ? sinkPeriod
- : MathUtils.gcd(confPeriod, sinkPeriod);
+ : ArithmeticUtils.gcd(confPeriod, sinkPeriod);
String clsName = conf.getClassName("");
if (clsName == null) continue; // sink can be registered later on
String sinkName = entry.getKey();
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/GenericOptionsParser.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/GenericOptionsParser.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/GenericOptionsParser.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/GenericOptionsParser.java Thu Nov 14 23:56:56 2013
@@ -431,6 +431,9 @@ public class GenericOptionsParser {
if (!Shell.WINDOWS) {
return args;
}
+ if (args == null) {
+ return null;
+ }
List<String> newArgs = new ArrayList<String>(args.length);
for (int i=0; i < args.length; i++) {
String prop = null;
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ZKUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ZKUtil.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ZKUtil.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ZKUtil.java Thu Nov 14 23:56:56 2013
@@ -72,6 +72,20 @@ public class ZKUtil {
}
/**
+ * Helper method to remove a subset of permissions (remove) from a
+ * given set (perms).
+ * @param perms The permissions flag to remove from. Should be an OR of a
+ * some combination of {@link ZooDefs.Perms}
+ * @param remove The permissions to be removed. Should be an OR of a
+ * some combination of {@link ZooDefs.Perms}
+ * @return A permissions flag that is an OR of {@link ZooDefs.Perms}
+ * present in perms and not present in remove
+ */
+ public static int removeSpecificPerms(int perms, int remove) {
+ return perms ^ remove;
+ }
+
+ /**
* Parse comma separated list of ACL entries to secure generated nodes, e.g.
* <code>sasl:hdfs/host1@MY.DOMAIN:cdrwa,sasl:hdfs/host2@MY.DOMAIN:cdrwa</code>
*
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java Thu Nov 14 23:56:56 2013
@@ -44,12 +44,19 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Random;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import javax.net.SocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.commons.logging.impl.Log4JLogger;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
@@ -64,8 +71,10 @@ import org.apache.hadoop.ipc.protobuf.Rp
import org.apache.hadoop.net.ConnectTimeoutException;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.Level;
import org.junit.Assert;
import org.junit.Assume;
+import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@@ -79,7 +88,7 @@ public class TestIPC {
public static final Log LOG =
LogFactory.getLog(TestIPC.class);
- final private static Configuration conf = new Configuration();
+ private static Configuration conf;
final static private int PING_INTERVAL = 1000;
final static private int MIN_SLEEP_TIME = 1000;
/**
@@ -89,7 +98,9 @@ public class TestIPC {
static boolean WRITABLE_FAULTS_ENABLED = true;
static int WRITABLE_FAULTS_SLEEP = 0;
- static {
+ @Before
+ public void setupConf() {
+ conf = new Configuration();
Client.setPingInterval(conf, PING_INTERVAL);
}
@@ -613,6 +624,255 @@ public class TestIPC {
server.stop();
}
+ private static class TestServerQueue extends Server {
+ final CountDownLatch firstCallLatch = new CountDownLatch(1);
+ final CountDownLatch callBlockLatch = new CountDownLatch(1);
+
+ TestServerQueue(int expectedCalls, int readers, int callQ, int handlers,
+ Configuration conf) throws IOException {
+ super(ADDRESS, 0, LongWritable.class, handlers, readers, callQ, conf, null, null);
+ }
+
+ @Override
+ public Writable call(RPC.RpcKind rpcKind, String protocol, Writable param,
+ long receiveTime) throws IOException {
+ firstCallLatch.countDown();
+ try {
+ callBlockLatch.await();
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ return param;
+ }
+ }
+
+ /**
+ * Check that reader queueing works
+ * @throws BrokenBarrierException
+ * @throws InterruptedException
+ */
+ @Test(timeout=60000)
+ public void testIpcWithReaderQueuing() throws Exception {
+ // 1 reader, 1 connectionQ slot, 1 callq
+ for (int i=0; i < 10; i++) {
+ checkBlocking(1, 1, 1);
+ }
+ // 4 readers, 5 connectionQ slots, 2 callq
+ for (int i=0; i < 10; i++) {
+ checkBlocking(4, 5, 2);
+ }
+ }
+
+ // goal is to jam a handler with a connection, fill the callq with
+ // connections, in turn jamming the readers - then flood the server and
+ // ensure that the listener blocks when the reader connection queues fill
+ private void checkBlocking(int readers, int readerQ, int callQ) throws Exception {
+ int handlers = 1; // makes it easier
+
+ final Configuration conf = new Configuration();
+ conf.setInt(CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_KEY, readerQ);
+
+ // send in enough clients to block up the handlers, callq, and readers
+ int initialClients = readers + callQ + handlers;
+ // max connections we should ever end up accepting at once
+ int maxAccept = initialClients + readers*readerQ + 1; // 1 = listener
+ // stress it with 2X the max
+ int clients = maxAccept*2;
+
+ final AtomicInteger failures = new AtomicInteger(0);
+ final CountDownLatch callFinishedLatch = new CountDownLatch(clients);
+
+ // start server
+ final TestServerQueue server =
+ new TestServerQueue(clients, readers, callQ, handlers, conf);
+ final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+ server.start();
+
+ Client.setConnectTimeout(conf, 10000);
+
+ // instantiate the threads, will start in batches
+ Thread[] threads = new Thread[clients];
+ for (int i=0; i<clients; i++) {
+ threads[i] = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ Client client = new Client(LongWritable.class, conf);
+ try {
+ client.call(new LongWritable(Thread.currentThread().getId()),
+ addr, null, null, 60000, conf);
+ } catch (Throwable e) {
+ LOG.error(e);
+ failures.incrementAndGet();
+ return;
+ } finally {
+ callFinishedLatch.countDown();
+ client.stop();
+ }
+ }
+ });
+ }
+
+ // start enough clients to block up the handler, callq, and each reader;
+ // let the calls sequentially slot in to avoid some readers blocking
+ // and others not blocking in the race to fill the callq
+ for (int i=0; i < initialClients; i++) {
+ threads[i].start();
+ if (i==0) {
+ // let first reader block in a call
+ server.firstCallLatch.await();
+ } else if (i <= callQ) {
+ // let subsequent readers jam the callq, will happen immediately
+ while (server.getCallQueueLen() != i) {
+ Thread.sleep(1);
+ }
+ } // additional threads block the readers trying to add to the callq
+ }
+
+ // wait till everything is slotted, should happen immediately
+ Thread.sleep(10);
+ if (server.getNumOpenConnections() < initialClients) {
+ LOG.info("(initial clients) need:"+initialClients+" connections have:"+server.getNumOpenConnections());
+ Thread.sleep(100);
+ }
+ LOG.info("ipc layer should be blocked");
+ assertEquals(callQ, server.getCallQueueLen());
+ assertEquals(initialClients, server.getNumOpenConnections());
+
+ // now flood the server with the rest of the connections, the reader's
+ // connection queues should fill and then the listener should block
+ for (int i=initialClients; i<clients; i++) {
+ threads[i].start();
+ }
+ Thread.sleep(10);
+ if (server.getNumOpenConnections() < maxAccept) {
+ LOG.info("(max clients) need:"+maxAccept+" connections have:"+server.getNumOpenConnections());
+ Thread.sleep(100);
+ }
+ // check a few times to make sure we didn't go over
+ for (int i=0; i<4; i++) {
+ assertEquals(maxAccept, server.getNumOpenConnections());
+ Thread.sleep(100);
+ }
+
+ // sanity check that no calls have finished
+ assertEquals(clients, callFinishedLatch.getCount());
+ LOG.info("releasing the calls");
+ server.callBlockLatch.countDown();
+ callFinishedLatch.await();
+ for (Thread t : threads) {
+ t.join();
+ }
+ assertEquals(0, failures.get());
+ server.stop();
+ }
+
+ @Test(timeout=30000)
+ public void testConnectionIdleTimeouts() throws Exception {
+ ((Log4JLogger)Server.LOG).getLogger().setLevel(Level.DEBUG);
+ final int maxIdle = 1000;
+ final int cleanupInterval = maxIdle*3/4; // stagger cleanups
+ final int killMax = 3;
+ final int clients = 1 + killMax*2; // 1 to block, 2 batches to kill
+
+ conf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY, maxIdle);
+ conf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_IDLETHRESHOLD_KEY, 0);
+ conf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_KILL_MAX_KEY, killMax);
+ conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECTION_IDLESCANINTERVAL_KEY, cleanupInterval);
+
+ final CyclicBarrier firstCallBarrier = new CyclicBarrier(2);
+ final CyclicBarrier callBarrier = new CyclicBarrier(clients);
+ final CountDownLatch allCallLatch = new CountDownLatch(clients);
+ final AtomicBoolean error = new AtomicBoolean();
+
+ final TestServer server = new TestServer(clients, false);
+ Thread[] threads = new Thread[clients];
+ try {
+ server.callListener = new Runnable(){
+ AtomicBoolean first = new AtomicBoolean(true);
+ @Override
+ public void run() {
+ try {
+ allCallLatch.countDown();
+ // block first call
+ if (first.compareAndSet(true, false)) {
+ firstCallBarrier.await();
+ } else {
+ callBarrier.await();
+ }
+ } catch (Throwable t) {
+ LOG.error(t);
+ error.set(true);
+ }
+ }
+ };
+ server.start();
+
+ // start client
+ final CountDownLatch callReturned = new CountDownLatch(clients-1);
+ final InetSocketAddress addr = NetUtils.getConnectAddress(server);
+ final Configuration clientConf = new Configuration();
+ clientConf.setInt(CommonConfigurationKeysPublic.IPC_CLIENT_CONNECTION_MAXIDLETIME_KEY, 10000);
+ for (int i=0; i < clients; i++) {
+ threads[i] = new Thread(new Runnable(){
+ @Override
+ public void run() {
+ Client client = new Client(LongWritable.class, clientConf);
+ try {
+ client.call(new LongWritable(Thread.currentThread().getId()),
+ addr, null, null, 0, clientConf);
+ callReturned.countDown();
+ Thread.sleep(10000);
+ } catch (IOException e) {
+ LOG.error(e);
+ } catch (InterruptedException e) {
+ }
+ }
+ });
+ threads[i].start();
+ }
+
+ // all calls blocked in handler so all connections made
+ allCallLatch.await();
+ assertFalse(error.get());
+ assertEquals(clients, server.getNumOpenConnections());
+
+ // wake up blocked calls and wait for client call to return, no
+ // connections should have closed
+ callBarrier.await();
+ callReturned.await();
+ assertEquals(clients, server.getNumOpenConnections());
+
+ // server won't close till maxIdle*2, so give scanning thread time to
+ // be almost ready to close idle connection. after which it should
+ // close max connections on every cleanupInterval
+ Thread.sleep(maxIdle*2-cleanupInterval);
+ for (int i=clients; i > 1; i -= killMax) {
+ Thread.sleep(cleanupInterval);
+ assertFalse(error.get());
+ assertEquals(i, server.getNumOpenConnections());
+ }
+
+ // connection for the first blocked call should still be open
+ Thread.sleep(cleanupInterval);
+ assertFalse(error.get());
+ assertEquals(1, server.getNumOpenConnections());
+
+ // wake up call and ensure connection times out
+ firstCallBarrier.await();
+ Thread.sleep(maxIdle*2);
+ assertFalse(error.get());
+ assertEquals(0, server.getNumOpenConnections());
+ } finally {
+ for (Thread t : threads) {
+ if (t != null) {
+ t.interrupt();
+ t.join();
+ }
+ server.stop();
+ }
+ }
+ }
+
/**
* Make a call from a client and verify if header info is changed in server side
*/
@@ -622,7 +882,7 @@ public class TestIPC {
client.call(new LongWritable(RANDOM.nextLong()),
addr, null, null, MIN_SLEEP_TIME, serviceClass, conf);
- Connection connection = server.getConnections().get(0);
+ Connection connection = server.getConnections()[0];
int serviceClass2 = connection.getServiceClass();
assertFalse(noChanged ^ serviceClass == serviceClass2);
client.stop();
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestGenericOptionsParser.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestGenericOptionsParser.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestGenericOptionsParser.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestGenericOptionsParser.java Thu Nov 14 23:56:56 2013
@@ -282,4 +282,12 @@ public class TestGenericOptionsParser ex
Arrays.toString(remainingArgs) + Arrays.toString(expectedRemainingArgs),
expectedRemainingArgs, remainingArgs);
}
+
+ /** Test passing null as args. Some classes still call
+ * Tool interface from java passing null.
+ */
+ public void testNullArgs() throws IOException {
+ GenericOptionsParser parser = new GenericOptionsParser(conf, null);
+ parser.getRemainingArgs();
+ }
}
Modified: hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestZKUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestZKUtil.java?rev=1542125&r1=1542124&r2=1542125&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestZKUtil.java (original)
+++ hadoop/common/branches/HDFS-2832/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestZKUtil.java Thu Nov 14 23:56:56 2013
@@ -24,7 +24,6 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.List;
-import org.apache.hadoop.util.ZKUtil;
import org.apache.hadoop.util.ZKUtil.BadAclFormatException;
import org.apache.hadoop.util.ZKUtil.ZKAuthInfo;
import org.apache.zookeeper.ZooDefs.Perms;
@@ -76,6 +75,14 @@ public class TestZKUtil {
}
@Test
+ public void testRemoveSpecificPerms() {
+ int perms = Perms.ALL;
+ int remove = Perms.CREATE;
+ int newPerms = ZKUtil.removeSpecificPerms(perms, remove);
+ assertEquals("Removal failed", 0, newPerms & Perms.CREATE);
+ }
+
+ @Test
public void testGoodACLs() {
List<ACL> result = ZKUtil.parseACLs(
"sasl:hdfs/host1@MY.DOMAIN:cdrwa, sasl:hdfs/host2@MY.DOMAIN:ca");