[
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, 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{}}}.
was:In filtering mode, the pre-filter source buffer used by
`RecoveredChannelStateHandler.getBuffer()` is now allocated from heap (bounded
to 5 per gate) 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.
> 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, 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{}}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
