[ISSUE-380] Refactor the flush process to fix fallback fail #383

zuston · 2022-12-02T03:25:34Z

What changes were proposed in this pull request?

Avoid partial events being pushed into pending queue when using MultipleStorageManager
Introduce the cleanupCallback of ShuffleDataFlushEvent to scope all cleanup operation with event lifecycle.
Make the concrete under storage bound to data-flush event to avoid duplicate invoking selectStorage method after writing.
Refactor the whole flush process to a while loop, more simple and clear.
Add the metric of total_failed_written_event_num

Why are the changes needed?

To fix the fallback invalid when data-flush event enters into pending queue

Does this PR introduce any user-facing change?

No

How was this patch tested?

Simple UTs
Integration tests

jerqi · 2022-12-02T03:28:12Z

server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java

  private final ShuffleBuffer shuffleBuffer;
  private final AtomicInteger retryTimes = new AtomicInteger();
+  private boolean isPended = false;
+  private StorageManager underStorageManager;


Could we avoid using storageManager in a ShuffleDataFlushEvent? It's a little weird.

Yes. I also want to avoid this, but if we don't, we have to maintain this relation mapping in multipleStorageManager. However we couldn't remove this structure when event's lifecycle finished, because we don't have similar interface in storage manager

Could we only have Storage field. Could we use Storage to find its StorageManager?

Currently it can't. But this solution sounds great. We could extend the interface of storage

Currently it can't. But this solution sounds great. We could extend the interface of storage

I think we need add a interface for StorageManager#isBelongto(Storage storage). Storage is in the module storage. StorageManager is in the module server. Storage shouldn't depend on StorageManager.

Emmm…. I still think cache is not suitable. Because it holds the reference of events and will cause memory leak. We should implicitly remove this event reference when event is flushed.

It's not acceptable for me to put StorageManager into ShuffleDataFlushEvent. I would rather give up this strategy.

I know. I don’t insist on keeping this into event. I just explain the cache is also unreasonable.

And this is also a potential bug due to memory leak.

Could we add a callback for ShuffleFlushEvent. In the callback, it will remove the data. When we call the cleanupFlushEventData(event) method. We call event.clearCallback();

Sounds great!

The relation will help event find the concrete under-storage-manager selected by method of selectStorage when invoked by write method. Right? @xianjingfeng

Yes, if we remove pending queue, we don't need maintain this relation.

jerqi · 2022-12-02T09:49:42Z

We should refactor the ShuffleFlushEvent. We need a method getShuffleWriteHandler(ShuffleFlushEvent). So we can remove Storage.getCreateWriterHandlerRequest().

jerqi · 2022-12-02T09:54:55Z

To implement what you want
We just modify here

  private StorageManager selectStorageManager(ShuffleDataFlushEvent event) {
    StorageManager storageManager = storageManagerCache.getIfPresent(event);
    if (storageManager != null) {
      // modify here if storageManager is not canWrite we fallback.
      return storageManager;
    }
    if (event.getSize() > flushColdStorageThresholdSize) {
      storageManager = coldStorageManager;
    } else {
      storageManager = warmStorageManager;
    }

    if (!storageManager.canWrite(event)) {
      storageManager = storageManagerFallbackStrategy.tryFallback(
          storageManager, event, warmStorageManager, coldStorageManager);
    }
    storageManagerCache.put(event, storageManager);
    return storageManager;
  }

codecov-commenter · 2022-12-02T09:56:27Z

Codecov Report

Merging #383 (48082a4) into master (897613b) will increase coverage by 0.18%.
The diff coverage is 64.89%.

@@             Coverage Diff              @@
##             master     #383      +/-   ##
============================================
+ Coverage     58.59%   58.78%   +0.18%     
- Complexity     1589     1602      +13     
============================================
  Files           193      193              
  Lines         10910    10938      +28     
  Branches        957      955       -2     
============================================
+ Hits           6393     6430      +37     
+ Misses         4139     4131       -8     
+ Partials        378      377       -1

Impacted Files	Coverage Δ
...rg/apache/uniffle/storage/common/LocalStorage.java	`43.22% <0.00%> (-0.57%)`	⬇️
...orage/handler/impl/LocalFileClientReadHandler.java	`61.29% <ø> (+1.91%)`	⬆️
...org/apache/uniffle/server/ShuffleFlushManager.java	`78.01% <61.11%> (-1.14%)`	⬇️
...g/apache/uniffle/server/ShuffleDataFlushEvent.java	`83.33% <66.66%> (-16.67%)`	⬇️
...he/uniffle/server/storage/MultiStorageManager.java	`64.15% <66.66%> (+16.38%)`	⬆️
...rg/apache/uniffle/server/ShuffleServerMetrics.java	`97.19% <100.00%> (+0.96%)`	⬆️
...rg/apache/uniffle/server/buffer/ShuffleBuffer.java	`93.38% <100.00%> (+0.05%)`	⬆️
...he/uniffle/server/buffer/ShuffleBufferManager.java	`82.74% <100.00%> (+0.06%)`	⬆️
...che/uniffle/server/storage/HdfsStorageManager.java	`95.23% <100.00%> (+0.15%)`	⬆️
...he/uniffle/server/storage/LocalStorageManager.java	`91.24% <100.00%> (+0.06%)`	⬆️
... and 10 more

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

