You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "rmcyang (via GitHub)" <gi...@apache.org> on 2023/01/24 20:14:33 UTC
[GitHub] [spark] rmcyang opened a new pull request, #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
rmcyang opened a new pull request, #39725:
URL: https://github.com/apache/spark/pull/39725
<!--
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.
-->
Currently, the `ignoredBlockBytes` of server side metrics for push-based shuffle does not take the late-pushed blocks or the colliding blocks into account. The proposed PR tries to account the late-pushed blocks and the colliding blocks into `ignoredBlockBytes`, given they are also the blocks being ignored.
### 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.
2. If you fix a bug, you can clarify why it is a bug.
-->
This would enhance the `ignoredBlockBytes` of server side metrics for push-based shuffle to be more accurate.
### 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.
-->
Modified existing UTs to capture `ignoredBlockBytes`.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088265331
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
@mridulm I didn't get your point. I thought you pointed that in `onComplete` when we detect `tooLate` or `tooStale`, we should update the metric. However, when we detect `tooLate` or `tooStale` we throw an exception, so that will trigger `onFailure`.
@rmcyang Can you check if those tests trigger `onFailure` after `onComplete` throws an exception? If they don't then that's why the metric didn't get updated. If `onComplete` throws an exception, then `onFailure` should be invoked.
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086459331
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1382,6 +1395,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
freeDeferredBufs();
Review Comment:
I think you are right. Have accommodated the suggestions below hence in this case it would be
`updateIgnoredBytes(deferredBytes + buf.remaining())`
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086238663
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1382,6 +1395,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
freeDeferredBufs();
Review Comment:
When it comes here because of `isTooLateBlockPush` then the whole block is going to be ignored. That means the deferredBufs which are getting freed will be ignored as well so we need to add those bytes to the ignoreBlockBytes as well, correct?
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086454074
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1393,6 +1405,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
freeDeferredBufs();
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
return;
}
abortIfNecessary();
Review Comment:
Fixed.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088313746
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
> However, when we detect tooLate or tooStale we throw an exception, so that will trigger onFailure.
I see what you mean, I read it as this wont happen and should have triggered onFailures from onData directly.
Agree - since `TransportRequestHandler.processStreamRequest` is explicitly doing the delegation to `onFailures` when `onComplete` fails, that should be fine.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088266355
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1478,13 +1478,15 @@ public void onComplete(String streamId) throws IOException {
if (isTooLate(info, partitionInfo.reduceId)) {
freeDeferredBufs();
mergeManager.pushMergeMetrics.lateBlockPushes.mark();
+ updateIgnoredBlockBytes();
throw new BlockPushNonFatalFailure(
new BlockPushReturnCode(ReturnCode.TOO_LATE_BLOCK_PUSH.id(), streamId).toByteBuffer(),
BlockPushNonFatalFailure.getErrorMsg(streamId, ReturnCode.TOO_LATE_BLOCK_PUSH));
}
if (isStale(info, partitionInfo.appAttemptShuffleMergeId.shuffleMergeId)) {
freeDeferredBufs();
mergeManager.pushMergeMetrics.staleBlockPushes.mark();
+ updateIgnoredBlockBytes();
Review Comment:
This is not needed. There is an exception thrown in the next line which triggers `onFailure`
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086433746
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1315,6 +1315,7 @@ private void freeDeferredBufs() {
*/
Review Comment:
This helps reduce iteration on `deferredBufs` which looks better. Have updated the PR, thanks for the suggestions.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1404045687
Good callout - I do not want to include failed writes, just those which were actually ignored.
We can try/finally the onData, and ignore update in case of IOException
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088214955
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
It need not necessarily be detected during onData - it can go stale subsequently as well.
(Else we wont need these two conditions here at all :-) )
--
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] rmcyang commented on pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1404725389
Have updated the PR to reflect the suggestions above and refactored a bit. Thanks @otterc, please let me know if anything 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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086008666
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1382,6 +1395,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
freeDeferredBufs();
if (isTooLateBlockPush) {
mergeManager.pushMergeMetrics.lateBlockPushes.mark();
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
Review Comment:
should be` buf.remaining()`
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1357,18 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * For the block bytes in the deferred buffers that are ignored, capture them
+ * and update pushMergeMetrics's ignoredBlockBytes.
+ */
+ private void updateIgnoredBytesWithDeferredBufs() {
+ if (deferredBufs != null && !deferredBufs.isEmpty()) {
+ for (ByteBuffer buf : deferredBufs) {
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.remaining());
Review Comment:
I made a mistake with the earlier comment. For deferredBufs, we can use `buf.limit`.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1405757802
Will wait for CI to pass before merging.
--
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 pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1404040435
> Sum up the total number of bytes received in `onData` - does not matter if it was written, deferred, etc.
> In onComplete/onFailure, add it to ignored bytes if we are deciding not to use data from this push.
> Will keep the change much more constrained, and avoid issues where a push failed/was aborted/etc after it has partially written data out.
So will we account for bytes that the server failed to write as well in `ignoredBytes`? I was thinking about this while making the suggestions and decided that the bytes that the server fails to write (throws an exception) are **not** `ignored`.
In the current PR, even when it aborts, only deferred bytes are included in `ignoredBytes`. The bytes during which the server encounters IOException while writing are not included in `ignoredBytes`.
I am actually okay with your suggestion but I just want to be clear on what we are capturing as ignored. If we are capturing the failed bytes as ignored, then maybe we should either rename it or at least add the comment to make it clear.
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086005947
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1356,17 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * If necessary, look for the ignored bock bytes and capture them in pushMergeMetrics.
+ */
+ private void lookForIgnoredBlockBytesIfNecessary() {
Review Comment:
Thanks for the suggestions, Chandni. Renamed it to `updateIgnoredBytesWithDeferredBufs` based on the former, as it sounds more smooth to me. Hopefully it looks better. Please let me know otherwise.
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088250874
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
The UTs `testBlockPushWithOlderShuffleMergeId` and `testFinalizeWithOlderShuffleMergeId ` have some stale blocks. Before applying `updateIgnoredBlockBytes` for `isStale` and `isTooLate` conditions above, they are not taken account into the ignored bytes. They are able to be captured after. So I think we should update ignored block bytes there. Updated the PR, 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] otterc commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088265331
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
@mridulm I didn't get your point. I thought you pointed that in `onComplete` when we detect `tooLate` or `tooStale`, we should update the metric. However, when we detect `tooLate` or `tooStale` we throw an exception, so that will trigger `onFailure`.
@rmcyang Can you check if those tests trigger `onFailure` after `onComplete` throws an exception? If they don't then that's why the metric didn't get updated.
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088338506
##########
docs/monitoring.md:
##########
@@ -1433,8 +1433,9 @@ Note: applies to the shuffle service
- deferredBlocks - the number of the current deferred block parts buffered in memory
- deferredBlockBytes - the size of the current deferred block parts buffered in memory
- staleBlockPushes - the number of stale shuffle block push requests
-- ignoredBlockBytes - the size of the pushed block data that are ignored after the shuffle
- file is finalized or when a request is for a duplicate block
+- ignoredBlockBytes - the size of the pushed block data that are ignored. The pushed block data are
+ considered as ignored for these cases: 1. received after the shuffle file is finalized; 2. when a
+ request is for a duplicate block; 3. the part that ESS failed to write.
Review Comment:
Thanks for the suggestion, also trimmed a little bit for other metrics above to be consistent :-)
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086246214
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1357,18 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * For the block bytes in the deferred buffers that are ignored, capture them
+ * and update pushMergeMetrics's ignoredBlockBytes.
+ */
+ private void updateIgnoredBytesWithDeferredBufs() {
+ if (deferredBufs != null && !deferredBufs.isEmpty()) {
+ for (ByteBuffer buf : deferredBufs) {
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.remaining());
+ }
+ }
+ }
+
Review Comment:
```suggestion
private void updateIgnoredBytes(long numBytes) {
if (numBytes > 0) {
mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(numBytes);
}
}
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1315,6 +1315,7 @@ private void freeDeferredBufs() {
*/
private void abortIfNecessary() {
if (partitionInfo.shouldAbort(mergeManager.ioExceptionsThresholdDuringMerge)) {
+ updateIgnoredBytesWithDeferredBufs();
freeDeferredBufs();
Review Comment:
```suggestion
updateIgnoredBytes(freeDeferredBufs());
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1461,6 +1476,7 @@ public void onComplete(String streamId) throws IOException {
AppShuffleMergePartitionsInfo info =
appShuffleInfo.shuffles.get(partitionInfo.appAttemptShuffleMergeId.shuffleId);
if (isTooLate(info, partitionInfo.reduceId)) {
+ updateIgnoredBytesWithDeferredBufs();
freeDeferredBufs();
Review Comment:
```suggestion
updateIgnoredBytes(freeDeferredBufs());
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1393,6 +1407,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
freeDeferredBufs();
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
Review Comment:
```suggestion
updateIgnoredBytes(freeDeferredBufs() + buf.remaining());
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1520,6 +1537,7 @@ public void onComplete(String streamId) throws IOException {
partitionInfo.resetChunkTracker();
}
} else {
+ updateIgnoredBytesWithDeferredBufs();
freeDeferredBufs();
Review Comment:
```suggestion
updateIgnoredBytes(freeDeferredBufs());
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1480,6 +1496,7 @@ public void onComplete(String streamId) throws IOException {
// Identify duplicate block generated by speculative tasks. We respond success to
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
+ updateIgnoredBytesWithDeferredBufs();
freeDeferredBufs();
Review Comment:
```suggestion
updateIgnoredBytes(freeDeferredBufs());
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1382,6 +1395,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
freeDeferredBufs();
if (isTooLateBlockPush) {
mergeManager.pushMergeMetrics.lateBlockPushes.mark();
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
Review Comment:
```suggestion
long deferredBytes = freeDeferredBufs();
if (isTooLateBlockPush) {
mergeManager.pushMergeMetrics.lateBlockPushes.mark();
updateIgnoredBytes(deferredBytes + buf.remaining());
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1315,6 +1315,7 @@ private void freeDeferredBufs() {
*/
Review Comment:
Wasn't able to comment on the `freeDeferredBufs()` lines since it is not changed. We can change it to below to apply my other suggestions.
```
/**
* @return total number of deferred bytes
*/
private long freeDeferredBufs() {
if (deferredBufs == null || deferredBufs.isEmpty()) {
deferredBufs = null;
return 0;
}
long totalSize = 0;
for (ByteBuffer deferredBuf : deferredBufs) {
totalSize += deferredBuf.limit();
mergeManager.pushMergeMetrics.deferredBlocks.mark(-1);
}
mergeManager.pushMergeMetrics.deferredBlockBytes.dec(totalSize);
deferredBufs = null;
return totalSize;
}
```
--
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 #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088326936
##########
docs/monitoring.md:
##########
@@ -1433,8 +1433,9 @@ Note: applies to the shuffle service
- deferredBlocks - the number of the current deferred block parts buffered in memory
- deferredBlockBytes - the size of the current deferred block parts buffered in memory
- staleBlockPushes - the number of stale shuffle block push requests
-- ignoredBlockBytes - the size of the pushed block data that are ignored after the shuffle
- file is finalized or when a request is for a duplicate block
+- ignoredBlockBytes - the size of the pushed block data that are ignored. The pushed block data are
+ considered as ignored for these cases: 1. received after the shuffle file is finalized; 2. when a
+ request is for a duplicate block; 3. the part that ESS failed to write.
Review Comment:
```suggestion
- ignoredBlockBytes - size of the pushed block data that was transferred to ESS, but ignored.
The pushed block data are considered as ignored when: 1. it was received after the shuffle
was finalized; 2. when a push request is for a duplicate block; 3. ESS was unable to write the block.
```
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1404062208
Deferred block writes makes it tricker, yes.
--
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 #39725: [SPARK-33573][CORE][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1406296111
Merged to master and branch-3.4
Thanks for working on this @rmcyang !
Thanks for the review @otterc :-)
+CC @xinrong-meng as this was merged to 3.4 branch
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086966365
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1362,15 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * Update ignoredBlockBytes in pushMergeMetrics.
+ */
+ private void updateIgnoredBytes(Long numBytes) {
Review Comment:
`Long` -> `long`
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1085930450
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1356,17 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * If necessary, look for the ignored bock bytes and capture them in pushMergeMetrics.
+ */
+ private void lookForIgnoredBlockBytesIfNecessary() {
Review Comment:
nit: rename to `updatedIgnoreBytesWithDeferredBufs` or `deferredBufsIgnored` or something similar.
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1393,6 +1405,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
freeDeferredBufs();
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
return;
}
abortIfNecessary();
Review Comment:
Even when the server aborts, it ignores the bytes.
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1356,17 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * If necessary, look for the ignored bock bytes and capture them in pushMergeMetrics.
+ */
+ private void lookForIgnoredBlockBytesIfNecessary() {
+ if (deferredBufs != null && !deferredBufs.isEmpty()) {
+ for (ByteBuffer buf : deferredBufs) {
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
Review Comment:
I think we should use `buf.remaining` here
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088149074
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
We throw the exception so `onFailure` will be called and they will be updated there.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088266577
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1478,13 +1478,15 @@ public void onComplete(String streamId) throws IOException {
if (isTooLate(info, partitionInfo.reduceId)) {
freeDeferredBufs();
mergeManager.pushMergeMetrics.lateBlockPushes.mark();
+ updateIgnoredBlockBytes();
Review Comment:
This is not needed. There is an exception thrown in the next line which triggers onFailure
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1405107741
+CC @xinrong-meng, I plan to include this in 3.4 as well.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088159016
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1554,6 +1572,10 @@ public void onFailure(String streamId, Throwable throwable) throws IOException {
}
}
isWriting = false;
+ // The failed bytes (and the subsequent bytes) landed at ESS, hence we considered them
+ // also as the bytes being ignored. Including them as ignoredBlockBytes would also help
+ // measure any server side improvement.
+ updateIgnoredBlockBytes();
Review Comment:
Nit: Can we move this block to line 1559.
Also reword: `The block was received by ESS but didn't get merged so it is considered as "ignored"`.
--
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 #39725: [SPARK-33573][CORE][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm closed pull request #39725: [SPARK-33573][CORE][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
URL: https://github.com/apache/spark/pull/39725
--
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] rmcyang commented on pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1402573328
cc @mridulm @otterc 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] otterc commented on pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1404061436
If we don't want to include failed writes, then I think your suggestion requires much more refactoring of the current code. I thought it was slightly riskier to make those changes just because of the metric update. I may be wrong. @rmcyang can look into it and see if it is not too much.
--
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] rmcyang commented on pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1404317258
Thanks @otterc and @mridulm !
Have updated the PR trying to cover below proposal:
- For `onComplete` and `onFailure`, add the data not being used from the push to ignored bytes
- For `onData`, add the received data not being used to ignored bytes (include the bytes during which the server encounters IOException)
- updated the documents for `ignoredBytes` to be clearer on what it tries to collect
Please let me know if you if any thoughts.
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1087470017
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1362,15 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * Update ignoredBlockBytes in pushMergeMetrics.
+ */
+ private void updateIgnoredBytes(long numBytes) {
+ if (numBytes > 0) {
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(numBytes);
+ }
+ }
+
Review Comment:
```suggestion
private long totalBytesReceived = 0;
/**
* Update ignoredBlockBytes in pushMergeMetrics.
*/
private void receivedBytesIgnored() {
if (totalBytesReceived > 0) {
mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(totalBytesReceived);
totalBytesReceived = 0;
}
}
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1298,24 +1298,30 @@ private void writeDeferredBufs() throws IOException {
deferredBufs = null;
}
- private void freeDeferredBufs() {
- if (deferredBufs != null && !deferredBufs.isEmpty()) {
- long totalSize = 0;
- for (ByteBuffer deferredBuf : deferredBufs) {
- totalSize += deferredBuf.limit();
- mergeManager.pushMergeMetrics.deferredBlocks.mark(-1);
- }
- mergeManager.pushMergeMetrics.deferredBlockBytes.dec(totalSize);
+ /**
+ * @return total number of deferred bytes
+ */
+ private long freeDeferredBufs() {
+ if (deferredBufs == null || deferredBufs.isEmpty()) {
+ deferredBufs = null;
+ return 0;
+ }
+ long totalSize = 0;
+ for (ByteBuffer deferredBuf : deferredBufs) {
+ totalSize += deferredBuf.limit();
+ mergeManager.pushMergeMetrics.deferredBlocks.mark(-1);
}
+ mergeManager.pushMergeMetrics.deferredBlockBytes.dec(totalSize);
deferredBufs = null;
+ return totalSize;
Review Comment:
revert this change
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1362,15 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * Update ignoredBlockBytes in pushMergeMetrics.
+ */
+ private void updateIgnoredBytes(long numBytes) {
+ if (numBytes > 0) {
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(numBytes);
+ }
+ }
+
@Override
public void onData(String streamId, ByteBuffer buf) throws IOException {
Review Comment:
```suggestion
public void onData(String streamId, ByteBuffer buf) throws IOException {
totalBytesReceived += buf.remaining();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1298,24 +1298,30 @@ private void writeDeferredBufs() throws IOException {
deferredBufs = null;
}
- private void freeDeferredBufs() {
- if (deferredBufs != null && !deferredBufs.isEmpty()) {
- long totalSize = 0;
- for (ByteBuffer deferredBuf : deferredBufs) {
- totalSize += deferredBuf.limit();
- mergeManager.pushMergeMetrics.deferredBlocks.mark(-1);
- }
- mergeManager.pushMergeMetrics.deferredBlockBytes.dec(totalSize);
+ /**
+ * @return total number of deferred bytes
+ */
+ private long freeDeferredBufs() {
+ if (deferredBufs == null || deferredBufs.isEmpty()) {
+ deferredBufs = null;
+ return 0;
+ }
+ long totalSize = 0;
+ for (ByteBuffer deferredBuf : deferredBufs) {
+ totalSize += deferredBuf.limit();
+ mergeManager.pushMergeMetrics.deferredBlocks.mark(-1);
}
+ mergeManager.pushMergeMetrics.deferredBlockBytes.dec(totalSize);
deferredBufs = null;
+ return totalSize;
}
/**
* @throws IllegalStateException if the number of IOExceptions have exceeded threshold.
*/
private void abortIfNecessary() {
if (partitionInfo.shouldAbort(mergeManager.ioExceptionsThresholdDuringMerge)) {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs());
Review Comment:
Revert
```suggestion
freeDeferredBufs();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1410,6 +1425,10 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
}
writeBuf(buf);
} catch (IOException ioe) {
+ // When server failed to write, capture the number of failed bytes to ignoreBlckBytes
+ // of pushMergeMetrics. The number of deferred bytes are already captured in
+ // incrementIOExceptionsAndAbortIfNecessary().
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.remaining());
Review Comment:
Revert this change
```suggestion
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1461,14 +1480,14 @@ public void onComplete(String streamId) throws IOException {
AppShuffleMergePartitionsInfo info =
appShuffleInfo.shuffles.get(partitionInfo.appAttemptShuffleMergeId.shuffleId);
if (isTooLate(info, partitionInfo.reduceId)) {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs());
Review Comment:
Revert
```suggestion
freeDeferredBufs();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1520,7 +1539,7 @@ public void onComplete(String streamId) throws IOException {
partitionInfo.resetChunkTracker();
}
} else {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs());
Review Comment:
Revert
```suggestion
freeDeferredBufs();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1554,6 +1573,7 @@ public void onFailure(String streamId, Throwable throwable) throws IOException {
}
}
isWriting = false;
+ updateIgnoredBytes(freeDeferredBufs());
Review Comment:
```suggestion
receivedBytesIgnored();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1392,7 +1407,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// Identify duplicate block generated by speculative tasks. We respond success to
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs() + buf.remaining());
Review Comment:
Revert
```suggestion
freeDeferredBufs();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1379,7 +1394,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
isStale(info, partitionInfo.appAttemptShuffleMergeId.shuffleMergeId);
boolean isTooLateBlockPush = isTooLate(info, partitionInfo.reduceId);
if (isStaleBlockPush || isTooLateBlockPush) {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs() + buf.remaining());
Review Comment:
Revert
```suggestion
freeDeferredBufs();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1461,14 +1480,14 @@ public void onComplete(String streamId) throws IOException {
AppShuffleMergePartitionsInfo info =
appShuffleInfo.shuffles.get(partitionInfo.appAttemptShuffleMergeId.shuffleId);
if (isTooLate(info, partitionInfo.reduceId)) {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs());
mergeManager.pushMergeMetrics.lateBlockPushes.mark();
throw new BlockPushNonFatalFailure(
new BlockPushReturnCode(ReturnCode.TOO_LATE_BLOCK_PUSH.id(), streamId).toByteBuffer(),
BlockPushNonFatalFailure.getErrorMsg(streamId, ReturnCode.TOO_LATE_BLOCK_PUSH));
}
if (isStale(info, partitionInfo.appAttemptShuffleMergeId.shuffleMergeId)) {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs());
Review Comment:
Revert
```suggestion
freeDeferredBufs();
```
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1480,7 +1499,7 @@ public void onComplete(String streamId) throws IOException {
// Identify duplicate block generated by speculative tasks. We respond success to
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
- freeDeferredBufs();
+ updateIgnoredBytes(freeDeferredBufs());
Review Comment:
```suggestion
freeDeferredBufs();
// Since we don't throw an exception and just return, but the received bytes are ignored we need to update the ignored bytes metric.
receivedBytesIgnored();
```
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "otterc (via GitHub)" <gi...@apache.org>.
otterc commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086007667
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1393,6 +1405,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
freeDeferredBufs();
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
return;
}
abortIfNecessary();
Review Comment:
Here it should be `buf.remaining`
--
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 #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088313746
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
> However, when we detect tooLate or tooStale we throw an exception, so that will trigger onFailure.
I see what you mean, I read it as this wont happen and should have triggered `onFailure` from `onData` directly.
Agree - since `TransportRequestHandler.processStreamRequest` is explicitly doing the delegation to `onFailure` when `onComplete` fails, that should be fine.
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1086006244
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1393,6 +1405,7 @@ public void onData(String streamId, ByteBuffer buf) throws IOException {
// the client in cases of duplicate even though no data is written.
if (isDuplicateBlock()) {
freeDeferredBufs();
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
return;
}
abortIfNecessary();
Review Comment:
Make sense, have updated the PR to call `updateIgnoredBytesWithDeferredBufs()` in `abortIfNecessary()`.
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1356,6 +1356,17 @@ private boolean isTooLate(
!appShuffleMergePartitionsInfo.shuffleMergePartitions.containsKey(reduceId);
}
+ /**
+ * If necessary, look for the ignored bock bytes and capture them in pushMergeMetrics.
+ */
+ private void lookForIgnoredBlockBytesIfNecessary() {
+ if (deferredBufs != null && !deferredBufs.isEmpty()) {
+ for (ByteBuffer buf : deferredBufs) {
+ mergeManager.pushMergeMetrics.ignoredBlockBytes.mark(buf.limit());
Review Comment:
Sounds good, have updated the PR.
--
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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088250874
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
The UTs `testBlockPushWithOlderShuffleMergeId` and `testFinalizeWithOlderShuffleMergeId ` have some stale blocks. Before applying updateIgnoredBlockBytes for isStale and isTooLate conditions above, they are not taken account into the ignored bytes. They are able to be captured after. So I think we should update ignored block bytes there. Updated the PR, 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] rmcyang commented on a diff in pull request #39725: [SPARK-33573][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1088315899
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
@otterc I checked those UTs. We didn't call onFailure explicitly for those steams, I have explicitly call onFailure now and the ignored bytes metric is able to be updated.
--
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] rmcyang commented on pull request #39725: [SPARK-33573][CORE][FOLLOW-UP] Enhance ignoredBlockBytes in pushMergeMetrics to cover more scenarios
Posted by "rmcyang (via GitHub)" <gi...@apache.org>.
rmcyang commented on PR #39725:
URL: https://github.com/apache/spark/pull/39725#issuecomment-1406942817
Thank you for the reviews, @otterc and @mridulm !
--
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 #39725: [SPARK-33573][FOLLOW-UP] Increment ignoredBlockBytes when shuffle push blocks are late or colliding
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39725:
URL: https://github.com/apache/spark/pull/39725#discussion_r1087934163
##########
common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java:
##########
@@ -1481,6 +1496,9 @@ public void onComplete(String streamId) throws IOException {
// the client in cases of duplicate even though no data is written.
Review Comment:
`updateIgnoredBlockBytes` for `isStale` and `isTooLate` conditions above ?
--
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