[incubator-celeborn] branch main updated: [CELEBORN-342] Fix the wrong avg produce bytes in Congestion control (#1279)

[an...@apache.org]

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

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


The following commit(s) were added to refs/heads/main by this push:
     new 798ff90bb [CELEBORN-342] Fix the wrong avg produce bytes in Congestion control (#1279)
798ff90bb is described below

commit 798ff90bb7bf1011dc1cd7745c0b064c92524953
Author: Rex(Hui) An <an...@126.com>
AuthorDate: Mon Feb 27 16:29:37 2023 +0800

    [CELEBORN-342] Fix the wrong avg produce bytes in Congestion control (#1279)
---
 .../deploy/worker/congestcontrol/BufferStatusHub.java     |  8 ++++++++
 .../worker/congestcontrol/CongestionController.java       | 15 +++++++--------
 .../deploy/worker/congestcontrol/TimeSlidingHub.java      |  4 ++++
 3 files changed, 19 insertions(+), 8 deletions(-)

diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java
index cd0b5ebdc..38c7282f4 100644
--- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java
+++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java
@@ -71,4 +71,12 @@ public class BufferStatusHub extends TimeSlidingHub<BufferStatusHub.BufferStatus
   protected BufferStatusNode newEmptyNode() {
     return new BufferStatusNode();
   }
+
+  public long avgBytesPerSec() {
+    long currentNumBytes = sum().numBytes();
+    if (currentNumBytes > 0) {
+      return currentNumBytes * 1000 / (long) getCurrentTimeWindowsInMills();
+    }
+    return 0L;
+  }
 }
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 4d49f5932..f6a13ef13 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
@@ -135,8 +135,6 @@ public class CongestionController {
     }
 
     long pendingConsumed = getTotalPendingBytes();
-
-    // The potential consume speed in average
     long avgConsumeSpeed = getPotentialConsumeSpeed();
 
     if (pendingConsumed > highWatermark && overHighWatermark.compareAndSet(false, true)) {
@@ -208,21 +206,22 @@ public class CongestionController {
     MemoryManager.instance().trimAllListeners();
   }
 
+  /**
+   * Get avg consumed bytes in a configured time window, and divide with the number of active users
+   * to determine the potential consume speed.
+   */
   public long getPotentialConsumeSpeed() {
     if (userBufferStatuses.size() == 0) {
       return 0;
     }
 
-    BufferStatusHub.BufferStatusNode totalBufferStatus = consumedBufferStatusHub.sum();
-    // The potential consume speed in average
-    return totalBufferStatus.numBytes()
-        / ((long) sampleTimeWindowSeconds * userBufferStatuses.size());
+    return consumedBufferStatusHub.avgBytesPerSec() / userBufferStatuses.size();
   }
 
+  /** Get the avg user produce speed, the unit is bytes/sec. */
   private long getUserProduceSpeed(UserBufferInfo userBufferInfo) {
     if (userBufferInfo != null) {
-      BufferStatusHub.BufferStatusNode userBufferStatus = userBufferInfo.getBufferStatusHub().sum();
-      return userBufferStatus.numBytes() / sampleTimeWindowSeconds;
+      return userBufferInfo.getBufferStatusHub().avgBytesPerSec();
     }
 
     return 0L;
diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java
index d24ac7ebb..ef34f07c6 100644
--- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java
+++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java
@@ -144,6 +144,10 @@ public abstract class TimeSlidingHub<N extends TimeSlidingHub.TimeSlidingNode> {
 
   protected abstract N newEmptyNode();
 
+  protected int getCurrentTimeWindowsInMills() {
+    return _deque.size() * intervalPerBucketInMills;
+  }
+
   @VisibleForTesting
   protected long currentTimeMillis() {
     return System.currentTimeMillis();