ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-24029][CORE] Set SO_REUSEADDR on listen sockets.

Browse source 
This allows sockets to be bound even if there are sockets
from a previous application that are still pending closure. It
avoids bind issues when, for example, re-starting the SHS.

Don't enable the option on Windows though. The following page
explains some odd behavior that this option can have there:
https://msdn.microsoft.com/en-us/library/windows/desktop/ms740621%28v=vs.85%29.aspx

I intentionally ignored server sockets that always bind to
ephemeral ports, since those don't benefit from this option.

Author: Marcelo Vanzin <vanzin@cloudera.com>

Closes #21110 from vanzin/SPARK-24029.
This commit is contained in:
Marcelo Vanzin 2018-04-21 23:14:58 +08:00 committed by hyukjinkwon
parent 1d758dc73b
commit 32b4bcd6d3
3 changed files with 5 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java
View file

@ -32,6 +32,7 @@ import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import org.apache.commons.lang3.SystemUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -98,7 +99,8 @@ public class TransportServer implements Closeable {
.group(bossGroup, workerGroup)
.channel(NettyUtils.getServerChannelClass(ioMode))
.option(ChannelOption.ALLOCATOR, allocator)
.childOption(ChannelOption.ALLOCATOR, allocator);
.childOption(ChannelOption.ALLOCATOR, allocator)
.childOption(ChannelOption.SO_REUSEADDR, !SystemUtils.IS_OS_WINDOWS);
this.metrics = new NettyMemoryMetrics(
allocator, conf.getModuleName() + "-server", conf);

1
core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionServer.scala
View file

@ -94,6 +94,7 @@ private[spark] abstract class RestSubmissionServer(
new HttpConnectionFactory())
connector.setHost(host)
connector.setPort(startPort)
connector.setReuseAddress(!Utils.isWindows)
server.addConnector(connector)
val mainHandler = new ServletContextHandler

1
core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
View file

@ -344,6 +344,7 @@ private[spark] object JettyUtils extends Logging {
connectionFactories: _*)
connector.setPort(port)
connector.setHost(hostName)
connector.setReuseAddress(!Utils.isWindows)
// Currently we only use "SelectChannelConnector"
// Limit the max acceptor number to 8 so that we don't waste a lot of threads