You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "leixm (via GitHub)" <gi...@apache.org> on 2024/10/09 11:29:12 UTC
[PR] [WIP][CELEBORN-1487] CongestionController support control traffic by user/worker traffic speed [celeborn]
leixm opened a new pull request, #2797:
URL: https://github.com/apache/celeborn/pull/2797
### What changes were proposed in this pull request?
Introduce support control traffic by user/worker traffic speed.
### Why are the changes needed?
Currently, Celeborn only supports quota management based on disk file bytes/count, and this quota management cannot cope with sudden increases in traffic, which will cause corrupt to the cluster.
### Does this PR introduce _any_ user-facing change?
Yes.
### How was this patch tested?
UTs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#discussion_r1796497994
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -122,25 +151,40 @@ public static CongestionController instance() {
* <p>3. If the pending bytes doesn't exceed the high watermark, will allow all users to try to
* get max throughout capacity.
*/
- public boolean isUserCongested(UserIdentifier userIdentifier) {
- if (userBufferStatuses.size() == 0) {
+ public boolean isUserCongested(UserCongestionControlContext userCongestionControlContext) {
+ if (userBufferStatuses.isEmpty()) {
return false;
}
- if (overHighWatermark.get()) {
- // If the user produce speed is higher that the avg consume speed, will congest it
- long userProduceSpeed = getUserProduceSpeed(userBufferStatuses.get(userIdentifier));
- long avgConsumeSpeed = getPotentialConsumeSpeed();
+
+ UserIdentifier userIdentifier = userCongestionControlContext.getUserIdentifier();
+ long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
+ long avgConsumeSpeed = getPotentialProduceSpeed();
+ // If the user produce speed is higher that the avg consume speed, will congest it
+ if (overHighWatermark.get() && userProduceSpeed > avgConsumeSpeed) {
if (logger.isDebugEnabled()) {
Review Comment:
Can I do it in the next PR? I will introduce dynamicConfig for CongestionController and support workers to synchronize dynamicConfig from the master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on code in PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#discussion_r1796489779
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -205,6 +257,7 @@ private void removeInactiveUsers() {
UserBufferInfo userBufferInfo = next.getValue();
if (currentTimeMillis - userBufferInfo.getTimestamp() >= userInactiveTimeMills) {
userBufferStatuses.remove(userIdentifier);
+ userCongestionContextMap.remove(userIdentifier);
Review Comment:
it will be removed only reach userInactiveTimeMills.
If getUserCongestionContext is called again after it is removed, a new context will be created.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#issuecomment-2404338854
Flaky test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "leixm (via GitHub)" <gi...@apache.org>.
leixm commented on PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#issuecomment-2404339358
@RexXiong @AngersZhuuuu Can you help review?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#discussion_r1796606625
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/UserCongestionControlContext.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker.congestcontrol;
+
+import com.codahale.metrics.Gauge;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.metrics.source.AbstractSource;
+import org.apache.celeborn.service.deploy.worker.WorkerSource;
+
+public class UserCongestionControlContext {
+
+ private volatile boolean congestionControlFlag;
+
+ private final UserBufferInfo userBufferInfo;
+
+ private final BufferStatusHub workerBufferStatusHub;
+
+ private final UserIdentifier userIdentifier;
+
+ public UserCongestionControlContext(
+ BufferStatusHub workerBufferStatusHub,
+ UserBufferInfo userBufferInfo,
+ AbstractSource workerSource,
+ UserIdentifier userIdentifier) {
+ this.congestionControlFlag = false;
+ this.userBufferInfo = userBufferInfo;
+ this.userIdentifier = userIdentifier;
+ this.workerBufferStatusHub = workerBufferStatusHub;
+ workerSource.addGauge(
+ WorkerSource.USER_PRODUCE_SPEED(),
+ userIdentifier.toJMap(),
+ (Gauge<Long>) () -> userBufferInfo.getBufferStatusHub().avgBytesPerSec());
+ }
+
+ public void onCongestionControl() {
+ congestionControlFlag = true;
+ }
+
+ public void offCongestionControl() {
+ congestionControlFlag = false;
+ }
+
+ public boolean inCongestionControl() {
+ return congestionControlFlag;
+ }
+
+ public void updateProduceBytes(long numBytes) {
+ BufferStatusHub.BufferStatusNode node = new BufferStatusHub.BufferStatusNode(numBytes);
+ userBufferInfo.updateInfo(System.currentTimeMillis(), node);
Review Comment:
compute current time `System.currentTimeMillis()` twice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#discussion_r1796602224
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -122,25 +151,43 @@ public static CongestionController instance() {
* <p>3. If the pending bytes doesn't exceed the high watermark, will allow all users to try to
* get max throughout capacity.
*/
- public boolean isUserCongested(UserIdentifier userIdentifier) {
- if (userBufferStatuses.size() == 0) {
+ public boolean isUserCongested(UserCongestionControlContext userCongestionControlContext) {
+ if (userBufferStatuses.isEmpty()) {
return false;
}
+
+ UserIdentifier userIdentifier = userCongestionControlContext.getUserIdentifier();
+ // If the user produce speed is higher that the avg consume speed, will congest it
if (overHighWatermark.get()) {
- // If the user produce speed is higher that the avg consume speed, will congest it
- long userProduceSpeed = getUserProduceSpeed(userBufferStatuses.get(userIdentifier));
- long avgConsumeSpeed = getPotentialConsumeSpeed();
+ long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
+ long avgConsumeSpeed = getPotentialProduceSpeed();
+ if (userProduceSpeed > avgConsumeSpeed) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "The user {}, produceSpeed is {}, while consumeSpeed is {}, need to congest it.",
+ userIdentifier,
+ userProduceSpeed,
+ avgConsumeSpeed);
+ }
+ return true;
+ }
+ }
+
+ long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
Review Comment:
userProduceSpeed may compute twice...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#discussion_r1796399830
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -122,25 +151,40 @@ public static CongestionController instance() {
* <p>3. If the pending bytes doesn't exceed the high watermark, will allow all users to try to
* get max throughout capacity.
*/
- public boolean isUserCongested(UserIdentifier userIdentifier) {
- if (userBufferStatuses.size() == 0) {
+ public boolean isUserCongested(UserCongestionControlContext userCongestionControlContext) {
+ if (userBufferStatuses.isEmpty()) {
return false;
}
- if (overHighWatermark.get()) {
- // If the user produce speed is higher that the avg consume speed, will congest it
- long userProduceSpeed = getUserProduceSpeed(userBufferStatuses.get(userIdentifier));
- long avgConsumeSpeed = getPotentialConsumeSpeed();
+
+ UserIdentifier userIdentifier = userCongestionControlContext.getUserIdentifier();
+ long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
+ long avgConsumeSpeed = getPotentialProduceSpeed();
+ // If the user produce speed is higher that the avg consume speed, will congest it
+ if (overHighWatermark.get() && userProduceSpeed > avgConsumeSpeed) {
if (logger.isDebugEnabled()) {
Review Comment:
maybe compute userProduceSpeed/avgConsumeSpeed after check overHighWatermark
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -40,36 +40,57 @@ public class CongestionController {
private final WorkerSource workerSource;
private final int sampleTimeWindowSeconds;
- private final long highWatermark;
- private final long lowWatermark;
+ private final long diskBufferHighWatermark;
+ private final long diskBufferLowWatermark;
+ private final long userProduceSpeedHighWatermark;
+ private final long userProduceSpeedLowWatermark;
+ private final long workerProduceSpeedHighWatermark;
+ private final long workerProduceSpeedLowWatermark;
private final long userInactiveTimeMills;
private final AtomicBoolean overHighWatermark = new AtomicBoolean(false);
private final BufferStatusHub consumedBufferStatusHub;
+ private final BufferStatusHub producedBufferStatusHub;
+
private final ConcurrentHashMap<UserIdentifier, UserBufferInfo> userBufferStatuses;
private final ScheduledExecutorService removeUserExecutorService;
private final ScheduledExecutorService checkService;
+ private final ConcurrentHashMap<UserIdentifier, UserCongestionControlContext>
+ userCongestionContextMap;
+
protected CongestionController(
WorkerSource workerSource,
int sampleTimeWindowSeconds,
- long highWatermark,
- long lowWatermark,
+ long diskBufferHighWatermark,
Review Comment:
Can we introduce ConfigService for Worker to support dynamic change these configuration? like `configService.getSystemConfigFromCache....`
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -205,6 +257,7 @@ private void removeInactiveUsers() {
UserBufferInfo userBufferInfo = next.getValue();
if (currentTimeMillis - userBufferInfo.getTimestamp() >= userInactiveTimeMills) {
userBufferStatuses.remove(userIdentifier);
+ userCongestionContextMap.remove(userIdentifier);
Review Comment:
this will potentially remove the UserCongestionControlContext which just put into userCongestionContextMap?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong closed pull request #2797: [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed
URL: https://github.com/apache/celeborn/pull/2797
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1487][Phase1] CongestionController support control traffic by user/worker traffic speed [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #2797:
URL: https://github.com/apache/celeborn/pull/2797#discussion_r1796523733
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -122,25 +151,40 @@ public static CongestionController instance() {
* <p>3. If the pending bytes doesn't exceed the high watermark, will allow all users to try to
* get max throughout capacity.
*/
- public boolean isUserCongested(UserIdentifier userIdentifier) {
- if (userBufferStatuses.size() == 0) {
+ public boolean isUserCongested(UserCongestionControlContext userCongestionControlContext) {
+ if (userBufferStatuses.isEmpty()) {
return false;
}
- if (overHighWatermark.get()) {
- // If the user produce speed is higher that the avg consume speed, will congest it
- long userProduceSpeed = getUserProduceSpeed(userBufferStatuses.get(userIdentifier));
- long avgConsumeSpeed = getPotentialConsumeSpeed();
+
+ UserIdentifier userIdentifier = userCongestionControlContext.getUserIdentifier();
+ long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
+ long avgConsumeSpeed = getPotentialProduceSpeed();
+ // If the user produce speed is higher that the avg consume speed, will congest it
+ if (overHighWatermark.get() && userProduceSpeed > avgConsumeSpeed) {
if (logger.isDebugEnabled()) {
Review Comment:
> Can I do it in the next PR? I will introduce dynamicConfig for CongestionController and support workers to synchronize dynamicConfig from the master.
sure, but imo we support dynamic configs same way like master does, it means worker can refresh/use local fs configs/get from db configs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org