Copilot commented on code in PR #3756:
URL: https://github.com/apache/celeborn/pull/3756#discussion_r3584282247
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -77,6 +82,7 @@ protected CongestionController(
this.workerSource = workerSource;
this.sampleTimeWindowSeconds = sampleTimeWindowSeconds;
this.userInactiveTimeMills =
conf.workerCongestionControlUserInactiveIntervalMs();
+ this.activePendingBytesCacheIntervalMs =
conf.workerPinnedMemoryCheckIntervalMs();
Review Comment:
activePendingBytesCacheIntervalMs is currently tied to
workerPinnedMemoryCheckIntervalMs, whose default is 10s
(CelebornConf.WORKER_PINNED_MEMORY_CHECK_INTERVAL). Since congestion checks run
every workerCongestionControlCheckIntervalMs (default 10ms), this can delay
exiting congestion control by up to 10 seconds even after pending bytes drop
below the low watermark, and can cause unnecessary repeated trimMemoryUsage()
calls while waiting for the cache to refresh.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -265,20 +285,22 @@ protected void checkCongestion() {
try {
long pendingConsume = getTotalPendingBytes();
long workerProduceSpeed = producedBufferStatusHub.avgBytesPerSec();
- if (pendingConsume < workerTrafficQuota.diskBufferLowWatermark()
- && workerProduceSpeed <
workerTrafficQuota.workerProduceSpeedLowWatermark()) {
- if (overHighWatermark.compareAndSet(true, false)) {
- logger.info(
- "Pending consume and produce speed is lower than low watermark,
exit congestion control");
+ if (overHighWatermark.get()) {
+ long activePendingBytes = getActivePendingBytes();
+ if (activePendingBytes < workerTrafficQuota.diskBufferLowWatermark()
+ && workerProduceSpeed <
workerTrafficQuota.workerProduceSpeedLowWatermark()) {
+ if (overHighWatermark.compareAndSet(true, false)) {
+ logger.info(
+ "Pending consume and produce speed is lower than low
watermark, exit congestion control");
+ }
Review Comment:
The log message says "Pending consume" but the exit condition now uses
activePendingBytes (pinned memory when pooled). Logging the measured values and
thresholds would make diagnosing congestion/resume behavior much easier and
avoid confusion.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -265,20 +285,22 @@ protected void checkCongestion() {
try {
long pendingConsume = getTotalPendingBytes();
long workerProduceSpeed = producedBufferStatusHub.avgBytesPerSec();
- if (pendingConsume < workerTrafficQuota.diskBufferLowWatermark()
- && workerProduceSpeed <
workerTrafficQuota.workerProduceSpeedLowWatermark()) {
- if (overHighWatermark.compareAndSet(true, false)) {
- logger.info(
- "Pending consume and produce speed is lower than low watermark,
exit congestion control");
+ if (overHighWatermark.get()) {
+ long activePendingBytes = getActivePendingBytes();
+ if (activePendingBytes < workerTrafficQuota.diskBufferLowWatermark()
+ && workerProduceSpeed <
workerTrafficQuota.workerProduceSpeedLowWatermark()) {
+ if (overHighWatermark.compareAndSet(true, false)) {
Review Comment:
checkCongestion() now decides when to exit congestion using
getActivePendingBytes(), but existing unit tests for this class override
getTotalPendingBytes() only. This leaves the new pooled-vs-unpooled resume
logic (and the caching behavior) untested and can also make tests inadvertently
depend on MemoryManager/Netty allocator state. Consider updating
TestCongestionController to override getActivePendingBytes() (or provide a test
seam) and add coverage for both pooled and unpooled paths.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
