You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "shuwang21 (via GitHub)" <gi...@apache.org> on 2023/06/07 06:30:09 UTC
[GitHub] [spark] shuwang21 opened a new pull request, #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
shuwang21 opened a new pull request, #41489:
URL: https://github.com/apache/spark/pull/41489
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
In this PR, we proposed to separate finalizeShuffleMerge processing into dedicated thread pools.
1. We introduce `ShuffleTransportContext` extend `TransportContext`.
2. We override the `initializePipeline` by adding `FinalizeHandler` when the newly added configuration `spark.shuffle.server.finalizeShuffleMergeThreads` is positive.
3. We override the `decode` within `ShuffleMessageDecoder` so that an `FINALIZE_SHUFFLE_MERGE` type `RpcRequest` will not be processed by current IO threads. We will further encapsulate it as `RpcRequestInternal`.
4. A dedicated `FinalizedHandler` will be attached to the channel pipeline, which only handles `RpcRequestInternal` type.
authors: @otterc @shuwang21
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
5. If you fix a bug, you can clarify why it is a bug.
-->
In our production environment, `finalizeShuffleMerge` processing took longer time (p90 is around 20s) than other PRC requests. This is due to `finalizeShuffleMerge` invoking IO operations like truncate and file open/close.
More importantly, processing this `finalizeShuffleMerge` can block other critical lightweight messages like authentications, which can cause authentication timeout as well as fetch failures. Those timeout and fetch failures affect the stability of the Spark job executions.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
1. We add a few related UTs.
2. We also tested internally with this patch. Under the same setting, without this patch, the shuffle server can encounter authentication timeouts thus fetch failures. With this path, `finalizeShuffleMerge` is processed by `finalizeHandler` we introduced, and authentication RPC requests are not blocked.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #41489:
URL: https://github.com/apache/spark/pull/41489#issuecomment-1675546034
Merged to master.
Thanks for fixing this @shuwang21 !
Thanks for the review @otterc :-)
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1278808341
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+public class ShuffleTransportContext extends TransportContext {
+ private static final Logger logger = LoggerFactory.getLogger(ShuffleTransportContext.class);
+ private final EventLoopGroup finalizeWorkers;
+
+ public ShuffleTransportContext(
+ TransportConf conf,
+ ExternalBlockHandler rpcHandler,
+ boolean closeIdleConnections) {
+ this(conf, rpcHandler, closeIdleConnections, false);
+ }
+
+ public ShuffleTransportContext(TransportConf conf,
+ RpcHandler rpcHandler,
+ boolean closeIdleConnections,
+ boolean isClientOnly) {
+ super(conf, rpcHandler, closeIdleConnections, isClientOnly);
+
+ if (conf.getModuleName() != null &&
+ conf.getModuleName().equalsIgnoreCase("shuffle") && conf.separateFinalizeShuffleMerge()) {
Review Comment:
super nit:
```suggestion
if ("shuffle".equalsIgnoreCase(conf.getModuleName()) && conf.separateFinalizeShuffleMerge()) {
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+public class ShuffleTransportContext extends TransportContext {
+ private static final Logger logger = LoggerFactory.getLogger(ShuffleTransportContext.class);
+ private final EventLoopGroup finalizeWorkers;
+
+ public ShuffleTransportContext(
+ TransportConf conf,
+ ExternalBlockHandler rpcHandler,
+ boolean closeIdleConnections) {
+ this(conf, rpcHandler, closeIdleConnections, false);
+ }
+
+ public ShuffleTransportContext(TransportConf conf,
+ RpcHandler rpcHandler,
+ boolean closeIdleConnections,
+ boolean isClientOnly) {
+ super(conf, rpcHandler, closeIdleConnections, isClientOnly);
+
+ if (conf.getModuleName() != null &&
+ conf.getModuleName().equalsIgnoreCase("shuffle") && conf.separateFinalizeShuffleMerge()) {
+ finalizeWorkers = NettyUtils.createEventLoop(
+ IOMode.valueOf(conf.ioMode()),
+ conf.finalizeShuffleMergeHandlerThreads(),
+ "shuffle-finalize-merge-handler");
+ logger.info("finalize shuffle merged workers created");
+ } else {
+ finalizeWorkers = null;
+ }
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel) {
+ TransportChannelHandler ch = super.initializePipeline(channel);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel,
+ RpcHandler channelRpcHandler) {
+ TransportChannelHandler ch = super.initializePipeline(channel, channelRpcHandler);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ /**
+ * Add finalize handler to pipeline if needed. This is needed only when separateFinalizeShuffleMerge
+ * is enabled.
+ */
+ private void addFinalizeHandlerToPipelineIfNeeded(SocketChannel channel,
+ TransportChannelHandler ch) {
+ if (finalizeWorkers != null) {
+ channel.pipeline().addLast(finalizeWorkers, FinalizedHandler.HANDLER_NAME,
+ new FinalizedHandler(ch.getRequestHandler()));
+ }
+ }
+
+ @Override
+ protected MessageToMessageDecoder<ByteBuf> getDecoder() {
+ if (finalizeWorkers != null) {
+ return new ShuffleMessageDecoder(MessageDecoder.INSTANCE);
Review Comment:
This is invoked for each pipeline. Make this a static final variable in this class instead of creating each time.
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+public class ShuffleTransportContext extends TransportContext {
+ private static final Logger logger = LoggerFactory.getLogger(ShuffleTransportContext.class);
+ private final EventLoopGroup finalizeWorkers;
+
+ public ShuffleTransportContext(
+ TransportConf conf,
+ ExternalBlockHandler rpcHandler,
+ boolean closeIdleConnections) {
+ this(conf, rpcHandler, closeIdleConnections, false);
+ }
+
+ public ShuffleTransportContext(TransportConf conf,
+ RpcHandler rpcHandler,
+ boolean closeIdleConnections,
+ boolean isClientOnly) {
+ super(conf, rpcHandler, closeIdleConnections, isClientOnly);
+
+ if (conf.getModuleName() != null &&
+ conf.getModuleName().equalsIgnoreCase("shuffle") && conf.separateFinalizeShuffleMerge()) {
+ finalizeWorkers = NettyUtils.createEventLoop(
+ IOMode.valueOf(conf.ioMode()),
+ conf.finalizeShuffleMergeHandlerThreads(),
+ "shuffle-finalize-merge-handler");
+ logger.info("finalize shuffle merged workers created");
+ } else {
+ finalizeWorkers = null;
+ }
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel) {
+ TransportChannelHandler ch = super.initializePipeline(channel);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel,
+ RpcHandler channelRpcHandler) {
+ TransportChannelHandler ch = super.initializePipeline(channel, channelRpcHandler);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ /**
+ * Add finalize handler to pipeline if needed. This is needed only when separateFinalizeShuffleMerge
+ * is enabled.
+ */
+ private void addFinalizeHandlerToPipelineIfNeeded(SocketChannel channel,
+ TransportChannelHandler ch) {
+ if (finalizeWorkers != null) {
+ channel.pipeline().addLast(finalizeWorkers, FinalizedHandler.HANDLER_NAME,
+ new FinalizedHandler(ch.getRequestHandler()));
+ }
+ }
+
+ @Override
+ protected MessageToMessageDecoder<ByteBuf> getDecoder() {
+ if (finalizeWorkers != null) {
+ return new ShuffleMessageDecoder(MessageDecoder.INSTANCE);
+ }
+ return super.getDecoder();
+ }
+
+ static class ShuffleMessageDecoder extends MessageToMessageDecoder<ByteBuf> {
+
+ private final MessageDecoder delegate;
+ public ShuffleMessageDecoder(MessageDecoder delegate) {
+ super();
+ this.delegate = delegate;
+ }
+
+ /**
+ * Decode the message and check if it is a finalize merge request. If yes, then create a
+ * internal rpc request message and add it to the list of messages to be handled by
+ * {@link TransportChannelHandler}
+ */
+ @Override
+ protected void decode(ChannelHandlerContext channelHandlerContext,
+ ByteBuf byteBuf,
+ List<Object> list) throws Exception {
+ delegate.decode(channelHandlerContext, byteBuf, list);
+ Object msg = list.get(list.size() - 1);
+ if (msg instanceof RpcRequest) {
+ RpcRequest req = (RpcRequest) msg;
+ ByteBuffer buffer = req.body().nioByteBuffer();
+ byte type = Unpooled.wrappedBuffer(buffer).readByte();
+ if (type == BlockTransferMessage.Type.FINALIZE_SHUFFLE_MERGE.id()) {
+ list.remove(list.size() - 1);
+ RpcRequestInternal rpcRequestInternal =
+ new RpcRequestInternal(BlockTransferMessage.Type.FINALIZE_SHUFFLE_MERGE, req);
+ logger.trace("Created internal rpc request msg with rpcId {} for finalize merge req",
+ req.requestId);
+ list.add(rpcRequestInternal);
+ }
+ }
+ }
+ }
+
+ /**
+ * Internal message to handle rpc requests that should not be accepted by
+ * {@link TransportChannelHandler}. Since, this message doesn't extend {@link Message}, it will
+ * not be accepted by {@link TransportChannelHandler}.
Review Comment:
```suggestion
* Internal message to handle rpc requests that is not accepted by
* {@link TransportChannelHandler} as this message doesn't extend {@link Message}. It will
* be accepted by {@link FinalizedHandler} instead, which is configured to execute in a separate EventLoopGroup
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+public class ShuffleTransportContext extends TransportContext {
+ private static final Logger logger = LoggerFactory.getLogger(ShuffleTransportContext.class);
+ private final EventLoopGroup finalizeWorkers;
+
+ public ShuffleTransportContext(
+ TransportConf conf,
+ ExternalBlockHandler rpcHandler,
+ boolean closeIdleConnections) {
+ this(conf, rpcHandler, closeIdleConnections, false);
+ }
+
+ public ShuffleTransportContext(TransportConf conf,
+ RpcHandler rpcHandler,
+ boolean closeIdleConnections,
+ boolean isClientOnly) {
+ super(conf, rpcHandler, closeIdleConnections, isClientOnly);
+
+ if (conf.getModuleName() != null &&
+ conf.getModuleName().equalsIgnoreCase("shuffle") && conf.separateFinalizeShuffleMerge()) {
+ finalizeWorkers = NettyUtils.createEventLoop(
+ IOMode.valueOf(conf.ioMode()),
+ conf.finalizeShuffleMergeHandlerThreads(),
+ "shuffle-finalize-merge-handler");
+ logger.info("finalize shuffle merged workers created");
+ } else {
+ finalizeWorkers = null;
+ }
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel) {
+ TransportChannelHandler ch = super.initializePipeline(channel);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel,
+ RpcHandler channelRpcHandler) {
+ TransportChannelHandler ch = super.initializePipeline(channel, channelRpcHandler);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ /**
+ * Add finalize handler to pipeline if needed. This is needed only when separateFinalizeShuffleMerge
+ * is enabled.
+ */
+ private void addFinalizeHandlerToPipelineIfNeeded(SocketChannel channel,
+ TransportChannelHandler ch) {
+ if (finalizeWorkers != null) {
+ channel.pipeline().addLast(finalizeWorkers, FinalizedHandler.HANDLER_NAME,
+ new FinalizedHandler(ch.getRequestHandler()));
+ }
+ }
+
+ @Override
+ protected MessageToMessageDecoder<ByteBuf> getDecoder() {
+ if (finalizeWorkers != null) {
+ return new ShuffleMessageDecoder(MessageDecoder.INSTANCE);
+ }
+ return super.getDecoder();
+ }
+
+ static class ShuffleMessageDecoder extends MessageToMessageDecoder<ByteBuf> {
+
+ private final MessageDecoder delegate;
+ public ShuffleMessageDecoder(MessageDecoder delegate) {
+ super();
+ this.delegate = delegate;
+ }
+
+ /**
+ * Decode the message and check if it is a finalize merge request. If yes, then create a
+ * internal rpc request message and add it to the list of messages to be handled by
+ * {@link TransportChannelHandler}
+ */
+ @Override
+ protected void decode(ChannelHandlerContext channelHandlerContext,
+ ByteBuf byteBuf,
+ List<Object> list) throws Exception {
+ delegate.decode(channelHandlerContext, byteBuf, list);
+ Object msg = list.get(list.size() - 1);
+ if (msg instanceof RpcRequest) {
+ RpcRequest req = (RpcRequest) msg;
+ ByteBuffer buffer = req.body().nioByteBuffer();
+ byte type = Unpooled.wrappedBuffer(buffer).readByte();
+ if (type == BlockTransferMessage.Type.FINALIZE_SHUFFLE_MERGE.id()) {
+ list.remove(list.size() - 1);
+ RpcRequestInternal rpcRequestInternal =
+ new RpcRequestInternal(BlockTransferMessage.Type.FINALIZE_SHUFFLE_MERGE, req);
+ logger.trace("Created internal rpc request msg with rpcId {} for finalize merge req",
+ req.requestId);
+ list.add(rpcRequestInternal);
+ }
+ }
+ }
+ }
+
+ /**
+ * Internal message to handle rpc requests that should not be accepted by
+ * {@link TransportChannelHandler}. Since, this message doesn't extend {@link Message}, it will
+ * not be accepted by {@link TransportChannelHandler}.
+ */
+ static class RpcRequestInternal {
+ public final BlockTransferMessage.Type messageType;
+ public final RpcRequest rpcRequest;
+
+ public RpcRequestInternal(BlockTransferMessage.Type messageType,
+ RpcRequest rpcRequest) {
+ this.messageType = messageType;
+ this.rpcRequest = rpcRequest;
+ }
+ }
+
+ static class FinalizedHandler extends SimpleChannelInboundHandler<RpcRequestInternal> {
+ private static final Logger logger = LoggerFactory.getLogger(FinalizedHandler.class);
+ public static final String HANDLER_NAME = "finalizeHandler";
+ private final TransportRequestHandler transportRequestHandler;
+
+ @Override
+ public boolean acceptInboundMessage(Object msg) throws Exception {
+ if (msg instanceof RpcRequestInternal) {
+ RpcRequestInternal rpcRequestInternal = (RpcRequestInternal) msg;
+ return rpcRequestInternal.messageType == BlockTransferMessage.Type.FINALIZE_SHUFFLE_MERGE;
+ }
+ return false;
+ }
+
+ public FinalizedHandler(TransportRequestHandler transportRequestHandler) {
+ this.transportRequestHandler = transportRequestHandler;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext channelHandlerContext,
+ RpcRequestInternal req) throws Exception {
+ logger.debug("finalize handler invoked for rpc request {}", req.rpcRequest.requestId);
Review Comment:
Include additional context ... something like:
```suggestion
if (logger.isTraceEnabled()) {
logger.trace("Finalize shuffle req from {} for rpc request {}",
getRemoteAddress(channelHandlerContext.channel()), req.rpcRequest.requestId);
}
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+public class ShuffleTransportContext extends TransportContext {
Review Comment:
Class javadoc
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+public class ShuffleTransportContext extends TransportContext {
+ private static final Logger logger = LoggerFactory.getLogger(ShuffleTransportContext.class);
+ private final EventLoopGroup finalizeWorkers;
+
+ public ShuffleTransportContext(
+ TransportConf conf,
+ ExternalBlockHandler rpcHandler,
+ boolean closeIdleConnections) {
+ this(conf, rpcHandler, closeIdleConnections, false);
+ }
+
+ public ShuffleTransportContext(TransportConf conf,
+ RpcHandler rpcHandler,
+ boolean closeIdleConnections,
+ boolean isClientOnly) {
+ super(conf, rpcHandler, closeIdleConnections, isClientOnly);
+
+ if (conf.getModuleName() != null &&
+ conf.getModuleName().equalsIgnoreCase("shuffle") && conf.separateFinalizeShuffleMerge()) {
+ finalizeWorkers = NettyUtils.createEventLoop(
+ IOMode.valueOf(conf.ioMode()),
+ conf.finalizeShuffleMergeHandlerThreads(),
+ "shuffle-finalize-merge-handler");
+ logger.info("finalize shuffle merged workers created");
+ } else {
+ finalizeWorkers = null;
+ }
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel) {
+ TransportChannelHandler ch = super.initializePipeline(channel);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel,
+ RpcHandler channelRpcHandler) {
+ TransportChannelHandler ch = super.initializePipeline(channel, channelRpcHandler);
+ addFinalizeHandlerToPipelineIfNeeded(channel, ch);
+ return ch;
+ }
+
+ /**
+ * Add finalize handler to pipeline if needed. This is needed only when separateFinalizeShuffleMerge
+ * is enabled.
+ */
+ private void addFinalizeHandlerToPipelineIfNeeded(SocketChannel channel,
Review Comment:
Rename this to `addHandlersToPipeline` or some such.
Currently, this is limited to finalization, but can have other expensive IO operations in future as well.
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +326,33 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Percentage of io.serverThreads used by netty to process FinalizeShuffleMerge. When the config
+ * `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is set, shuffle server will use a
+ * separate EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions. The number of threads used for handling
+ * finalizeShuffleMerge requests are percentage of io.serverThreads (if defined) else it is a
+ * percentage of 2 * #cores.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ if (!this.getModuleName().equalsIgnoreCase("shuffle")) {
+ return 0;
+ }
+ int finalizeShuffleMergeThreadsPercent =
+ Integer.parseInt(conf.get("spark.shuffle.server.finalizeShuffleMergeThreadsPercent"));
Review Comment:
Currently, this method should not be called when `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is not specified - but that is relying on current code flow in order for it to hold, which is brittle as project evolved.
Instead, handle explicitly this contract:
a) Precondition check for `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` to be present, or
b) (more preferably) Handle the case when `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is missing.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1221921518
##########
common/network-common/src/main/java/org/apache/spark/network/protocol/MessageWithHeader.java:
##########
@@ -35,7 +35,8 @@
*
* The header must be a ByteBuf, while the body can be a ByteBuf or a FileRegion.
*/
-class MessageWithHeader extends AbstractFileRegion {
+
+public class MessageWithHeader extends AbstractFileRegion {
Review Comment:
Note for reviewers: Need to make this public for the UTs added for overridden decoder in `ShuffleTransportContext`
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +326,28 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Number of threads to process FinalizeShuffleMerge. Shuffle server will use a separate
+ * EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ if (!this.getModuleName().equalsIgnoreCase("shuffle")) {
+ return 0;
+ }
+ Preconditions.checkArgument(separateFinalizeShuffleMerge(),
+ "Please set spark.shuffle.server.finalizeShuffleMergeThreads to a positive value");
+ return Integer.parseInt(conf.get("spark.shuffle.server.finalizeShuffleMergeThreads"));
+ }
+
Review Comment:
Should this be a percentage like chunkFetchHandlerThreads config?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] shuwang21 commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "shuwang21 (via GitHub)" <gi...@apache.org>.
shuwang21 commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1284977056
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +326,33 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Percentage of io.serverThreads used by netty to process FinalizeShuffleMerge. When the config
+ * `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is set, shuffle server will use a
+ * separate EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions. The number of threads used for handling
+ * finalizeShuffleMerge requests are percentage of io.serverThreads (if defined) else it is a
+ * percentage of 2 * #cores.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ if (!this.getModuleName().equalsIgnoreCase("shuffle")) {
+ return 0;
+ }
+ int finalizeShuffleMergeThreadsPercent =
+ Integer.parseInt(conf.get("spark.shuffle.server.finalizeShuffleMergeThreadsPercent"));
Review Comment:
I see your point. Add Preconditions. Please check. I also fixed all other suggestions. Thanks.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] shuwang21 commented on pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "shuwang21 (via GitHub)" <gi...@apache.org>.
shuwang21 commented on PR #41489:
URL: https://github.com/apache/spark/pull/41489#issuecomment-1581480409
> Let us mark it as DRAFT until we are ready to review/merge.
Sure.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] shuwang21 commented on pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "shuwang21 (via GitHub)" <gi...@apache.org>.
shuwang21 commented on PR #41489:
URL: https://github.com/apache/spark/pull/41489#issuecomment-1580011361
cc: @otterc @mridulm @zhouyejoe
Please take a look. Thanks.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1284975014
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +326,33 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Percentage of io.serverThreads used by netty to process FinalizeShuffleMerge. When the config
+ * `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is set, shuffle server will use a
+ * separate EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions. The number of threads used for handling
+ * finalizeShuffleMerge requests are percentage of io.serverThreads (if defined) else it is a
+ * percentage of 2 * #cores.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ if (!this.getModuleName().equalsIgnoreCase("shuffle")) {
+ return 0;
+ }
+ int finalizeShuffleMergeThreadsPercent =
+ Integer.parseInt(conf.get("spark.shuffle.server.finalizeShuffleMergeThreadsPercent"));
Review Comment:
I covered this above :-)
What you are describing is how it currently works - but as project and code evolves, these assumption need not hold.
Let us make it either an explicit precondition (which will fail if/when the assumption is broken), or handle it such that we dont need to this assumption.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] shuwang21 commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "shuwang21 (via GitHub)" <gi...@apache.org>.
shuwang21 commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1284961300
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +326,33 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Percentage of io.serverThreads used by netty to process FinalizeShuffleMerge. When the config
+ * `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is set, shuffle server will use a
+ * separate EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions. The number of threads used for handling
+ * finalizeShuffleMerge requests are percentage of io.serverThreads (if defined) else it is a
+ * percentage of 2 * #cores.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ if (!this.getModuleName().equalsIgnoreCase("shuffle")) {
+ return 0;
+ }
+ int finalizeShuffleMergeThreadsPercent =
+ Integer.parseInt(conf.get("spark.shuffle.server.finalizeShuffleMergeThreadsPercent"));
Review Comment:
I think this is handled by `separateFinalizeShuffleMerge` we current have.
1. When `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is missing, `separateFinalizeShuffleMerge` return false, so we will not call `finalizeShuffleMergeHandlerThreads` to create finalize handler.
2. Only when `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is larger than 0, we will create finalize handler.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1284975014
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +326,33 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Percentage of io.serverThreads used by netty to process FinalizeShuffleMerge. When the config
+ * `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is set, shuffle server will use a
+ * separate EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions. The number of threads used for handling
+ * finalizeShuffleMerge requests are percentage of io.serverThreads (if defined) else it is a
+ * percentage of 2 * #cores.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ if (!this.getModuleName().equalsIgnoreCase("shuffle")) {
+ return 0;
+ }
+ int finalizeShuffleMergeThreadsPercent =
+ Integer.parseInt(conf.get("spark.shuffle.server.finalizeShuffleMergeThreadsPercent"));
Review Comment:
I covered this above :-)
What you are describing is how it currently works, and you are right currently it does not violate this assumption - but as project and code evolves, these assumption need not hold.
Let us make it either an explicit precondition (which will fail if/when the assumption is broken), or handle it such that we dont need to this assumption.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #41489:
URL: https://github.com/apache/spark/pull/41489#issuecomment-1581479205
Let us mark it as DRAFT until we are ready to review/merge.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] otterc commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1287913820
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+import static org.apache.spark.network.util.NettyUtils.getRemoteAddress;
+
+/**
+ * Extends {@link TransportContext} to support customized shuffle service. Specifically, we
+ * modified the Netty Channel Pipeline so that IO expensive messages such as FINALIZE_SHUFFLE_MERGE
+ * are processed in the separate handlers.
+ * */
+public class ShuffleTransportContext extends TransportContext {
+ private static final Logger logger = LoggerFactory.getLogger(ShuffleTransportContext.class);
+ private static final ShuffleMessageDecoder SHUFFLE_DECODER =
+ new ShuffleMessageDecoder(MessageDecoder.INSTANCE);
+ private final EventLoopGroup finalizeWorkers;
+
+ public ShuffleTransportContext(
+ TransportConf conf,
+ ExternalBlockHandler rpcHandler,
+ boolean closeIdleConnections) {
+ this(conf, rpcHandler, closeIdleConnections, false);
+ }
+
+ public ShuffleTransportContext(TransportConf conf,
+ RpcHandler rpcHandler,
+ boolean closeIdleConnections,
+ boolean isClientOnly) {
+ super(conf, rpcHandler, closeIdleConnections, isClientOnly);
+
+ if ("shuffle".equalsIgnoreCase(conf.getModuleName()) && conf.separateFinalizeShuffleMerge()) {
+ finalizeWorkers = NettyUtils.createEventLoop(
+ IOMode.valueOf(conf.ioMode()),
Review Comment:
Nit: indentation
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ShuffleTransportContext.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.spark.network.TransportContext;
+import org.apache.spark.network.protocol.Message;
+import org.apache.spark.network.protocol.MessageDecoder;
+import org.apache.spark.network.protocol.RpcRequest;
+import org.apache.spark.network.server.RpcHandler;
+import org.apache.spark.network.server.TransportChannelHandler;
+import org.apache.spark.network.server.TransportRequestHandler;
+import org.apache.spark.network.shuffle.protocol.BlockTransferMessage;
+import org.apache.spark.network.util.IOMode;
+import org.apache.spark.network.util.NettyUtils;
+import org.apache.spark.network.util.TransportConf;
+
+import static org.apache.spark.network.util.NettyUtils.getRemoteAddress;
+
+/**
+ * Extends {@link TransportContext} to support customized shuffle service. Specifically, we
+ * modified the Netty Channel Pipeline so that IO expensive messages such as FINALIZE_SHUFFLE_MERGE
+ * are processed in the separate handlers.
+ * */
+public class ShuffleTransportContext extends TransportContext {
+ private static final Logger logger = LoggerFactory.getLogger(ShuffleTransportContext.class);
+ private static final ShuffleMessageDecoder SHUFFLE_DECODER =
+ new ShuffleMessageDecoder(MessageDecoder.INSTANCE);
+ private final EventLoopGroup finalizeWorkers;
+
+ public ShuffleTransportContext(
+ TransportConf conf,
+ ExternalBlockHandler rpcHandler,
+ boolean closeIdleConnections) {
+ this(conf, rpcHandler, closeIdleConnections, false);
+ }
+
+ public ShuffleTransportContext(TransportConf conf,
+ RpcHandler rpcHandler,
+ boolean closeIdleConnections,
+ boolean isClientOnly) {
+ super(conf, rpcHandler, closeIdleConnections, isClientOnly);
+
+ if ("shuffle".equalsIgnoreCase(conf.getModuleName()) && conf.separateFinalizeShuffleMerge()) {
+ finalizeWorkers = NettyUtils.createEventLoop(
+ IOMode.valueOf(conf.ioMode()),
+ conf.finalizeShuffleMergeHandlerThreads(),
+ "shuffle-finalize-merge-handler");
+ logger.info("finalize shuffle merged workers created");
+ } else {
+ finalizeWorkers = null;
+ }
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel) {
+ TransportChannelHandler ch = super.initializePipeline(channel);
+ addHandlerToPipeline(channel, ch);
+ return ch;
+ }
+
+ @Override
+ public TransportChannelHandler initializePipeline(SocketChannel channel,
+ RpcHandler channelRpcHandler) {
+ TransportChannelHandler ch = super.initializePipeline(channel, channelRpcHandler);
+ addHandlerToPipeline(channel, ch);
+ return ch;
+ }
+
+ /**
+ * Add finalize handler to pipeline if needed. This is needed only when
+ * separateFinalizeShuffleMerge is enabled.
+ */
+ private void addHandlerToPipeline(SocketChannel channel,
+ TransportChannelHandler ch) {
Review Comment:
Nit: rename `ch` to `transportChannelHandler`
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm closed pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm closed pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
URL: https://github.com/apache/spark/pull/41489
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] shuwang21 commented on pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "shuwang21 (via GitHub)" <gi...@apache.org>.
shuwang21 commented on PR #41489:
URL: https://github.com/apache/spark/pull/41489#issuecomment-1652775139
Feel free to take a look! Thank you. cc: @Ngone51 @tgravescs
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1285148356
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +324,35 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Percentage of io.serverThreads used by netty to process FinalizeShuffleMerge. When the config
+ * `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is set, shuffle server will use a
+ * separate EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions. The number of threads used for handling
+ * finalizeShuffleMerge requests are percentage of io.serverThreads (if defined) else it is a
+ * percentage of 2 * #cores.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ Preconditions.checkArgument(separateFinalizeShuffleMerge(),
+ "Please set spark.shuffle.server.finalizeShuffleMergeThreadsPercent to a positive value");
Review Comment:
nit: move this below 'shuffle' module check below.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] shuwang21 commented on a diff in pull request #41489: [SPARK-43987][Shuffle] Separate finalizeShuffleMerge Processing to Dedicated Thread Pools
Posted by "shuwang21 (via GitHub)" <gi...@apache.org>.
shuwang21 commented on code in PR #41489:
URL: https://github.com/apache/spark/pull/41489#discussion_r1285151943
##########
common/network-common/src/main/java/org/apache/spark/network/util/TransportConf.java:
##########
@@ -324,6 +324,35 @@ public boolean separateChunkFetchRequest() {
return conf.getInt("spark.shuffle.server.chunkFetchHandlerThreadsPercent", 0) > 0;
}
+ /**
+ * Percentage of io.serverThreads used by netty to process FinalizeShuffleMerge. When the config
+ * `spark.shuffle.server.finalizeShuffleMergeThreadsPercent` is set, shuffle server will use a
+ * separate EventLoopGroup to process FinalizeShuffleMerge messages, which are I/O intensive and
+ * could take long time to process due to disk contentions. The number of threads used for handling
+ * finalizeShuffleMerge requests are percentage of io.serverThreads (if defined) else it is a
+ * percentage of 2 * #cores.
+ */
+ public int finalizeShuffleMergeHandlerThreads() {
+ Preconditions.checkArgument(separateFinalizeShuffleMerge(),
+ "Please set spark.shuffle.server.finalizeShuffleMergeThreadsPercent to a positive value");
Review Comment:
Thanks. Fixed with some format issue report by test results.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org