SteNicholas commented on code in PR #3824:
URL: https://github.com/apache/celeborn/pull/3824#discussion_r3845596436
##########
common/src/main/java/org/apache/celeborn/common/network/util/TransportFrameDecoder.java:
##########
@@ -51,11 +51,37 @@ public class TransportFrameDecoder extends
ChannelInboundHandlerAdapter implemen
private static final int MAX_FRAME_SIZE = Integer.MAX_VALUE;
private static final int UNKNOWN_FRAME_SIZE = -1;
+ /**
+ * Netty's {@code AdaptiveRecvByteBufAllocator} caps a single {@code
channelRead} at this many
+ * bytes. Used to spot likely-large stuck frames; see {@link
#hasLikelyLargeIncompleteFrame()}.
+ */
+ public static final long MAX_SINGLE_READ_BYTES = 65536;
+
private final LinkedList<ByteBuf> buffers = new LinkedList<>();
- private long totalSize = 0;
+ // Written by the Netty I/O thread; read by checkService via
hasLikelyLargeIncompleteFrame().
+ private volatile long totalSize = 0;
private long nextFrameSize = UNKNOWN_FRAME_SIZE;
+ /**
+ * When set, this channel is resumed only to drain its stuck half-received
frame (in {@link
+ * #buffers}) and release the memory, not to resume normal traffic.
+ */
+ private volatile boolean frameDrain = false;
+
+ /** Enables single-frame drain mode: the next decoded frame fires {@link
FrameDrainCompleted}. */
+ public void enableFrameDrain() {
+ this.frameDrain = true;
+ }
+
+ /**
+ * True if the stuck frame's buffered bytes exceed one {@code channelRead}
worth ({@link
+ * #MAX_SINGLE_READ_BYTES}), making it a strong candidate for the
backpressure root cause.
+ */
+ public boolean hasLikelyLargeIncompleteFrame() {
+ return totalSize > MAX_SINGLE_READ_BYTES;
Review Comment:
A typical 256 KiB frame paused after its first 64 KiB read has less than 64
KiB in `totalSize` after the header is consumed, so this predicate stays false
forever while `autoRead=false`. Those partial frames can still pin allocator
chunks and cause the reported deadlock. Please use the decoded frame length or
an explicit incomplete-large-frame flag instead of requiring more than one read
to have accumulated.
##########
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();
Review Comment:
`pendingReplicateBytesCounter` is decremented only by replication response
callbacks, but the replication limiter also pauses the outbound client channels
that must read those responses. In `PUSH_AND_REPLICATE_PAUSED`, if this counter
alone exceeds the watermark, draining never starts, acknowledgements remain
unread, and the counter cannot fall—recreating the deadlock. Please track bytes
only until the outbound write completes, exclude this ack-dependent counter
from the drain gate, or exempt response channels.
--
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]
