HADOOP-4346. Implement blocking connect so that Hadoop is not affected

by selector problem with JDK default implementation. (Raghu Angadi)


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@708773 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -80,6 +80,9 @@ Trunk (unreleased changes)
     HADOOP-4482. Make the JMX monitoring use predictable names for the 
     datanodes to enable Nagios monitoring. (Brian Bockelman via omalley)
 
+    HADOOP-4346. Implement blocking connect so that Hadoop is not affected
+    selector problem with JDK default implementation. (Raghu Angadi)
+
 Release 0.19.0 - Unreleased
 
   INCOMPATIBLE CHANGES

src/core/org/apache/hadoop/ipc/Client.java

@@ -296,7 +296,7 @@ public class Client {
             this.socket = socketFactory.createSocket();
             this.socket.setTcpNoDelay(tcpNoDelay);
             // connection time out is 20s
-            this.socket.connect(remoteId.getAddress(), 20000);
+            NetUtils.connect(this.socket, remoteId.getAddress(), 20000);
             this.socket.setSoTimeout(pingInterval);
             break;
           } catch (SocketTimeoutException toe) {

src/core/org/apache/hadoop/net/NetUtils.java

@@ -23,8 +23,10 @@ import java.io.OutputStream;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.net.SocketAddress;
 import java.net.URI;
 import java.net.UnknownHostException;
+import java.nio.channels.SocketChannel;
 import java.util.Map.Entry;
 import java.util.*;
 
@@ -369,6 +371,40 @@ public class NetUtils {
             socket.getOutputStream() : new SocketOutputStream(socket, timeout);            
   }
   
+  /**
+   * This is a drop-in replacement for 
+   * {@link Socket#connect(SocketAddress, int)}.
+   * In the case of normal sockets that don't have associated channels, this 
+   * just invokes <code>socket.connect(endpoint, timeout)</code>. If 
+   * <code>socket.getChannel()</code> returns a non-null channel,
+   * connect is implemented using Hadoop's selectors. This is done mainly
+   * to avoid Sun's connect implementation from creating thread-local 
+   * selectors, since Hadoop does not have control on when these are closed
+   * and could end up taking all the available file descriptors.
+   * 
+   * @see java.net.Socket#connect(java.net.SocketAddress, int)
+   * 
+   * @param socket
+   * @param endpoint 
+   * @param timeout - timeout in milliseconds
+   */
+  public static void connect(Socket socket, 
+                             SocketAddress endpoint, 
+                             int timeout) throws IOException {
+    if (socket == null || endpoint == null || timeout < 0) {
+      throw new IllegalArgumentException("Illegal argument for connect()");
+    }
+    
+    SocketChannel ch = socket.getChannel();
+    
+    if (ch == null) {
+      // let the default implementation handle it.
+      socket.connect(endpoint, timeout);
+    } else {
+      SocketIOWithTimeout.connect(ch, endpoint, timeout);
+    }
+  }
+  
   /** 
    * Given a string representation of a host, return its ip address
    * in textual presentation.

src/core/org/apache/hadoop/net/SocketIOWithTimeout.java

@@ -20,11 +20,13 @@ package org.apache.hadoop.net;
 
 import java.io.IOException;
 import java.io.InterruptedIOException;
+import java.net.SocketAddress;
 import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
 import java.nio.channels.spi.SelectorProvider;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -159,7 +161,8 @@ abstract class SocketIOWithTimeout {
       } 
 
       if (count == 0) {
-        throw new SocketTimeoutException(timeoutExceptionString(ops));
+        throw new SocketTimeoutException(timeoutExceptionString(channel,
+                                                                timeout, ops));
       }
       // otherwise the socket should be ready for io.
     }
@@ -167,6 +170,64 @@ abstract class SocketIOWithTimeout {
     return 0; // does not reach here.
   }
   
+  /**
+   * The contract is similar to {@link SocketChannel#connect(SocketAddress)} 
+   * with a timeout.
+   * 
+   * @see SocketChannel#connect(SocketAddress)
+   * 
+   * @param channel - this should be a {@link SelectableChannel}
+   * @param endpoint
+   * @throws IOException
+   */
+  static void connect(SocketChannel channel, 
+                      SocketAddress endpoint, int timeout) throws IOException {
+    
+    boolean blockingOn = channel.isBlocking();
+    if (blockingOn) {
+      channel.configureBlocking(false);
+    }
+    
+    try { 
+      if (channel.connect(endpoint)) {
+        return;
+      }
+
+      long timeoutLeft = timeout;
+      long endTime = (timeout > 0) ? (System.currentTimeMillis() + timeout): 0;
+      
+      while (true) {
+        // we might have to call finishConnect() more than once
+        // for some channels (with user level protocols)
+        
+        int ret = selector.select((SelectableChannel)channel, 
+                                  SelectionKey.OP_CONNECT, timeoutLeft);
+        
+        if (ret > 0 && channel.finishConnect()) {
+          return;
+        }
+        
+        if (ret == 0 ||
+            (timeout > 0 &&  
+              (timeoutLeft = (endTime - System.currentTimeMillis())) <= 0)) {
+          throw new SocketTimeoutException(
+                    timeoutExceptionString(channel, timeout, 
+                                           SelectionKey.OP_CONNECT));
+        }
+      }
+    } catch (IOException e) {
+      // javadoc for SocketChannel.connect() says channel should be closed.
+      try {
+        channel.close();
+      } catch (IOException ignored) {}
+      throw e;
+    } finally {
+      if (blockingOn && channel.isOpen()) {
+        channel.configureBlocking(true);
+      }
+    }
+  }
+
   /**
    * This is similar to {@link #doIO(ByteBuffer, int)} except that it
    * does not perform any I/O. It just waits for the channel to be ready
@@ -182,17 +243,28 @@ abstract class SocketIOWithTimeout {
   void waitForIO(int ops) throws IOException {
     
     if (selector.select(channel, ops, timeout) == 0) {
-      throw new SocketTimeoutException(timeoutExceptionString(ops)); 
+      throw new SocketTimeoutException(timeoutExceptionString(channel, timeout,
+                                                              ops)); 
     }
   }
-  
-  private String timeoutExceptionString(int ops) {
     
-    String waitingFor = "" + ops;
-    if (ops == SelectionKey.OP_READ) {
-      waitingFor = "read";
-    } else if (ops == SelectionKey.OP_WRITE) {
-      waitingFor = "write";
+  private static String timeoutExceptionString(SelectableChannel channel,
+                                               long timeout, int ops) {
+    
+    String waitingFor;
+    switch(ops) {
+    
+    case SelectionKey.OP_READ :
+      waitingFor = "read"; break;
+      
+    case SelectionKey.OP_WRITE :
+      waitingFor = "write"; break;      
+      
+    case SelectionKey.OP_CONNECT :
+      waitingFor = "connect"; break;
+      
+    default :
+      waitingFor = "" + ops;  
     }
     
     return timeout + " millis timeout while " +

src/hdfs/org/apache/hadoop/hdfs/DFSClient.java

@@ -619,7 +619,9 @@ public class DFSClient implements FSConstants, java.io.Closeable {
       for(int j = 0; !done && j < datanodes.length; j++) {
         //connect to a datanode
         final Socket sock = socketFactory.createSocket();
-        sock.connect(NetUtils.createSocketAddr(datanodes[j].getName()), timeout);
+        NetUtils.connect(sock, 
+                         NetUtils.createSocketAddr(datanodes[j].getName()),
+                         timeout);
         sock.setSoTimeout(timeout);
 
         DataOutputStream out = new DataOutputStream(
@@ -1531,7 +1533,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
 
         try {
           s = socketFactory.createSocket();
-          s.connect(targetAddr, socketTimeout);
+          NetUtils.connect(s, targetAddr, socketTimeout);
           s.setSoTimeout(socketTimeout);
           Block blk = targetBlock.getBlock();
           
@@ -1733,7 +1735,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
             
         try {
           dn = socketFactory.createSocket();
-          dn.connect(targetAddr, socketTimeout);
+          NetUtils.connect(dn, targetAddr, socketTimeout);
           dn.setSoTimeout(socketTimeout);
               
           int len = (int) (end - start + 1);
@@ -2737,7 +2739,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
         InetSocketAddress target = NetUtils.createSocketAddr(nodes[0].getName());
         s = socketFactory.createSocket();
         int timeoutValue = 3000 * nodes.length + socketTimeout;
-        s.connect(target, timeoutValue);
+        NetUtils.connect(s, target, timeoutValue);
         s.setSoTimeout(timeoutValue);
         s.setSendBufferSize(DEFAULT_DATA_SOCKET_SIZE);
         LOG.debug("Send buf size " + s.getSendBufferSize());

src/hdfs/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -1033,7 +1033,7 @@ public class DataNode extends Configured
         InetSocketAddress curTarget = 
           NetUtils.createSocketAddr(targets[0].getName());
         sock = newSocket();
-        sock.connect(curTarget, socketTimeout);
+        NetUtils.connect(sock, curTarget, socketTimeout);
         sock.setSoTimeout(targets.length * socketTimeout);
 
         long writeTimeout = socketWriteTimeout + 

src/hdfs/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java

@@ -278,7 +278,7 @@ class DataXceiver implements Runnable, FSConstants {
           int timeoutValue = numTargets * datanode.socketTimeout;
           int writeTimeout = datanode.socketWriteTimeout + 
                              (HdfsConstants.WRITE_TIMEOUT_EXTENSION * numTargets);
-          mirrorSock.connect(mirrorTarget, timeoutValue);
+          NetUtils.connect(mirrorSock, mirrorTarget, timeoutValue);
           mirrorSock.setSoTimeout(timeoutValue);
           mirrorSock.setSendBufferSize(DEFAULT_DATA_SOCKET_SIZE);
           mirrorOut = new DataOutputStream(
@@ -557,7 +557,7 @@ class DataXceiver implements Runnable, FSConstants {
       InetSocketAddress proxyAddr = NetUtils.createSocketAddr(
           proxySource.getName());
       proxySock = datanode.newSocket();
-      proxySock.connect(proxyAddr, datanode.socketTimeout);
+      NetUtils.connect(proxySock, proxyAddr, datanode.socketTimeout);
       proxySock.setSoTimeout(datanode.socketTimeout);
 
       OutputStream baseStream = NetUtils.getOutputStream(proxySock,