update

apache · Oct 10, 2024 · 843f009 · 843f009
1 parent 332de4a
commit 843f009
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 5 deletions.
diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java
@@ -70,7 +70,7 @@ public class MemoryManager {
   private final AtomicLong diskBufferCounter = new AtomicLong(0);
   private final LongAdder pausePushDataCounter = new LongAdder();
   private final LongAdder pausePushDataAndReplicateCounter = new LongAdder();
-  private ServingState servingState = ServingState.NONE_PAUSED;
+  public ServingState servingState = ServingState.NONE_PAUSED;
   private long pausePushDataStartTime = -1L;
   private long pausePushDataTime = 0L;
   private long pausePushDataAndReplicateStartTime = -1L;
@@ -286,7 +286,7 @@ private MemoryManager(CelebornConf conf, StorageManager storageManager) {
   }
 
   public boolean shouldEvict(boolean aggressiveMemoryFileEvictEnabled, double evictRatio) {
-    return currentServingState() != ServingState.NONE_PAUSED
+    return servingState != ServingState.NONE_PAUSED
         && (aggressiveMemoryFileEvictEnabled
             || (memoryFileStorageCounter.sum() >= evictRatio * memoryFileStorageThreshold));
   }
@@ -405,8 +405,7 @@ public void reserveSortMemory(long fileLen) {
 
   public boolean sortMemoryReady() {
     return bypassMemoryCheckForSortingFiles
-        || (currentServingState() == ServingState.NONE_PAUSED
-            && sortMemoryCounter.get() < maxSortMemory);
+        || (servingState == ServingState.NONE_PAUSED && sortMemoryCounter.get() < maxSortMemory);
   }
 
   public void releaseSortMemory(long size) {

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
@@ -440,7 +440,7 @@ private[celeborn] class Worker(
   }
 
   private def highWorkload: Boolean = {
-    (memoryManager.currentServingState, conf.workerActiveConnectionMax) match {
+    (memoryManager.servingState, conf.workerActiveConnectionMax) match {
       case (ServingState.PUSH_AND_REPLICATE_PAUSED, _) => true
       case (ServingState.PUSH_PAUSED, _) => true
       case (_, Some(activeConnectionMax)) =>