Copilot commented on code in PR #3824:
URL: https://github.com/apache/celeborn/pull/3824#discussion_r3845713350
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/ChannelsLimiter.java:
##########
@@ -156,5 +174,75 @@ public void onTrim() {
}
}
+ /**
+ * Resumes {@code ratio} fraction of paused channels that have a
likely-large stuck half-frame
+ * ({@link TransportFrameDecoder#hasLikelyLargeIncompleteFrame()}), each in
frame-drain mode.
+ */
+ @Override
+ public int drainIncompleteFrame(double ratio) {
+ List<Channel> candidates = new ArrayList<>();
+ for (Channel ch : channels) {
+ if (!ch.isActive() || ch.config().isAutoRead()) {
+ continue;
+ }
+ TransportFrameDecoder decoder = frameDecoderOf(ch);
+ if (decoder != null && decoder.hasLikelyLargeIncompleteFrame()) {
+ candidates.add(ch);
+ }
+ }
+
+ if (candidates.isEmpty()) {
+ // Only worth logging when backpressure is actually active; if isPaused
is false there are
+ // no paused channels to scan, so a zero result is expected and
uninteresting.
+ if (isPaused.get()) {
+ logger.info(
+ "{} drainIncompleteFrame skipped this tick: no paused channel
found with a stuck "
+ + "half-frame larger than {} bytes.",
+ moduleName,
+ TransportFrameDecoder.MAX_SINGLE_READ_BYTES);
+ }
Review Comment:
The "no candidate" path logs at INFO every drain tick while backpressure is
active. With the default 5ms interval this can flood logs under normal
backpressure episodes where no stuck half-frames exist. Consider downgrading to
DEBUG or adding a time-based throttle.
This issue also appears on line 229 of the same file.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala:
##########
@@ -291,16 +291,28 @@ class PushDataHandler(val workerSource: WorkerSource)
extends BaseMessageHandler
}
pushData.body().retain()
+ val bodySize: Long = pushData.body().size()
+ MemoryManager.instance().incrementPendingReplicateBytes(bodySize)
+ val bytesReleased = new AtomicBoolean(false)
+ // Ensures the matching decrement below fires exactly once no matter
which of the several
+ // failure/success branches below is taken.
+ def releasePendingReplicateBytes(): Unit = {
+ if (bytesReleased.compareAndSet(false, true)) {
+ MemoryManager.instance().releasePendingReplicateBytes(bodySize)
+ }
+ }
replicateThreadPool.submit(new Runnable {
Review Comment:
`replicateThreadPool.submit(...)` is not guarded. If the worker is shutting
down (executor is terminated) this can throw `RejectedExecutionException`,
leaking the retained body and leaving `pendingReplicateBytes` incremented, and
the callback may never be completed. Consider catching submission failures and
performing the same cleanup as the in-task failure paths (release body,
decrement pendingReplicateBytes, mark peer unavailable / fail the callback, and
decrement pendingWrites as appropriate).
This issue also appears on line 643 of the same file.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java:
##########
@@ -411,6 +443,41 @@ public void switchServingState() {
}
}
+ /**
+ * Fires a drainIncompleteFrame tick when backpressure is active and
appActiveMemory is at/below
+ * the watermark.
+ */
+ private void checkAndTriggerDrainIncompleteFrame() {
+ if (!drainIncompleteFrameEnabled) return;
+
+ long appActiveMemory =
+ sortMemoryCounter.get()
+ + diskBufferCounter.get()
+ + memoryFileStorageCounter.sum()
+ + pendingReplicateBytesCounter.get();
+ if (appActiveMemory > drainIncompleteFrameWatermarkBytes) return;
+
+ long now = System.currentTimeMillis();
+ if (drainIncompleteFrameLastTickTime >= 0
+ && now - drainIncompleteFrameLastTickTime <
drainIncompleteFrameIntervalMs) return;
+ drainIncompleteFrameLastTickTime = now;
+
+ int resumedChannels = 0;
+ for (MemoryPressureListener listener : memoryPressureListeners) {
+ resumedChannels +=
listener.drainIncompleteFrame(drainIncompleteFrameRatio);
+ }
+ if (resumedChannels > 0) {
+ logger.info(
+ "DrainIncompleteFrame resumed {} channels (ratio={})",
+ resumedChannels,
+ drainIncompleteFrameRatio);
+ }
Review Comment:
`checkAndTriggerDrainIncompleteFrame()` logs the "resumed" message at INFO
whenever it resumes any channels. With the default 5ms drain interval this can
produce extremely high log volume during prolonged backpressure recovery.
Consider downgrading this to DEBUG or adding a rate limit (similar to the
pinned-memory log throttle just below).
--
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]
