[CELEBORN-1472] Reduce CongestionController#userBufferStatuses call times. #2583
Conversation
|
|
| @@ -1282,6 +1282,8 @@ class PushDataHandler(val workerSource: WorkerSource) extends BaseMessageHandler | |||
| fileWriter.decrementPendingWrites() | |||
| } | |||
| } | |||
|
|
|||
| updateBytesProduced(fileWriters.head, body.readableBytes()) | |||
There was a problem hiding this comment.
Seems this changes the behavior, PartitionDataWriters in fileWriters are different objects for handlePushMergedData and body contains data for those objects. IIUC we need to update each object for its own body, instead of only updating the first one with the whole body.
There was a problem hiding this comment.
You are right, I will modify another place, it seems that there is no need to call userBufferStatuses.computeIfAbsent every time
|
Found through stress testing, we reduced the CPU usage of produceBytes from 25.07% to 16.58% After this PR |
leixm
changed the title
leixm
changed the title
|
Pls update pr title and pr desc |
leixm
changed the title
|
@AngersZhuuuu @waitinfuture PTAL. |
...main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java
Outdated
Show resolved
Hide resolved
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriter.java
Show resolved
Hide resolved
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## main #2583 +/- ##
==========================================
- Coverage 40.49% 39.98% -0.50%
==========================================
Files 222 233 +11
Lines 14289 14695 +406
Branches 1291 1338 +47
==========================================
+ Hits 5785 5875 +90
- Misses 8173 8487 +314
- Partials 331 333 +2 ☔ View full report in Codecov by Sentry. |
| @@ -153,6 +156,10 @@ public PartitionDataWriter( | |||
| this.mapIdBitMap = new RoaringBitmap(); | |||
| } | |||
| takeBuffer(); | |||
| CongestionController congestionController = CongestionController.instance(); | |||
| if (!isMemoryShuffleFile.get() && congestionController != null) { | |||
| userBufferInfo = congestionController.getUserBuffer(getDiskFileInfo().getUserIdentifier()); | |||
There was a problem hiding this comment.
IMO the CongestionController should control all file write operations including memory file ?
There was a problem hiding this comment.
I keep the original code logic unchanged. In the original logic, CongestionController does not manage memory files. Maybe I can correct this logic in the next PR?
There was a problem hiding this comment.
I keep the original code logic unchanged. In the original logic, CongestionController does not manage memory files. Maybe I can correct this logic in the next PR?
LGTM
| congestionController -> | ||
| congestionController.produceBytes(diskFileInfo.getUserIdentifier(), numBytes)); | ||
| if (userBufferInfo != null) { | ||
| userBufferInfo.updateInfo( |
…imes ### What changes were proposed in this pull request? Reduce CongestionController#userBufferStatuses call times. ### Why are the changes needed? When we use sort based shuffle writer, The number of PushMergedData requests has increased which make CongestionController#produceBytes taking up much cpu time. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing uts. Closes #2583 from leixm/issue_1472. Authored-by: Xianming Lei <[email protected]> Signed-off-by: Shuang <[email protected]> (cherry picked from commit d362d9f) Signed-off-by: Shuang <[email protected]>
|
Merge to main(v0.6.0) and branch-0.5(v0.5.1) |
|
Thank you for you review. @AngersZhuuuu @RexXiong |
What changes were proposed in this pull request?
Reduce CongestionController#userBufferStatuses call times.
Why are the changes needed?
When we use sort based shuffle writer, The number of PushMergedData requests has increased which make CongestionController#produceBytes taking up much cpu time.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Existing uts.