[
https://issues.apache.org/jira/browse/FLINK-39519?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-39519:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Labels: pull-request-available
>
> 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)