zuston · 2022-12-02T10:00:22Z

To implement what you want We just modify here

  private StorageManager selectStorageManager(ShuffleDataFlushEvent event) {
    StorageManager storageManager = storageManagerCache.getIfPresent(event);
    if (storageManager != null) {
      // modify here if storageManager is not canWrite we fallback.
      return storageManager;
    }
    if (event.getSize() > flushColdStorageThresholdSize) {
      storageManager = coldStorageManager;
    } else {
      storageManager = warmStorageManager;
    }

    if (!storageManager.canWrite(event)) {
      storageManager = storageManagerFallbackStrategy.tryFallback(
          storageManager, event, warmStorageManager, coldStorageManager);
    }
    storageManagerCache.put(event, storageManager);
    return storageManager;
  }

Make sense.

zuston · 2022-12-02T10:00:49Z

We should refactor the ShuffleFlushEvent. We need a method getShuffleWriteHandler(ShuffleFlushEvent). So we can remove Storage.getCreateWriterHandlerRequest().

For what?

jerqi · 2022-12-02T10:32:44Z

We should refactor the ShuffleFlushEvent. We need a method getShuffleWriteHandler(ShuffleFlushEvent). So we can remove Storage.getCreateWriterHandlerRequest().

For what?

I think Storage.getOrCreateWriterHandlerRequest isn't reasonable.

zuston · 2022-12-02T10:58:03Z

I think Storage.getOrCreateWriterHandlerRequest isn't reasonable.

This is not related with this PR. Let's focus on this.

jerqi · 2022-12-02T11:42:10Z

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

