[#757] feat(server): separate flush thread pools for different storage types #775

leixm · 2023-03-28T12:48:24Z

What changes were proposed in this pull request?

Separate flush thread pools for different storage type.

Why are the changes needed?

Writing local files requires less concurrency, while writing hdfs requires more concurrency, it is best to separate thread pools.

Does this PR introduce any user-facing change?

Yes.

How was this patch tested?

existing UTs.

codecov-commenter · 2023-03-28T13:22:03Z

Codecov Report

Merging #775 (1a28453) into master (b759588) will increase coverage by 1.21%.
The diff coverage is 74.24%.

@@             Coverage Diff              @@
##             master     #775      +/-   ##
============================================
+ Coverage     53.57%   54.78%   +1.21%     
- Complexity     2511     2518       +7     
============================================
  Files           378      359      -19     
  Lines         20528    18170    -2358     
  Branches       1773     1776       +3     
============================================
- Hits          10997     9955    -1042     
+ Misses         8835     7585    -1250     
+ Partials        696      630      -66

Impacted Files	Coverage Δ
...org/apache/uniffle/server/ShuffleFlushManager.java	`76.47% <58.20%> (-5.77%)`	⬇️
...pache/uniffle/server/DefaultFlushEventHandler.java	`88.88% <88.88%> (ø)`
...g/apache/uniffle/server/ShuffleDataFlushEvent.java	`94.54% <100.00%> (+3.97%)`	⬆️
...a/org/apache/uniffle/server/ShuffleServerConf.java	`99.43% <100.00%> (+<0.01%)`	⬆️
...rg/apache/uniffle/server/ShuffleServerMetrics.java	`95.83% <100.00%> (+0.08%)`	⬆️

... and 23 files with indirect coverage changes

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

jerqi · 2023-03-28T14:22:23Z

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

@@ -118,14 +123,24 @@ private void startEventProcessor() {
    processEventThread.start();
  }

-  protected Executor createFlushEventExecutor() {
+  protected void initFlushEventExecutor() {


We shouldn't expose the concept of storageType in the FlushManager.

zuston · 2023-03-29T02:01:38Z

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

@@ -145,7 +160,14 @@ protected void eventLoop() {
  protected void processNextEvent() {
    try {
      ShuffleDataFlushEvent event = flushQueue.take();
-      threadPoolExecutor.execute(() -> processEvent(event));
+      Storage storage = storageManager.selectStorage(event);


It looks really strange for selecting storage here, which will select again before doing flush.

And the selected storage will be changed if writing failed. I think we'd better refactor flushing part.

It looks really strange for selecting storage here, which will select again before doing flush.

And the selected storage will be changed if writing failed. I think we'd better refactor flushing part.

You're right.

I think that the storageType should not be switched in the flushToFile method. When the writing fails or the storage is corrupted, the event should be put into pendingEvents, and flushToFile should be performed again.

leixm · 2023-03-29T11:31:54Z

It seems to be caused by concurrency issues, which cannot be reproduced in the development environment.

jerqi · 2023-03-29T14:31:15Z

It seems to be caused by concurrency issues, which cannot be reproduced in the development environment.

Maybe you can read the logs of ut. Do you know how to find the logs of ut?

leixm · 2023-03-30T07:37:59Z

@jerqi @smallzhongfeng PTAL.

leixm · 2023-03-30T07:38:19Z

It seems to be caused by concurrency issues, which cannot be reproduced in the development environment.

Maybe you can read the logs of ut. Do you know how to find the logs of ut?

Yes, i know, already fixed.

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

smallzhongfeng · 2023-03-30T07:42:39Z

server/src/test/java/org/apache/uniffle/server/buffer/ShuffleBufferManagerTest.java

@@ -587,6 +587,7 @@ public void shuffleFlushThreshold() throws Exception {
    shuffleBufferManager.cacheShuffleData(appId, smallShuffleId, false, createData(0, 31));
    assertEquals(96 + 63, shuffleBufferManager.getUsedMemory());
    shuffleFlushManager.flush();
+    Thread.sleep(100);


Why do we need to sleep here?

server/src/test/java/org/apache/uniffle/server/TestShuffleFlushManager.java#createFlushEventExecutor method make here sync, so we do not sleep here before.

smallzhongfeng · 2023-03-30T07:43:31Z

server/src/test/java/org/apache/uniffle/server/ShuffleFlushManagerTest.java

+    shuffleServerConf.set(RssBaseConf.RSS_STORAGE_TYPE, StorageType.LOCALFILE_HDFS.toString());
+    shuffleServerConf.set(RssBaseConf.RSS_STORAGE_BASE_PATH, Arrays.asList(tempDir.getAbsolutePath()));
+    shuffleServerConf.set(ShuffleServerConf.DISK_CAPACITY, 100L);
+    shuffleServerConf.set(ShuffleServerConf.SERVER_FLUSH_HDFS_THREAD_POOL_SIZE, 1);


Maybe we can change it to default value now.

zuston · 2023-03-30T11:01:49Z

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

+import org.apache.uniffle.storage.common.Storage;
+import org.apache.uniffle.storage.util.StorageType;
+
+public class StorageTypeFlushEventHandler implements FlushEventHandler {


Rename to DefaultFlushEventHandler ?

zuston · 2023-03-30T11:16:17Z

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

-          LOG.error("Storage selected is null and this should not happen. event: {}", event);
-          break;
-        }
+      Storage storage = storageManager.selectStorage(event);


I'm still not satisfied with this selectStorage again.

zuston · 2023-03-30T11:18:30Z

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

          event.increaseRetryTimes();
          ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+          event.markPended();
+          addPendingEvents(event);


How about re-pushing to flushEventHandler queue? I think the pending queue could be removed, which is strange after this PR refactoring.

Make sense.

zuston · 2023-03-30T11:19:17Z

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

-  public static final ConfigOption<Integer> SERVER_FLUSH_THREAD_POOL_SIZE = ConfigOptions
-      .key("rss.server.flush.threadPool.size")
+  public static final ConfigOption<Integer> SERVER_FLUSH_LOCALFILE_THREAD_POOL_SIZE = ConfigOptions
+      .key("rss.server.flush.local-file.threadPool.size")


rss.server.flush.local-file.threadPool.size -> rss.server.flush.localfile.threadPool.size ?

zuston · 2023-03-30T11:19:38Z

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

+      .key("rss.server.flush.hdfs.threadPool.size")
+      .intType()
+      .defaultValue(10)
+      .withDescription("thread pool for flush data to hdfs");

  public static final ConfigOption<Integer> SERVER_FLUSH_THREAD_POOL_QUEUE_SIZE = ConfigOptions


Why not removing this?

This is useful to flush the waiting queue of the thread pool.

Let's eliminate this pending queue and unify it with the event handler.

I do not understand.

zuston · 2023-03-30T11:26:36Z

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

+  private final ShuffleServerConf shuffleServerConf;
+  private final ShuffleFlushManager shuffleFlushManager;
+  private final StorageManager storageManager;
+  private Executor localFileThreadPoolExecutor;


Can we design a unified mechanism to avoid hard code ? That means we don't need to change this part when introducing a new storage type like object store.

Can we design a unified mechanism to avoid hard code ? That means we don't need to change this part when introducing a new storage type like object store.

Do you have a better suggestion? We need to determine which thread pool to allocate according to the type of Storage. MultiStorageManager also uses hard coding.

private final StorageManager warmStorageManager; private final StorageManager coldStorageManager; warmStorageManager = new LocalStorageManager(conf); coldStorageManager = new HdfsStorageManager(conf);

Not yet. OK. let's reserve current implementation.

leixm · 2023-03-30T11:44:21Z

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

+
+  protected Executor createFlushEventExecutor(int poolSize, String threadFactoryName) {
+    int waitQueueSize = shuffleServerConf.getInteger(
+        ShuffleServerConf.SERVER_FLUSH_THREAD_POOL_QUEUE_SIZE);


See here @zuston

leixm · 2023-03-31T02:42:20Z

@zuston @smallzhongfeng PTAL.

leixm · 2023-03-31T06:38:37Z

@jerqi PTAL.

smallzhongfeng

LGTM. Just one minor comment. Waiting for @jerqi and @zuston .cc please :-)

smallzhongfeng · 2023-03-31T09:27:51Z

...ration-test/common/src/test/java/org/apache/uniffle/test/MultiStorageFaultToleranceBase.java

@@ -66,6 +66,7 @@ public void closeClient() {

  @Test
  public void fallbackTest() throws Exception {
+    REMOTE_STORAGE = HDFS_URI + "rss/multi_storage_fault";


Change to local variable maybe better.

Change to local variable maybe better.

Other method use also.

smallzhongfeng · 2023-03-31T09:45:22Z

We should add this change to uniffle-migration-guide.md.

leixm · 2023-03-31T11:08:00Z

We should add this change to uniffle-migration-guide.md.

Ok.

leixm · 2023-03-31T11:29:47Z

We should add this change to uniffle-migration-guide.md.

Done.

zuston · 2023-03-31T12:00:38Z

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

-          LOG.error("Storage selected is null and this should not happen. event: {}", event);
-          break;
-        }
+      Storage storage = event.getUnderStorage();


When to set this storage of event ?

zuston · 2023-03-31T12:01:12Z

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

+        // todo: Could we add an interface supportPending for storageManager
+        //       to unify following logic of multiple different storage managers
+        if (event.getRetryTimes() <= retryMax) {
+          if (event.isPended()) {


How about removing this pending queue? WDYT @jerqi

zuston · 2023-03-31T12:02:56Z

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

+      } else if (event.getRetryTimes() <= retryMax) {
+        event.increaseRetryTimes();
+        ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+        event.getUnderEventHandler().handle(event);


I think this is wrong. That means once the storage is selected for one event, it will always try this, which will not give chance to change storage for event

And this re-pushing by itself looks strange.

zuston · 2023-03-31T12:15:48Z

The design looks not clear, the dependencies are complex. I think it should refactor like following steps

Introducing the eventHandler like this PR does and using a queue in here. But I hope it should do more, including pushing events directly to this queue rather than using the processEventThread
Remove the pending queue or unify to eventHandler. If using the latter, the multiple priority queues should be introduced in eventHandler

Plz let me know If I'm wrong.

cc @jerqi I remember you want to remove the pending queue. How about doing this in this PR?

jerqi · 2023-05-23T14:31:15Z

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

-          break;
-        } else {
+      if (!storage.canWrite()) {
+        // todo: Could we add an interface supportPending for storageManager


Should we remove the todo?

jerqi · 2023-05-23T14:32:55Z

@zuston Could you help this pr again?

jerqi · 2023-05-23T14:35:49Z

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

+  }
+
+  protected void eventLoop() {
+    while (true) {


Could we have a variable to control whether stop the eventloop? Could we add the method stop?

zuston · 2023-05-24T01:55:09Z

@zuston Could you help this pr again?

Yes, let me take a look in next few days

leixm · 2023-05-31T03:07:25Z

@zuston PTAL.

zuston

I'm so sorry for this later review

LGTM. The refactor looks good to me . Please fix the conflict.

zuston · 2023-06-02T09:54:48Z

server/src/main/java/org/apache/uniffle/server/buffer/ShuffleBuffer.java

-    event.addCleanupCallback(() -> this.clearInFlushBuffer(event.getEventId()));
+    event.addCleanupCallback(() -> {
+      this.clearInFlushBuffer(event.getEventId());
+      spBlocks.forEach(spb -> spb.getData().release());


spBlocks.forEach(spb -> spb.getData().release());

This is for netty?

My mistake.

zuston · 2023-06-02T09:55:55Z

server/src/main/java/org/apache/uniffle/server/storage/LocalStorageManager.java

@@ -181,6 +181,7 @@ public Storage selectStorage(ShuffleDataFlushEvent event) {
              storage.getBasePath(), event);
        }
      } else {
+        event.setUnderStorage(storage);


This is to check the storage is hdfs or localfile to put its corresponding thread pool ? Right?

#881 already add.

if (event.getUnderStorage() == null) { event.setUnderStorage(storage); }

leixm · 2023-06-28T05:16:28Z

I will fix later.

…storage types.

leixm · 2023-07-11T09:12:41Z

All done. @jerqi @zuston

leixm · 2023-07-11T09:13:13Z

I renamed hdfs thread pool to hadoop thread pool.

jerqi

LGTM. let @zuston take another look

zuston

LGTM. Thanks for your effort @leixm

jerqi reviewed Mar 28, 2023

View reviewed changes

zuston reviewed Mar 29, 2023

View reviewed changes

smallzhongfeng reviewed Mar 30, 2023

View reviewed changes

zuston reviewed Mar 30, 2023

View reviewed changes

zuston changed the title ~~[#757] Separate flush thread pools for different storage type.~~ [#757] feat(server): separate flush thread pools for different storage types Mar 30, 2023

leixm commented Mar 30, 2023

View reviewed changes

smallzhongfeng reviewed Mar 31, 2023

View reviewed changes

zuston reviewed Mar 31, 2023

View reviewed changes

leixm closed this Apr 17, 2023

leixm force-pushed the master branch from da0b9cb to c6cde5d Compare April 17, 2023 02:43

leixm reopened this May 22, 2023

jerqi closed this May 22, 2023

jerqi reopened this May 22, 2023

leixm closed this May 23, 2023

leixm reopened this May 23, 2023

leixm closed this May 23, 2023

leixm reopened this May 23, 2023

jerqi reviewed May 23, 2023

View reviewed changes

jerqi requested a review from zuston May 23, 2023 14:32

jerqi reviewed May 23, 2023

View reviewed changes

zuston reviewed Jun 2, 2023

View reviewed changes

jerqi mentioned this pull request Jun 16, 2023

[DOCS] Improving the description of several parameters #955

Open

3 tasks

[apache#757] feat(server): separate flush thread pools for different …

1a28453

…storage types.

leixm force-pushed the master branch from afb12e6 to 1a28453 Compare July 11, 2023 08:29

jerqi approved these changes Jul 11, 2023

View reviewed changes

jerqi requested a review from zuston July 11, 2023 09:45

zuston approved these changes Jul 11, 2023

View reviewed changes

zuston merged commit 0040c4a into apache:master Jul 11, 2023
27 checks passed

zuston mentioned this pull request Jul 11, 2023

[Improvement] Separate flush thread pools for different storage type. #757

Closed

3 tasks

zuston mentioned this pull request Jan 17, 2024

[Improvement] Refactor DefaultFlushEventHandler and ShuffleFlushManager#flushToFile #1459

Closed

3 tasks

[#757] feat(server): separate flush thread pools for different storage types #775

[#757] feat(server): separate flush thread pools for different storage types #775

Conversation

leixm commented Mar 28, 2023

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 Mar 28, 2023 • edited Loading

Codecov Report

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

leixm Mar 29, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leixm commented Mar 29, 2023

jerqi commented Mar 29, 2023

leixm commented Mar 30, 2023

leixm commented Mar 30, 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

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

leixm commented Mar 31, 2023

leixm commented Mar 31, 2023

smallzhongfeng left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

smallzhongfeng commented Mar 31, 2023

leixm commented Mar 31, 2023

leixm commented Mar 31, 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

zuston commented Mar 31, 2023

Choose a reason for hiding this comment

jerqi commented May 23, 2023

Choose a reason for hiding this comment

zuston commented May 24, 2023

leixm commented May 31, 2023

zuston 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

leixm commented Jun 28, 2023

leixm commented Jul 11, 2023

leixm commented Jul 11, 2023

jerqi left a comment

Choose a reason for hiding this comment

zuston left a comment

Choose a reason for hiding this comment

codecov-commenter commented Mar 28, 2023 •

edited

Loading

leixm Mar 29, 2023 •

edited

Loading