iframe-proxy

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

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

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

roryqi · 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.

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

  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.

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

  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.

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

    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

+    return sentBlocks;
+  }
+
+  /**


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

roryqi · 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.

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

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

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

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

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

-    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

roryqi

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.

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

  private List<WrappedBuffer> buffers = Lists.newArrayList();
  private int dataLength = 0;
  private int memoryUsed = 0;
+  private boolean isAvailable = false;


When do we change this value?

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

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

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

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

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

-        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

roryqi

LGTM. Now it's ok for me.

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

…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

roryqi reviewed Apr 28, 2023

View reviewed changes

zuston commented Apr 28, 2023

View reviewed changes

roryqi reviewed May 4, 2023

View reviewed changes

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

roryqi requested changes May 4, 2023

View reviewed changes

roryqi reviewed May 4, 2023

View reviewed changes

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

roryqi reviewed May 5, 2023

View reviewed changes

roryqi 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 roryqi July 20, 2023 07:45

roryqi 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

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Conversation

zuston commented Apr 28, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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?

Uh oh!

codecov-commenter commented Apr 28, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

zuston commented Apr 28, 2023

Uh oh!

roryqi commented Apr 28, 2023

Uh oh!

zuston commented Apr 28, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zuston Apr 28, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

roryqi commented May 4, 2023

Uh oh!

zuston commented May 4, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

zuston commented Apr 28, 2023 •

edited

Loading

codecov-commenter commented Apr 28, 2023 •

edited

Loading

zuston Apr 28, 2023 •

edited

Loading

roryqi May 4, 2023 •

edited

Loading