[
https://issues.apache.org/jira/browse/FLINK-39519?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rui Fan updated FLINK-39519:
----------------------------
Description:
In filtering mode, the pre-filter source buffer used by
{{RecoveredChannelStateHandler.getBuffer()}} is now allocated from heap instead
of the Network Buffer Pool. This eliminates the deadlock between pre-filter and
post-filter buffers competing for the same pool, which used to be masked by an
unbounded heap fallback in {{{}requestBufferBlocking(){}}}. Memory is bounded
to a single segment per task (see invariant below), so no semaphore or per-gate
limit is needed. The {{requestBufferBlocking()}} fallback is kept unchanged in
this commit and will be removed in a follow-up, once FLINK-38544's OutputWriter
provides the disk-spilling replacement for the post-filter path.
h2. Source Buffer One-At-A-Time Invariant
In filtering-mode channel state recovery, {*}at most one pre-filter (source)
heap buffer is in flight per task at any moment{*}. Worst-case source-buffer
memory = {{1 × memorySegmentSize}} (≈ 32 KB). No OOM risk, no semaphore or
per-gate limit needed.
*Why it holds:*
# {{ChannelStateChunkReader.readChunk()}} is a single-threaded while-loop:
{{{}getBuffer → fill → recover(finally recycle){}}}. The next {{getBuffer()}}
can only start after {{recover()}} returns. Structural concurrency = 1.
# {{SpillingAdaptiveSpanningRecordDeserializer.getNextRecord()}} recycles
{{currentBuffer}} the instant {{{}isBufferConsumed=true{}}}, which happens on
both {{PARTIAL_RECORD}} (record spans buffers) and {{LAST_RECORD_FROM_BUFFER}}
(buffer exhausted). {{filterAndRewrite()}} breaks its loop on the same
condition.
# Cross-buffer bytes are always *copied out* before recycle:
{{SpanningWrapper.transferFrom}} / {{addNextChunkFromMemorySegment}} copy
partial bytes into an internal {{byte[]}} (or into {{{}SpanningWrapper{}}}'s
own spill file for records ≥ 5 MB). The only retained segment pointer
({{{}leftOverData{}}} / {{{}NonSpanningWrapper.segment{}}}) is read via
{{segment.get(...)}} — copies into the target — and only while
{{{}isBufferConsumed=false{}}}.
# Exception path converges through {{{}ChannelStateFilteringHandler.close() →
VirtualChannel.clear() → deserializer.clear(){}}}, which recycles any buffer
the deserializer still holds.
*Implication:* the implementation is minimal — {{allocateUnpooledSegment}} +
{{{}FreeingBufferRecycler.INSTANCE{}}}, no semaphore, no per-gate limit, no
custom recycler. A test feeds a record spanning multiple source buffers and
asserts {{maxOutstanding == 1}} to guard the invariant against regression.
was:
In filtering mode, the pre-filter source buffer used by
`RecoveredChannelStateHandler.getBuffer()` is now allocated from heap instead
of the Network Buffer Pool, preventing deadlock between source and filtered
buffers. Also adds a non-blocking `RecoveredInputChannel.requestBuffer()` used
by OutputWriter, and removes the unbounded heap fallback from
`requestBufferBlocking()` in filtering mode.
In filtering-mode channel state recovery, {*}at most one pre-filter (source)
heap buffer is in flight per task at any moment{*}. Worst-case source-buffer
memory = {{1 × memorySegmentSize}} (≈ 32 KB). No OOM risk, no semaphore or
per-gate limit needed.
*Why it holds:*
# {{ChannelStateChunkReader.readChunk()}} is a single-threaded while-loop:
{{{}getBuffer → fill → recover(finally recycle){}}}. The next {{getBuffer()}}
can only start after {{recover()}} returns. Structural concurrency = 1.
# {{SpillingAdaptiveSpanningRecordDeserializer.getNextRecord()}} recycles
{{currentBuffer}} the instant {{{}isBufferConsumed=true{}}}, which happens on
both {{PARTIAL_RECORD}} (record spans buffers) and {{LAST_RECORD_FROM_BUFFER}}
(buffer exhausted). {{filterAndRewrite()}} breaks its loop on the same
condition.
# Cross-buffer bytes are always *copied out* before recycle:
{{SpanningWrapper.transferFrom}} / {{addNextChunkFromMemorySegment}} copy
partial bytes into an internal {{byte[]}} (or into {{{}SpanningWrapper{}}}'s
own spill file for records ≥ 5 MB). The only retained segment pointer
({{{}leftOverData{}}} / {{{}NonSpanningWrapper.segment{}}}) is read via
{{segment.get(...)}} — copies into the target — and only while
{{{}isBufferConsumed=false{}}}.
# Exception path converges through {{{}ChannelStateFilteringHandler.close() →
VirtualChannel.clear() → deserializer.clear(){}}}, which recycles any buffer
the deserializer still holds.
*Implication:* the heap-isolation change needs only {{allocateUnpooledSegment}}
+ {{{}FreeingBufferRecycler.INSTANCE{}}}. Drop the semaphore, the
{{MAX_HEAP_BUFFERS_PER_GATE}} constant, and the custom recycler. Add one test
that feeds a record spanning multiple source buffers and asserts
{{{}maxOutstanding == 1{}}}.
> Allocate pre-filter source buffers from heap in filtering mode
> --------------------------------------------------------------
>
> Key: FLINK-39519
> URL: https://issues.apache.org/jira/browse/FLINK-39519
> Project: Flink
> Issue Type: Sub-task
> Components: Runtime / Checkpointing
> Reporter: Rui Fan
> Assignee: Rui Fan
> Priority: Major
>
> In filtering mode, the pre-filter source buffer used by
> {{RecoveredChannelStateHandler.getBuffer()}} is now allocated from heap
> instead of the Network Buffer Pool. This eliminates the deadlock between
> pre-filter and post-filter buffers competing for the same pool, which used to
> be masked by an unbounded heap fallback in {{{}requestBufferBlocking(){}}}.
> Memory is bounded to a single segment per task (see invariant below), so no
> semaphore or per-gate limit is needed. The {{requestBufferBlocking()}}
> fallback is kept unchanged in this commit and will be removed in a follow-up,
> once FLINK-38544's OutputWriter provides the disk-spilling replacement for
> the post-filter path.
> h2. Source Buffer One-At-A-Time Invariant
> In filtering-mode channel state recovery, {*}at most one pre-filter (source)
> heap buffer is in flight per task at any moment{*}. Worst-case source-buffer
> memory = {{1 × memorySegmentSize}} (≈ 32 KB). No OOM risk, no semaphore or
> per-gate limit needed.
> *Why it holds:*
> # {{ChannelStateChunkReader.readChunk()}} is a single-threaded while-loop:
> {{{}getBuffer → fill → recover(finally recycle){}}}. The next {{getBuffer()}}
> can only start after {{recover()}} returns. Structural concurrency = 1.
> # {{SpillingAdaptiveSpanningRecordDeserializer.getNextRecord()}} recycles
> {{currentBuffer}} the instant {{{}isBufferConsumed=true{}}}, which happens on
> both {{PARTIAL_RECORD}} (record spans buffers) and
> {{LAST_RECORD_FROM_BUFFER}} (buffer exhausted). {{filterAndRewrite()}} breaks
> its loop on the same condition.
> # Cross-buffer bytes are always *copied out* before recycle:
> {{SpanningWrapper.transferFrom}} / {{addNextChunkFromMemorySegment}} copy
> partial bytes into an internal {{byte[]}} (or into {{{}SpanningWrapper{}}}'s
> own spill file for records ≥ 5 MB). The only retained segment pointer
> ({{{}leftOverData{}}} / {{{}NonSpanningWrapper.segment{}}}) is read via
> {{segment.get(...)}} — copies into the target — and only while
> {{{}isBufferConsumed=false{}}}.
> # Exception path converges through {{{}ChannelStateFilteringHandler.close()
> → VirtualChannel.clear() → deserializer.clear(){}}}, which recycles any
> buffer the deserializer still holds.
> *Implication:* the implementation is minimal — {{allocateUnpooledSegment}} +
> {{{}FreeingBufferRecycler.INSTANCE{}}}, no semaphore, no per-gate limit, no
> custom recycler. A test feeds a record spanning multiple source buffers and
> asserts {{maxOutstanding == 1}} to guard the invariant against regression.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
