(celeborn) branch main updated: [CELEBORN-1782] Worker in congestion control should be in blacklist to avoid impact new shuffle

[rexxiong]

This is an automated email from the ASF dual-hosted git repository.

rexxiong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new cec88b2de [CELEBORN-1782] Worker in congestion control should be in 
blacklist to avoid impact new shuffle
cec88b2de is described below

commit cec88b2def8bf5a53dbaa0004f4818548e541855
Author: Xianming Lei <31424839+le...@users.noreply.github.com>
AuthorDate: Thu Dec 19 10:12:09 2024 +0800

    [CELEBORN-1782] Worker in congestion control should be in blacklist to 
avoid impact new shuffle
    
    ### What changes were proposed in this pull request?
    Set highWorkload=true when worker in congestion control.
    
    ### Why are the changes needed?
    Worker in congestion control should be in blacklist to avoid impact new 
shuffle.
    
    ### Does this PR introduce _any_ user-facing change?
    No
    
    ### How was this patch tested?
    Existing UTS.
    
    Closes #3003 from leixm/CELEBORN-1782.
    
    Authored-by: Xianming Lei <31424839+le...@users.noreply.github.com>
    Signed-off-by: Shuang <lvshuang....@alibaba-inc.com>
---
 .../deploy/worker/congestcontrol/CongestionController.java   |  4 ++++
 .../org/apache/celeborn/service/deploy/worker/Worker.scala   | 12 ++++++++----
 2 files changed, 12 insertions(+), 4 deletions(-)

diff --git 
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java
 
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java
index 11e68f4c3..2fbe8d637 100644
--- 
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java
+++ 
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java
@@ -285,6 +285,10 @@ public class CongestionController {
     }
   }
 
+  public Boolean isOverHighWatermark() {
+    return overHighWatermark.get();
+  }
+
   public void close() {
     logger.info("Closing {}", this.getClass().getSimpleName());
     this.removeUserExecutorService.shutdownNow();
diff --git 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index eac9a5ef4..fd0c0889b 100644
--- 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++ 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -447,10 +447,14 @@ private[celeborn] class Worker(
   }
 
   private def highWorkload: Boolean = {
-    (memoryManager.servingState, conf.workerActiveConnectionMax) match {
-      case (ServingState.PUSH_AND_REPLICATE_PAUSED, _) => true
-      case (ServingState.PUSH_PAUSED, _) => true
-      case (_, Some(activeConnectionMax)) =>
+    (
+      memoryManager.currentServingState,
+      Option(CongestionController.instance()),
+      conf.workerActiveConnectionMax) match {
+      case (_, Some(instance), _) if instance.isOverHighWatermark => true
+      case (ServingState.PUSH_AND_REPLICATE_PAUSED, _, _) => true
+      case (ServingState.PUSH_PAUSED, _, _) => true
+      case (_, _, Some(activeConnectionMax)) =>
         workerSource.getCounterCount(ACTIVE_CONNECTION_COUNT) >= 
activeConnectionMax
       case _ => false
     }