+            ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+            continue;
+          } else {
+            if (event.isPended()) {


Why do we need isPended?

jerqi · 2022-12-02T11:43:07Z

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

-        } else if (!event.isValid()) {
-          //  avoid printing error log
+
+    while (true) {


Why do need while(true)?

If not, how to handle?

while (retryTimes < maxRetry)

For me, it is the same with this PR’s current implementation.

OK, the exit condition in while will be more clear.

I will correct it

jerqi · 2022-12-02T11:53:39Z

We should refactor the ShuffleFlushEvent. We need a method getShuffleWriteHandler(ShuffleFlushEvent). So we can remove Storage.getCreateWriterHandlerRequest().

For what?

I think Storage.getOrCreateWriterHandlerRequest isn't reasonable.

My mistake. It's necessary.

jerqi · 2022-12-02T12:09:14Z

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

    long start = System.currentTimeMillis();
-    List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
    boolean writeSuccess = false;
-    try {


Why do we need to modify the flush process?

I think this logic need to be refactored.

jerqi · 2022-12-02T12:12:16Z

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

              ShuffleServerMetrics.gaugeEventQueueSize.set(flushQueue.size());
              ShuffleServerMetrics.gaugeWriteHandler.inc();
-              flushToFile(event);
+              flushToFileImpl(event);


Why do we rename to flushToFIleImpl?
Usually we use Impl method , because we have already had a interface method.

Yes. I’ll correct it

jerqi · 2022-12-02T12:16:01Z

If we can't make the process clear, I would better to remove pending queue. There are too many bugs caused by pending queue.

server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java

zuston · 2022-12-02T12:31:38Z

If we can't make the process clear, I would better to remove pending queue. There are too many bugs caused by pending queue.

Removing is OK for me.

jerqi · 2022-12-02T12:33:40Z

If we can't make the process clear, I would better to remove pending queue. There are too many bugs caused by pending queue.

Removing is OK for me.

Let's remove it. It's more simple.

jerqi · 2022-12-02T13:07:02Z

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

+        }
+
+        if (!storage.canWrite()) {
+          if (storageManager instanceof MultiStorageManager) {


Could we avoid judging the MultiStorageManager?

If we want to remove pending queue, this is unnecessary.

Could we add an interface supportPending for storageManager?

Emm... It's weird, let me think again.

Let us leave a todo comment for later optimization, and make current PR small.

Is there any process about this optimization?

After supporting multiple disk selection #435 , I think we could remove this pending queue.

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java

jerqi · 2022-12-06T06:04:34Z

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java


+    if (event.getRetryTimes() > retryMax) {
+      LOG.error("Failed to write data for {} in {} times, shuffle data will be lost", event, retryMax);
+      ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());


getUnderStorage may be null.

Nice catch. Fixed.

jerqi

Except for MultiStorageManager, LGTM.

jerqi · 2022-12-06T07:18:23Z

Merge. thanks @zuston

…rs cache (#1627) ### What changes were proposed in this pull request? Remove the meaningless eventOfUnderStorageManagers. ### Why are the changes needed? Fix #1626 & #1620. It's also a follow-up PR for #383. This cache only makes sense when the event retries after a failure. However, after the event fails, it is not appropriate to continue taking the original storageManager from the cache(because events usually fail due to high IO pressure or disk damage or disk full). In this case, the cache seems to be meaningless, so there is a contradiction here, we should remove it. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Tested in our env

[BUG] Overlook fallback when data-flush event enters into pending queue

30f6b9b

zuston requested review from jerqi and xianjingfeng December 2, 2022 03:25

jerqi reviewed Dec 2, 2022

View reviewed changes

zuston added 5 commits December 2, 2022 14:09

fix test bug

df1281c

Add test cases

1e63047

Add more tests

1ddfe9c

fix

8be9265

remove

42f883f

fix

31cbb9b

zuston changed the title ~~[WIP][BUG] Overlook fallback when data-flush event enters into pending queue~~ [BUG] Overlook fallback when data-flush event enters into pending queue Dec 2, 2022

reuse the cache

09c143e

jerqi reviewed Dec 2, 2022

View reviewed changes

server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java Show resolved Hide resolved

jerqi reviewed Dec 2, 2022

View reviewed changes

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java Show resolved Hide resolved

zuston added 3 commits December 3, 2022 10:57

Use the callback to cleanup data-flush-event

ead6348

optimize while exit condition

6e7c46f

fix size

896955a

zuston changed the title ~~[BUG] Overlook fallback when data-flush event enters into pending queue~~ [ISSUE-380] Refactor the flush process to fix fallback fail Dec 3, 2022

zuston added 3 commits December 4, 2022 20:19

Invalidate client cache to avoid being closed

25ba191

add the metric of total_failed_written_event_num and add comments

2aa338a

fix bug

e0d6740

jerqi reviewed Dec 4, 2022

View reviewed changes

internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java Show resolved Hide resolved

jerqi reviewed Dec 4, 2022

View reviewed changes

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java Outdated Show resolved Hide resolved

jerqi reviewed Dec 4, 2022

View reviewed changes

server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java Outdated Show resolved Hide resolved

zuston added 3 commits December 5, 2022 16:56

fix comments

58bfe2e

fix

9fa503e

Make event callback function as runnable

992dcb3

zuston requested a review from jerqi December 6, 2022 05:50

jerqi reviewed Dec 6, 2022

View reviewed changes

fix

48082a4

jerqi approved these changes Dec 6, 2022

View reviewed changes

jerqi merged commit c09cbd1 into apache:master Dec 6, 2022

zuston mentioned this pull request Dec 8, 2022

[Bug] Overlook fallback when data-flush event enters into pending queue #380

Closed

3 tasks

This was referenced Apr 8, 2024

[Bug] The StorageManager cache might not function effectively under heavy IO pressure #1626

Closed

[#1626] fix(server): Remove the meaningless eventOfUnderStorageManagers cache #1627

Merged

[ISSUE-380] Refactor the flush process to fix fallback fail #383

[ISSUE-380] Refactor the flush process to fix fallback fail #383

Uh oh!

Conversation

zuston commented Dec 2, 2022 • 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!

Choose a reason for hiding this comment

Uh oh!

zuston Dec 2, 2022 • 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!

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!

jerqi commented Dec 2, 2022

Uh oh!

jerqi commented Dec 2, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov-commenter commented Dec 2, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

zuston commented Dec 2, 2022

Uh oh!

zuston commented Dec 2, 2022

Uh oh!

jerqi commented Dec 2, 2022

Uh oh!

zuston commented Dec 2, 2022

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!

jerqi commented Dec 2, 2022

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!

jerqi commented Dec 2, 2022

Uh oh!

Uh oh!

zuston commented Dec 2, 2022

Uh oh!

jerqi commented Dec 2, 2022

Uh oh!

Choose a reason for hiding this comment

zuston commented Dec 2, 2022 •

edited

Loading

zuston Dec 2, 2022 •

edited

Loading

jerqi commented Dec 2, 2022 •

edited

Loading

codecov-commenter commented Dec 2, 2022 •

edited

Loading

zuston Dec 3, 2022 •

edited

Loading