Removed OIO and added num threads settings.

Author: rxin

core/src/main/scala/org/apache/spark/network/netty/BlockClientFactory.scala

@@ -25,10 +25,8 @@ import io.netty.buffer.PooledByteBufAllocator
 import io.netty.channel._
 import io.netty.channel.epoll.{Epoll, EpollEventLoopGroup, EpollSocketChannel}
 import io.netty.channel.nio.NioEventLoopGroup
-import io.netty.channel.oio.OioEventLoopGroup
 import io.netty.channel.socket.SocketChannel
 import io.netty.channel.socket.nio.NioSocketChannel
-import io.netty.channel.socket.oio.OioSocketChannel
 import io.netty.util.internal.PlatformDependent
 
 import org.apache.spark.{Logging, SparkConf}
@@ -65,23 +63,18 @@ class BlockClientFactory(val conf: NettyConfig) extends Logging with Closeable {
 
   /** Initialize [[socketChannelClass]] and [[workerGroup]] based on ioMode. */
   private def init(): Unit = {
-    def initOio(): Unit = {
-      socketChannelClass = classOf[OioSocketChannel]
-      workerGroup = new OioEventLoopGroup(0, threadFactory)
-    }
     def initNio(): Unit = {
       socketChannelClass = classOf[NioSocketChannel]
-      workerGroup = new NioEventLoopGroup(0, threadFactory)
+      workerGroup = new NioEventLoopGroup(conf.clientThreads, threadFactory)
     }
     def initEpoll(): Unit = {
       socketChannelClass = classOf[EpollSocketChannel]
-      workerGroup = new EpollEventLoopGroup(0, threadFactory)
+      workerGroup = new EpollEventLoopGroup(conf.clientThreads, threadFactory)
     }
 
     // For auto mode, first try epoll (only available on Linux), then nio.
     conf.ioMode match {
       case "nio" => initNio()
-      case "oio" => initOio()
       case "epoll" => initEpoll()
       case "auto" => if (Epoll.isAvailable) initEpoll() else initNio()
     }
@@ -102,7 +95,7 @@ class BlockClientFactory(val conf: NettyConfig) extends Logging with Closeable {
       return cachedClient
     }
 
-    logInfo(s"Creating new connection to $remoteHost:$remotePort")
+    logDebug(s"Creating new connection to $remoteHost:$remotePort")
 
     // There is a chance two threads are creating two different clients connecting to the same host.
     // But that's probably ok ...

core/src/main/scala/org/apache/spark/network/netty/BlockServer.scala

@@ -24,10 +24,8 @@ import io.netty.bootstrap.ServerBootstrap
 import io.netty.buffer.PooledByteBufAllocator
 import io.netty.channel.epoll.{Epoll, EpollEventLoopGroup, EpollServerSocketChannel}
 import io.netty.channel.nio.NioEventLoopGroup
-import io.netty.channel.oio.OioEventLoopGroup
 import io.netty.channel.socket.SocketChannel
 import io.netty.channel.socket.nio.NioServerSocketChannel
-import io.netty.channel.socket.oio.OioServerSocketChannel
 import io.netty.channel.{ChannelInitializer, ChannelFuture, ChannelOption}
 
 import org.apache.spark.Logging
@@ -60,24 +58,18 @@ class BlockServer(conf: NettyConfig, dataProvider: BlockDataManager)
 
     // Use only one thread to accept connections, and 2 * num_cores for worker.
     def initNio(): Unit = {
-      val bossGroup = new NioEventLoopGroup(0, threadFactory)
+      val bossGroup = new NioEventLoopGroup(conf.serverThreads, threadFactory)
       val workerGroup = bossGroup
       bootstrap.group(bossGroup, workerGroup).channel(classOf[NioServerSocketChannel])
     }
-    def initOio(): Unit = {
-      val bossGroup = new OioEventLoopGroup(0, threadFactory)
-      val workerGroup = bossGroup
-      bootstrap.group(bossGroup, workerGroup).channel(classOf[OioServerSocketChannel])
-    }
     def initEpoll(): Unit = {
-      val bossGroup = new EpollEventLoopGroup(0, threadFactory)
+      val bossGroup = new EpollEventLoopGroup(conf.serverThreads, threadFactory)
       val workerGroup = bossGroup
       bootstrap.group(bossGroup, workerGroup).channel(classOf[EpollServerSocketChannel])
     }
 
     conf.ioMode match {
       case "nio" => initNio()
-      case "oio" => initOio()
       case "epoll" => initEpoll()
       case "auto" => if (Epoll.isAvailable) initEpoll() else initNio()
     }

core/src/main/scala/org/apache/spark/network/netty/NettyConfig.scala

@@ -31,18 +31,20 @@ class NettyConfig(conf: SparkConf) {
   /** IO mode: nio, oio, epoll, or auto (try epoll first and then nio). */
   private[netty] val ioMode = conf.get("spark.shuffle.io.mode", "nio").toLowerCase
 
-  /** Connect timeout in secs. Default 60 secs. */
-  private[netty] val connectTimeoutMs = conf.getInt("spark.shuffle.io.connectionTimeout", 60) * 1000
-
-  /**
-   * Percentage of the desired amount of time spent for I/O in the child event loops.
-   * Only applicable in nio and epoll.
-   */
-  private[netty] val ioRatio = conf.getInt("spark.shuffle.io.netty.ioRatio", 80)
+  /** Connect timeout in secs. Default 120 secs. */
+  private[netty] val connectTimeoutMs = {
+    conf.getInt("spark.shuffle.io.connectionTimeout", 120) * 1000
+  }
 
   /** Requested maximum length of the queue of incoming connections. */
   private[netty] val backLog: Option[Int] = conf.getOption("spark.shuffle.io.backLog").map(_.toInt)
 
+  /** Number of threads used in the server thread pool. Default to 0, which is 2x#cores. */
+  private[netty] val serverThreads: Int = conf.getInt("spark.shuffle.io.serverThreads", 0)
+
+  /** Number of threads used in the client thread pool. Default to 0, which is 2x#cores. */
+  private[netty] val clientThreads: Int = conf.getInt("spark.shuffle.io.clientThreads", 0)
+
   /**
    * Receive buffer size (SO_RCVBUF).
    * Note: the optimal size for receive buffer and send buffer should be