This is an automated email from the ASF dual-hosted git repository.
zhouky 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 1b74d85fb [CELEBORN-725][MINOR] Refine congestion code
1b74d85fb is described below
commit 1b74d85fb1f2aceb05f137417468e71b267aa0bd
Author: onebox-li <[email protected]>
AuthorDate: Wed Jun 28 18:31:40 2023 +0800
[CELEBORN-725][MINOR] Refine congestion code
### What changes were proposed in this pull request?
Refine the congestion relevant code/log/comments
### Why are the changes needed?
ditto
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
manually test
Closes #1637 from onebox-li/improve-congestion.
Authored-by: onebox-li <[email protected]>
Signed-off-by: zky.zhoukeyong <[email protected]>
---
.../apache/celeborn/client/ShuffleClientImpl.java | 12 ++++----
.../common/write/SlowStartPushStrategy.java | 34 +++++++++++++---------
.../org/apache/celeborn/common/CelebornConf.scala | 6 ++--
docs/configuration/client.md | 4 +--
.../congestcontrol/CongestionController.java | 8 ++---
5 files changed, 35 insertions(+), 29 deletions(-)
diff --git
a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
index 1324db481..c7c07a1ad 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -579,7 +579,10 @@ public class ShuffleClientImpl extends ShuffleClient {
if (reachLimit) {
throw new CelebornIOException(
- "Waiting timeout for task " + mapKey, pushState.exception.get());
+ String.format(
+ "Waiting timeout for task %s while limiting max in-flight
requests to %s",
+ mapKey, hostAndPushPort),
+ pushState.exception.get());
}
}
@@ -588,7 +591,9 @@ public class ShuffleClientImpl extends ShuffleClient {
if (reachLimit) {
throw new CelebornIOException(
- "Waiting timeout for task " + mapKey, pushState.exception.get());
+ String.format(
+ "Waiting timeout for task %s while limiting zero in-flight
requests", mapKey),
+ pushState.exception.get());
}
}
@@ -861,8 +866,6 @@ public class ShuffleClientImpl extends ShuffleClient {
@Override
public void onSuccess(ByteBuffer response) {
pushState.removeBatch(nextBatchId, loc.hostAndPushPort());
- // TODO Need to adjust maxReqsInFlight if server response is
congested, see
- // CELEBORN-62
if (response.remaining() > 0 && response.get() ==
StatusCode.STAGE_ENDED.getValue()) {
stageEndShuffleSet.add(shuffleId);
}
@@ -1247,7 +1250,6 @@ public class ShuffleClientImpl extends ShuffleClient {
groupedBatchId,
Arrays.toString(batchIds));
pushState.removeBatch(groupedBatchId, hostPort);
- // TODO Need to adjust maxReqsInFlight if server response is
congested, see CELEBORN-62
if (response.remaining() > 0 && response.get() ==
StatusCode.STAGE_ENDED.getValue()) {
stageEndShuffleSet.add(shuffleId);
}
diff --git
a/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
b/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
index e5be6d650..b22630b02 100644
---
a/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
+++
b/common/src/main/java/org/apache/celeborn/common/write/SlowStartPushStrategy.java
@@ -29,6 +29,24 @@ import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.exception.CelebornIOException;
import org.apache.celeborn.common.util.JavaUtils;
+/**
+ * Similar to the TCP congestion control algorithm, this strategy adjusts
`currentMaxReqsInFlight`,
+ * equivalent to the congestion window size in TCP, to limit max in-flight
push data requests to
+ * each worker host. Use separate `CongestControlContext` for each worker to
track the congestion
+ * control state. Note that here we define one RTT period: one
batch(`currentMaxReqsInFlight`) of
+ * push data requests.
+ *
+ * <p>"slow start" mechanism is applied to increase `currentMaxReqsInFlight`
while first request to
+ * the worker comes or the value is under slow start threshold
(`reqsInFlightBlockThreshold`).
+ * `currentMaxReqsInFlight` can be increased by 1 with each request succeed,
effectively be doubled
+ * each RTT.
+ *
+ * <p>If slow start threshold (`reqsInFlightBlockThreshold`) is reached, it
will change to
+ * congestion avoidance algorithm. During congestion avoidance,
`currentMaxReqsInFlight` will be
+ * increased by 1 every RTT period, which may more than
`celeborn.client.push.maxReqsInFlight`.
+ *
+ * <p>If congestion happens, `currentMaxReqsInFlight` will be halved.
+ */
public class SlowStartPushStrategy extends PushStrategy {
protected static class CongestControlContext {
@@ -102,17 +120,6 @@ public class SlowStartPushStrategy extends PushStrategy {
hostAndPushPort, host -> new CongestControlContext(maxInFlight));
}
- /**
- * If `pushDataSlowStart` is enabled, will increase `currentMaxReqsInFlight`
gradually to meet the
- * max push speed.
- *
- * <p>1. slow start period: every RTT period, `currentMaxReqsInFlight` is
doubled.
- *
- * <p>2. congestion avoidance: every RTT period, `currentMaxReqsInFlight`
plus 1.
- *
- * <p>Note that here we define one RTT period: one
batch(currentMaxReqsInFlight) of push data
- * requests.
- */
@Override
public void onSuccess(String hostAndPushPort) {
CongestControlContext congestControlContext =
@@ -129,7 +136,7 @@ public class SlowStartPushStrategy extends PushStrategy {
protected long getSleepTime(CongestControlContext context) {
int currentMaxReqs = context.getCurrentMaxReqsInFlight();
- if (currentMaxReqs >= conf.clientPushMaxReqsInFlight()) {
+ if (currentMaxReqs >= maxInFlight) {
return 0;
}
@@ -152,7 +159,8 @@ public class SlowStartPushStrategy extends PushStrategy {
long sleepInterval = getSleepTime(congestControlContext);
if (sleepInterval > 0L) {
try {
- logger.debug("Will sleep {} ms to control the push speed.",
sleepInterval);
+ logger.debug(
+ "Will sleep {} ms to control the push speed to {}.",
sleepInterval, hostAndPushPort);
Thread.sleep(sleepInterval);
} catch (InterruptedException e) {
pushState.exception.set(new CelebornIOException(e));
diff --git
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 1b2af255c..85717075f 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -2665,7 +2665,7 @@ object CelebornConf extends Logging {
.version("0.3.0")
.doc("Amount of Netty in-flight requests per worker. The maximum memory
is " +
"`celeborn.client.push.maxReqsInFlight` *
`celeborn.push.buffer.max.size` * " +
- "compression ratio(1 in worst case), default: 64Kib * 32 = 2Mib")
+ "compression ratio(1 in worst case), default: 64KiB * 4 = 256KiB")
.intConf
.createWithDefault(4)
@@ -2706,8 +2706,8 @@ object CelebornConf extends Logging {
buildConf("celeborn.client.push.limit.strategy")
.categories("client")
.doc("The strategy used to control the push speed. " +
- "Valid strategies are SIMPLE and SLOWSTART. the SLOWSTART strategy is
usually cooperate with " +
- "congest control mechanism in the worker side.")
+ "Valid strategies are SIMPLE and SLOWSTART. The SLOWSTART strategy
usually works with " +
+ "congestion control mechanism on the worker side.")
.version("0.3.0")
.stringConf
.transform(_.toUpperCase(Locale.ROOT))
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
index 0bc1f79bb..f93ba9f97 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -42,8 +42,8 @@ license: |
| celeborn.client.push.excludeWorkerOnFailure.enabled | false | Whether to
enable shuffle client-side push exclude workers on failures. | 0.3.0 |
| celeborn.client.push.limit.inFlight.sleepInterval | 50ms | Sleep interval
when check netty in-flight requests to be done. | 0.3.0 |
| celeborn.client.push.limit.inFlight.timeout | <undefined> | Timeout
for netty in-flight requests to be done.Default value should be
`celeborn.client.push.timeout * 2`. | 0.3.0 |
-| celeborn.client.push.limit.strategy | SIMPLE | The strategy used to control
the push speed. Valid strategies are SIMPLE and SLOWSTART. the SLOWSTART
strategy is usually cooperate with congest control mechanism in the worker
side. | 0.3.0 |
-| celeborn.client.push.maxReqsInFlight | 4 | Amount of Netty in-flight
requests per worker. The maximum memory is
`celeborn.client.push.maxReqsInFlight` * `celeborn.push.buffer.max.size` *
compression ratio(1 in worst case), default: 64Kib * 32 = 2Mib | 0.3.0 |
+| celeborn.client.push.limit.strategy | SIMPLE | The strategy used to control
the push speed. Valid strategies are SIMPLE and SLOWSTART. The SLOWSTART
strategy usually works with congestion control mechanism on the worker side. |
0.3.0 |
+| celeborn.client.push.maxReqsInFlight | 4 | Amount of Netty in-flight
requests per worker. The maximum memory is
`celeborn.client.push.maxReqsInFlight` * `celeborn.push.buffer.max.size` *
compression ratio(1 in worst case), default: 64KiB * 4 = 256KiB | 0.3.0 |
| celeborn.client.push.queue.capacity | 512 | Push buffer queue size for a
task. The maximum memory is `celeborn.push.buffer.max.size` *
`celeborn.push.queue.capacity`, default: 64KiB * 512 = 32MiB | 0.3.0 |
| celeborn.client.push.replicate.enabled | false | When true, Celeborn worker
will replicate shuffle data to another Celeborn worker asynchronously to ensure
the pushed shuffle data won't be lost after the node failure. It's recommended
to set `false` when `HDFS` is enabled in `celeborn.storage.activeTypes`. |
0.3.0 |
| celeborn.client.push.retry.threads | 8 | Thread number to process shuffle
re-send push data requests. | 0.3.0 |
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 ec2a2e2c6..4249d896d 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
@@ -139,8 +139,6 @@ public class CongestionController {
}
long pendingConsumed = getTotalPendingBytes();
- long avgConsumeSpeed = getPotentialConsumeSpeed();
-
if (pendingConsumed > highWatermark &&
overHighWatermark.compareAndSet(false, true)) {
logger.info(
"Pending consume bytes: {} higher than high watermark, need to
congest it",
@@ -161,6 +159,7 @@ public class CongestionController {
// If the user produce speed is higher that the avg consume speed, will
congest it
long userProduceSpeed =
getUserProduceSpeed(userBufferStatuses.get(userIdentifier));
+ long avgConsumeSpeed = getPotentialConsumeSpeed();
if (logger.isDebugEnabled()) {
logger.debug(
"The user {}, produceSpeed is {},"
@@ -172,7 +171,6 @@ public class CongestionController {
}
return userProduceSpeed > avgConsumeSpeed;
}
-
return false;
}
@@ -244,9 +242,7 @@ public class CongestionController {
if (currentTimeMillis - userBufferInfo.getTimestamp() >=
userInactiveTimeMills) {
userBufferStatuses.remove(userIdentifier);
workerSource.removeGauge(WorkerSource.UserProduceSpeed(),
userIdentifier.toMap());
- logger.info(
- String.format(
- "User: %s has been expired, remove it from rate limit list",
userIdentifier));
+ logger.info("User {} has been expired, remove from rate limit list",
userIdentifier);
}
}
} catch (Exception e) {
