You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2023/04/28 03:01:22 UTC
[GitHub] [incubator-uniffle] zuston opened a new pull request, #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
zuston opened a new pull request, #848:
URL: https://github.com/apache/incubator-uniffle/pull/848
### What changes were proposed in this pull request?
(Please outline the changes and how this PR fixes the issue.)
### 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, describe the bug.)
Fix: # (issue)
### Does this PR introduce _any_ user-facing change?
(Please list the user-facing changes introduced by your change, including
1. Change in user-facing APIs.
2. Addition or removal of property keys.)
No.
### How was this patch tested?
(Please test your changes, and provide instructions on how to test it:
1. If you add a feature or fix a bug, add a test to cover your changes.
2. If you fix a flaky test, repeat it for many times to prove it works.)
--
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] zuston commented on pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1526942596
I propose a new way to keep thread-safe and data consistency. PTAL @jerqi
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184589092
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
It's weird synchroinization.
--
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 #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1526920970
## [Codecov](https://codecov.io/gh/apache/incubator-uniffle/pull/848?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 [#848](https://codecov.io/gh/apache/incubator-uniffle/pull/848?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (44315da) into [master](https://codecov.io/gh/apache/incubator-uniffle/commit/8edefdf84c4393f3e8faf7d58b2fb93d02ac975c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (8edefdf) will **increase** coverage by `1.62%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #848 +/- ##
============================================
+ Coverage 57.14% 58.77% +1.62%
+ Complexity 2164 2041 -123
============================================
Files 323 289 -34
Lines 15722 12474 -3248
Branches 1247 1172 -75
============================================
- Hits 8984 7331 -1653
+ Misses 6232 4740 -1492
+ Partials 506 403 -103
```
[see 42 files with indirect coverage changes](https://codecov.io/gh/apache/incubator-uniffle/pull/848/indirect-changes?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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180080935
##########
client-spark/spark3/src/test/java/org/apache/spark/shuffle/writer/RssShuffleWriterTest.java:
##########
@@ -289,7 +355,6 @@ public void writeTest() throws Exception {
assertEquals(2, partitionToBlockIds.get(0).size());
assertEquals(2, partitionToBlockIds.get(2).size());
partitionToBlockIds.clear();
- sc.stop();
Review Comment:
Not. I could remove 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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1179982397
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java:
##########
@@ -65,7 +69,7 @@
private final String appId;
private final int shuffleId;
- private final WriteBufferManager bufferManager;
+ private WriteBufferManager bufferManager;
Review Comment:
Why do we remove `final`?
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
WIll it be accessed by multi threads?
##########
client-spark/spark3/src/test/java/org/apache/spark/shuffle/writer/RssShuffleWriterTest.java:
##########
@@ -289,7 +355,6 @@ public void writeTest() throws Exception {
assertEquals(2, partitionToBlockIds.get(0).size());
assertEquals(2, partitionToBlockIds.get(2).size());
partitionToBlockIds.clear();
- sc.stop();
Review Comment:
Why do we remove this?
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180009836
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
No. Buffer will only be access in a single 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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184585162
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
Do we change this value in the spill 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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184708607
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
> Optimistic lock will. But I still think this is not complex. @advancedxy If you have time, could you help check this?
In my opinion, there is a bug, it reflects that the mechinsm is too complex.
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184602314
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
Why don't we use lock directly?
--
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] zuston commented on a diff in pull request #848: [#808] feat(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1269068025
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
Remove this part.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184611254
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
Optimistic lock will. But I still think this is not complex. @advancedxy If you have time, could you help check this?
--
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] zuston commented on pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1534344625
I want to re-explain this design to make more people understand. @jerqi
As we know, in the origin codebase, concurrency doesn't exist. Buffer flushing operation will only be acted in the processing of adding records, which is owned by one thread.
But when we want to flush buffer to release memory in spill method, it's thread unsafe. Because spill function will be invoked by `TaskMemoryManager` . So does it means the spill method and addRecord are acted in the different threads?
I think no. Reasons are as follows.
1. When adding records, if memory is tight, it will request memory from `TaskMemoryManager` . And then the free memory is empty, manager will spill the underlying `MemoryConsumer` . So the current `WriteBufferManager` 's spill method will be invoked in this process. From this example, the two methods will be invoked in the same thread, so this is thread safe. Right?
2. If the current `WriteBufferManager` 's spill method is invoked by `TaskMemoryManager` , due to others' memory consumers' unsufficient memory. This is thread unsafe.
Based on above examples, for one `WriteBufferManager` , if one consumer is not itself, it should avoid doing flushing buffer due to the concurrency problem of adding records. If the consumer is itself, it could doing buffer flushing, it's thread safe.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184653943
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
> if `SpillTrigger` is syncrhonization value, why do we put the variable after `||`? It's strange.
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184715672
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -321,7 +350,36 @@ public List<AddBlockEvent> buildBlockEvents(List<ShuffleBlockInfo> shuffleBlockI
@Override
public long spill(long size, MemoryConsumer trigger) {
- return 0L;
+ if (!memorySpillEnabled) {
+ return 0L;
+ }
+
+ // If it is triggered by other consumers, it will spilled async by current consumer.
+ if (trigger != this) {
+ spillTriggered.set(true);
+ return 0L;
+ }
+
+ List<CompletableFuture<Long>> futures = spillFunc.apply(clear());
+ CompletableFuture<Void> allOfFutures =
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
+ try {
+ allOfFutures.get(memorySpillTimeoutSec, TimeUnit.SECONDS);
+ } catch (TimeoutException timeoutException) {
+ // A best effort strategy to wait.
+ // If timeout exception occurs, the underlying tasks won't be cancelled.
+ } finally {
+ long releasedSize = futures.stream().filter(x -> x.isDone()).mapToLong(x -> {
Review Comment:
How to handle the rpc failure?
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
When do we change this value?
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184586702
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
No, it won't
Any buffer flush operation in spill function only will do in the adding record 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] zuston commented on pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1534148472
> We would better have a config option to control whether to use this function. It seems a little complex.
It's OK.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184599137
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
I think it's simple that all sync operation has been changed to one thread to reduce cost of sync.
The key point is to send signal to the thread of adding records when spilling in another thread.
Feel free to propose your idea
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184742935
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -321,7 +350,36 @@ public List<AddBlockEvent> buildBlockEvents(List<ShuffleBlockInfo> shuffleBlockI
@Override
public long spill(long size, MemoryConsumer trigger) {
- return 0L;
+ if (!memorySpillEnabled) {
+ return 0L;
+ }
+
+ // If it is triggered by other consumers, it will spilled async by current consumer.
+ if (trigger != this) {
+ spillTriggered.set(true);
+ return 0L;
+ }
+
+ List<CompletableFuture<Long>> futures = spillFunc.apply(clear());
Review Comment:
If I'm correct, I think you don't understand this design. Please see the below explaintion.
--
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] zuston commented on pull request #848: [#808] feat(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1646534871
If nobody have any ideas, I will merge this in next 2 hour. 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: 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] zuston commented on pull request #848: [#808] feat(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1644918210
Although this has a long history, do you mind take a look? @leixm @smallzhongfeng @xianjingfeng
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184605370
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
Bring too much cost. Every record with one lock is unreasonable.
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184609261
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
> Bring too much cost. Every record with one lock is unreasonable.
If we don't have lock conflicts, it won't bring too much cost. It is about several nano seconds.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180010998
##########
client-spark/spark3/src/test/java/org/apache/spark/shuffle/writer/RssShuffleWriterTest.java:
##########
@@ -289,7 +355,6 @@ public void writeTest() throws Exception {
assertEquals(2, partitionToBlockIds.get(0).size());
assertEquals(2, partitionToBlockIds.get(2).size());
partitionToBlockIds.clear();
- sc.stop();
Review Comment:
No necessary.
##########
client-spark/spark3/src/test/java/org/apache/spark/shuffle/writer/RssShuffleWriterTest.java:
##########
@@ -289,7 +355,6 @@ public void writeTest() throws Exception {
assertEquals(2, partitionToBlockIds.get(0).size());
assertEquals(2, partitionToBlockIds.get(2).size());
partitionToBlockIds.clear();
- sc.stop();
Review Comment:
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180009132
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java:
##########
@@ -65,7 +69,7 @@
private final String appId;
private final int shuffleId;
- private final WriteBufferManager bufferManager;
+ private WriteBufferManager bufferManager;
Review Comment:
Because the partial test cases use the original `RssShuffleWriter` constructor. To reduce the test changes, remove the `final` modifier.
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184596969
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
You design too complex synchronization mechnism. It will be buggy usually.
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1185670840
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
+ // we should re-insert the record.
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
Review Comment:
Could we simplify the logic here? It's weird that we use recursion 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: 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 #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1537665314
The idea seems simple enough and should work.
I'm a bit of concerning that whether would it work sell for production cases, since it doesn't respond to other tasks' spill request.
But it's better than current impl.
--
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] zuston commented on pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1534232761
> We would better have a config option to control whether to use this function. It seems a little complex.
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] zuston commented on pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1527031441
> Have you verify the patch in your production environment?
No. I hope community could review this. If OK, I will put into prod env.
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180008620
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
Is it necessary？
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180081552
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
Review Comment:
Do you have any thought on this part? @jerqi This is the main 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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180079890
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
When adding record, the memory is not sufficient. This will trigger the spill operation. Once buffer is flushed to remote server, the current buffer held by the writer should be dropped. So this is a mark for write to analyze whether it is available.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184595609
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
Nice catch. It's bug
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184595638
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -321,7 +349,32 @@ public List<AddBlockEvent> buildBlockEvents(List<ShuffleBlockInfo> shuffleBlockI
@Override
public long spill(long size, MemoryConsumer trigger) {
- return 0L;
+ // If it is triggered by other consumers, it will spilled async by current consumer.
+ if (trigger != this) {
+ spillTriggered.set(true);
Review Comment:
Will the variable accessed by multi threads?
--
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] zuston merged pull request #848: [#808] feat(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston merged PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848
--
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] zuston commented on pull request #848: [#808] feat(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1643442367
Sorry for the late reply. I have updated this code! PTAL @jerqi @advancedxy 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: 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] zuston commented on a diff in pull request #848: [#808] feat(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1269068483
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
+ // we should re-insert the record.
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
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] advancedxy commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1186981887
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
+ // we should re-insert the record.
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
Review Comment:
+1. This logic should be simplified.
It seems obvious that it could be split into two functions:
`request memory`: wait until enough memory is spilled.
`insert records`: only insert records when memory is satisfied.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1185677770
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +174,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
Review Comment:
Got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184596637
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -321,7 +349,32 @@ public List<AddBlockEvent> buildBlockEvents(List<ShuffleBlockInfo> shuffleBlockI
@Override
public long spill(long size, MemoryConsumer trigger) {
- return 0L;
+ // If it is triggered by other consumers, it will spilled async by current consumer.
+ if (trigger != this) {
+ spillTriggered.set(true);
Review Comment:
Yes, It will
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184594158
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
if `SpillTrigger` is syncrhonization value, why do we put the variable after `||`? It's strange.
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184586224
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
+ // we should re-insert the record.
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
Review Comment:
This seems bad code. It may cause more concurrency problems.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184587815
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
+ // we should re-insert the record.
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
Review Comment:
This will do in a thread. It's thread safe
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184725101
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -321,7 +350,36 @@ public List<AddBlockEvent> buildBlockEvents(List<ShuffleBlockInfo> shuffleBlockI
@Override
public long spill(long size, MemoryConsumer trigger) {
- return 0L;
+ if (!memorySpillEnabled) {
+ return 0L;
+ }
+
+ // If it is triggered by other consumers, it will spilled async by current consumer.
+ if (trigger != this) {
+ spillTriggered.set(true);
+ return 0L;
+ }
+
+ List<CompletableFuture<Long>> futures = spillFunc.apply(clear());
Review Comment:
You will call the method `clear` in different threads. Is it thread safe?
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1185671179
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +174,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
Review Comment:
The comment isn't clear enough. You should explain more information.
--
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] jerqi commented on pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1534140906
We would better have a config option to control whether to use this function. It seems a little complex.
--
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] jerqi commented on pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#issuecomment-1527027881
Have you verify the patch in your production environment?
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180080334
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java:
##########
@@ -65,7 +69,7 @@
private final String appId;
private final int shuffleId;
- private final WriteBufferManager bufferManager;
+ private WriteBufferManager bufferManager;
Review Comment:
No. This is not a refactor.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1188350251
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
+ sentBlocks.addAll(clear());
+ }
+ writeTime += System.currentTimeMillis() - start;
+ return sentBlocks;
+ }
+
+ /**
+ * Before inserting a record into its corresponding buffer, the system should check if there is
+ * sufficient buffer memory available. If there isn't enough memory, it will request additional
+ * memory from the {@link TaskMemoryManager}. In the event that the JVM is low on memory,
+ * a spill operation will be triggered. If any memory consumer managed by the {@link TaskMemoryManager}
+ * fails to meet its memory requirements, it will also be triggered one by one.
+ *
+ * If the current buffer manager requests memory and triggers a spill operation,
+ * the buffer that is currently being held should be dropped, and then re-inserted.
+ */
+ private void requestMemoryAndInsert(int partitionId, byte[] serializedData, int serializedDataLength,
+ List<ShuffleBlockInfo> sentBlocks) {
+ long required = Math.max(bufferSegmentSize, serializedDataLength);
if (buffers.containsKey(partitionId)) {
WriterBuffer wb = buffers.get(partitionId);
if (wb.askForMemory(serializedDataLength)) {
- requestMemory(Math.max(bufferSegmentSize, serializedDataLength));
+ requestMemory(required);
+ if (wb.isAvailable()) {
+ // The readonly of writer buffer means the data has been sent when spilling. To avoid data lost,
+ // we should re-insert the record.
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
Review Comment:
Got it. I will check this PR in our internal env. Let's freeze this until all test finish.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184590703
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
emm, sorry I don't get your point.
--
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] zuston commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180010608
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
> Is it necessary？
Yes. This is a mark to indicate the buffer has been processed.
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1180074700
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java:
##########
@@ -65,7 +69,7 @@
private final String appId;
private final int shuffleId;
- private final WriteBufferManager bufferManager;
+ private WriteBufferManager bufferManager;
Review Comment:
You can raise a refactor pr after this pr. I prefer using final here.
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriterBuffer.java:
##########
@@ -33,6 +33,7 @@ public class WriterBuffer {
private List<WrappedBuffer> buffers = Lists.newArrayList();
private int dataLength = 0;
private int memoryUsed = 0;
+ private boolean isAvailable = false;
Review Comment:
Why do we need this?
##########
client-spark/spark3/src/test/java/org/apache/spark/shuffle/writer/RssShuffleWriterTest.java:
##########
@@ -289,7 +355,6 @@ public void writeTest() throws Exception {
assertEquals(2, partitionToBlockIds.get(0).size());
assertEquals(2, partitionToBlockIds.get(2).size());
partitionToBlockIds.clear();
- sc.stop();
Review Comment:
Is it necessary?
--
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] jerqi commented on a diff in pull request #848: [#808] fix(spark): ensure thread safe and data consistency when spilling
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #848:
URL: https://github.com/apache/incubator-uniffle/pull/848#discussion_r1184596969
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java:
##########
@@ -170,35 +175,59 @@ public List<ShuffleBlockInfo> addRecord(int partitionId, Object key, Object valu
if (serializedDataLength == 0) {
return null;
}
- List<ShuffleBlockInfo> result = Lists.newArrayList();
+
+ List<ShuffleBlockInfo> sentBlocks = Lists.newArrayList();
+ requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+ shuffleWriteMetrics.incRecordsWritten(1L);
+
+ if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {
Review Comment:
You design too complex synchronization mechnism. It will be buggy.
--
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