You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/11/03 05:18:53 UTC
(incubator-celeborn) branch main updated: [CELEBORN-1107] Make the max default number of netty threads configurable
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new c8b5384ba [CELEBORN-1107] Make the max default number of netty threads configurable
c8b5384ba is described below
commit c8b5384baf93f79664664670b797314bb1604d4f
Author: Chandni Singh <si...@gmail.com>
AuthorDate: Fri Nov 3 13:18:44 2023 +0800
[CELEBORN-1107] Make the max default number of netty threads configurable
### What changes were proposed in this pull request?
This change makes the maximum default number of Netty threads configurable. Previously, this value was hardcoded to 64, which could be small for certain environments. While it's possible to configure the number of Netty server and client threads individually for each module, providing an option to increase the default value offers greater convenience.
### Why are the changes needed?
The change offers convenience.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added a UT
Closes #2065 from otterc/CELEBORN-1107.
Authored-by: Chandni Singh <si...@gmail.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../main/scala/org/apache/celeborn/common/CelebornConf.scala | 10 ++++++++++
.../main/scala/org/apache/celeborn/common/util/Utils.scala | 6 +++---
.../scala/org/apache/celeborn/common/util/UtilsSuite.scala | 12 +++++++++++-
docs/configuration/network.md | 1 +
4 files changed, 25 insertions(+), 4 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 0879a6107..194c7e039 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -507,6 +507,8 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
getTimeAsMs(key, FETCH_TIMEOUT_CHECK_INTERVAL.defaultValueString)
}
+ def maxDefaultNettyThreads: Int = get(MAX_DEFAULT_NETTY_THREADS)
+
// //////////////////////////////////////////////////////
// Master //
// //////////////////////////////////////////////////////
@@ -3993,4 +3995,12 @@ object CelebornConf extends Logging {
.version("0.3.1")
.booleanConf
.createWithDefault(false)
+
+ val MAX_DEFAULT_NETTY_THREADS: ConfigEntry[Int] =
+ buildConf("celeborn.io.maxDefaultNettyThreads")
+ .categories("network")
+ .doc("Max default netty threads")
+ .version("0.3.2")
+ .intConf
+ .createWithDefault(64)
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
index 41271b04c..2e6c0af67 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
@@ -476,14 +476,14 @@ object Utils extends Logging {
hostPortParseResults.get(hostPort)
}
- private val MAX_DEFAULT_NETTY_THREADS = 64
+ private var maxDefaultNettyThreads = 64
def fromCelebornConf(
_conf: CelebornConf,
module: String,
numUsableCores: Int = 0): TransportConf = {
val conf = _conf.clone
-
+ maxDefaultNettyThreads = conf.maxDefaultNettyThreads
// Specify thread configuration based on our JVM's allocation of cores (rather than necessarily
// assuming we have all the machine's cores).
// NB: Only set if serverThreads/clientThreads not already set.
@@ -500,7 +500,7 @@ object Utils extends Logging {
private def defaultNumThreads(numUsableCores: Int): Int = {
val availableCores =
if (numUsableCores > 0) numUsableCores else Runtime.getRuntime.availableProcessors()
- math.min(availableCores, MAX_DEFAULT_NETTY_THREADS)
+ math.min(availableCores, maxDefaultNettyThreads)
}
def getClassLoader: ClassLoader = getClass.getClassLoader
diff --git a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
index bce77cc68..d33eeb69c 100644
--- a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
+++ b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
@@ -20,7 +20,8 @@ package org.apache.celeborn.common.util
import java.util
import org.apache.celeborn.CelebornFunSuite
-import org.apache.celeborn.common.protocol.PartitionLocation
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.protocol.{PartitionLocation, TransportModuleConstants}
import org.apache.celeborn.common.protocol.message.ControlMessages.{GetReducerFileGroupResponse, MapperEnd}
import org.apache.celeborn.common.protocol.message.StatusCode
@@ -168,6 +169,15 @@ class UtilsSuite extends CelebornFunSuite {
assert(set.size == 0)
}
+ test("validate number of client/server netty threads") {
+ val celebornConf = new CelebornConf()
+ celebornConf.set("celeborn.io.maxDefaultNettyThreads", "100")
+ val transportConf =
+ Utils.fromCelebornConf(celebornConf, TransportModuleConstants.PUSH_MODULE, 120)
+ assert(transportConf.serverThreads() == 100)
+ assert(transportConf.clientThreads() == 100)
+ }
+
def partitionLocation(partitionId: Int): util.HashSet[PartitionLocation] = {
val partitionSet = new util.HashSet[PartitionLocation]
for (i <- 0 until 3) {
diff --git a/docs/configuration/network.md b/docs/configuration/network.md
index f63fc8bf7..241f19836 100644
--- a/docs/configuration/network.md
+++ b/docs/configuration/network.md
@@ -39,6 +39,7 @@ license: |
| celeborn.<module>.push.timeoutCheck.interval | 5s | Interval for checking push data timeout. If setting <module> to `data`, it works for shuffle client push data and should be configured on client side. If setting <module> to `replicate`, it works for worker replicate data to peer worker and should be configured on worker side. | 0.3.0 |
| celeborn.<module>.push.timeoutCheck.threads | 4 | Threads num for checking push data timeout. If setting <module> to `data`, it works for shuffle client push data and should be configured on client side. If setting <module> to `replicate`, it works for worker replicate data to peer worker and should be configured on worker side. | 0.3.0 |
| celeborn.<role>.rpc.dispatcher.threads | <value of celeborn.rpc.dispatcher.threads> | Threads number of message dispatcher event loop for roles | |
+| celeborn.io.maxDefaultNettyThreads | 64 | Max default netty threads | 0.3.2 |
| celeborn.network.bind.preferIpAddress | true | When `ture`, prefer to use IP address, otherwise FQDN. This configuration only takes effects when the bind hostname is not set explicitly, in such case, Celeborn will find the first non-loopback address to bind. | 0.3.0 |
| celeborn.network.connect.timeout | 10s | Default socket connect timeout. | 0.2.0 |
| celeborn.network.memory.allocator.numArenas | <undefined> | Number of arenas for pooled memory allocator. Default value is Runtime.getRuntime.availableProcessors, min value is 2. | 0.3.0 |