[#808] feat(spark): ensure thread safe and data consistency when spilling #848

zuston · 2023-04-28T03:01:19Z

What changes were proposed in this pull request?

Guarantees thread safe by only allowing spills to be triggered by the current thread
Using the same logic of processing blocks in the RssShuffleWriter and WriteBufferManager to ensure the data consistency

Why are the changes needed?

Fix: #808

In this PR, we use the two ways to solve the concurrent problem for addRecord and spill function

For the same thread, the spill will be invoked when adding records and unsuffcient memory. This case could ensure
thread safe. So it will do the spill sync.
When spill is invoked by other consumers, it will do nothing in this thread and just set a signal to let owner to release when adding record.

After this, we could avoid lock(may cause performance regression, like #811 did) to keep thread safe

Does this PR introduce any user-facing change?

No.

How was this patch tested?

UTs

codecov-commenter · 2023-04-28T03:10:59Z

Codecov Report

Merging #848 (448a5a9) into master (d6b43f0) will increase coverage by 1.28%.
The diff coverage is 90.90%.

@@             Coverage Diff              @@
##             master     #848      +/-   ##
============================================
+ Coverage     53.66%   54.94%   +1.28%     
- Complexity     2523     2539      +16     
============================================
  Files           382      362      -20     
  Lines         21672    19346    -2326     
  Branches       1795     1798       +3     
============================================
- Hits          11630    10630    -1000     
+ Misses         9338     8085    -1253     
+ Partials        704      631      -73

Impacted Files	Coverage Δ
...pache/spark/shuffle/writer/WriteBufferManager.java	`86.48% <89.74%> (+6.69%)`	⬆️
.../java/org/apache/spark/shuffle/RssSparkConfig.java	`98.67% <100.00%> (+0.03%)`	⬆️

... and 25 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

zuston · 2023-04-28T03:44:16Z

I propose a new way to keep thread-safe and data consistency. PTAL @jerqi

jerqi · 2023-04-28T05:56:53Z

Have you verify the patch in your production environment?

zuston · 2023-04-28T06:01:13Z

Have you verify the patch in your production environment?

No. I hope community could review this. If OK, I will put into prod env.

jerqi · 2023-04-28T06:31:32Z

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;


Why do we remove final?

Because the partial test cases use the original RssShuffleWriter constructor. To reduce the test changes, remove the final modifier.

You can raise a refactor pr after this pr. I prefer using final here.

No. This is not a refactor.

jerqi · 2023-04-28T06:38:52Z

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;


WIll it be accessed by multi threads?

Is it necessary？

No. Buffer will only be access in a single thread.

Is it necessary？

Yes. This is a mark to indicate the buffer has been processed.

Why do we need this?

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.

Do we change this value in the spill thread?

No, it won't

Any buffer flush operation in spill function only will do in the adding record thread.

When do we change this value?

Remove this part.

jerqi · 2023-04-28T06:43:44Z

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();


Why do we remove this?

Is it necessary?

Not. I could remove this change.

zuston · 2023-04-28T08:17:23Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

+    return sentBlocks;
+  }
+
+  /**


Do you have any thought on this part? @jerqi This is the main logic

jerqi · 2023-05-04T06:08:58Z

We would better have a config option to control whether to use this function. It seems a little complex.

zuston · 2023-05-04T06:19:46Z

We would better have a config option to control whether to use this function. It seems a little complex.

It's OK.

jerqi · 2023-05-04T06:24:03Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

+        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);


This seems bad code. It may cause more concurrency problems.

This will do in a thread. It's thread safe

Could we simplify the logic here? It's weird that we use recursion here.

+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.

Got it. I will check this PR in our internal env. Let's freeze this until all test finish.

jerqi · 2023-05-04T06:28:29Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

+    requestMemoryAndInsert(partitionId, serializedData, serializedDataLength, sentBlocks);
+    shuffleWriteMetrics.incRecordsWritten(1L);
+
+    if (usedBytes.get() - inSendListBytes.get() > spillSize || spillTriggered.compareAndSet(true, false)) {


It's weird synchroinization.

emm, sorry I don't get your point.

if SpillTrigger is syncrhonization value, why do we put the variable after ||? It's strange.

Nice catch. It's bug

You design too complex synchronization mechnism. It will be buggy usually.

Bring too much cost. Every record with one lock is unreasonable.

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.

Optimistic lock will. But I still think this is not complex. @advancedxy If you have time, could you help check this?

if SpillTrigger is syncrhonization value, why do we put the variable after ||? It's strange.

Done.

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.

jerqi · 2023-05-04T06:38:03Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

-    return 0L;
+    // If it is triggered by other consumers, it will spilled async by current consumer.
+    if (trigger != this) {
+      spillTriggered.set(true);


Will the variable accessed by multi threads?

Yes, It will

zuston · 2023-05-04T07:41:26Z

We would better have a config option to control whether to use this function. It seems a little complex.

Done

jerqi

In general, the pr don't propose a simple or clear method to solve the conflicts of multi threads. You can't use several atomic variables to guarantee the thread safe unless it's learned from other successful projects. It is verified by many production environments.

jerqi · 2023-05-04T08:35:10Z

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;


When do we change this value?

jerqi · 2023-05-04T08:38:08Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

+      // 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 -> {


How to handle the rpc failure?

jerqi · 2023-05-04T08:45:57Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

+      return 0L;
+    }
+
+    List<CompletableFuture<Long>> futures = spillFunc.apply(clear());


You will call the method clear in different threads. Is it thread safe?

If I'm correct, I think you don't understand this design. Please see the below explaintion.

zuston · 2023-05-04T09:00:00Z

I want to re-explain this design to make more people understand. @jerqi

As we know, in the master codebase of WriteBufferManager , 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.

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?
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.

jerqi · 2023-05-05T03:32:51Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

-        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,


The comment isn't clear enough. You should explain more information.

advancedxy · 2023-05-08T02:57:03Z

The idea seems simple enough and should work.
I'm a bit of concerning that whether would it work well for production cases, since it doesn't respond to other tasks' spill request.

But it's better than current impl.

…n spill

zuston · 2023-07-20T07:47:30Z

Sorry for the late reply. I have updated this code! PTAL @jerqi @advancedxy Thanks!

zuston · 2023-07-21T03:00:02Z

Although this has a long history, do you mind take a look? @leixm @smallzhongfeng @xianjingfeng

jerqi

LGTM. Now it's ok for me.

zuston · 2023-07-22T09:02:10Z

If nobody have any ideas, I will merge this in next 2 hour. Thanks ~

…sure data correctness (#1558) ### What changes were proposed in this pull request? Verify the number of written records to enhance data accuracy. Make sure all data records are sent by clients. Make sure bugs like #714 will never be introduced into the code. ### Why are the changes needed? A follow-up PR for #848. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.

zuston marked this pull request as draft April 28, 2023 03:11

zuston marked this pull request as ready for review April 28, 2023 03:43

zuston mentioned this pull request Apr 28, 2023

[#808] fix(client): use synchronized to make spill and addRecord thread-safe #811

Closed

jerqi reviewed Apr 28, 2023

View reviewed changes

zuston commented Apr 28, 2023

View reviewed changes

jerqi reviewed May 4, 2023

View reviewed changes

zuston requested a review from jerqi May 4, 2023 07:41

jerqi requested changes May 4, 2023

View reviewed changes

jerqi reviewed May 4, 2023

View reviewed changes

jerqi self-requested a review May 4, 2023 11:02

jerqi reviewed May 5, 2023

View reviewed changes

jerqi changed the title ~~[#808] fix(spark): ensure thread safe and data consistency when spilling~~ [#808] feat(spark): ensure thread safe and data consistency when spilling May 10, 2023

[apache#808] feat(spark): ensure thread safe and data consistency whe…

dbedf57

…n spill

zuston force-pushed the issue-808-2 branch from dde5255 to dbedf57 Compare July 20, 2023 06:51

checkstyle fix

448a5a9

zuston requested review from advancedxy and jerqi July 20, 2023 07:45

jerqi approved these changes Jul 21, 2023

View reviewed changes

zuston merged commit 03b04b9 into apache:master Jul 22, 2023

zuston deleted the issue-808-2 branch July 22, 2023 13:10

rickyma mentioned this pull request Mar 6, 2024

[#808] improvement(spark): Verify the number of written records to enhance data correctness #1558

Merged

[#808] feat(spark): ensure thread safe and data consistency when spilling #848

[#808] feat(spark): ensure thread safe and data consistency when spilling #848

Conversation

zuston commented Apr 28, 2023 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

codecov-commenter commented Apr 28, 2023 • edited Loading

Codecov Report

zuston commented Apr 28, 2023

jerqi commented Apr 28, 2023

zuston commented Apr 28, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zuston Apr 28, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerqi commented May 4, 2023

zuston commented May 4, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerqi May 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zuston commented May 4, 2023

jerqi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zuston commented May 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

advancedxy commented May 8, 2023 • edited Loading

zuston commented Jul 20, 2023

zuston commented Jul 21, 2023

jerqi left a comment

Choose a reason for hiding this comment

zuston commented Jul 22, 2023

zuston commented Apr 28, 2023 •

edited

Loading

codecov-commenter commented Apr 28, 2023 •

edited

Loading

zuston Apr 28, 2023 •

edited

Loading

jerqi May 4, 2023 •

edited

Loading

zuston commented May 4, 2023 •

edited

Loading

advancedxy commented May 8, 2023 •

edited

Loading