Victsm commented on a change in pull request #25907: [SPARK-29206][SHUFFLE]
Make number of shuffle server threads a multiple of number of chunk fetch
handler threads.
URL: https://github.com/apache/spark/pull/25907#discussion_r328300250
##########
File path:
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java
##########
@@ -320,25 +342,25 @@ public long maxChunksBeingTransferred() {
* threads for the completion of sending back responses, we are able to put
a limit on
* the max number of threads from TransportServer's default EventLoopGroup
that are
* going to be consumed by writing response to ChunkFetchRequest, which are
I/O intensive
- * and could take long time to process due to disk contentions. By
configuring a slightly
+ * and could take long time to process due to disk contentions. By
configuring a
* higher number of shuffler server threads, we are able to reserve some
threads for
* handling other RPC messages, thus making the Client less likely to
experience timeout
- * when sending RPC messages to the shuffle server. The number of threads
used for handling
- * chunked fetch requests are percentage of io.serverThreads (if defined)
else it is a percentage
- * of 2 * #cores. However, a percentage of 0 means netty default number of
threads which
- * is 2 * #cores ignoring io.serverThreads. The percentage here is
configured via
- * spark.shuffle.server.chunkFetchHandlerThreadsPercent. The returned value
is rounded off to
- * ceiling of the nearest integer.
+ * when sending RPC messages to the shuffle server. The number of server
threads needs to
+ * be a multiple of the number of threads used for handling chunked fetch
requests. See
+ * SPARK-29206 for more details. This is controlled via configuring a ratio
between
+ * io.serverThreads and chunk fetch handler threads, such that # server
threads = ratio *
+ * # chunk fetch handler threads. The default value of this ratio is 1,
which means the
+ * numbers of threads in both thread pools are equal. If io.serverThreads is
not defined,
+ * then 2 * #cores threads will be used for both server threads and chunk
fetch handler
+ * threads, ignoring the ratio configuration.
*/
public int chunkFetchHandlerThreads() {
if (!this.getModuleName().equalsIgnoreCase("shuffle")) {
return 0;
}
- int chunkFetchHandlerThreadsPercent =
- conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 100);
- int threads =
- this.serverThreads() > 0 ? this.serverThreads() : 2 *
NettyRuntime.availableProcessors();
- return (int) Math.ceil(threads * (chunkFetchHandlerThreadsPercent /
100.0));
+ int chunkFetchHandlerThreadsRatio = getChunkFetchHandlerThreadsRatio();
+ return this.serverThreads() > 0 ? this.serverThreads() /
chunkFetchHandlerThreadsRatio :
Review comment:
It won't be 9/2, it would be 10/2.
this.serverThreads will change the configured 9 server threads to 10 to make
sure it can be divided by the int ratio.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
