You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "onebox-li (via GitHub)" <gi...@apache.org> on 2023/10/12 03:29:37 UTC
[PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
onebox-li opened a new pull request, #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978
### What changes were proposed in this pull request?
In our test jobs, we found few map tasks may hang at InFlightRequestTracker#limitZeroInFlight (both
prepareForMergeData and mapEndInternal can occurs) when worker unexpected shutdown. We add logs to trace InFlightRequestTracker#totalInflightReqs and found this adder may become negative In the above case.
When worker suddenly shutdown, the channel connection raise exception.
If NioEventLoop.processSelectedKeys is doing read, the exceptionCaught will be called. In TransportResponseHandler#exceptionCaught will call failOutstandingRequests and each request‘s onFailure callback.
```
WARN [data-client-5-9] TransportChannelHandler: Exception in connection from /xx
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
at sun.nio.ch.IOUtil.read(IOUtil.java:192)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
at org.apache.celeborn.shaded.io.netty.buffer.PooledByteBuf.setBytes(PooledByteBuf.java:256)
at org.apache.celeborn.shaded.io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1132)
at org.apache.celeborn.shaded.io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:357)
at org.apache.celeborn.shaded.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:151)
at org.apache.celeborn.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
at org.apache.celeborn.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
at org.apache.celeborn.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
at org.apache.celeborn.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
at org.apache.celeborn.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.celeborn.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at org.apache.celeborn.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:745)
ERROR [data-client-5-9] ShuffleClientImpl: Push data to xx failed for shuffle 0 map 11 attempt 0 partition 178 batch 634, remain revive times 5.
```
Next NioEventLoop start to `runAllTasks` in the finally block.If there is push write task, PushChannelListener.handleFailure will be called because of the closing channel. Here callback.onFailure may have a data race on `outstandingPushes`.
```
ERROR [data-client-5-9] ShuffleClientImpl: Push data to xx failed for shuffle 0 map 11 attempt 0 partition 178 batch 634, remain revive times 4.
org.apache.celeborn.common.exception.CelebornIOException: Failed to send request PUSH 1264 to /xx: org.apache.celeborn.shaded.io.netty.channel.StacklessClosedChannelException, channel will be closed
at org.apache.celeborn.common.network.client.TransportClient$PushChannelListener.handleFailure(TransportClient.java:382)
at org.apache.celeborn.common.network.client.TransportClient$StdChannelListener.operationComplete(TransportClient.java:325)
at org.apache.celeborn.common.network.client.TransportClient$PushChannelListener.operationComplete(TransportClient.java:373)
at org.apache.celeborn.shaded.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590)
at org.apache.celeborn.shaded.io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557)
at org.apache.celeborn.shaded.io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492)
at org.apache.celeborn.shaded.io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636)
at org.apache.celeborn.shaded.io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:629)
at org.apache.celeborn.shaded.io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:118)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannel$AbstractUnsafe.safeSetFailure(AbstractChannel.java:999)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannel$AbstractUnsafe.write(AbstractChannel.java:860)
at org.apache.celeborn.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.write(DefaultChannelPipeline.java:1367)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:877)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:863)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:968)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:856)
at org.apache.celeborn.shaded.io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:113)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:881)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:863)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:968)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:856)
at org.apache.celeborn.shaded.io.netty.handler.timeout.IdleStateHandler.write(IdleStateHandler.java:302)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:879)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeWriteAndFlush(AbstractChannelHandlerContext.java:940)
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannelHandlerContext$WriteTask.run(AbstractChannelHandlerContext.java:1247)
at org.apache.celeborn.shaded.io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
at org.apache.celeborn.shaded.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
at org.apache.celeborn.shaded.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
at org.apache.celeborn.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
at org.apache.celeborn.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at org.apache.celeborn.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at org.apache.celeborn.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.celeborn.shaded.io.netty.channel.StacklessClosedChannelException
at org.apache.celeborn.shaded.io.netty.channel.AbstractChannel$AbstractUnsafe.write(Object, ChannelPromise)(Unknown Source)
```
Duplicate callback.onFailure will lead to totalInflightReqs count exception.
Here race will not be too severe and only occur under exception situation. So I think synchronize a lock is enough to avoid race.
### Why are the changes needed?
Increase robustness.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Manual test.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "onebox-li (via GitHub)" <gi...@apache.org>.
onebox-li commented on code in PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#discussion_r1357723352
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java:
##########
@@ -70,6 +70,8 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
private static ScheduledExecutorService fetchTimeoutChecker = null;
private ScheduledFuture fetchCheckerScheduleFuture;
+ private final Object handleFailureLock = new Object();
Review Comment:
Now outstandingFetches/outstandingRpcs is still Map. Change them to ConcurrentHashMap and `failOutstandingRequests` handle requests one by one seems to work. I will have a try later.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "onebox-li (via GitHub)" <gi...@apache.org>.
onebox-li commented on code in PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#discussion_r1357973678
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java:
##########
@@ -155,24 +161,26 @@ public void failExpiredFetchRequest() {
long currentTime = System.currentTimeMillis();
Iterator<Map.Entry<StreamChunkSlice, FetchRequestInfo>> iter =
outstandingFetches.entrySet().iterator();
- while (iter.hasNext()) {
- Map.Entry<StreamChunkSlice, FetchRequestInfo> entry = iter.next();
- if (entry.getValue().dueTime <= currentTime) {
- FetchRequestInfo info = outstandingFetches.remove(entry.getKey());
- if (info != null) {
- if (info.channelFuture != null) {
- info.channelFuture.cancel(true);
+ synchronized (handleFailureLock) {
+ while (iter.hasNext()) {
+ Map.Entry<StreamChunkSlice, FetchRequestInfo> entry = iter.next();
+ if (entry.getValue().dueTime <= currentTime) {
+ FetchRequestInfo info = outstandingFetches.remove(entry.getKey());
Review Comment:
Done
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture closed pull request #1978: [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called
URL: https://github.com/apache/incubator-celeborn/pull/1978
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#discussion_r1357717758
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java:
##########
@@ -155,24 +161,26 @@ public void failExpiredFetchRequest() {
long currentTime = System.currentTimeMillis();
Iterator<Map.Entry<StreamChunkSlice, FetchRequestInfo>> iter =
outstandingFetches.entrySet().iterator();
- while (iter.hasNext()) {
- Map.Entry<StreamChunkSlice, FetchRequestInfo> entry = iter.next();
- if (entry.getValue().dueTime <= currentTime) {
- FetchRequestInfo info = outstandingFetches.remove(entry.getKey());
- if (info != null) {
- if (info.channelFuture != null) {
- info.channelFuture.cancel(true);
+ synchronized (handleFailureLock) {
+ while (iter.hasNext()) {
+ Map.Entry<StreamChunkSlice, FetchRequestInfo> entry = iter.next();
+ if (entry.getValue().dueTime <= currentTime) {
+ FetchRequestInfo info = outstandingFetches.remove(entry.getKey());
Review Comment:
nit: call `removeFetchRequest`
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#issuecomment-1758865127
## [Codecov](https://app.codecov.io/gh/apache/incubator-celeborn/pull/1978?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#1978](https://app.codecov.io/gh/apache/incubator-celeborn/pull/1978?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (53836b6) into [main](https://app.codecov.io/gh/apache/incubator-celeborn/commit/6fa669748c565454602d4d44ae179d889fa00495?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (6fa6697) will **decrease** coverage by `0.17%`.
> Report is 2 commits behind head on main.
> The diff coverage is `30.00%`.
```diff
@@ Coverage Diff @@
## main #1978 +/- ##
==========================================
- Coverage 47.01% 46.83% -0.17%
==========================================
Files 164 164
Lines 10368 10403 +35
Branches 957 963 +6
==========================================
- Hits 4873 4871 -2
- Misses 5178 5214 +36
- Partials 317 318 +1
```
| [Files](https://app.codecov.io/gh/apache/incubator-celeborn/pull/1978?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...eleborn/common/network/client/TransportClient.java](https://app.codecov.io/gh/apache/incubator-celeborn/pull/1978?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9jb21tb24vbmV0d29yay9jbGllbnQvVHJhbnNwb3J0Q2xpZW50LmphdmE=) | `40.75% <0.00%> (+0.89%)` | :arrow_up: |
| [...ommon/network/client/TransportResponseHandler.java](https://app.codecov.io/gh/apache/incubator-celeborn/pull/1978?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9jb21tb24vbmV0d29yay9jbGllbnQvVHJhbnNwb3J0UmVzcG9uc2VIYW5kbGVyLmphdmE=) | `53.53% <30.85%> (-5.29%)` | :arrow_down: |
... and [2 files with indirect coverage changes](https://app.codecov.io/gh/apache/incubator-celeborn/pull/1978/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#discussion_r1357725876
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java:
##########
@@ -70,6 +70,8 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
private static ScheduledExecutorService fetchTimeoutChecker = null;
private ScheduledFuture fetchCheckerScheduleFuture;
+ private final Object handleFailureLock = new Object();
Review Comment:
@onebox-li They are already `ConcurrentHashMap`, see
```
this.outstandingFetches = JavaUtils.newConcurrentHashMap();
this.outstandingRpcs = JavaUtils.newConcurrentHashMap();
this.outstandingPushes = JavaUtils.newConcurrentHashMap();
```
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "onebox-li (via GitHub)" <gi...@apache.org>.
onebox-li commented on code in PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#discussion_r1357973229
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -361,8 +359,7 @@ private class RpcChannelListener extends StdChannelListener {
@Override
protected void handleFailure(String errorMsg, Throwable cause) {
- handler.removeRpcRequest(rpcRequestId);
- callback.onFailure(new IOException(errorMsg, cause));
+ handler.handleRpcFailure(rpcRequestId, errorMsg, cause);
Review Comment:
Done
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -393,8 +390,7 @@ public void operationComplete(Future<? super Void> future) throws Exception {
@Override
protected void handleFailure(String errorMsg, Throwable cause) {
- handler.removePushRequest(pushRequestId);
- callback.onFailure(new CelebornIOException(errorMsg, cause));
+ handler.handlePushFailure(pushRequestId, errorMsg, cause);
Review Comment:
Done
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#discussion_r1357714633
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java:
##########
@@ -70,6 +70,8 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
private static ScheduledExecutorService fetchTimeoutChecker = null;
private ScheduledFuture fetchCheckerScheduleFuture;
+ private final Object handleFailureLock = new Object();
Review Comment:
IMO we don't need the lock here, because `outstandingFetches/outstandingRpcs/outstandingPushes` are `ConcurrentHashMap`s, they can guarantee `remove(requestId)` is atomic. Except for `failOutstandingRequests`, the `callback` will only be called after it is removed from the `ConcurrentHashMap`, so we only need to consider refactoring `failOutstandingRequests`. IMO, we can loop over the key iterator, inside each loop we call `remove(key)`, if the returned value is not null, call `onFailure` on it.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -361,8 +359,7 @@ private class RpcChannelListener extends StdChannelListener {
@Override
protected void handleFailure(String errorMsg, Throwable cause) {
- handler.removeRpcRequest(rpcRequestId);
- callback.onFailure(new IOException(errorMsg, cause));
+ handler.handleRpcFailure(rpcRequestId, errorMsg, cause);
Review Comment:
Seems we don't need `callback` anymore.
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -393,8 +390,7 @@ public void operationComplete(Future<? super Void> future) throws Exception {
@Override
protected void handleFailure(String errorMsg, Throwable cause) {
- handler.removePushRequest(pushRequestId);
- callback.onFailure(new CelebornIOException(errorMsg, cause));
+ handler.handlePushFailure(pushRequestId, errorMsg, cause);
Review Comment:
Seems we don't need `callback` anymore.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1036] Map task hangs at limitZeroInFlight due to duplicate onFailure called [incubator-celeborn]
Posted by "onebox-li (via GitHub)" <gi...@apache.org>.
onebox-li commented on code in PR #1978:
URL: https://github.com/apache/incubator-celeborn/pull/1978#discussion_r1357726426
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java:
##########
@@ -70,6 +70,8 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
private static ScheduledExecutorService fetchTimeoutChecker = null;
private ScheduledFuture fetchCheckerScheduleFuture;
+ private final Object handleFailureLock = new Object();
Review Comment:
Thanks, got it.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org