[CELEBORN-2378] Fix CongestionController never resuming due to Netty pool overhead in getTotalPendingBytes()#3756
[CELEBORN-2378] Fix CongestionController never resuming due to Netty pool overhead in getTotalPendingBytes()#3756Deegue wants to merge 1 commit into
Conversation
… getTotalPendingBytes()
There was a problem hiding this comment.
Pull request overview
Updates worker congestion-control pending-bytes calculation to avoid Netty pooled allocator overhead so the worker can correctly resume once actual pending data drops below the low watermark.
Changes:
- Switch
CongestionController.getTotalPendingBytes()fromMemoryManager.getMemoryUsage()toMemoryManager.getPinnedMemory().
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| public long getTotalPendingBytes() { | ||
| return MemoryManager.instance().getMemoryUsage(); | ||
| return MemoryManager.instance().getPinnedMemory(); | ||
| } |
| public long getTotalPendingBytes() { | ||
| return MemoryManager.instance().getMemoryUsage(); | ||
| return MemoryManager.instance().getPinnedMemory(); | ||
| } |
|
It's ok for me, but if we change |
zaynt4606
left a comment
zaynt4606
left a comment
There was a problem hiding this comment.
In non-POOLED mode (UNPOOLED or ADAPTIVE), pooledByteBufAllocators stays empty because both fill sites guard with if (allocatorType == POOLED).
So getNettyPinnedDirectMemory() sums an empty list and returns 0, making getTotalPendingBytes() underestimate pending data.
Suggested fix:
fall back to PlatformDependent.usedDirectMemory() when no pooled allocators exist, so non-POOLED configs retain the old behavior.
What changes were proposed in this pull request?
Change CongestionController.getTotalPendingBytes() from MemoryManager.instance().getMemoryUsage() to MemoryManager.instance().getPinnedMemory().
Why are the changes needed?
getMemoryUsage() returns nettyUsedDirectMemory + sortMemory, which includes Netty's pooled allocator overhead.
Change to getPinnedMemory() whcih returns nettyPinnedDirectMemory + sortMemory instead.
Does this PR resolve a correctness bug?
Does this PR introduce any user-facing change?
No API change.
Workers with congestion control will now correctly resume from congestion state when actual pending data drops below the low watermark.
How was this patch tested?
UTs