You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "xianjingfeng (via GitHub)" <gi...@apache.org> on 2023/01/28 08:17:11 UTC
[GitHub] [incubator-uniffle] xianjingfeng opened a new pull request, #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
xianjingfeng opened a new pull request, #510:
URL: https://github.com/apache/incubator-uniffle/pull/510
<!--
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://github.com/apache/incubator-uniffle/blob/master/CONTRIBUTING.md
2. Ensure you have added or run the appropriate tests for your PR
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]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.
-->
### 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.
-->
Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold.
Add suspend mark to `ShuffleFlushManager`.
### 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.
-->
Flaky Test
### 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 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.
-->
No need
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089902117
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
OK, I got it. If the flushQueue is empty, `processEvents()` will be called again and again.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089908574
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -50,8 +51,8 @@ public class ShuffleFlushManager {
private static final Logger LOG = LoggerFactory.getLogger(ShuffleFlushManager.class);
public static final AtomicLong ATOMIC_EVENT_ID = new AtomicLong(0);
private final ShuffleServer shuffleServer;
- private final BlockingQueue<ShuffleDataFlushEvent> flushQueue = Queues.newLinkedBlockingQueue();
- private final ThreadPoolExecutor threadPoolExecutor;
+ protected final BlockingQueue<ShuffleDataFlushEvent> flushQueue = Queues.newLinkedBlockingQueue();
Review Comment:
Please revert this change.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089897680
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
```java
while (true) {
while (!flushQueue.isEmpty()) {
}
}
```
is same with
```java
while (true || !flushQueue.isEmpty()) {
}
```
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089904214
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
> We need overwrite `startEventProcessor` in `TestShuffleFlushManager`, Because if we start EventProcesser, events will be flushed automatically and this pr is used to prevent automatic flush.
I have fixed the infinite loop and squashed the commits, so you can see all changes in one commit.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089889652
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -137,6 +113,14 @@ public void run() {
thread.start();
}
+ protected void startEventProcesser() {
+ // the thread for flush data
+ Thread processEventThread = new Thread(() -> processEvents());
Review Comment:
Got
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] codecov-commenter commented on pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#issuecomment-1407341448
# [Codecov](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#510](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (098903f) into [master](https://codecov.io/gh/apache/incubator-uniffle/commit/33c4fe4208b640a71d4ecbb2f9ad3f20f4ba1cc9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (33c4fe4) will **increase** coverage by `1.26%`.
> The diff coverage is `50.00%`.
```diff
@@ Coverage Diff @@
## master #510 +/- ##
============================================
+ Coverage 62.86% 64.13% +1.26%
+ Complexity 1766 1634 -132
============================================
Files 199 186 -13
Lines 10366 9094 -1272
Branches 1033 912 -121
============================================
- Hits 6517 5832 -685
+ Misses 3504 2945 -559
+ Partials 345 317 -28
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...org/apache/uniffle/server/ShuffleFlushManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9TaHVmZmxlRmx1c2hNYW5hZ2VyLmphdmE=) | `83.33% <50.00%> (-0.71%)` | :arrow_down: |
| [...he/uniffle/server/storage/LocalStorageManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9zdG9yYWdlL0xvY2FsU3RvcmFnZU1hbmFnZXIuamF2YQ==) | `85.87% <0.00%> (-1.13%)` | :arrow_down: |
| [...java/org/apache/hadoop/mapred/SortWriteBuffer.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkL1NvcnRXcml0ZUJ1ZmZlci5qYXZh) | | |
| [...apache/hadoop/mapreduce/v2/app/RssMRAppMaster.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkdWNlL3YyL2FwcC9Sc3NNUkFwcE1hc3Rlci5qYXZh) | | |
| [.../hadoop/mapreduce/task/reduce/RssBypassWriter.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkdWNlL3Rhc2svcmVkdWNlL1Jzc0J5cGFzc1dyaXRlci5qYXZh) | | |
| [.../java/org/apache/hadoop/mapreduce/RssMRConfig.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkdWNlL1Jzc01SQ29uZmlnLmphdmE=) | | |
| [...n/java/org/apache/hadoop/mapreduce/MRIdHelper.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkdWNlL01SSWRIZWxwZXIuamF2YQ==) | | |
| [...pache/hadoop/mapreduce/task/reduce/RssShuffle.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkdWNlL3Rhc2svcmVkdWNlL1Jzc1NodWZmbGUuamF2YQ==) | | |
| [...g/apache/hadoop/mapred/SortWriteBufferManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkL1NvcnRXcml0ZUJ1ZmZlck1hbmFnZXIuamF2YQ==) | | |
| [...n/java/org/apache/hadoop/mapreduce/RssMRUtils.java](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LW1yL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9oYWRvb3AvbWFwcmVkdWNlL1Jzc01SVXRpbHMuamF2YQ==) | | |
| ... and [5 more](https://codecov.io/gh/apache/incubator-uniffle/pull/510?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
: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=The+Apache+Software+Foundation)
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#issuecomment-1407606959
Thanks @xianjingfeng for updating the PR, @advancedxy what do you think?
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089897932
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -137,6 +108,24 @@ public void run() {
thread.start();
}
+ protected void startEventProcesser() {
Review Comment:
Typo: startEventProcesser -> startEventProcessor
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#issuecomment-1407354006
> Could you help describe more about this fix in description? From current code, this fix looks a little bit strange.
Because the flush operation is asynchronous, the first block maybe been flushed when judging the memory size.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089907950
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
Update:
Actually I'm splitting `processEvents()` into `eventLoop()` and `flush()`.
```java
protected void eventLoop() {
while (true) {
processNextEvent();
}
}
@VisibleForTesting
public void flush() {
while (!flushQueue.isEmpty()) {
processNextEvent();
}
}
private void processNextEvent() {
try {
ShuffleDataFlushEvent event = flushQueue.take();
threadPoolExecutor.execute(() -> {
processEvent(event);
});
} catch (Exception e) {
LOG.error("Exception happened when process event.", e);
}
}
```
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen merged pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen merged PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089891556
##########
server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java:
##########
@@ -567,7 +568,25 @@ public void shuffleFlushThreshold() throws Exception {
StorageManager storageManager = StorageManagerFactory.getInstance().createStorageManager(conf);
ShuffleFlushManager shuffleFlushManager = new ShuffleFlushManager(conf,
- "serverId", mockShuffleServer, storageManager);
+ "serverId", mockShuffleServer, storageManager) {
+
+ @Override
+ protected void startEventProcesser() {
Review Comment:
https://github.com/apache/incubator-uniffle/pull/510/files/a803bf816ff79e16834ab6f42258373b71e71f5e
If you apply this code to your test class method, the event flush could be controlled externally?
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089899642
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
I mean `flushQueue.take()` will block the thread until `flushQueue` is not empty and it will not cost too much cpu resources. But `flushQueue.take()` will not block the thread in the code you implement and it will not cost a lot of cpu resources.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089856504
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -95,6 +96,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
while (true) {
try {
ShuffleDataFlushEvent event = flushQueue.take();
+ // Only for test
+ while (isSuspend) {
Review Comment:
To be honest, I don't think this is a good fix. Especially it complicates the shuffle manager's logic just for test and the `setSuspend` method is opened as public. It couples the normal logic and test specific logic.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089858541
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -95,6 +96,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
while (true) {
try {
ShuffleDataFlushEvent event = flushQueue.take();
+ // Only for test
+ while (isSuspend) {
Review Comment:
You are right. Do you have any good ideas?
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089887569
##########
server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java:
##########
@@ -567,7 +568,25 @@ public void shuffleFlushThreshold() throws Exception {
StorageManager storageManager = StorageManagerFactory.getInstance().createStorageManager(conf);
ShuffleFlushManager shuffleFlushManager = new ShuffleFlushManager(conf,
- "serverId", mockShuffleServer, storageManager);
+ "serverId", mockShuffleServer, storageManager) {
+
+ @Override
+ protected void startEventProcesser() {
Review Comment:
In this UT, if we start EventProcesser, events will be flushed automatically.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089899638
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
For reference: https://github.com/kaijchen/incubator-uniffle/tree/issue_509_fork
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#issuecomment-1407613119
> Thanks @xianjingfeng for updating the PR, @advancedxy what do you think?
Thanks @xianjingfeng for your help. This code generally looks good to me with one minor comment. And I believe after introduce the flush method, some other tests may also be refactored to use the flush method instead of `waitForFlush`. But that should be addressed in another issue/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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089917908
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -81,40 +82,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
this.maxConcurrencyOfSingleOnePartition =
shuffleServerConf.get(ShuffleServerConf.SERVER_MAX_CONCURRENCY_OF_ONE_PARTITION);
- int waitQueueSize = shuffleServerConf.getInteger(
- ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_QUEUE_SIZE);
- BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(waitQueueSize);
- int poolSize = shuffleServerConf.getInteger(ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_SIZE);
- long keepAliveTime = shuffleServerConf.getLong(ShuffleServerConf.SERVER_FLUSH_THREAD_ALIVE);
- threadPoolExecutor = new ThreadPoolExecutor(poolSize, poolSize, keepAliveTime, TimeUnit.SECONDS, waitQueue,
- ThreadUtils.getThreadFactory("FlushEventThreadPool"));
+ threadPoolExecutor = createFlushEventExecutor();
+ startEventProcessor();
Review Comment:
I think line L85-86 should be placed after line L87-88.
Because `storageBasePaths` is accessed in `processEvent() -> flushToFile()`
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#issuecomment-1407615360
> Maybe it's better to not expose `flush()` to production code: [kaijchen@cb9a0be](https://github.com/kaijchen/incubator-uniffle/commit/cb9a0be5734825e9e89f62ff50052ddb32fd90b8)
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089897280
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
```java
protected void eventLoop() {
while (true) {
processEvents();
}
}
```
The CPU maybe will full?
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
+ while (endless || !flushQueue.isEmpty()) {
+ try {
+ ShuffleDataFlushEvent event = flushQueue.take();
+ threadPoolExecutor.execute(() -> {
+ processEvent(event);
+ });
+ } catch (Exception e) {
+ LOG.error("Exception happened when process event.", e);
+ }
+ }
+ }
+
+ protected void processEvent(ShuffleDataFlushEvent event) {
Review Comment:
Got
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089897280
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
```java
protected void eventLoop() {
while (true) {
processEvents();
}
}
```
The CPU maybe will be full?
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089898151
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
Got. But in this pr, if `endless=true`, `flushQueue.take()` will block the thread.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089909261
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -50,8 +51,8 @@ public class ShuffleFlushManager {
private static final Logger LOG = LoggerFactory.getLogger(ShuffleFlushManager.class);
public static final AtomicLong ATOMIC_EVENT_ID = new AtomicLong(0);
private final ShuffleServer shuffleServer;
- private final BlockingQueue<ShuffleDataFlushEvent> flushQueue = Queues.newLinkedBlockingQueue();
- private final ThreadPoolExecutor threadPoolExecutor;
+ protected final BlockingQueue<ShuffleDataFlushEvent> flushQueue = Queues.newLinkedBlockingQueue();
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089897280
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
```java
protected void eventLoop() {
while (true) {
processEvents();
}
}
```
The CPU utilization maybe will be full?
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089682669
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -67,6 +67,7 @@ public class ShuffleFlushManager {
private final long pendingEventTimeoutSec;
private int processPendingEventIndex = 0;
private final int maxConcurrencyOfSingleOnePartition;
+ private boolean isSuspend;
Review Comment:
Add `volatile`?
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089893072
##########
server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java:
##########
@@ -567,7 +568,25 @@ public void shuffleFlushThreshold() throws Exception {
StorageManager storageManager = StorageManagerFactory.getInstance().createStorageManager(conf);
ShuffleFlushManager shuffleFlushManager = new ShuffleFlushManager(conf,
- "serverId", mockShuffleServer, storageManager);
+ "serverId", mockShuffleServer, storageManager) {
+
+ @Override
+ protected void startEventProcesser() {
Review Comment:
> https://github.com/apache/incubator-uniffle/pull/510/files/a803bf816ff79e16834ab6f42258373b71e71f5e If you apply this code to your test class method, the event flush could be controlled externally?
Not exactly. At least we can control when to begin flush.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089911947
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -81,40 +82,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
this.maxConcurrencyOfSingleOnePartition =
shuffleServerConf.get(ShuffleServerConf.SERVER_MAX_CONCURRENCY_OF_ONE_PARTITION);
- int waitQueueSize = shuffleServerConf.getInteger(
- ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_QUEUE_SIZE);
- BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(waitQueueSize);
- int poolSize = shuffleServerConf.getInteger(ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_SIZE);
- long keepAliveTime = shuffleServerConf.getLong(ShuffleServerConf.SERVER_FLUSH_THREAD_ALIVE);
- threadPoolExecutor = new ThreadPoolExecutor(poolSize, poolSize, keepAliveTime, TimeUnit.SECONDS, waitQueue,
- ThreadUtils.getThreadFactory("FlushEventThreadPool"));
+ threadPoolExecutor = createFlushEventExecutor();
storageBasePaths = shuffleServerConf.get(ShuffleServerConf.RSS_STORAGE_BASE_PATH);
pendingEventTimeoutSec = shuffleServerConf.getLong(ShuffleServerConf.PENDING_EVENT_TIMEOUT_SEC);
- // the thread for flush data
- Runnable processEventRunnable = () -> {
- while (true) {
- try {
- ShuffleDataFlushEvent event = flushQueue.take();
- threadPoolExecutor.execute(() -> {
- try {
- ShuffleServerMetrics.gaugeWriteHandler.inc();
- flushToFile(event);
- } catch (Exception e) {
- LOG.error("Exception happened when flush data for " + event, e);
- } finally {
- ShuffleServerMetrics.gaugeWriteHandler.dec();
- ShuffleServerMetrics.gaugeEventQueueSize.dec();
- }
- });
- } catch (Exception e) {
- LOG.error("Exception happened when process event.", e);
- }
- }
- };
- Thread processEventThread = new Thread(processEventRunnable);
- processEventThread.setName("ProcessEventThread");
- processEventThread.setDaemon(true);
- processEventThread.start();
+ startEventProcessor();
Review Comment:
nit: let's move L85 before this line since these two lines are related.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089869885
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -95,6 +96,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
while (true) {
try {
ShuffleDataFlushEvent event = flushQueue.take();
+ // Only for test
+ while (isSuspend) {
Review Comment:
I did a quick overview of related code, and there might be two proposals:
1. modify unit test to use Hamcrest match API to match multiple possible values, since the first block could already be flushed, we can consider that in the unit test
2. refactor `ShuffleFlushManager` processEventRunnable construct, extract related logic into one method, for example `processEvent`.
In `ShuffleBufferManagerTest#shuffleFlushThreshold`, a new TestShuffleFlushManager should be created and overrides the `processEvent` to inject test related logic.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089870547
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -95,6 +96,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
while (true) {
try {
ShuffleDataFlushEvent event = flushQueue.take();
+ // Only for test
+ while (isSuspend) {
Review Comment:
Thanks for the investigation @advancedxy, I prefer the 2nd way. It's always better to improve testability.
If this test fails very often, we can merge this PR first and do the refactor 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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089877302
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -95,6 +96,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
while (true) {
try {
ShuffleDataFlushEvent event = flushQueue.take();
+ // Only for test
+ while (isSuspend) {
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089894549
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
+ while (endless || !flushQueue.isEmpty()) {
+ try {
+ ShuffleDataFlushEvent event = flushQueue.take();
+ threadPoolExecutor.execute(() -> {
+ processEvent(event);
+ });
+ } catch (Exception e) {
+ LOG.error("Exception happened when process event.", e);
+ }
+ }
+ }
+
+ protected void processEvent(ShuffleDataFlushEvent event) {
Review Comment:
This method can be private.
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
Maybe we can create another method `eventLoop()`:
```java
private void startEventProcesser() {
// the thread for flush data
Thread processEventThread = new Thread(this::eventLoop);
processEventThread.setName("ProcessEventThread");
processEventThread.setDaemon(true);
processEventThread.start();
}
protected void eventLoop() {
while (true) {
processEvents();
}
}
public void processEvents() {
while (!flushQueue.isEmpty()) {
try {
ShuffleDataFlushEvent event = flushQueue.take();
threadPoolExecutor.execute(() -> {
processEvent(event);
});
} catch (Exception e) {
LOG.error("Exception happened when process event.", e);
}
}
}
```
Then override it in `TestShuffleFlushManager`:
```java
@Override
protected void eventLoop() {
// do nothing
}
```
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089898414
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
I didn't mean to change the `flushQueue.take()`.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089902831
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +134,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents(boolean endless) {
Review Comment:
> For reference: https://github.com/kaijchen/incubator-uniffle/tree/issue_509_fork
We need overwrite `startEventProcessor` in `TestShuffleFlushManager`, Because if we start EventProcesser, events will be flushed automatically and this pr is used to prevent automatic flush.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089878670
##########
server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java:
##########
@@ -567,7 +568,25 @@ public void shuffleFlushThreshold() throws Exception {
StorageManager storageManager = StorageManagerFactory.getInstance().createStorageManager(conf);
ShuffleFlushManager shuffleFlushManager = new ShuffleFlushManager(conf,
- "serverId", mockShuffleServer, storageManager);
+ "serverId", mockShuffleServer, storageManager) {
+
+ @Override
+ protected void startEventProcesser() {
+ // do nothing
+ }
+
+ @Override
+ public void processEvents() {
Review Comment:
I think we shouldn't override `processEvents()`, because if `ShuffleFlushManager#processEvents()` is changed, the test will pass no matter if the change is 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: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] advancedxy commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089880578
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -137,6 +113,14 @@ public void run() {
thread.start();
}
+ protected void startEventProcesser() {
+ // the thread for flush data
+ Thread processEventThread = new Thread(() -> processEvents());
Review Comment:
In java8, method reference is preferred, such as: `new Thread(this::processEvents)`
##########
server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java:
##########
@@ -567,7 +568,25 @@ public void shuffleFlushThreshold() throws Exception {
StorageManager storageManager = StorageManagerFactory.getInstance().createStorageManager(conf);
ShuffleFlushManager shuffleFlushManager = new ShuffleFlushManager(conf,
- "serverId", mockShuffleServer, storageManager);
+ "serverId", mockShuffleServer, storageManager) {
+
+ @Override
+ protected void startEventProcesser() {
Review Comment:
I don't think we should override `startEventProcesser`?
The `processEvents` could be override with your previous change to the `ShuffleBufferManager`.
##########
server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java:
##########
@@ -567,7 +568,25 @@ public void shuffleFlushThreshold() throws Exception {
StorageManager storageManager = StorageManagerFactory.getInstance().createStorageManager(conf);
ShuffleFlushManager shuffleFlushManager = new ShuffleFlushManager(conf,
- "serverId", mockShuffleServer, storageManager);
+ "serverId", mockShuffleServer, storageManager) {
Review Comment:
I believe create an anonymous subclass is generally avoided in practice.
Better way would be creating a `TestShuffleFlushManager` class in a new file or as a private inner class.
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +129,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents() {
Review Comment:
this could be protect?
##########
server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java:
##########
@@ -567,7 +568,25 @@ public void shuffleFlushThreshold() throws Exception {
StorageManager storageManager = StorageManagerFactory.getInstance().createStorageManager(conf);
ShuffleFlushManager shuffleFlushManager = new ShuffleFlushManager(conf,
- "serverId", mockShuffleServer, storageManager);
+ "serverId", mockShuffleServer, storageManager) {
+
+ @Override
+ protected void startEventProcesser() {
+ // do nothing
+ }
+
+ @Override
+ public void processEvents() {
Review Comment:
`processEvents` in the main file is simple enough, it can be override. But we should override it with similar logic with the origin one.
This override seems to be quite different.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089885934
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -145,6 +129,31 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
+ public void processEvents() {
Review Comment:
No, it will be invoked in UT.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] kaijchen commented on pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "kaijchen (via GitHub)" <gi...@apache.org>.
kaijchen commented on PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#issuecomment-1407612891
Maybe it's better to not expose `flush()` to production code: https://github.com/kaijchen/incubator-uniffle/commit/cb9a0be5734825e9e89f62ff50052ddb32fd90b8
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089916797
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -81,40 +82,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
this.maxConcurrencyOfSingleOnePartition =
shuffleServerConf.get(ShuffleServerConf.SERVER_MAX_CONCURRENCY_OF_ONE_PARTITION);
- int waitQueueSize = shuffleServerConf.getInteger(
- ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_QUEUE_SIZE);
- BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(waitQueueSize);
- int poolSize = shuffleServerConf.getInteger(ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_SIZE);
- long keepAliveTime = shuffleServerConf.getLong(ShuffleServerConf.SERVER_FLUSH_THREAD_ALIVE);
- threadPoolExecutor = new ThreadPoolExecutor(poolSize, poolSize, keepAliveTime, TimeUnit.SECONDS, waitQueue,
- ThreadUtils.getThreadFactory("FlushEventThreadPool"));
+ threadPoolExecutor = createFlushEventExecutor();
storageBasePaths = shuffleServerConf.get(ShuffleServerConf.RSS_STORAGE_BASE_PATH);
pendingEventTimeoutSec = shuffleServerConf.getLong(ShuffleServerConf.PENDING_EVENT_TIMEOUT_SEC);
- // the thread for flush data
- Runnable processEventRunnable = () -> {
- while (true) {
- try {
- ShuffleDataFlushEvent event = flushQueue.take();
- threadPoolExecutor.execute(() -> {
- try {
- ShuffleServerMetrics.gaugeWriteHandler.inc();
- flushToFile(event);
- } catch (Exception e) {
- LOG.error("Exception happened when flush data for " + event, e);
- } finally {
- ShuffleServerMetrics.gaugeWriteHandler.dec();
- ShuffleServerMetrics.gaugeEventQueueSize.dec();
- }
- });
- } catch (Exception e) {
- LOG.error("Exception happened when process event.", e);
- }
- }
- };
- Thread processEventThread = new Thread(processEventRunnable);
- processEventThread.setName("ProcessEventThread");
- processEventThread.setDaemon(true);
- processEventThread.start();
+ startEventProcessor();
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089918777
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -81,40 +82,10 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
this.maxConcurrencyOfSingleOnePartition =
shuffleServerConf.get(ShuffleServerConf.SERVER_MAX_CONCURRENCY_OF_ONE_PARTITION);
- int waitQueueSize = shuffleServerConf.getInteger(
- ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_QUEUE_SIZE);
- BlockingQueue<Runnable> waitQueue = Queues.newLinkedBlockingQueue(waitQueueSize);
- int poolSize = shuffleServerConf.getInteger(ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_SIZE);
- long keepAliveTime = shuffleServerConf.getLong(ShuffleServerConf.SERVER_FLUSH_THREAD_ALIVE);
- threadPoolExecutor = new ThreadPoolExecutor(poolSize, poolSize, keepAliveTime, TimeUnit.SECONDS, waitQueue,
- ThreadUtils.getThreadFactory("FlushEventThreadPool"));
+ threadPoolExecutor = createFlushEventExecutor();
+ startEventProcessor();
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #510: [ISSUE-509] Fix Flaky Test: ShuffleBufferManagerTest#shuffleFlushThreshold
Posted by "xianjingfeng (via GitHub)" <gi...@apache.org>.
xianjingfeng commented on code in PR #510:
URL: https://github.com/apache/incubator-uniffle/pull/510#discussion_r1089688253
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -67,6 +67,7 @@ public class ShuffleFlushManager {
private final long pendingEventTimeoutSec;
private int processPendingEventIndex = 0;
private final int maxConcurrencyOfSingleOnePartition;
+ private boolean isSuspend;
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org