SteNicholas commented on code in PR #3824:
URL: https://github.com/apache/celeborn/pull/3824#discussion_r3845595211
##########
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:
[P1] `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.
##########
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:
[P1] 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.
--
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]
