[CELEBORN-2441] Add drainIncompleteFrame mechanism to recover from backpressure deadlock - #3824
[CELEBORN-2441] Add drainIncompleteFrame mechanism to recover from backpressure deadlock#3824buska88 wants to merge 4 commits into
Conversation
…ckpressure deadlock
|
@RexXiong @FMX @SteNicholas |
| sortMemoryCounter.get() | ||
| + diskBufferCounter.get() | ||
| + memoryFileStorageCounter.sum() | ||
| + pendingReplicateBytesCounter.get(); |
There was a problem hiding this 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.
There was a problem hiding this comment.
Thanks for pointing this out! I agree this can create the deadlock you described.
One caveat: it's not a permanent deadlock — failExpiredPushRequest() runs on
an independent scheduled thread that doesn't depend on the paused channel's
readability, so once shufflePushDataTimeout elapses, onFailure fires and the
counter is released regardless. That said, recovery then falls back to
timeout-based retry (tens of seconds+) instead of the millisecond-level drain
this PR aims for, and repeats each cycle if backpressure persists — so I still
think it's worth fixing.
On the three options: I'm hesitant about the first two (release on write-complete,
or exclude the counter from the drain gate), since pendingReplicateBytes
reflects real retained memory — relaxing it could let drain resume more push
channels while replicate data is still genuinely occupying memory, risking OOM.
I'd lean towards option 3 (exempting response/ack channels from the pause), which
keeps the counter accurate while unblocking acks. Did you have a specific
approach in mind for identifying response channels, or is a fresh registration
flag/separate limiter the way to go?
There was a problem hiding this comment.
Good catch, thanks! You're right that pendingReplicateBytesCounter alone crossing the watermark shouldn't gate draining, since it only decrements via replication ack callbacks, which in turn depend on this worker's own channels being resumed to read those acks — a circular dependency that never resolves on its own.
Fixed in the latest commit: we now split the watermark check into two independent gates —
PUSH_MODULE drains only when localActiveMemory + pendingReplicateBytesCounter is at/below the watermark (unchanged, full accounting), and
REPLICATE_MODULE drains based on localActiveMemory alone (excluding pendingReplicateBytesCounter), since that counter reflects a cluster-wide/peer-dependent state rather than this worker's own memory footprint, and letting it block REPLICATE_MODULE would recreate exactly the deadlock you described.
This way, even if pendingReplicateBytesCounter stays high, the replicate channels can still be drained first to unblock the peer's push side, which lets acks flow back and the counter naturally fall.
| * #MAX_SINGLE_READ_BYTES}), making it a strong candidate for the backpressure root cause. | ||
| */ | ||
| public boolean hasLikelyLargeIncompleteFrame() { | ||
| return totalSize > MAX_SINGLE_READ_BYTES; |
There was a problem hiding this 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.
There was a problem hiding this comment.
Good point, thanks! You're right that for a typical 256 KiB frame paused right after its first 64 KiB read, totalSize alone can stay under the cap and never flag it.
Fixed in the latest commit by making hasLikelyLargeIncompleteFrame take a byFrameSize flag and picking the criterion per module, since the "large frame" signal means different things on each side:
For PUSH_MODULE (draining channels where this worker is the receiver), we keep using totalSize — that's the actual bytes already buffered in this worker's memory, so it's the most precise/direct signal for relieving this worker's own memory pressure.
For REPLICATE_MODULE (draining channels where this worker is the sender to a peer), we now use the decoded frame length (nextFrameSize) instead, because in this direction the large frame is actually piling up on the peer's (primary's) push-receiving side, not locally. Ranking by nextFrameSize correctly identifies which peer connections are stuck behind a large frame, so draining them unblocks the peer and relieves memory pressure on the primary side faster — which matters a lot for us since our jobs with Gluten enabled can push frames of several hundred MB.
Let me know if this addresses the concern, or if you'd prefer a different split.
| // 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( |
There was a problem hiding this comment.
With the default 10 ms memory check and 5 ms drain interval, sustained backpressure with no qualifying candidate emits this INFO message about 100 times per second per paused limiter—potentially 200/s when push and replication are both paused. This can exacerbate an incident and consume substantial log storage. Please rate-limit it or lower it to DEBUG.
There was a problem hiding this comment.
Pull request overview
This PR introduces a controlled “drain incomplete frame” mechanism to break backpressure deadlocks caused by large half-received frames stuck inside Netty’s TransportFrameDecoder buffers, and adds accounting for replication bytes so the “app-layer active memory” watermark logic is more accurate.
Changes:
- Added
TransportFrameDecodersupport for detecting likely-large incomplete frames and a single-frame “drain mode” that emits aFrameDrainCompleteduser event. - Added
MemoryManagerlogic to periodically trigger draining under backpressure when app-layer active memory is below a configurable watermark, plus tracking/metrics for pending replicate bytes. - Added
ChannelsLimiter#drainIncompleteFrame(ratio)to resume a small subset of paused channels in drain mode, with unit tests and new worker configs/docs.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| worker/src/test/scala/org/apache/celeborn/service/deploy/memory/MemoryManagerSuite.scala | Adds unit tests for watermark-based drain triggering and pending replicate bytes accounting. |
| worker/src/test/java/org/apache/celeborn/service/deploy/worker/memory/ChannelsLimiterSuiteJ.java | New unit tests for paused-channel selection and drain ratio behavior. |
| worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala | Adds a metric name for pending replicate bytes. |
| worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala | Exposes PendingReplicateBytes as a worker gauge. |
| worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala | Tracks pending replicate bytes around replication send/ack paths. |
| worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java | Implements drain tick triggering and pending replicate bytes tracking. |
| worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/ChannelsLimiter.java | Adds drain scanning/resume logic and re-pausing on FrameDrainCompleted. |
| docs/configuration/worker.md | Documents new celeborn.worker.monitor.drainIncompleteFrame.* configs. |
| common/src/test/java/org/apache/celeborn/common/network/util/TransportFrameDecoderSuiteJ.java | New unit tests for large-incomplete-frame detection and drain event behavior. |
| common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala | Adds config entries and getters for drain settings. |
| common/src/main/java/org/apache/celeborn/common/network/util/TransportFrameDecoder.java | Adds heuristics and drain event support in the decoder. |
Suppressed comments (2)
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala:654
- Same submission-risk as the PushData replication path above:
replicateThreadPool.submit(...)can throw (e.g., executor shutdown) which would leak the retained body and leavependingReplicateBytesincremented. Add a submission failure handler that releases the body, decrements pending replicate bytes, and fails the callback / marks peer unavailable so the request doesn't hang.
pushMergedData.body().retain()
val bodySize: Long = pushMergedData.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 {
worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/ChannelsLimiter.java:239
- The "resumed" log is also emitted at INFO on every drain tick. With a 5ms default tick interval, this can generate very high log volume while draining is active. Consider downgrading to DEBUG or throttling (similar to the pinned-memory log throttling in MemoryManager).
if (resumed > 0) {
logger.info(
"{} drainIncompleteFrame resumed {}/{} channels with a stuck half-frame larger than "
+ "{} bytes (ratio={})",
moduleName,
resumed,
candidates.size(),
TransportFrameDecoder.MAX_SINGLE_READ_BYTES,
ratio);
}
return resumed;
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| 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 { |
| 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); | ||
| } |
| if (resumedChannels > 0) { | ||
| logger.info( | ||
| "DrainIncompleteFrame resumed {} channels (ratio={})", | ||
| resumedChannels, | ||
| drainIncompleteFrameRatio); | ||
| } |
What changes were proposed in this pull request?
Introduce a
drainIncompleteFramemechanism toMemoryManager/ChannelsLimiter/TransportFrameDecoderthat lets a worker recover from a backpressure deadlock causedby stuck half-frames in Netty's pipeline:
TransportFrameDecoder#hasLikelyLargeIncompleteFrame()heuristically detectschannels stuck with a large (> 64KB) buffered-but-undecoded frame.
TransportFrameDecoder#enableFrameDrain()puts a channel into single-frame "drain"mode: once exactly one frame is decoded, it fires
FrameDrainCompletedso thechannel can be re-paused immediately without resuming full traffic.
MemoryManager#checkAndTriggerDrainIncompleteFrame()periodically checks, whilebackpressure is active, whether application-layer active memory (sort memory + disk
buffer + memory file storage + pending replicate bytes) is at/below a configurable
watermark, and if so triggers a drain tick at a configurable interval.
ChannelsLimiter#drainIncompleteFrame(ratio)scans paused channels for likely-stuckhalf-frames and resumes a configurable ratio of them in drain mode to release the
stuck memory.
New configs added to
CelebornConf:celeborn.worker.monitor.drainIncompleteFrame.enabled(defaulttrue)celeborn.worker.monitor.drainIncompleteFrame.interval(default5ms)celeborn.worker.monitor.drainIncompleteFrame.ratio(default0.05)celeborn.worker.monitor.drainIncompleteFrame.watermark.ratio(default0.1)Why are the changes needed?
Worker monitors direct memory usage and pauses reading from all business channels
(
autoRead=false) once usage crosses a threshold, waiting for buffered data to beconsumed before resuming. However, if
autoRead=falsetakes effect exactly when achannel is mid-way through an incomplete frame, the buffered bytes stay stuck inside
TransportFrameDecoder— never dispatched, never released, and invisible to everyapplication-layer memory counter. This can produce a deadlock: application-layer
counters report near-zero usage while the underlying direct memory doesn't drop below
the resume threshold, because it's still holding onto these stuck half-frames — and
since reads stay globally paused, the stuck half-frame never receives the remaining
bytes needed to complete decoding. The worker becomes stuck and requires a manual
restart to recover.
This is not a rare corner case in practice: with the common frame size around a few
hundred KB (e.g.
celeborn.client.push.buffer.max.size=256KB) and Netty'sAdaptiveRecvByteBufAllocatorcapping a singlechannelReadat64KB in steady state,5 reads to complete, making a stuck half-frame at the momenta frame typically needs 4
of pause fairly likely under sustained high throughput.
This issue was first discovered in production: some workers got stuck in backpressure
and never recovered even though application-layer memory counters had dropped to near
zero. Using Arthas thread/stack inspection together with internal memory metrics, we
confirmed the stuck memory was sitting inside the Netty pipeline (
TransportFrameDecoder)rather than any application-layer counter, matching the root cause above.
See CELEBORN-2441 for more details.
Does this PR resolve a correctness bug?
Does this PR introduce any user-facing change?
New configs are added under
celeborn.worker.monitor.drainIncompleteFrame.*(documented in
docs/configuration/worker.md), all with safe defaults; behavior isunchanged from prior versions when
celeborn.worker.monitor.drainIncompleteFrame.enabled=false.How was this patch tested?
Added unit tests:
TransportFrameDecoderSuiteJcovershasLikelyLargeIncompleteFrame()and framedrain behavior.
ChannelsLimiterSuiteJcovers candidate selection and drain ratio behavior.MemoryManagerSuitecovers watermark-based triggering and pending replicatebytes accounting.
Verified in production on a fleet of ~500 worker machines: after enabling
drainIncompleteFrame, workers that previously got stuck in backpressure recoveredautomatically, with no observed CPU regression during normal operation.
celeborn.PausePushData.Deltametric before/after rollout — a backpressure spikethat used to stay stuck is now resolved within the same minute:
drainIncompleteFrameactively resuming stuck channels andthe serving state transitioning back to
NONE_PAUSEDright after:celeborn.PausePushDataTime.Deltametric dropping back to0oncedrainIncompleteFrametakes effect, meaning the worker is no longer inbackpressure for that minute: