Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-9669. TcpPeerServer should respect ipc.server.listen.queue.size (Elliot Clark via cmccabe) 

(cherry picked from commit 2da03b48eb)
(cherry picked from commit 60d3a3c30b)
(cherry picked from commit c4c94e1cf2)

Conflicts:
	hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
	hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/SecureDataNodeStarter.java
This commit is contained in:
Colin Patrick Mccabe 2016-02-02 10:59:10 -08:00
parent 2f031830e8
commit 449d2ad85c
4 changed files with 17 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View File

@ -27,6 +27,9 @@ Release 2.7.3 - UNRELEASED
HDFS-9654. Code refactoring for HDFS-8578. (szetszwo) HDFS-9654. Code refactoring for HDFS-8578. (szetszwo)
HDFS-9669. TcpPeerServer should respect ipc.server.listen.queue.size
(Elliot Clark via cmccabe)
OPTIMIZATIONS OPTIMIZATIONS
BUG FIXES BUG FIXES

6
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/TcpPeerServer.java
View File

@ -102,13 +102,15 @@ public class TcpPeerServer implements PeerServer {
* *
* @param socketWriteTimeout The Socket write timeout in ms. * @param socketWriteTimeout The Socket write timeout in ms.
* @param bindAddr The address to bind to. * @param bindAddr The address to bind to.
* @param backlogLength The length of the tcp accept backlog
* @throws IOException * @throws IOException
*/ */
public TcpPeerServer(int socketWriteTimeout, public TcpPeerServer(int socketWriteTimeout,
InetSocketAddress bindAddr) throws IOException { InetSocketAddress bindAddr,
int backlogLength) throws IOException {
this.serverSocket = (socketWriteTimeout > 0) ? this.serverSocket = (socketWriteTimeout > 0) ?
ServerSocketChannel.open().socket() : new ServerSocket(); ServerSocketChannel.open().socket() : new ServerSocket();
Server.bind(serverSocket, bindAddr, 0); Server.bind(serverSocket, bindAddr, backlogLength);
} }
/** /**

6
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
View File

@ -47,6 +47,7 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_STARTUP_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MAX_NUM_BLOCKS_TO_LOG_DEFAULT; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MAX_NUM_BLOCKS_TO_LOG_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MAX_NUM_BLOCKS_TO_LOG_KEY; import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_MAX_NUM_BLOCKS_TO_LOG_KEY;
import static org.apache.hadoop.util.ExitUtil.terminate; import static org.apache.hadoop.util.ExitUtil.terminate;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import java.io.BufferedOutputStream; import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream; import java.io.ByteArrayInputStream;
@ -890,8 +891,11 @@ public class DataNode extends ReconfigurableBase
if (secureResources != null) { if (secureResources != null) {
tcpPeerServer = new TcpPeerServer(secureResources); tcpPeerServer = new TcpPeerServer(secureResources);
} else { } else {
int backlogLength = conf.getInt(
CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_KEY,
CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_DEFAULT);
tcpPeerServer = new TcpPeerServer(dnConf.socketWriteTimeout, tcpPeerServer = new TcpPeerServer(dnConf.socketWriteTimeout,
DataNode.getStreamingAddr(conf)); DataNode.getStreamingAddr(conf), backlogLength);
} }
tcpPeerServer.setReceiveBufferSize(HdfsConstants.DEFAULT_DATA_SOCKET_SIZE); tcpPeerServer.setReceiveBufferSize(HdfsConstants.DEFAULT_DATA_SOCKET_SIZE);
streamingAddr = tcpPeerServer.getStreamingAddr(); streamingAddr = tcpPeerServer.getStreamingAddr();

6
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/SecureDataNodeStarter.java
View File

@ -20,6 +20,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.daemon.Daemon; import org.apache.commons.daemon.Daemon;
import org.apache.commons.daemon.DaemonContext; import org.apache.commons.daemon.DaemonContext;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.hdfs.DFSConfigKeys; import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSUtil; import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.HdfsConfiguration; import org.apache.hadoop.hdfs.HdfsConfiguration;
@ -95,10 +96,13 @@ public class SecureDataNodeStarter implements Daemon {
int socketWriteTimeout = conf.getInt( int socketWriteTimeout = conf.getInt(
DFSConfigKeys.DFS_DATANODE_SOCKET_WRITE_TIMEOUT_KEY, DFSConfigKeys.DFS_DATANODE_SOCKET_WRITE_TIMEOUT_KEY,
HdfsServerConstants.WRITE_TIMEOUT); HdfsServerConstants.WRITE_TIMEOUT);
int backlogLength = conf.getInt(
CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_KEY,
CommonConfigurationKeysPublic.IPC_SERVER_LISTEN_QUEUE_SIZE_DEFAULT);
ServerSocket ss = (socketWriteTimeout > 0) ? ServerSocket ss = (socketWriteTimeout > 0) ?
ServerSocketChannel.open().socket() : new ServerSocket(); ServerSocketChannel.open().socket() : new ServerSocket();
ss.bind(streamingAddr, 0); ss.bind(streamingAddr, backlogLength);
// Check that we got the port we need // Check that we got the port we need
if (ss.getLocalPort() != streamingAddr.getPort()) { if (ss.getLocalPort() != streamingAddr.getPort()) {