Copilot commented on code in PR #3756:
URL: https://github.com/apache/celeborn/pull/3756#discussion_r3577829191
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -265,20 +275,22 @@ protected void checkCongestion() {
try {
long pendingConsume = getTotalPendingBytes();
Review Comment:
`checkCongestion()` uses `getTotalPendingBytes()` (which still includes
Netty pool overhead) to enter congestion, but uses `getActivePendingBytes()` to
exit. If the pool overhead alone keeps `getTotalPendingBytes()` above the high
watermark, the controller can exit congestion and then immediately re-enter on
the next check. Using the same metric for both entry/exit avoids this
oscillation and better matches the PR description.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -265,20 +275,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:
This log message says "Pending consume" but the decision is based on
`activePendingBytes` (pinned memory) rather than the `pendingConsume` computed
above. Updating the wording will make logs accurately reflect the condition
used to exit congestion.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java:
##########
@@ -206,6 +207,15 @@ public long getTotalPendingBytes() {
return MemoryManager.instance().getMemoryUsage();
}
+ public long getActivePendingBytes() {
+ MemoryManager memoryManager = MemoryManager.instance();
+ long pinnedMemory = memoryManager.getPinnedMemory();
+ if (pinnedMemory == 0 &&
NettyUtils.getAllPooledByteBufAllocators().isEmpty()) {
+ return memoryManager.getMemoryUsage();
+ }
+ return pinnedMemory;
+ }
Review Comment:
`getActivePendingBytes()` falls back to `getMemoryUsage()` only when
`pinnedMemory == 0`, but `pinnedMemory` includes `sortMemoryCounter`. If there
are no pooled allocators but sort memory is non-zero, this method will return
only sort memory and ignore Netty direct memory tracked by `getMemoryUsage()`,
undercounting pending bytes and potentially exiting congestion too early.
Consider falling back to `getMemoryUsage()` whenever there are no pooled
allocators, regardless of the pinned value.
--
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]
